feat: assemble IOCs in the manifest's shape

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KphFXTc2QajxXsHWyvGJ4R

2 files changed, 168 insertions, 0 deletions

diff --git a/abusectl/parse.py b/abusectl/parse.py
index 6d6789a..eaf6165 100644
--- a/abusectl/parse.py
+++ b/abusectl/parse.py
@@ -38,6 +38,7 @@ import re
 from dataclasses import dataclass
 from email import policy
 from email.parser import BytesParser
+from urllib.parse import parse_qsl, urlsplit
 
 from abusectl import redact
 
@@ -299,3 +300,107 @@ def attachments(raw: bytes) -> list[Attachment]:
             )
         )
     return found
+
+
+def _suspect_segments(raw_url: str) -> list[str]:
+    """Return suspect-looking segments from raw_url's path and query values.
+
+    redact.suspect_path_segments() only looks at the path, but a phishing
+    kit's identifier just as often sits in a query value (the fixture case
+    is exactly this: ``?id=<base64 of the address>``). Checking it there
+    too is what lets review see the shape of the token redaction removes.
+
+    Must be called on the ORIGINAL url, before redact.url() replaces every
+    query value with REDACTED: past that point there is nothing left to
+    recognise as suspect.
+    """
+    found = list(redact.suspect_path_segments(raw_url))
+    query = urlsplit(raw_url).query
+    for _, value in parse_qsl(query, keep_blank_values=True):
+        found.extend(redact.suspect_path_segments(value))
+    return found
+
+
+def _original_urls(message) -> dict[str, str]:
+    """Return {redacted_url: original_url} for every URL in the message.
+
+    urls() returns only the redacted form, which is right for the report
+    but useless for flagging a suspect query value: by the time a value is
+    REDACTED there is nothing left to look at. This mirrors urls()'s own
+    scan so the two stay in step, keyed by the redacted form since that is
+    what the caller already has in hand.
+    """
+    mapping = {}
+    for text in _text_parts(message):
+        for match in _URL.findall(text):
+            original = match.rstrip(".,;:!?")
+            mapping[redact.url(original)] = original
+    return mapping
+
+
+def iocs(raw: bytes, trusted: list[str]) -> list[dict]:
+    """Assemble every indicator this module can extract into one flat list.
+
+    Two decisions shape every entry:
+
+    Every IOC carries an ``id`` (``ioc-1``, ``ioc-2``, ...), assigned in
+    the order entries are appended, so contacts and destinations elsewhere
+    in a case manifest can reference an indicator by id rather than
+    repeating its value. One place to correct a value, and a destination
+    can never drift from the IOC it reports.
+
+    Every IOC carries an ``origin`` saying where it came from
+    (``received-chain``, ``header-from``, ``body``, ``redirect-target``,
+    ``attachment``, ...). During review the user must be able to tell an
+    IP taken from a header we trust from one the attacker wrote; without
+    this, review is guesswork.
+
+    Hops additionally carry a ``confidence``: the Received hop AT the trust
+    boundary (what sending_ip() returns) is ``boundary-hop``, the one we can
+    stand behind, and every hop below it is ``untrusted-hop``, recorded
+    because it may be useful but never presented as fact. Hops INSIDE the
+    boundary are our own infrastructure, not indicators, and are skipped.
+
+    sending_ip() is called first so an empty `trusted` raises
+    NoTrustBoundary before any other extraction runs.
+    """
+    boundary_ip = sending_ip(raw, trusted)
+    originals = _original_urls(_message(raw))
+
+    result: list[dict] = []
+
+    def add(**fields) -> None:
+        result.append({"id": f"ioc-{len(result) + 1}", **fields})
+
+    for hop in received_hops(raw):
+        if _in_any(hop.ip, trusted):
+            continue
+        add(
+            type="ipv6" if ":" in hop.ip else "ipv4",
+            value=hop.ip,
+            origin="received-chain",
+            confidence="boundary-hop" if hop.ip == boundary_ip else "untrusted-hop",
+        )
+
+    for key, domain in sender_domains(raw).items():
+        add(type="domain", value=domain, origin=f"header-{key}")
+
+    for url in urls(raw):
+        entry = {"type": "url", "value": url, "origin": "body"}
+        segments = _suspect_segments(originals.get(url, url))
+        if segments:
+            entry["suspect_path_segments"] = segments
+        add(**entry)
+
+    for source, target in redirect_chains(raw):
+        add(type="url", value=target, origin="redirect-target", redirect_from=source)
+
+    for attachment in attachments(raw):
+        add(
+            type="sha256",
+            value=attachment.sha256,
+            origin="attachment",
+            filename=attachment.filename,
+        )
+
+    return result
diff --git a/tests/test_parse.py b/tests/test_parse.py
index 0fac304..28c8609 100644
--- a/tests/test_parse.py
+++ b/tests/test_parse.py
@@ -157,5 +157,68 @@ class TestAttachments(unittest.TestCase):
         self.assertEqual(parse.attachments(load("simple.eml")), [])
 
 
