feat(profiler): write ASN + AS name onto attacker rows

Adds asn (int), as_name (varchar 128), asn_source (varchar 16) to
the Attacker SQLModel — direct columns, no _migrate_* helper per
feedback_no_new_migrations_prev1.

Profiler worker now calls decnet.asn.enrich_ip alongside the existing
geoip enrich_ip; both feed the upsert payload. Failure is total — if
either lookup throws or the IP is private/unannounced, the field stays
None and the row still writes.

Both lookups are independent: a CGNAT address can have a country (RIR
allocation) but no ASN (no BGP origin), and vice-versa for unrouted
RIR-allocated space. Storing them separately preserves that signal.

decnet/profiler/worker.py

@@ -29,6 +29,7 @@ from decnet.bus.publish import (
 )
 from decnet.correlation.engine import CorrelationEngine
 from decnet.correlation.parser import LogEvent
+from decnet.asn import enrich_ip as enrich_ip_asn
 from decnet.geoip import enrich_ip
 from decnet.geoip.ptr import resolve_ptr_record
 from decnet.logging import get_logger
@@ -307,6 +308,7 @@ def _build_record(
     fingerprints = [b for b in bounties if b.get("bounty_type") == "fingerprint"]
     credential_count = sum(1 for b in bounties if b.get("bounty_type") == "credential")
     country_code, country_source = enrich_ip(ip)
+    asn, as_name, asn_source = enrich_ip_asn(ip)
 
     record: dict[str, Any] = {
         "ip": ip,
@@ -325,6 +327,9 @@ def _build_record(
         "commands": json.dumps(commands),
         "country_code": country_code,
         "country_source": country_source,
+        "asn": asn,
+        "as_name": as_name,
+        "asn_source": asn_source,
         "updated_at": datetime.now(timezone.utc),
     }
     # ptr_record is omitted from the dict entirely when the caller didn't

decnet/web/db/models/attackers.py

@@ -63,6 +63,12 @@ class Attacker(SQLModel, table=True):
     # Nullable because private / loopback / IPv6 sources never resolve.
     country_code: Optional[str] = Field(default=None, max_length=2, index=True)
     country_source: Optional[str] = Field(default=None, max_length=16)
+    # ASN enrichment (populated by the profiler from decnet.asn.enrich_ip).
+    # Nullable for the same reasons as country_code, plus IPs not currently
+    # announced in the global BGP table (e.g. CGNAT, dark space).
+    asn: Optional[int] = Field(default=None, index=True)
+    as_name: Optional[str] = Field(default=None, max_length=128)
+    asn_source: Optional[str] = Field(default=None, max_length=16)
     # Reverse-DNS (PTR) name, one-shot resolved by the profiler at first
     # sighting. Nullable — many attackers run infra with no rDNS, and
     # private/loopback addresses never resolve.  256 chars matches

tests/asn/test_profiler_integration.py

@@ -0,0 +1,50 @@
+"""_build_record must thread ASN fields through to the upsert payload."""
+from __future__ import annotations
+
+import gzip
+from datetime import datetime, timezone
+from pathlib import Path
+
+from decnet.correlation.parser import LogEvent
+from decnet.profiler.worker import _build_record
+
+
+def _evt(ip: str) -> LogEvent:
+    return LogEvent(
+        timestamp=datetime(2026, 4, 23, tzinfo=timezone.utc),
+        attacker_ip=ip,
+        decky="decky-01",
+        service="ssh",
+        event_type="conn",
+        fields={},
+        raw="",
+    )
+
+
+def _seed(root: Path) -> None:
+    target = root / "ip2asn-v4.tsv.gz"
+    with gzip.open(target, "wt", encoding="utf-8") as fh:
+        fh.write("8.8.8.0\t8.8.8.255\t15169\tUS\tGOOGLE\n")
+
+
+def test_build_record_includes_asn_when_resolved(tmp_path: Path) -> None:
+    _seed(tmp_path)
+    record = _build_record("8.8.8.8", [_evt("8.8.8.8")], None, [], [])
+    assert record["asn"] == 15169
+    assert record["as_name"] == "GOOGLE"
+    assert record["asn_source"] == "iptoasn"
+
+
+def test_build_record_asn_none_for_private(tmp_path: Path) -> None:
+    _seed(tmp_path)
+    record = _build_record("10.0.0.1", [_evt("10.0.0.1")], None, [], [])
+    assert record["asn"] is None
+    assert record["as_name"] is None
+    assert record["asn_source"] is None
+
+
+def test_build_record_asn_none_for_unannounced(tmp_path: Path) -> None:
+    _seed(tmp_path)
+    # 9.0.0.0 isn't in the seeded fixture range — no BGP origin we know of.
+    record = _build_record("9.0.0.0", [_evt("9.0.0.0")], None, [], [])
+    assert record["asn"] is None