feat(intel): worker shell + attacker.intel.enriched bus topic

run_intel_loop fans out across configured providers per IP, writes the
aggregate row, and publishes attacker.intel.enriched. Mirrors the
correlation/reuse_worker.py wake-on pattern: subscribes to
attacker.observed and attacker.scored for sub-second latency, falls back
to a 60s poll when the bus is unavailable. Heartbeat + control-listener
wired so the workers panel sees it like every other supervised worker.

Aggregate verdict picks the strongest provider tier (malicious >
suspicious > benign > unknown). Provider-level errors land in
IntelResult.error and are logged without poisoning the row — partial
success is the expected case for free-tier providers under their daily
caps.

Concrete provider impls land in follow-up commits; the worker is fully
exercised here against fake providers so the framing is locked in.

decnet/bus/topics.py

@@ -77,6 +77,11 @@ ATTACKER_SCORED = "scored"
 ATTACKER_FINGERPRINTED = "fingerprinted"
 ATTACKER_SESSION_STARTED = "session.started"
 ATTACKER_SESSION_ENDED = "session.ended"
+# Published by the ``decnet enrich`` worker after an enrichment pass
+# succeeds for an attacker IP (one or more 3rd-party intel providers
+# returned a verdict).  Payload carries the aggregate verdict + per-
+# provider summary so SIEM-bound webhooks don't need to re-query the DB.
+ATTACKER_INTEL_ENRICHED = "intel.enriched"
 
 # Credential event types (second/third tokens under ``credential``).
 # ``credential.captured`` fires once per upserted Credential row — the

decnet/intel/worker.py

@@ -0,0 +1,220 @@
+"""Long-running threat-intel enrichment worker.
+
+Fans out per attacker IP across the configured intel providers
+(GreyNoise / AbuseIPDB / abuse.ch Feodo + ThreatFox), writes the
+combined verdict to ``attacker_intel``, and publishes
+``attacker.intel.enriched`` for downstream consumers (SIEM webhooks,
+dashboard).
+
+Mirrors :mod:`decnet.correlation.reuse_worker` — bus-woken on
+``attacker.scored`` and ``attacker.observed`` for sub-second latency,
+falls back to a slow tick (default 60s) when the bus is unavailable so
+operators with bus disabled still get periodic backfills.
+
+A single worker instance handles all providers; provider-level
+concurrency is bounded by the per-provider semaphore on each
+:class:`~decnet.intel.base.IntelProvider`. The worker itself does not
+hold a global lock — each IP runs through its providers concurrently.
+"""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from datetime import datetime, timedelta, timezone
+from typing import Any, Optional
+
+from decnet.bus import topics as _topics
+from decnet.bus.base import BaseBus
+from decnet.bus.factory import get_bus
+from decnet.bus.publish import (
+    publish_safely,
+    run_control_listener_signal as _run_control_listener_signal,
+    run_health_heartbeat as _run_health_heartbeat,
+)
+from decnet.intel.base import IntelProvider, IntelResult
+from decnet.intel.factory import get_intel_providers
+from decnet.logging import get_logger
+from decnet.web.db.repository import BaseRepository
+
+log = get_logger("intel.worker")
+
+_DEFAULT_POLL_SECS = 60.0
+_DEFAULT_TTL_HOURS = 24
+_BACKFILL_BATCH = 50
+
+# Aggregate-verdict precedence: most-confident first. Any provider
+# returning the higher tier wins regardless of how many lower-tier
+# verdicts exist alongside it.
+_VERDICT_PRECEDENCE = ("malicious", "suspicious", "benign", "unknown")
+
+
+def _aggregate(verdicts: list[Optional[str]]) -> Optional[str]:
+    """Pick the strongest provider verdict, or ``None`` if all silent."""
+    seen = {v for v in verdicts if v}
+    if not seen:
+        return None
+    for tier in _VERDICT_PRECEDENCE:
+        if tier in seen:
+            return tier
+    return None
+
+
+async def _enrich_one(
+    ip: str,
+    providers: list[IntelProvider],
+    ttl_hours: int,
+) -> dict[str, Any]:
+    """Fan out across providers for a single IP and assemble the row update."""
+    results: list[IntelResult] = await asyncio.gather(
+        *(p.lookup(ip) for p in providers),
+        return_exceptions=False,  # providers contractually never raise
+    )
+
+    now = datetime.now(timezone.utc)
+    row: dict[str, Any] = {
+        "attacker_ip": ip,
+        "cached_at": now,
+        "expires_at": now + timedelta(hours=ttl_hours),
+    }
+    verdicts: list[Optional[str]] = []
+    for result in results:
+        if result.error:
+            log.warning(
+                "intel: provider %s failed for ip=%s: %s",
+                result.provider, ip, result.error,
+            )
+            continue
+        row.update(result.column_updates)
+        verdicts.append(result.verdict)
+    row["aggregate_verdict"] = _aggregate(verdicts)
+    return row
+
+
+async def run_intel_loop(
+    repo: BaseRepository,
+    *,
+    poll_interval_secs: float = _DEFAULT_POLL_SECS,
+    ttl_hours: int = _DEFAULT_TTL_HOURS,
+    backfill_batch: int = _BACKFILL_BATCH,
+    providers: Optional[list[IntelProvider]] = None,
+    shutdown: Optional[asyncio.Event] = None,
+) -> None:
+    """Run the intel-enrichment loop until cancelled.
+
+    *providers* defaults to :func:`get_intel_providers` — tests pass a
+    list of fakes. *shutdown* is an optional external stop signal; the
+    loop also exits cleanly on ``CancelledError`` and ``KeyboardInterrupt``.
+    """
+    if providers is None:
+        providers = get_intel_providers()
+    log.info(
+        "intel worker started providers=%s poll=%ss ttl=%sh",
+        [p.name for p in providers], poll_interval_secs, ttl_hours,
+    )
+
+    bus: Optional[BaseBus] = None
+    wake = asyncio.Event()
+    wake_tasks: list[asyncio.Task] = []
+    heartbeat_task: Optional[asyncio.Task] = None
+    try:
+        candidate = get_bus(client_name="intel")
+        await candidate.connect()
+        bus = candidate
+        wake_tasks.append(asyncio.create_task(
+            _wake_on(bus, wake, _topics.attacker(_topics.ATTACKER_OBSERVED)),
+        ))
+        wake_tasks.append(asyncio.create_task(
+            _wake_on(bus, wake, _topics.attacker(_topics.ATTACKER_SCORED)),
+        ))
+        heartbeat_task = asyncio.create_task(
+            _run_health_heartbeat(bus, "intel"),
+        )
+        wake_tasks.append(asyncio.create_task(
+            _run_control_listener_signal(bus, "intel"),
+        ))
+    except Exception as exc:  # noqa: BLE001
+        log.warning(
+            "intel worker: bus unavailable, running in poll-only mode: %s",
+            exc,
+        )
+
+    if shutdown is None:
+        shutdown = asyncio.Event()
+
+    try:
+        while not shutdown.is_set():
+            try:
+                pending = await repo.get_unenriched_attacker_ips(
+                    limit=backfill_batch,
+                )
+            except Exception:  # noqa: BLE001
+                log.exception("intel worker: backfill query failed")
+                pending = []
+
+            if pending and providers:
+                for ip in pending:
+                    if shutdown.is_set():
+                        break
+                    try:
+                        row = await _enrich_one(ip, providers, ttl_hours)
+                        await repo.upsert_attacker_intel(row)
+                        await publish_safely(
+                            bus,
+                            _topics.attacker(_topics.ATTACKER_INTEL_ENRICHED),
+                            {
+                                "attacker_ip": ip,
+                                "aggregate_verdict": row.get("aggregate_verdict"),
+                                "providers": [p.name for p in providers],
+                            },
+                            event_type=_topics.ATTACKER_INTEL_ENRICHED,
+                        )
+                    except Exception:  # noqa: BLE001
+                        log.exception(
+                            "intel worker: enrichment failed for ip=%s", ip,
+                        )
+
+            try:
+                await asyncio.wait_for(
+                    wake.wait(), timeout=float(poll_interval_secs),
+                )
+            except asyncio.TimeoutError:
+                pass
+            wake.clear()
+    except (asyncio.CancelledError, KeyboardInterrupt):
+        log.info("intel worker stopped")
+    finally:
+        for t in wake_tasks:
+            t.cancel()
+        if heartbeat_task is not None:
+            heartbeat_task.cancel()
+        for t in (*wake_tasks, heartbeat_task):
+            if t is None:
+                continue
+            with contextlib.suppress(asyncio.CancelledError, Exception):
+                await t
+        if bus is not None:
+            with contextlib.suppress(Exception):
+                await bus.close()
+
+
+async def _wake_on(bus: BaseBus, wake: asyncio.Event, pattern: str) -> None:
+    """Flip *wake* every time *pattern* fires on the bus.
+
+    Survives transient subscriber errors by logging and exiting; the
+    poll-interval fallback keeps the loop alive in poll-only mode.
+    """
+    try:
+        sub = bus.subscribe(pattern)
+        async with sub:
+            async for _event in sub:
+                wake.set()
+    except asyncio.CancelledError:
+        raise
+    except Exception as exc:  # noqa: BLE001
+        log.warning(
+            "intel worker: subscriber for %s died (%s); falling back to poll",
+            pattern, exc,
+        )
+
+
+__all__ = ["run_intel_loop"]