+class TestIocAssembly(unittest.TestCase):
+    def test_every_ioc_has_a_unique_id_and_an_origin(self):
+        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
+        ids = [i["id"] for i in iocs]
+        self.assertEqual(len(ids), len(set(ids)))
+        self.assertTrue(all(i["origin"] for i in iocs))
+
+    def test_the_sending_ip_is_present_and_marked_boundary_hop(self):
+        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
+        ips = [i for i in iocs if i["type"] == "ipv4"]
+        self.assertEqual(ips[0]["value"], "203.0.113.42")
+        self.assertEqual(ips[0]["confidence"], "boundary-hop")
+
+    def test_hops_below_the_boundary_are_marked_untrusted(self):
+        iocs = parse.iocs(load("forged-chain.eml"), trusted=["192.0.2.0/24"])
+        ips = {i["value"]: i for i in iocs if i["type"] == "ipv4"}
+        self.assertEqual(ips["203.0.113.99"]["confidence"], "boundary-hop")
+        self.assertEqual(ips["198.51.100.7"]["confidence"], "untrusted-hop")
+
+    def test_our_own_relays_are_not_reported_as_indicators(self):
+        # Inside the boundary is our own infrastructure, not an indicator.
+        iocs = parse.iocs(load("forged-chain.eml"), trusted=["192.0.2.0/24"])
+        values = [i["value"] for i in iocs]
+        self.assertNotIn("192.0.2.11", values)
+
+    def test_urls_carry_their_redacted_form(self):
+        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
+        urls = [i for i in iocs if i["type"] == "url"]
+        self.assertEqual(len(urls), 1)
+        self.assertIn("REDACTED", urls[0]["value"])
+
+    def test_a_suspect_path_segment_is_flagged_on_the_ioc(self):
+        iocs = parse.iocs(load("simple.eml"), trusted=["192.0.2.0/24"])
+        url = next(i for i in iocs if i["type"] == "url")
+        self.assertEqual(url["suspect_path_segments"], ["dGVzdEBleGFtcGxlLm9yZw"])
+
+    def test_a_redirect_target_is_its_own_ioc(self):
+        iocs = parse.iocs(load("redirector.eml"), trusted=["192.0.2.0/24"])
+        targets = [i for i in iocs if i["origin"] == "redirect-target"]
+        self.assertEqual(len(targets), 1)
+        self.assertEqual(targets[0]["value"],
+                         "http://evil.example.invalid/pay?ref=REDACTED")
+
+    def test_an_attachment_becomes_a_hash_ioc(self):
+        iocs = parse.iocs(load("with-attachment.eml"), trusted=["192.0.2.0/24"])
+        hashes = [i for i in iocs if i["type"] == "sha256"]
+        self.assertEqual(len(hashes), 1)
+        self.assertEqual(hashes[0]["filename"], "invoice.pdf")
+
+    def test_no_trusted_relays_still_refuses(self):
+        with self.assertRaises(parse.NoTrustBoundary):
+            parse.iocs(load("simple.eml"), trusted=[])
+
+    def test_no_ioc_holds_a_recipient_address(self):
+        # The safety property, asserted over the whole output.
+        for name in ("simple.eml", "forged-chain.eml", "with-attachment.eml",
+                     "redirector.eml"):
+            iocs = parse.iocs(load(name), trusted=["192.0.2.0/24"])
+            blob = repr(iocs)
+            self.assertNotIn("you@example.org", blob)
+            self.assertNotIn("example.org", blob)
+
+
 if __name__ == "__main__":
     unittest.main()