feat(intel): persist per-provider taxonomy on AttackerIntel for TTP dispatch

The 2026-05-02 ship-time audit of the R0054-R0058 intel rule pack found
that AbuseIPDB / GreyNoise / ThreatFox stored only the aggregate verdict
(score / classification / listed-bool) plus the raw response blob. The
TTP IntelLifter expects per-provider taxonomy fields (categories, tags,
threat_types) that were never populated, so R0054 / R0055 / R0057
emitted zero tags in production despite passing unit tests.

Add typed columns: abuseipdb_categories, greynoise_tags, greynoise_name,
feodo_malware_family, threatfox_threat_types, threatfox_ioc_types,
threatfox_malware_families. Each provider now parses the relevant
taxonomy out of the upstream response and writes it through
column_updates. JSON-list columns ride as TEXT with default "[]" to
keep the SQLite/MySQL backend split honest, deserialised back to native
lists by the repo on read.
This commit is contained in:
2026-05-02 18:07:57 -04:00
parent d1c4a48963
commit 999d3494b4
10 changed files with 272 additions and 1 deletions

View File

@@ -123,6 +123,45 @@ async def test_429_returns_error_no_writes():
assert result.column_updates == {}
@pytest.mark.anyio
async def test_actor_name_and_tags_persisted_when_present():
"""Post-2026-05-02 audit: ``name`` (actor label) and any ``tags``
list returned by the upstream survive into ``column_updates``.
The Community endpoint does not return ``tags`` in practice; the
test seeds the field anyway so non-Community provider plans that
do (paid / Enterprise) work without further code changes.
"""
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
json={
"classification": "malicious",
"name": "Tor",
"tags": ["tor_exit_node", "ssh_bruteforcer"],
},
)
provider = GreyNoiseProvider()
_install_transport(provider, handler)
result = await provider.lookup("1.2.3.4")
assert result.column_updates["greynoise_name"] == "Tor"
tags = json.loads(result.column_updates["greynoise_tags"])
assert tags == ["tor_exit_node", "ssh_bruteforcer"]
@pytest.mark.anyio
async def test_404_clears_actor_and_tags():
async def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(404, json={"message": "not seen"})
provider = GreyNoiseProvider()
_install_transport(provider, handler)
result = await provider.lookup("10.0.0.5")
assert result.column_updates["greynoise_name"] is None
assert result.column_updates["greynoise_tags"] == "[]"
@pytest.mark.anyio
async def test_network_failure_becomes_error():
async def handler(request: httpx.Request) -> httpx.Response: