feat(stix): STIX→MISP download export (per-attacker + fleet)
Adds GET /api/v1/attackers/{uuid}/export/misp and
GET /api/v1/attackers/export/misp backed by misp_export.py, which
converts existing STIX bundles to MISP events via misp-stix
ExternalSTIX2toMISPParser. Fleet endpoint emits {response:[...]}
collection (one event per attacker). Frontend: STIX/MISP buttons on
AttackerDetail header and Attackers list. 13 new tests green.
This commit is contained in:
217
tests/web/test_api_export_attacker_misp.py
Normal file
217
tests/web/test_api_export_attacker_misp.py
Normal file
@@ -0,0 +1,217 @@
|
||||
"""Tests for GET /api/v1/attackers/{uuid}/export/misp."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from decnet.ttp import attack_stix
|
||||
from decnet.web.router.attackers.api_export_attacker_misp import (
|
||||
api_export_attacker_misp,
|
||||
)
|
||||
|
||||
_REPO_BUNDLE = Path(__file__).resolve().parents[2] / "enterprise-attack-19.0.json"
|
||||
_FAKE_USER: dict = {"uuid": "test-user", "role": "viewer"}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _pin_bundle(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
license_path = tmp_path / "LICENSE.txt"
|
||||
license_path.write_text("placeholder", encoding="utf-8")
|
||||
monkeypatch.setenv("DECNET_ATTACK_BUNDLE", str(_REPO_BUNDLE))
|
||||
monkeypatch.setenv("DECNET_ATTACK_LICENSE", str(license_path))
|
||||
attack_stix._data = None
|
||||
attack_stix._loaded_path = None
|
||||
attack_stix._attack_pattern_by_id.cache_clear()
|
||||
attack_stix._tactic_by_id.cache_clear()
|
||||
attack_stix._tactic_by_short_name.cache_clear()
|
||||
attack_stix.groups_using_technique.cache_clear()
|
||||
|
||||
|
||||
def _attacker(uuid: str = "att-aaaabbbbccccdddd") -> dict:
|
||||
return {
|
||||
"uuid": uuid,
|
||||
"ip": "1.2.3.4",
|
||||
"first_seen": datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
"last_seen": datetime(2026, 1, 31, tzinfo=timezone.utc),
|
||||
"event_count": 100,
|
||||
"identity_id": None,
|
||||
"country_code": "US",
|
||||
"asn": 15169,
|
||||
"as_name": "GOOGLE",
|
||||
}
|
||||
|
||||
|
||||
def _tag(technique_id: str) -> dict:
|
||||
return {
|
||||
"uuid": f"tag-{technique_id}",
|
||||
"attacker_uuid": "att-aaaabbbbccccdddd",
|
||||
"technique_id": technique_id,
|
||||
"sub_technique_id": None,
|
||||
"tactic": "TA0006",
|
||||
"confidence": 0.85,
|
||||
"rule_id": "R0001",
|
||||
"rule_version": 1,
|
||||
"evidence": {},
|
||||
"created_at": datetime(2026, 1, 15, tzinfo=timezone.utc),
|
||||
}
|
||||
|
||||
|
||||
def _artifact(sha256: str = "a" * 64) -> dict:
|
||||
return {
|
||||
"timestamp": datetime(2026, 1, 10, tzinfo=timezone.utc),
|
||||
"fields": json.dumps({"sha256": sha256, "filename": "payload.sh"}),
|
||||
}
|
||||
|
||||
|
||||
def _smtp_target() -> dict:
|
||||
return {
|
||||
"domain": "victim.example.com",
|
||||
"count": 5,
|
||||
"first_seen": datetime(2026, 1, 5, tzinfo=timezone.utc),
|
||||
"last_seen": datetime(2026, 1, 20, tzinfo=timezone.utc),
|
||||
}
|
||||
|
||||
|
||||
def _intel() -> dict:
|
||||
return {
|
||||
"aggregate_verdict": "malicious",
|
||||
"abuseipdb_score": 95,
|
||||
"greynoise_classification": "malicious",
|
||||
"greynoise_tags": json.dumps(["ssh_bruteforcer"]),
|
||||
"feodo_listed": False,
|
||||
"threatfox_listed": False,
|
||||
}
|
||||
|
||||
|
||||
def _mock_repo(*, attacker=None, intel=None, rollup=None, tags=None,
|
||||
artifacts=None, smtp=None, commands=None):
|
||||
m = type("M", (), {})()
|
||||
m.get_attacker_by_uuid = AsyncMock(return_value=attacker or _attacker())
|
||||
m.get_attacker_behavior = AsyncMock(return_value={})
|
||||
m.get_identity_by_uuid = AsyncMock(return_value=None)
|
||||
m.get_attacker_intel_by_uuid = AsyncMock(return_value=intel)
|
||||
from decnet.web.db.models.ttp import IdentityTechniqueRow
|
||||
m.list_techniques_by_attacker = AsyncMock(
|
||||
return_value=[
|
||||
IdentityTechniqueRow(
|
||||
technique_id=t, technique_name=None,
|
||||
sub_technique_id=None, sub_technique_name=None,
|
||||
tactic="TA0006", count=1, confidence_max=0.8,
|
||||
first_seen=datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
last_seen=datetime(2026, 1, 31, tzinfo=timezone.utc),
|
||||
mitre_url=None,
|
||||
)
|
||||
for t in (rollup or [])
|
||||
]
|
||||
)
|
||||
m.list_ttp_tags_by_attacker = AsyncMock(return_value=tags or [])
|
||||
m.get_attacker_artifacts = AsyncMock(return_value=artifacts or [])
|
||||
m.list_smtp_targets = AsyncMock(return_value=smtp or [])
|
||||
m.list_attacker_commands_deduped = AsyncMock(return_value=commands or [])
|
||||
return m
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_404_on_unknown_attacker():
|
||||
m = _mock_repo()
|
||||
m.get_attacker_by_uuid = AsyncMock(return_value=None)
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
from fastapi import HTTPException
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await api_export_attacker_misp("no-such-uuid", user=_FAKE_USER)
|
||||
assert exc.value.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_skinny_attacker_returns_misp_event():
|
||||
"""No TTPs, no artifacts — response is valid MISP event JSON."""
|
||||
m = _mock_repo()
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
resp = await api_export_attacker_misp("att-aaaabbbbccccdddd", user=_FAKE_USER)
|
||||
event = json.loads(resp.body)
|
||||
# Must be a MISP event dict with an Attribute list
|
||||
assert "Attribute" in event or "info" in event
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ip_attribute_present():
|
||||
"""Attacker IP appears as an ip-src or ip-dst attribute."""
|
||||
m = _mock_repo()
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
resp = await api_export_attacker_misp("att-aaaabbbbccccdddd", user=_FAKE_USER)
|
||||
event = json.loads(resp.body)
|
||||
attrs = event.get("Attribute", [])
|
||||
ip_attrs = [a for a in attrs if a.get("value") == "1.2.3.4"]
|
||||
assert len(ip_attrs) >= 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_hash_attribute_present():
|
||||
"""A captured artifact's SHA-256 appears in the MISP event."""
|
||||
m = _mock_repo(artifacts=[_artifact("b" * 64)])
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
resp = await api_export_attacker_misp("att-aaaabbbbccccdddd", user=_FAKE_USER)
|
||||
event = json.loads(resp.body)
|
||||
all_attrs = event.get("Attribute", [])
|
||||
# Also check inside Objects
|
||||
for obj in event.get("Object", []):
|
||||
all_attrs.extend(obj.get("Attribute", []))
|
||||
hash_attrs = [a for a in all_attrs if ("b" * 64) in str(a.get("value", ""))]
|
||||
assert len(hash_attrs) >= 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_domain_attribute_present():
|
||||
"""An SMTP target domain appears in the MISP event."""
|
||||
m = _mock_repo(smtp=[_smtp_target()])
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
resp = await api_export_attacker_misp("att-aaaabbbbccccdddd", user=_FAKE_USER)
|
||||
event = json.loads(resp.body)
|
||||
all_attrs = event.get("Attribute", [])
|
||||
for obj in event.get("Object", []):
|
||||
all_attrs.extend(obj.get("Attribute", []))
|
||||
domain_attrs = [
|
||||
a for a in all_attrs if "victim.example.com" in str(a.get("value", ""))
|
||||
]
|
||||
assert len(domain_attrs) >= 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mitre_galaxy_present_when_technique_tagged():
|
||||
"""A MITRE ATT&CK galaxy cluster appears when a technique is tagged."""
|
||||
m = _mock_repo(rollup=["T1059"], tags=[_tag("T1059")])
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
resp = await api_export_attacker_misp("att-aaaabbbbccccdddd", user=_FAKE_USER)
|
||||
event = json.loads(resp.body)
|
||||
galaxies = event.get("Galaxy", [])
|
||||
# misp-stix maps attack-pattern SDOs to "STIX 2.1 Attack Pattern" galaxies
|
||||
attack_pattern_galaxy = any(
|
||||
"attack pattern" in str(g.get("name", "")).lower() for g in galaxies
|
||||
)
|
||||
assert attack_pattern_galaxy
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response_headers():
|
||||
m = _mock_repo()
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
resp = await api_export_attacker_misp("att-aaaabbbbccccdddd", user=_FAKE_USER)
|
||||
assert resp.media_type == "application/json"
|
||||
assert ".misp.json" in resp.headers["content-disposition"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pymisp_round_trip():
|
||||
"""Event round-trips through pymisp.MISPEvent.from_dict without error."""
|
||||
import pymisp
|
||||
m = _mock_repo(rollup=["T1059"], tags=[_tag("T1059")], intel=_intel())
|
||||
with patch("decnet.web.router.attackers.api_export_attacker_misp.repo", m):
|
||||
resp = await api_export_attacker_misp("att-aaaabbbbccccdddd", user=_FAKE_USER)
|
||||
raw = json.loads(resp.body)
|
||||
e = pymisp.MISPEvent()
|
||||
e.from_dict(**raw)
|
||||
assert e.info
|
||||
112
tests/web/test_api_export_attackers_misp.py
Normal file
112
tests/web/test_api_export_attackers_misp.py
Normal file
@@ -0,0 +1,112 @@
|
||||
"""Tests for GET /api/v1/attackers/export/misp — fleet-wide MISP collection."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from decnet.ttp import attack_stix
|
||||
from decnet.web.router.attackers.api_export_attackers_misp import (
|
||||
api_export_attackers_misp,
|
||||
)
|
||||
|
||||
_REPO_BUNDLE = Path(__file__).resolve().parents[2] / "enterprise-attack-19.0.json"
|
||||
_FAKE_USER: dict = {"uuid": "test-user", "role": "viewer"}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _pin_bundle(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
license_path = tmp_path / "LICENSE.txt"
|
||||
license_path.write_text("placeholder", encoding="utf-8")
|
||||
monkeypatch.setenv("DECNET_ATTACK_BUNDLE", str(_REPO_BUNDLE))
|
||||
monkeypatch.setenv("DECNET_ATTACK_LICENSE", str(license_path))
|
||||
attack_stix._data = None
|
||||
attack_stix._loaded_path = None
|
||||
attack_stix._attack_pattern_by_id.cache_clear()
|
||||
attack_stix._tactic_by_id.cache_clear()
|
||||
attack_stix._tactic_by_short_name.cache_clear()
|
||||
attack_stix.groups_using_technique.cache_clear()
|
||||
|
||||
|
||||
def _attacker(uuid: str = "att-aaaa", ip: str = "1.2.3.4") -> dict:
|
||||
return {
|
||||
"uuid": uuid,
|
||||
"ip": ip,
|
||||
"first_seen": datetime(2026, 1, 1, tzinfo=timezone.utc),
|
||||
"last_seen": datetime(2026, 1, 31, tzinfo=timezone.utc),
|
||||
"event_count": 10,
|
||||
"identity_id": None,
|
||||
"country_code": "US",
|
||||
"asn": 15169,
|
||||
"as_name": "GOOGLE",
|
||||
"commands": [],
|
||||
"threat_intel": None,
|
||||
}
|
||||
|
||||
|
||||
def _mock_repo(*, rows=None, ttp_by_attacker=None):
|
||||
m = type("M", (), {})()
|
||||
m.get_all_attackers_for_export = AsyncMock(return_value=rows or [])
|
||||
m.get_all_ttp_rollups_for_export = AsyncMock(return_value=ttp_by_attacker or {})
|
||||
return m
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_fleet_returns_empty_collection():
|
||||
"""Zero attackers → {"response": []}."""
|
||||
m = _mock_repo()
|
||||
with patch("decnet.web.router.attackers.api_export_attackers_misp.repo", m):
|
||||
resp = await api_export_attackers_misp(user=_FAKE_USER)
|
||||
body = json.loads(resp.body)
|
||||
assert "response" in body
|
||||
assert body["response"] == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_single_attacker_one_event():
|
||||
"""One attacker → collection with one event."""
|
||||
m = _mock_repo(rows=[_attacker()])
|
||||
with patch("decnet.web.router.attackers.api_export_attackers_misp.repo", m):
|
||||
resp = await api_export_attackers_misp(user=_FAKE_USER)
|
||||
body = json.loads(resp.body)
|
||||
assert len(body["response"]) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_two_attackers_two_events():
|
||||
"""Two distinct attacker rows → two events in the collection."""
|
||||
rows = [_attacker("att-1111", "1.1.1.1"), _attacker("att-2222", "2.2.2.2")]
|
||||
m = _mock_repo(rows=rows)
|
||||
with patch("decnet.web.router.attackers.api_export_attackers_misp.repo", m):
|
||||
resp = await api_export_attackers_misp(user=_FAKE_USER)
|
||||
body = json.loads(resp.body)
|
||||
assert len(body["response"]) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_commands_in_fleet_event():
|
||||
"""Commands in the row's commands field end up in the event."""
|
||||
row = _attacker()
|
||||
row["commands"] = [
|
||||
{"command_text": "whoami", "ts": "2026-01-10T00:00:00"},
|
||||
{"command_text": "id", "ts": "2026-01-10T00:01:00"},
|
||||
]
|
||||
m = _mock_repo(rows=[row])
|
||||
with patch("decnet.web.router.attackers.api_export_attackers_misp.repo", m):
|
||||
resp = await api_export_attackers_misp(user=_FAKE_USER)
|
||||
body = json.loads(resp.body)
|
||||
assert len(body["response"]) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_response_headers():
|
||||
m = _mock_repo()
|
||||
with patch("decnet.web.router.attackers.api_export_attackers_misp.repo", m):
|
||||
resp = await api_export_attackers_misp(user=_FAKE_USER)
|
||||
assert resp.media_type == "application/json"
|
||||
cd = resp.headers["content-disposition"]
|
||||
assert "decnet-fleet-" in cd
|
||||
assert ".misp.json" in cd
|
||||
Reference in New Issue
Block a user