tests/intel/test_worker.py

@@ -0,0 +1,205 @@
+"""End-to-end tests for the intel worker shell.
+
+Covers — without any real provider impls — that the loop:
+
+* exits cleanly on shutdown signal (and via cancel)
+* does nothing when no providers are configured
+* fans out across fake providers and writes the aggregate row
+* aggregate_verdict picks the strongest provider verdict
+* a provider returning ``error`` is logged but does not poison the row
+* gates IPs through ``get_unenriched_attacker_ips`` (TTL respected)
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import datetime, timezone
+from typing import Optional
+
+import pytest
+
+from decnet.intel.base import IntelProvider, IntelResult
+from decnet.intel.worker import run_intel_loop, _aggregate
+from decnet.web.db.factory import get_repository
+
+
+class _FakeProvider(IntelProvider):
+    """Test double — instantly returns a canned :class:`IntelResult`."""
+
+    concurrency = 1
+    min_dispatch_interval_s = 0.0
+
+    def __init__(
+        self,
+        name: str,
+        *,
+        verdict: Optional[str] = None,
+        error: Optional[str] = None,
+        column_updates: Optional[dict] = None,
+    ) -> None:
+        super().__init__()
+        self.name = name
+        self._verdict = verdict
+        self._error = error
+        self._cols = column_updates or {}
+        self.calls: list[str] = []
+
+    async def lookup(self, ip: str) -> IntelResult:
+        self.calls.append(ip)
+        return IntelResult(
+            provider=self.name,
+            verdict=self._verdict,
+            error=self._error,
+            column_updates=self._cols,
+        )
+
+
+@pytest.fixture
+async def repo(tmp_path):
+    r = get_repository(db_path=str(tmp_path / "intel_worker.db"))
+    await r.initialize()
+    return r
+
+
+# Disable bus connection in tests — workers under test should run in
+# poll-only mode without hitting a real Unix socket.
+@pytest.fixture(autouse=True)
+def _no_bus(monkeypatch):
+    monkeypatch.setenv("DECNET_BUS_ENABLED", "false")
+
+
+def test_aggregate_picks_strongest_verdict():
+    assert _aggregate(["benign", "malicious", None]) == "malicious"
+    assert _aggregate(["benign", "suspicious"]) == "suspicious"
+    assert _aggregate(["benign", None]) == "benign"
+    assert _aggregate([None, None]) is None
+    assert _aggregate([]) is None
+
+
+@pytest.mark.anyio
+async def test_loop_exits_on_shutdown_signal(repo):
+    shutdown = asyncio.Event()
+    task = asyncio.create_task(
+        run_intel_loop(
+            repo,
+            poll_interval_secs=0.05,
+            providers=[],
+            shutdown=shutdown,
+        )
+    )
+    await asyncio.sleep(0.1)
+    shutdown.set()
+    await asyncio.wait_for(task, timeout=2.0)
+
+
+@pytest.mark.anyio
+async def test_no_providers_skips_enrichment(repo):
+    now = datetime.now(timezone.utc)
+    await repo.upsert_attacker(
+        {"ip": "1.1.1.1", "first_seen": now, "last_seen": now, "event_count": 1}
+    )
+    shutdown = asyncio.Event()
+    task = asyncio.create_task(
+        run_intel_loop(
+            repo,
+            poll_interval_secs=0.05,
+            providers=[],
+            shutdown=shutdown,
+        )
+    )
+    await asyncio.sleep(0.15)
+    shutdown.set()
+    await asyncio.wait_for(task, timeout=2.0)
+    # No row written for 1.1.1.1.
+    assert await repo.get_attacker_intel_by_ip("1.1.1.1") is None
+
+
+@pytest.mark.anyio
+async def test_fan_out_writes_aggregate_row(repo):
+    now = datetime.now(timezone.utc)
+    await repo.upsert_attacker(
+        {"ip": "2.2.2.2", "first_seen": now, "last_seen": now, "event_count": 1}
+    )
+
+    gn = _FakeProvider(
+        "greynoise",
+        verdict="benign",
+        column_updates={
+            "greynoise_classification": "benign",
+            "greynoise_raw": json.dumps({"classification": "benign"}),
+            "greynoise_queried_at": datetime.now(timezone.utc),
+        },
+    )
+    aip = _FakeProvider(
+        "abuseipdb",
+        verdict="malicious",
+        column_updates={
+            "abuseipdb_score": 90,
+            "abuseipdb_raw": json.dumps({"abuseConfidenceScore": 90}),
+            "abuseipdb_queried_at": datetime.now(timezone.utc),
+        },
+    )
+
+    shutdown = asyncio.Event()
+    task = asyncio.create_task(
+        run_intel_loop(
+            repo,
+            poll_interval_secs=0.05,
+            providers=[gn, aip],
+            shutdown=shutdown,
+        )
+    )
+    # One tick is enough — both providers respond instantly.
+    await asyncio.sleep(0.15)
+    shutdown.set()
+    await asyncio.wait_for(task, timeout=2.0)
+
+    row = await repo.get_attacker_intel_by_ip("2.2.2.2")
+    assert row is not None
+    assert row["greynoise_classification"] == "benign"
+    assert row["abuseipdb_score"] == 90
+    # Strongest verdict wins.
+    assert row["aggregate_verdict"] == "malicious"
+    # Both providers were queried.
+    assert gn.calls == ["2.2.2.2"]
+    assert aip.calls == ["2.2.2.2"]
+
+
+@pytest.mark.anyio
+async def test_provider_error_does_not_poison_row(repo):
+    now = datetime.now(timezone.utc)
+    await repo.upsert_attacker(
+        {"ip": "3.3.3.3", "first_seen": now, "last_seen": now, "event_count": 1}
+    )
+
+    good = _FakeProvider(
+        "greynoise",
+        verdict="benign",
+        column_updates={
+            "greynoise_classification": "benign",
+            "greynoise_raw": "{}",
+            "greynoise_queried_at": datetime.now(timezone.utc),
+        },
+    )
+    broken = _FakeProvider("abuseipdb", error="HTTP 500")
+
+    shutdown = asyncio.Event()
+    task = asyncio.create_task(
+        run_intel_loop(
+            repo,
+            poll_interval_secs=0.05,
+            providers=[good, broken],
+            shutdown=shutdown,
+        )
+    )
+    await asyncio.sleep(0.15)
+    shutdown.set()
+    await asyncio.wait_for(task, timeout=2.0)
+
+    row = await repo.get_attacker_intel_by_ip("3.3.3.3")
+    assert row is not None
+    assert row["greynoise_classification"] == "benign"
+    # Broken provider's columns stay null; row is still written.
+    assert row["abuseipdb_score"] is None
+    # Aggregate reflects only the providers that responded.
+    assert row["aggregate_verdict"] == "benign"