anti
1d3086a5c7
Surfaces the intrusion-set reverse index from the loaded ATT&CK
bundle: given a technique, returns the list of groups MITRE has
documented as using it. Read-only — explicitly NOT an attribution
claim about a DECNET attacker. The frontend pulls this lazily when
the operator expands a technique panel; payload-size cost on every
TTPTagDetailRow makes embedding wasteful for techniques with 50+
documented groups.
- decnet/web/router/ttp/api_get_groups_for_technique.py exposes
GET /api/v1/ttp/techniques/{technique_id}/groups, response_model
list[GroupRef]. Same JWT-viewer auth gating as the rest of the
TTP router. 404 when the technique_id doesn't resolve in the
bundle.
- Sub-techniques are queried directly (no auto-union with parent)
to match ATT&CK Navigator semantics; callers that want a broader
view query the parent themselves.
- tests/ttp/test_groups_for_technique.py covers happy path, 404,
sub-technique attribution independence, empty-list-on-zero-groups,
and that responses include mitre_url + aliases.
- tests/web/test_api_attackers.py: fix pre-existing fixture drift
introduced by a2a61b63 — three TestGetAttackerDetail cases were
missing AsyncMock for repo.latest_observation_per_primitive,
causing TypeError on await of MagicMock. The new groups endpoint
doesn't share code with attacker_detail; this is a drive-by fix
surfaced by the same suite run.
644 lines
27 KiB
Python
644 lines
27 KiB
Python
"""
|
|
Tests for the attacker profile API routes.
|
|
|
|
Covers:
|
|
- GET /attackers: paginated list, search, sort_by
|
|
- GET /attackers/{uuid}: single profile detail, 404 on missing UUID
|
|
- Auth enforcement on both endpoints
|
|
"""
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from decnet.web.auth import create_access_token
|
|
from decnet.web.router.attackers.api_get_attackers import _reset_total_cache
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_attackers_cache():
|
|
_reset_total_cache()
|
|
yield
|
|
_reset_total_cache()
|
|
|
|
|
|
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def _auth_request(uuid: str = "test-user-uuid") -> MagicMock:
|
|
token = create_access_token({"uuid": uuid})
|
|
req = MagicMock()
|
|
req.headers = {"Authorization": f"Bearer {token}"}
|
|
return req
|
|
|
|
|
|
def _sample_attacker(uuid: str = "att-uuid-1", ip: str = "1.2.3.4") -> dict:
|
|
return {
|
|
"uuid": uuid,
|
|
"ip": ip,
|
|
"first_seen": datetime(2026, 4, 1, tzinfo=timezone.utc).isoformat(),
|
|
"last_seen": datetime(2026, 4, 10, tzinfo=timezone.utc).isoformat(),
|
|
"event_count": 42,
|
|
"service_count": 3,
|
|
"decky_count": 2,
|
|
"services": ["ssh", "http", "ftp"],
|
|
"deckies": ["decky-01", "decky-02"],
|
|
"traversal_path": "decky-01 → decky-02",
|
|
"is_traversal": True,
|
|
"bounty_count": 5,
|
|
"credential_count": 2,
|
|
"fingerprints": [{"type": "ja3", "hash": "abc"}],
|
|
"commands": [{"service": "ssh", "decky": "decky-01", "command": "id", "timestamp": "2026-04-01T10:00:00"}],
|
|
"updated_at": datetime(2026, 4, 10, tzinfo=timezone.utc).isoformat(),
|
|
}
|
|
|
|
|
|
# ─── GET /attackers ──────────────────────────────────────────────────────────
|
|
|
|
class TestGetAttackers:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_paginated_response(self):
|
|
from decnet.web.router.attackers.api_get_attackers import get_attackers
|
|
|
|
sample = _sample_attacker()
|
|
with patch("decnet.web.router.attackers.api_get_attackers.repo") as mock_repo:
|
|
mock_repo.get_attackers = AsyncMock(return_value=[sample])
|
|
mock_repo.get_total_attackers = AsyncMock(return_value=1)
|
|
mock_repo.get_behaviors_for_ips = AsyncMock(return_value={})
|
|
|
|
result = await get_attackers(
|
|
limit=50, offset=0, search=None, sort_by="recent",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert result["total"] == 1
|
|
assert result["limit"] == 50
|
|
assert result["offset"] == 0
|
|
assert len(result["data"]) == 1
|
|
assert result["data"][0]["uuid"] == "att-uuid-1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_parameter_forwarded(self):
|
|
from decnet.web.router.attackers.api_get_attackers import get_attackers
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attackers.repo") as mock_repo:
|
|
mock_repo.get_attackers = AsyncMock(return_value=[])
|
|
mock_repo.get_total_attackers = AsyncMock(return_value=0)
|
|
mock_repo.get_behaviors_for_ips = AsyncMock(return_value={})
|
|
|
|
await get_attackers(
|
|
limit=50, offset=0, search="192.168", sort_by="recent",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
mock_repo.get_attackers.assert_awaited_once_with(
|
|
limit=50, offset=0, search="192.168", sort_by="recent", service=None,
|
|
)
|
|
mock_repo.get_total_attackers.assert_awaited_once_with(search="192.168", service=None)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_null_search_normalized(self):
|
|
from decnet.web.router.attackers.api_get_attackers import get_attackers
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attackers.repo") as mock_repo:
|
|
mock_repo.get_attackers = AsyncMock(return_value=[])
|
|
mock_repo.get_total_attackers = AsyncMock(return_value=0)
|
|
mock_repo.get_behaviors_for_ips = AsyncMock(return_value={})
|
|
|
|
await get_attackers(
|
|
limit=50, offset=0, search="null", sort_by="recent",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
mock_repo.get_attackers.assert_awaited_once_with(
|
|
limit=50, offset=0, search=None, sort_by="recent", service=None,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sort_by_active(self):
|
|
from decnet.web.router.attackers.api_get_attackers import get_attackers
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attackers.repo") as mock_repo:
|
|
mock_repo.get_attackers = AsyncMock(return_value=[])
|
|
mock_repo.get_total_attackers = AsyncMock(return_value=0)
|
|
mock_repo.get_behaviors_for_ips = AsyncMock(return_value={})
|
|
|
|
await get_attackers(
|
|
limit=50, offset=0, search=None, sort_by="active",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
mock_repo.get_attackers.assert_awaited_once_with(
|
|
limit=50, offset=0, search=None, sort_by="active", service=None,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_search_normalized_to_none(self):
|
|
from decnet.web.router.attackers.api_get_attackers import get_attackers
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attackers.repo") as mock_repo:
|
|
mock_repo.get_attackers = AsyncMock(return_value=[])
|
|
mock_repo.get_total_attackers = AsyncMock(return_value=0)
|
|
mock_repo.get_behaviors_for_ips = AsyncMock(return_value={})
|
|
|
|
await get_attackers(
|
|
limit=50, offset=0, search="", sort_by="recent",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
mock_repo.get_attackers.assert_awaited_once_with(
|
|
limit=50, offset=0, search=None, sort_by="recent", service=None,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_filter_forwarded(self):
|
|
from decnet.web.router.attackers.api_get_attackers import get_attackers
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attackers.repo") as mock_repo:
|
|
mock_repo.get_attackers = AsyncMock(return_value=[])
|
|
mock_repo.get_total_attackers = AsyncMock(return_value=0)
|
|
mock_repo.get_behaviors_for_ips = AsyncMock(return_value={})
|
|
|
|
await get_attackers(
|
|
limit=50, offset=0, search=None, sort_by="recent",
|
|
service="https", user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
mock_repo.get_attackers.assert_awaited_once_with(
|
|
limit=50, offset=0, search=None, sort_by="recent", service="https",
|
|
)
|
|
mock_repo.get_total_attackers.assert_awaited_once_with(search=None, service="https")
|
|
|
|
|
|
# ─── GET /attackers/{uuid} ───────────────────────────────────────────────────
|
|
|
|
class TestGetAttackerDetail:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_attacker_by_uuid(self):
|
|
from decnet.web.router.attackers.api_get_attacker_detail import get_attacker_detail
|
|
|
|
sample = _sample_attacker()
|
|
with patch("decnet.web.router.attackers.api_get_attacker_detail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_behavior = AsyncMock(return_value=None)
|
|
mock_repo.get_attacker_service_activity = AsyncMock(return_value=[])
|
|
mock_repo.get_attacker_ip_leaks = AsyncMock(return_value=[])
|
|
mock_repo.count_attacker_ip_leaks = AsyncMock(return_value=0)
|
|
mock_repo.latest_observation_per_primitive = AsyncMock(return_value={})
|
|
|
|
result = await get_attacker_detail(uuid="att-uuid-1", user={"uuid": "test-user", "role": "viewer"})
|
|
|
|
assert result["uuid"] == "att-uuid-1"
|
|
assert result["ip"] == "1.2.3.4"
|
|
assert result["is_traversal"] is True
|
|
assert isinstance(result["commands"], list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_404_on_unknown_uuid(self):
|
|
from decnet.web.router.attackers.api_get_attacker_detail import get_attacker_detail
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attacker_detail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=None)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_attacker_detail(uuid="nonexistent", user={"uuid": "test-user", "role": "viewer"})
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deserialized_json_fields(self):
|
|
from decnet.web.router.attackers.api_get_attacker_detail import get_attacker_detail
|
|
|
|
sample = _sample_attacker()
|
|
with patch("decnet.web.router.attackers.api_get_attacker_detail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_behavior = AsyncMock(return_value=None)
|
|
mock_repo.get_attacker_service_activity = AsyncMock(return_value=[])
|
|
mock_repo.get_attacker_ip_leaks = AsyncMock(return_value=[])
|
|
mock_repo.count_attacker_ip_leaks = AsyncMock(return_value=0)
|
|
mock_repo.latest_observation_per_primitive = AsyncMock(return_value={})
|
|
|
|
result = await get_attacker_detail(uuid="att-uuid-1", user={"uuid": "test-user", "role": "viewer"})
|
|
|
|
assert isinstance(result["services"], list)
|
|
assert isinstance(result["deckies"], list)
|
|
assert isinstance(result["fingerprints"], list)
|
|
assert isinstance(result["commands"], list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_activity_splits_scanned_vs_interacted(self):
|
|
"""Attacker detail response buckets services by event-type signal."""
|
|
from decnet.web.router.attackers.api_get_attacker_detail import get_attacker_detail
|
|
|
|
sample = _sample_attacker()
|
|
pairs = [
|
|
("ssh", "connection"),
|
|
("ssh", "shell_input"), # promotes ssh to interacted
|
|
("http", "get_request"), # scan only
|
|
("ftp", "retr"), # interacted
|
|
("bus", "startup"), # noise — dropped
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_detail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_behavior = AsyncMock(return_value=None)
|
|
mock_repo.get_attacker_service_activity = AsyncMock(return_value=pairs)
|
|
mock_repo.get_attacker_ip_leaks = AsyncMock(return_value=[])
|
|
mock_repo.count_attacker_ip_leaks = AsyncMock(return_value=0)
|
|
mock_repo.latest_observation_per_primitive = AsyncMock(return_value={})
|
|
|
|
result = await get_attacker_detail(
|
|
uuid="att-uuid-1",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert result["service_activity"] == {
|
|
"interacted": ["ftp", "ssh"],
|
|
"scanned": ["http"],
|
|
}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ip_leaks_included_in_response(self):
|
|
"""Attacker detail surfaces ip_leak bounty rows for the UI."""
|
|
from decnet.web.router.attackers.api_get_attacker_detail import get_attacker_detail
|
|
|
|
sample = _sample_attacker()
|
|
leaks = [
|
|
{
|
|
"timestamp": "2026-04-24T12:00:00+00:00",
|
|
"decky": "http-01",
|
|
"service": "http",
|
|
"bounty_type": "ip_leak",
|
|
"payload": {
|
|
"source_ip": "203.0.113.42",
|
|
"real_ip_claim": "198.51.100.7",
|
|
"source_header": "X-Forwarded-For",
|
|
"headers_seen": {
|
|
"X-Forwarded-For": "198.51.100.7",
|
|
},
|
|
},
|
|
},
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_detail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_behavior = AsyncMock(return_value=None)
|
|
mock_repo.get_attacker_service_activity = AsyncMock(return_value=[])
|
|
mock_repo.get_attacker_ip_leaks = AsyncMock(return_value=leaks)
|
|
mock_repo.count_attacker_ip_leaks = AsyncMock(return_value=1)
|
|
mock_repo.latest_observation_per_primitive = AsyncMock(return_value={})
|
|
|
|
result = await get_attacker_detail(
|
|
uuid="att-uuid-1",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert result["ip_leaks"] == leaks
|
|
assert result["ip_leaks"][0]["payload"]["real_ip_claim"] == "198.51.100.7"
|
|
assert result["ip_leaks_total"] == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ip_leaks_total_reported_separately_from_list(self):
|
|
"""Rotation attack: DB has 100 ip_leak rows, endpoint caps the
|
|
returned list at 10 but reports the full count so the UI can
|
|
render a ROTATION DETECTED badge."""
|
|
from decnet.web.router.attackers.api_get_attacker_detail import get_attacker_detail
|
|
|
|
sample = _sample_attacker()
|
|
# Caller already limited; we just assert the shape of the response.
|
|
first_ten = [
|
|
{
|
|
"timestamp": "2026-04-24T12:00:00+00:00",
|
|
"decky": "http-01",
|
|
"service": "http",
|
|
"bounty_type": "ip_leak",
|
|
"payload": {
|
|
"source_ip": "203.0.113.42",
|
|
"real_ip_claim": f"198.51.100.{i}",
|
|
"source_header": "X-Forwarded-For",
|
|
"headers_seen": {},
|
|
},
|
|
}
|
|
for i in range(1, 11)
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_detail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_behavior = AsyncMock(return_value=None)
|
|
mock_repo.get_attacker_service_activity = AsyncMock(return_value=[])
|
|
mock_repo.get_attacker_ip_leaks = AsyncMock(return_value=first_ten)
|
|
mock_repo.count_attacker_ip_leaks = AsyncMock(return_value=100)
|
|
mock_repo.latest_observation_per_primitive = AsyncMock(return_value={})
|
|
|
|
result = await get_attacker_detail(
|
|
uuid="att-uuid-1",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert len(result["ip_leaks"]) == 10
|
|
assert result["ip_leaks_total"] == 100
|
|
|
|
|
|
# ─── GET /attackers/{uuid}/commands ──────────────────────────────────────────
|
|
|
|
class TestGetAttackerCommands:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_paginated_commands(self):
|
|
from decnet.web.router.attackers.api_get_attacker_commands import get_attacker_commands
|
|
|
|
sample = _sample_attacker()
|
|
cmds = [
|
|
{"service": "ssh", "decky": "decky-01", "command": "id", "timestamp": "2026-04-01T10:00:00"},
|
|
{"service": "ssh", "decky": "decky-01", "command": "whoami", "timestamp": "2026-04-01T10:01:00"},
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_commands.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_commands = AsyncMock(return_value={"total": 2, "data": cmds})
|
|
|
|
result = await get_attacker_commands(
|
|
uuid="att-uuid-1", limit=50, offset=0, service=None,
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert result["total"] == 2
|
|
assert len(result["data"]) == 2
|
|
assert result["limit"] == 50
|
|
assert result["offset"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_filter_forwarded(self):
|
|
from decnet.web.router.attackers.api_get_attacker_commands import get_attacker_commands
|
|
|
|
sample = _sample_attacker()
|
|
with patch("decnet.web.router.attackers.api_get_attacker_commands.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_commands = AsyncMock(return_value={"total": 0, "data": []})
|
|
|
|
await get_attacker_commands(
|
|
uuid="att-uuid-1", limit=50, offset=0, service="ssh",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
mock_repo.get_attacker_commands.assert_awaited_once_with(
|
|
uuid="att-uuid-1", limit=50, offset=0, service="ssh",
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_404_on_unknown_uuid(self):
|
|
from decnet.web.router.attackers.api_get_attacker_commands import get_attacker_commands
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attacker_commands.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=None)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_attacker_commands(
|
|
uuid="nonexistent", limit=50, offset=0, service=None,
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
# ─── GET /attackers/{uuid}/artifacts ─────────────────────────────────────────
|
|
|
|
class TestGetAttackerArtifacts:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_artifacts(self):
|
|
from decnet.web.router.attackers.api_get_attacker_artifacts import get_attacker_artifacts
|
|
|
|
sample = _sample_attacker()
|
|
rows = [
|
|
{
|
|
"id": 1,
|
|
"timestamp": "2026-04-18T02:22:56+00:00",
|
|
"decky": "decky-01",
|
|
"service": "ssh",
|
|
"event_type": "file_captured",
|
|
"attacker_ip": "1.2.3.4",
|
|
"raw_line": "",
|
|
"msg": "",
|
|
"fields": json.dumps({
|
|
"stored_as": "2026-04-18T02:22:56Z_abc123def456_drop.bin",
|
|
"sha256": "deadbeef" * 8,
|
|
"size": "4096",
|
|
"orig_path": "/root/drop.bin",
|
|
}),
|
|
},
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_artifacts.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_artifacts = AsyncMock(return_value=rows)
|
|
|
|
result = await get_attacker_artifacts(
|
|
uuid="att-uuid-1",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert result["total"] == 1
|
|
assert result["data"][0]["decky"] == "decky-01"
|
|
mock_repo.get_attacker_artifacts.assert_awaited_once_with("att-uuid-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_404_on_unknown_uuid(self):
|
|
from decnet.web.router.attackers.api_get_attacker_artifacts import get_attacker_artifacts
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attacker_artifacts.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=None)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_attacker_artifacts(
|
|
uuid="nonexistent",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
# ─── GET /attackers/{uuid}/transcripts ───────────────────────────────────────
|
|
|
|
class TestGetAttackerTranscripts:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_transcripts(self):
|
|
from decnet.web.router.attackers.api_get_attacker_transcripts import get_attacker_transcripts
|
|
|
|
sample = _sample_attacker()
|
|
rows = [
|
|
{
|
|
"id": 1,
|
|
"timestamp": "2026-04-18T02:22:56+00:00",
|
|
"decky": "decky-01",
|
|
"service": "ssh",
|
|
"event_type": "session_recorded",
|
|
"attacker_ip": "1.2.3.4",
|
|
"raw_line": "",
|
|
"msg": "",
|
|
"fields": json.dumps({
|
|
"sid": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
|
"service": "ssh",
|
|
"src_ip": "1.2.3.4",
|
|
"duration_s": "42",
|
|
"bytes": "1024",
|
|
"truncated": "false",
|
|
"shard_path": "/var/lib/systemd/coredump/transcripts/sessions-2026-04-18.jsonl",
|
|
}),
|
|
},
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_transcripts.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_transcripts = AsyncMock(return_value=rows)
|
|
|
|
result = await get_attacker_transcripts(
|
|
uuid="att-uuid-1",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert result["total"] == 1
|
|
assert result["data"][0]["service"] == "ssh"
|
|
mock_repo.get_attacker_transcripts.assert_awaited_once_with("att-uuid-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_404_on_unknown_uuid(self):
|
|
from decnet.web.router.attackers.api_get_attacker_transcripts import get_attacker_transcripts
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attacker_transcripts.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=None)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_attacker_transcripts(
|
|
uuid="nonexistent",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
# ─── GET /attackers/{uuid}/smtp-targets ──────────────────────────────────────
|
|
|
|
class TestGetAttackerSmtpTargets:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_smtp_targets(self):
|
|
from decnet.web.router.attackers.api_get_attacker_smtp_targets import (
|
|
get_attacker_smtp_targets,
|
|
)
|
|
|
|
sample = _sample_attacker()
|
|
rows = [
|
|
{
|
|
"id": 1, "attacker_uuid": "att-uuid-1",
|
|
"domain": "corp1.com", "count": 5,
|
|
"first_seen": "2026-04-18T02:22:56+00:00",
|
|
"last_seen": "2026-04-19T10:15:03+00:00",
|
|
},
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_smtp_targets.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.list_smtp_targets = AsyncMock(return_value=rows)
|
|
|
|
result = await get_attacker_smtp_targets(
|
|
uuid="att-uuid-1",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert result["total"] == 1
|
|
assert result["data"][0]["domain"] == "corp1.com"
|
|
mock_repo.list_smtp_targets.assert_awaited_once_with("att-uuid-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_404_on_unknown_uuid(self):
|
|
from decnet.web.router.attackers.api_get_attacker_smtp_targets import (
|
|
get_attacker_smtp_targets,
|
|
)
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attacker_smtp_targets.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=None)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_attacker_smtp_targets(
|
|
uuid="nonexistent",
|
|
user={"uuid": "test-user", "role": "viewer"},
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
# ─── GET /attackers/{uuid}/mail ──────────────────────────────────────────────
|
|
|
|
class TestGetAttackerMail:
|
|
@pytest.mark.asyncio
|
|
async def test_returns_stored_mail(self):
|
|
from decnet.web.router.attackers.api_get_attacker_mail import get_attacker_mail
|
|
|
|
sample = _sample_attacker()
|
|
rows = [
|
|
{
|
|
"id": 1,
|
|
"timestamp": "2026-04-18T02:22:56+00:00",
|
|
"decky": "decky-01", "service": "smtp",
|
|
"event_type": "message_stored",
|
|
"attacker_ip": "1.2.3.4",
|
|
"raw_line": "", "msg": "",
|
|
"fields": json.dumps({
|
|
"stored_as": "2026-04-18T02:22:56Z_abc123def456_ABC123.eml",
|
|
"sha256": "deadbeef" * 8,
|
|
"size": "1024",
|
|
"subject": "URGENT invoice",
|
|
"from_hdr": "spam@evil.com",
|
|
"rcpt_to": "<a@corp.com>",
|
|
"attachment_count": "1",
|
|
}),
|
|
},
|
|
]
|
|
with patch("decnet.web.router.attackers.api_get_attacker_mail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=sample)
|
|
mock_repo.get_attacker_stored_mail = AsyncMock(return_value=rows)
|
|
|
|
result = await get_attacker_mail(
|
|
uuid="att-uuid-1",
|
|
admin={"uuid": "test-admin", "role": "admin"},
|
|
)
|
|
|
|
assert result["total"] == 1
|
|
assert result["data"][0]["event_type"] == "message_stored"
|
|
mock_repo.get_attacker_stored_mail.assert_awaited_once_with("att-uuid-1")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_404_on_unknown_uuid(self):
|
|
from decnet.web.router.attackers.api_get_attacker_mail import get_attacker_mail
|
|
|
|
with patch("decnet.web.router.attackers.api_get_attacker_mail.repo") as mock_repo:
|
|
mock_repo.get_attacker_by_uuid = AsyncMock(return_value=None)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_attacker_mail(
|
|
uuid="nonexistent",
|
|
admin={"uuid": "test-admin", "role": "admin"},
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
|
|
# ─── Auth enforcement ────────────────────────────────────────────────────────
|
|
|
|
class TestAttackersAuth:
|
|
@pytest.mark.asyncio
|
|
async def test_list_requires_auth(self):
|
|
"""get_current_user dependency raises 401 when called without valid token."""
|
|
from decnet.web.dependencies import get_current_user
|
|
|
|
req = MagicMock()
|
|
req.headers = {}
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user(req)
|
|
assert exc_info.value.status_code == 401
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detail_requires_auth(self):
|
|
from decnet.web.dependencies import get_current_user
|
|
|
|
req = MagicMock()
|
|
req.headers = {"Authorization": "Bearer bad-token"}
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await get_current_user(req)
|
|
assert exc_info.value.status_code == 401
|