path: root/abusectl/report.py

# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""IOCs and abuse contacts to report bodies: the last step before anything
irreversible happens.

This module is PURE and OFFLINE. It opens no socket, sends no mail and reads
no file outside the case directory. What it produces is a document the user
reads, edits and approves, so the output is written for a human first and a
parser second.

It takes the reporting identity as an ARGUMENT rather than reading the
config, the way parse.py takes the trust boundary. The identity is the one
thing in a report disclosed deliberately, and a module that reaches for it
itself is a module that can disclose it in a code path nobody reviewed.
"""

import hashlib

from datetime import datetime, timezone
from email.header import Header
from email.headerregistry import Address
from email.message import EmailMessage
from email.policy import SMTP
from pathlib import Path


VERSION = "0.1.0"


# Every IOC type parse.py emits. Named rather than derived, so a new type
# arriving without a decision about which destinations accept it is a test
# failure rather than an indicator that quietly reaches MISP alone.
ALL_TYPES = ("ipv4", "ipv6", "domain", "url", "sha256")


# What each destination accepts, read from its own API documentation on
# 2026-09-10:
#
#   misp      MISP book, circl.lu/doc/misp/automation. Any attribute type.
#   abusedb   docs.abuseipdb.com, POST /api/v2/report: "a valid IPv4 or
#             IPv6 address", and nothing else.
#   urlhaus   abuse.ch's own submit_url.py, POST urlhaus.abuse.ch/api/,
#             whose submission array carries a url. The full docs are
#             behind a login at auth.abuse.ch; re-verify the PAYLOAD there
#             when submit is built. The accepted type is unambiguous.
#
# VirusTotal is deliberately absent: its only submission endpoints are
# POST /urls and file upload, so it accepts exactly what urlhaus accepts
# and takes no verdict with a submission. See backlog item 6.
#
# DO NOT EDIT THESE FROM MEMORY. init.PROVIDERS records what that costs:
# the first draft was written from memory and every range was wrong. A
# wrong "accepts" here means a row promising a submission the endpoint
# will refuse, or a missing row for something the vendor would have taken.
DESTINATIONS: dict[str, dict] = {
    "misp": {"kind": "misp", "accepts": ALL_TYPES},
    "abusedb": {"kind": "api", "accepts": ("ipv4", "ipv6")},
    "urlhaus": {"kind": "api", "accepts": ("url",)},
}


# Hard wrap column, from the spec: "Plain text, hard-wrapped at 72 columns."
# Abuse desks run ticketing systems that reflow or clip long lines, and a
# clipped line is the defect this whole wrapping scheme exists to prevent.
_WIDTH = 72

# The marker that says "this value continues on the next line". A trailing
# backslash is the convention shells, C and Makefiles all use, so it reads
# without a legend, and it is what makes a broken line UNAMBIGUOUS: a reader
# seeing no backslash knows the value ended there.
#
# The alternative, breaking silently, is the same defect as truncating. A
# desk that copies one line of a wrapped URL and acts on it has acted on a
# resource that was never reported. Something must mark the seam, and this
# marks it in the only direction that is safe: a fragment ANNOUNCES that it
# is a fragment, rather than a whole value having to prove it is whole.
#
# Because the marker is a character a VALUE may also contain, every literal
# backslash in a value is DOUBLED before wrapping and halved on the way
# back. Without that, a value ending in "\" is indistinguishable from a wrap
# marker, and the values here are attacker-supplied: a trailing backslash is
# legal in a URL path, and an attacker reading this source could append one
# to make their indicator garble itself in the report a desk reads.
#
# Doubling EVERY backslash rather than only a trailing one is what keeps the
# encoding unambiguous. Escaping just the last character leaves "x\\" (two
# literal backslashes) encoding to the same text as "x\" followed by a wrap,
# which is the same bug one character further along.
_CONTINUATION = "\\"
_ESCAPE = "\\"


def _escape(value: str) -> str:
    """Double every backslash so none can be read as a continuation marker.

    The inverse is _unescape(). Applied to the value only, never to the
    indent or the surrounding prose, so what a reader sees differs from the
    literal value in exactly one way and unwrap() undoes exactly that.
    """
    return value.replace(_ESCAPE, _ESCAPE + _ESCAPE)


def _unescape(value: str) -> str:
    """Halve the doubled backslashes _escape() produced.

    Scans left to right, consuming a doubled pair as one character. On text
    _escape() actually produced, every run is even and str.replace gives
    the same answer, which a mutation test confirmed over every backslash
    pattern up to length 11: this is NOT protecting the round trip, and
    saying otherwise would overstate what the tests hold down.

    It is kept because unwrap() is public and may be handed a line a user
    edited, where a run can be odd. The scan then consumes pairs strictly
    left to right and leaves the odd one alone, which is the reading that
    matches how the text was written; str.replace rescans its own output
    and would fold a stray backslash into the pair beside it.
    """
    out = []
    index = 0
    while index < len(value):
        if value.startswith(_ESCAPE + _ESCAPE, index):
            out.append(_ESCAPE)
            index += 2
        else:
            out.append(value[index])
            index += 1
    return "".join(out)


def _wrap_value(value: str, indent: str) -> list[str]:
    """Break one long value across lines so no line exceeds _WIDTH.

    Breaks at an arbitrary column rather than at a word or a punctuation
    boundary, DELIBERATELY. A URL has no whitespace, so a word-wrapper
    leaves it over-long and the column guarantee fails on exactly the value
    that matters most. Breaking after a "/" or a "&" instead would be
    prettier and is wrong: those characters are meaningful inside the value,
    so a break at one is a place a reader cannot tell a seam from content.
    An arbitrary break plus an explicit marker is legible precisely because
    the marker, not the position, carries the meaning.

    The value is ESCAPED first, so its own backslashes cannot be mistaken
    for the marker, and the wrap arithmetic then runs over the escaped text:
    the column limit governs what is PRINTED, and the escaped form is what
    is printed. Measuring the original instead would let a value full of
    backslashes overflow the line.

    A break must NEVER land between the two halves of an escaped pair, and
    the loop below backs off by one character to guarantee it. This is not
    tidiness: unwrap() decides whether a trailing backslash is a marker or
    content by the PARITY of the run it ends, and a break inside a pair
    splits that run across two lines, so both halves are counted wrongly.
    A genuine marker then reads as content, the continuation line is
    orphaned, and the tail of the value is silently dropped.

    That defect survived a first fix and a passing test suite, because it
    needs a backslash to land exactly on the break column: it appears in
    random adversarial values roughly one time in eight and in none of the
    hand-written cases. Backing off one character costs a column on a line
    that has a backslash at its edge, and buys an invariant that holds for
    every input rather than for the inputs someone thought of.

    The value is never altered, only divided and escaped; unwrap() is the
    exact inverse, asserted over adversarial values including trailing and
    doubled backslashes.
    """
    value = _escape(value)
    room = _WIDTH - len(indent) - len(_CONTINUATION)
    if len(indent) + len(value) <= _WIDTH:
        return [indent + value]
    lines = []
    while len(value) > room:
        cut = room
        # An odd run of backslashes ending at the cut means the last one is
        # the first half of a pair; move the break before it.
        if (len(value[:cut]) - len(value[:cut].rstrip(_ESCAPE))) % 2 == 1:
            cut -= 1
        lines.append(indent + value[:cut] + _CONTINUATION)
        value = value[cut:]
    lines.append(indent + value)
    return lines


def unwrap(text: str) -> str:
    """Rejoin lines a continuation marker broke, giving back the values.

    The inverse of the wrapping above, and the reason the wrapping is
    honest rather than merely tidy: a value that can be mechanically
    reassembled is a value that was not damaged by being displayed. A desk
    that scripts against the text part gets its indicators back exactly as
    reported, and the tests assert reassembly rather than mere presence,
    which is what rules out a truncation passing as a wrap.

    The leading whitespace of a continuation line is display indent, not
    content: no value this module emits begins with a space, because every
    one of them is an indicator, a header value or an identity, all of
    which are stripped before they arrive.

    A trailing backslash is a marker only when the run of backslashes it
    ends is ODD, because _escape() doubled every literal one. An even run is
    entirely escaped content and the line ends there. Testing endswith("\\")
    alone was the defect this parity check replaces: it read a value's own
    trailing backslash as a marker and swallowed the following line, which
    for a Subject meant absorbing the Date header beneath it.

    Rejoining happens BEFORE unescaping, so a doubled pair split across a
    break is whole again before it is decoded.
    """
    joined = []
    for line in text.splitlines():
        if joined and _ends_with_marker(joined[-1]):
            joined[-1] = joined[-1][:-len(_CONTINUATION)] + line.lstrip()
        else:
            joined.append(line)
    return "\n".join(_unescape(line) for line in joined)


def _ends_with_marker(line: str) -> bool:
    """Whether this line ends in a continuation marker rather than content.

    The marker is one unescaped backslash, and _escape() doubled every
    literal one, so the question is purely the PARITY of the trailing run:
    odd means the last backslash has no partner and is the marker, even
    means every one is half of an escaped pair and the line ends here.
    """
    run = len(line) - len(line.rstrip(_ESCAPE))
    return run % 2 == 1


# What each parse.py origin means in a sentence a desk can act on.
#
# The raw tokens are a PARSER's vocabulary: "header-list_unsubscribe" has an
# underscore in it and reads as debug output, which makes a careful report
# look machine-dumped and invites a desk to discount it. The mapping is
# deliberately small and flat, one line each, because the alternative, a
# sentence generated per indicator, is a second body of prose to keep true.
#
# An origin absent from this table is shown AS-IS rather than dropped: a
# newer parse.py may invent one, and losing the only line that says where an
# indicator was seen is worse than showing an ugly token. Ugly is also
# self-correcting, since it is visible to whoever reads the next report.
#
# Each phrase completes the sentence "seen ...", so every entry must read
# grammatically after that word. The first draft mixed "from the Received
# chain" with "the From header" and rendered "seen the From header", which
# reads as a typo and undercuts exactly the care the report is meant to
# show. Keep new entries in the same voice.
_ORIGINS = {
    "received-chain": "in the Received chain",
    "body": "in a link in the message body",
    "redirect-target": "as a redirect target declared by another link",
    "attachment": "as an attachment",
    "header-from": "in the From header",
    "header-reply_to": "in the Reply-To header",
    "header-return_path": "in the Return-Path header",
    "header-list_unsubscribe": "in the List-Unsubscribe header",
}


_REDACTION_NOTE = (
    "Recipient identifiers have been removed from this report by policy.",
    "Parameter names are preserved, parameter values are not. Full",
    "evidence is retained locally and is available on request.",
)


# The role mailboxes RFC 2142 mandates, which it also requires be matched
# case-insensitively. Only the ones an RDAP abuse entity plausibly
# publishes; this is not the full list and does not need to be.
_ROLE_MAILBOXES = frozenset({
    "abuse", "postmaster", "security", "noc", "hostmaster",
})


def _is_mailable(address: str) -> bool:
    """Whether this value can be the target of a mail at all.

    RDAP jCard data is third-party and occasionally malformed, so a
    published "abuse address" is not guaranteed to be one. A value with no
    "@", or with either half empty, cannot be delivered to anyone.

    The check exists because the alternative is silent and worse than a
    missing contact. A malformed value used to become a destination with
    status "pending" and an unsendable target, so the indicator appeared
    on its way to a desk, no desk would ever receive it, and it was
    excluded from unreportable() precisely because its contact HAD an
    abuse entry. That is the failure the unreportable array exists to
    prevent, arriving through the one door that array does not watch.

    Deliberately shallow: this is not address validation and must not
    become it. Whether a syntactically fine address reaches a live desk is
    the mail transport's answer, not a parser's, and rejecting an address
    a desk actually reads would drop a report. It rejects only what cannot
    be a mailbox under any reading.
    """
    local, at, domain = address.rpartition("@")
    return bool(at and local.strip() and domain.strip())


def _group_key(address: str) -> str:
    """The key two spellings of one desk must share, and no more than that.

    The DOMAIN is case-insensitive by every standard that touches it, so
    "abuse@Host.Invalid" and "abuse@host.invalid" are one desk and must
    not be mailed twice about one incident.

    The LOCAL PART folds only for the RFC 2142 ROLE MAILBOXES. Those are
    standardised names that the same RFC requires be recognised regardless
    of case, so no host runs "Abuse@" and "abuse@" as two different desks,
    and treating them as two is a duplicate mail with nothing on the other
    side of the trade. This is where the duplicate actually happens, since
    an abuse entity publishes a role mailbox nearly every time.

    Any other local part is left exactly as published. RFC 5321 leaves its
    interpretation to the receiving host, and only that host knows whether
    it folds case. For a NAMED mailbox the asymmetry that governs the role
    names reverses: folding two desks a host genuinely distinguishes would
    silently drop one of them, and a dropped desk is worse than a duplicate
    mail. So each half folds on the strength of its own standard, and
    neither borrows the other's.

    The first version of this folded the domain alone and shipped with a
    test that used a lowercase local part throughout, so the test passed
    while "Abuse@Host.Invalid" and "abuse@host.invalid" produced two
    destinations. A test that varies one half of its input proves nothing
    about the other.
    """
    local, at, domain = address.rpartition("@")
    if not at:
        # Not an address shape we can split. _is_mailable keeps these out
        # of the destinations, but the key stays defined for anything that
        # asks for an id directly; group it by its literal text rather
        # than inventing a domain for it.
        return address
    if local.lower() in _ROLE_MAILBOXES:
        local = local.lower()
    return f"{local}@{domain.lower()}"


def email_destination_id(address: str) -> str:
    """The id of the destination that reports to this desk.

    Derived from the ADDRESS, so an id names a desk rather than a position
    in whatever list this run happened to build.

    That distinction is the whole point. `report` may run again on a case,
    and `contacts` may have resolved a new indicator since; a positional id
    then renumbers every desk after the newcomer. Bodies are written to
    bodies/<id>.xarf and each body's SHA-256 is recorded against its id, so
    after a renumber the file on disk belongs to a DIFFERENT desk than the
    manifest entry sharing its id. The edit check would compare one desk's
    body against another's, reporting an edit nobody made, or, if the two
    happened to match, missing one that was.

    It hashes the same normalised form the grouping uses, so two spellings
    of one desk get one id. Deriving it from the raw text instead would let
    the spelling RDAP published first decide a body's filename.
    """
    digest = hashlib.sha256(_group_key(address).encode("utf-8")).hexdigest()
    # ponytail: 8 hex chars, a case has a handful of desks; widen if a
    # collision is ever observed. A short id keeps a case directory
    # readable to the human reviewing it, which is what it is for.
    return f"email-{digest[:8]}"


def email_destinations(contacts: list[dict]) -> list[dict]:
    """Group contacts into one destination per abuse ADDRESS.

    Contacts already fold by host, but two different contacts can still
    resolve to the same address, an IP and a domain at one hoster being the
    common case. One mail per address rather than per contact is what stops
    a desk receiving two mails about one incident.

    Each id is derived from its own target address, so it survives both a
    reordering and a change in the contacts: a desk keeps its id when a new
    indicator resolves to a new desk ahead of it. See email_destination_id
    for why that matters more than it first appears.

    The returned ORDER is still first-seen, because the destinations are a
    list a human reads during review and the order the indicators were
    found in is the most explicable one available. Nothing downstream may
    key off that order; the id is what identifies a destination.
    """
    by_address: dict[str, dict] = {}

    for contact in contacts:
        for address in contact.get("abuse", []):
            if not _is_mailable(address):
                # unreportable() applies the same test, so the indicator
                # is listed there rather than vanishing between the two.
                continue
            key = _group_key(address)
            # First spelling seen wins the target. Any spelling reaches the
            # desk, and picking one keeps the report stable across a re-run.
            destination = by_address.setdefault(key, {"target": address,
                                                      "iocs": []})
            for ioc in contact.get("iocs", []):
                if ioc not in destination["iocs"]:
                    destination["iocs"].append(ioc)

    return [
        {
            "id": email_destination_id(destination["target"]),
            "kind": "email",
            "target": destination["target"],
            "iocs": destination["iocs"],
            "body": None,
            "status": "pending",
        }
        for destination in by_address.values()
    ]


def vendor_destinations(iocs: list[dict], configured: set) -> list[dict]:
    """Build one row per configured destination that has something to send.

    BOTH conditions have to hold. A row for an unconfigured destination is a
    promise that can only fail; a row for a configured one with nothing it
    accepts is a promise with no content, an AbuseIPDB submission with
    nothing to put in its `ip` parameter.

    Each row carries ONLY the IOCs its own destination accepts. A row is
    what submit iterates, so a urlhaus row listing an IP is a submission
    that gets built wrong or dropped at send time, whichever the
    implementer notices first.

    Order comes from DESTINATIONS rather than from `configured`, which is a
    set and therefore has no order worth writing into a manifest twice.
    """
    rows = []
    for name, spec in DESTINATIONS.items():
        if name not in configured:
            continue
        accepted = [ioc["id"] for ioc in iocs if ioc.get("type") in spec["accepts"]]
        if not accepted:
            continue
        rows.append({
            "id": name,
            "kind": spec["kind"],
            "iocs": accepted,
            # Null until submit settles each payload shape. Writing a
            # vendor's JSON now would mean guessing an endpoint's contract.
            # No body_sha256 either: that hash records what was disclosed,
            # and nothing has been.
            "body": None,
            "status": "pending",
        })
    return rows


def unreportable(contacts: list[dict]) -> list[dict]:
    """List every IOC that reached no email destination, with the reason.

    A missing contact is a normal outcome, not an error: RDAP publishes no
    abuse role for many netblocks. Making it visible is what keeps review
    honest, since finding it any other way means diffing the IOC list
    against every destination's IOC list. Same instinct as
    suspect_path_segments flagging rather than redacting.

    The membership test is "reached no destination", not "sits in a
    contact that resolved nothing", and those differ. Contacts fold by
    HOST, so one indicator can appear in two contacts, a domain that
    resolved and an IP that did not. Listing it because one of its
    contacts failed would put it in the destinations AND in the list of
    things no desk was found for, in one manifest. A reviewer reads the
    second and hand-reports an indicator already on its way to a desk,
    which costs the exact diffing this array exists to spare them. So an
    indicator is unreportable only when NONE of its contacts produced a
    mailable address, and the mailability test is the one
    email_destinations applies, so no indicator can fall between them.

    Each indicator appears ONCE. Two failed contacts for one host are two
    rows about one indicator otherwise, possibly with different reasons.
    First reason seen wins, matching the first-seen ordering of the
    destinations: both lists then read in the order the indicators were
    found, which is the only ordering a human can explain.
    """
    reachable = set()
    for contact in contacts:
        if any(_is_mailable(address)
               for address in contact.get("abuse", [])):
            reachable.update(contact.get("iocs", []))

    result: list[dict] = []
    listed = set()
    for contact in contacts:
        if any(_is_mailable(address)
               for address in contact.get("abuse", [])):
            continue
        # A contact's own error says more than the fallback, which is why
        # it wins even when the contact published an unusable address.
        # An empty string is not a reason: a manifest is a file the user
        # edits, and a blank reason renders as a blank cell that tells
        # them nothing.
        reason = contact.get("error") or (
            "no usable abuse address published" if contact.get("abuse")
            else "no abuse address resolved")
        for ioc in contact.get("iocs", []):
            if ioc in reachable or ioc in listed:
                continue
            listed.add(ioc)
            result.append({"ioc": ioc, "reason": reason})
    return result


def _describe(entry: dict) -> str:
    """The one-line "why you are seeing this" under an indicator.

    A boundary hop wins over its origin because it is the strongest claim
    the tool makes: it is the hop sending_ip() resolved to, the last one we
    can stand behind, and a desk needs to know it is being told "your
    address sent this" rather than "your address appeared somewhere in a
    chain the attacker partly wrote".

    Every other hop in the chain is attacker-writable, so it gets the
    ordinary origin line and no claim of authorship. That distinction is
    the third property expressed to a reader.
    """
    if entry.get("confidence") == "boundary-hop":
        return "sending IP, first hop outside our trust boundary"
    origin = entry.get("origin", "")
    if not origin:
        return ""
    return "seen " + _ORIGINS.get(origin, origin)


def text_part(manifest: dict, destination: dict, identity: dict) -> str:
    """Build the human-readable part: the one that decides whether a desk
    acts on the report.

    The ask goes first, because a desk triaging a queue must know in one
    line what happened and what is wanted. Only THIS destination's own
    indicators appear: a desk shown three IPs that are not theirs stops
    reading, and, worse, has been told about a third party's infrastructure
    for no reason. The lookup is by id against the manifest, so a
    destination naming an id the manifest does not carry contributes no row
    rather than raising; a manifest is a file the user edits and the two can
    disagree.

    NOTHING here is truncated. Every value that does not fit is wrapped with
    an explicit continuation marker instead, because a cut value is a WRONG
    value rather than a short one: a desk acting on the first 72 characters
    of a URL acts on a resource nobody reported, and a cut header misstates
    what the message declared. See _wrap_value for why the marker is what
    makes that safe.

    Introduces nothing that did not come from the manifest or the identity.
    Everything it formats has already been through redact.py, so this is not
    a filter and must not become one, but it also must not add: the
    destination's target address is deliberately absent from the body, since
    a desk knows its own address and printing it only adds a string to a
    document whose whole discipline is that fewer strings leak less.
    """
    by_id = {entry["id"]: entry for entry in manifest.get("iocs", [])}
    mine = [by_id[i] for i in destination.get("iocs", []) if i in by_id]

    lines = [
        "Phishing message reported: infrastructure on your network was",
        "used to send or host it. Requesting takedown and customer",
        "notification.",
    ]

    if mine:
        lines += ["", "Observed on your infrastructure:", ""]
        for entry in mine:
            lines += _wrap_value(str(entry.get("value", "")), "  ")
            description = _describe(entry)
            if description:
                lines += _wrap_value(description, "      ")

    # Pairs, not a dict: JSON has no tuple, so case.load() hands these back
    # as lists. Both shapes destructure identically, and there is a test
    # driving the round-tripped one because that is what actually arrives.
    shown = [(name, value) for name, value in (manifest.get("headers") or [])
             if name in ("Date", "From", "Subject")]
    if shown:
        lines += ["", "Message as declared:", ""]
        for name, value in shown:
            lines += _wrap_value(f"{name}: {value}", "  ")

    auth = manifest.get("auth") or {}
    if auth:
        lines += ["", "Authentication results:", ""]
        lines += _wrap_value(
            "   ".join(f"{key.upper()}: {value}"
                       for key, value in sorted(auth.items())), "  ")

    lines += ["", *_REDACTION_NOTE, ""]
    # Each [reporter] key is individually skippable and config DROPS a
    # skipped one rather than storing "", so a partial identity is the
    # normal shape here. Subscripting raised KeyError on a case that had
    # parsed perfectly, and joining unconditionally left a stray comma with
    # nothing on one side of it.
    who = ", ".join(str(identity[key]) for key in ("name", "org")
                    if identity.get(key))
    lines += _wrap_value(
        f"Reported by: {who + ' ' if who else ''}"
        f"<{identity.get('email', '')}>", "")
    lines.append("Generated by abusectl.")

    return "\n".join(lines) + "\n"


# Every character that any reasonable reader of this part might treat as the
# end of a field, plus "%" itself.
#
# CR and LF are the ones that matter: a field value carrying one forges a
# field, and the forged field is read as something THIS TOOL asserted, on a
# document that carries the reporter's identity. That is header injection
# into mail we send, which the contacts spec already names as a hazard.
#
# The rest are here because "what counts as a line break" is not one answer.
# Python's own email module raises on U+2028 and U+2029 as readily as on LF,
# because it reaches for str.splitlines(), which also breaks on VT, FF, the
# three information separators and NEL. A value carrying one of those does
# not merely render oddly in the assembled document: it aborts the document.
# So the set is taken from splitlines() rather than from RFC 5322, on the
# principle that the defence must cover what the CONSUMERS break on, not what
# one specification says a break is.
#
# "%" is in the set for a different reason, and leaving it out is a second
# injection one step later: see _field_value().
_UNSAFE = "".join(chr(c) for c in (
    0x0A, 0x0B, 0x0C, 0x0D, 0x1C, 0x1D, 0x1E, 0x85, 0x2028, 0x2029,
)) + "%"


def _field_value(value) -> str:
    """Make one attacker-supplied value safe to emit as a field value.

    ENCODES rather than rejects or strips, and the choice is the whole point
    of this function.

    The value is reachable, not theoretical. redact.url_valued_parameters()
    URL-DECODES a redirector's destination parameter in order to recover it
    as an indicator in its own right, exactly as the second property
    intends. So a message body carrying

        http://r.invalid/go?next=http%3A%2F%2Fa.invalid%2Fx%0AFeedback-Type...

    produces, through parse.iocs() on a real message, an IOC whose value
    contains a literal newline followed by text shaped like a field. There
    is no upstream filter between that and here.

    REJECTING the indicator loses a genuine redirect target, which is one of
    the more actionable things a desk receives, over an attacker's choice of
    byte. STRIPPING the character silently rewrites the indicator into a
    different URL, and a desk that acts on the stripped form has acted on a
    resource nobody reported: that is the truncation defect from the text
    part wearing a different coat, and it is worse here because nothing in
    the output says it happened.

    Percent-encoding is the URL's own native encoding, so it reads as
    intended by the audience that receives it; it is EXACTLY REVERSIBLE, so
    a desk or a later submit path recovers what the message declared; and it
    is visible, so an altered value announces the alteration rather than
    passing as a clean one. Same instinct as the continuation marker in the
    text part: a fragment announces that it is a fragment.

    "%" must be encoded too, and this is not tidiness. A redacted URL
    legitimately contains percent signs, so an encoder that leaves them
    alone emits text whose reversal produces the very control character the
    encoding existed to remove: an attacker writes the literal five
    characters "%0A" into a path and unquote() hands the next reader a
    newline. An encoding that is not a true inverse is not a defence.

    The escape runs over UTF-8 BYTES, not over code points, and that is not
    a detail. Percent-encoding is defined on octets, so a character above
    U+007F has more than one byte to spell: encoding U+2028 as "%2028" from
    its ordinal produces text that unquote() reads back as "%20" followed by
    the literal "28", which is a SPACE and the digits, not the character
    that was there. The reversibility this function claims would then be
    false for exactly the two characters that are here because Python's
    email module breaks on them. Encoding each of its three UTF-8 bytes
    gives "%E2%80%A8", and unquote() returns U+2028.

    Non-strings are coerced rather than raising. A manifest is a file the
    user edits by hand and JSON has numbers; this module is the last step
    before a reviewed case becomes a sent mail, and a report that raises
    produces nothing at all.
    """
    text = str(value)
    if not any(ch in text for ch in _UNSAFE):
        return text
    out = []
    for ch in text:
        if ch in _UNSAFE:
            out.append("".join(f"%{b:02X}" for b in ch.encode("utf-8")))
        else:
            out.append(ch)
    return "".join(out)


def feedback_fields(manifest: dict, destination: dict) -> list[tuple[str, str]]:
    """Build the machine-readable part: an RFC 5965 envelope carrying x-arf
    fields inside it.

    RFC 5965 is an IETF standard and universally understood, but it was
    designed for feedback loops, where a report is ABOUT A MESSAGE. These
    reports are about INDICATORS, and 5965 has no field for "this specific
    host is the thing being reported". x-arf's Source does. The envelope is
    the standard's own extension point: 5965 requires an implementation to
    ignore fields it does not support, so a standards parser reads what it
    knows and x-arf tooling finds what it wants.

    Returned as PAIRS, not a dict, because Reported-URI and Reported-Domain
    repeat. Everything else does not, and that is enforced rather than
    assumed: RFC 5965 gives Source-IP and Arrival-Date "once maximum". A
    strict parser meeting a repeated single-occurrence field either rejects
    the part or keeps the last occurrence, so a second Source-IP does not
    add an address, it DISPLACES the primary one. Every address still
    travels: the text part lists all of this destination's indicators, and
    that is the part a human acts on.

    ARRIVAL-DATE IS DELIBERATELY ABSENT. 5965 defines it as when the
    generating ADMD's own MTA received the message; the manifest's Date
    header is when the SENDER CLAIMED to have sent it, which on a phishing
    message is attacker-controlled free text. Copying one into the other
    asserts an attacker's timestamp as our own observation, and a desk
    correlating it against their logs finds nothing and discounts the
    report. The honest source is the boundary Received hop's timestamp,
    which parse.report_headers() already publishes; extracting it needs a
    date parser, and 5965 makes the field optional, so it is omitted until
    something actually needs it. An absent optional field misstates nothing.

    SOURCE IS OMITTED WHEN THERE IS NOTHING TO PUT IN IT, rather than
    emitted empty. "Source:" with nothing after it tells a parser that the
    thing being reported is the empty string; no field tells it nothing,
    which is the truth. This happens when a desk's only indicators are a
    sha256 or an observation, and both of those belong to the human part.

    sha256 and observation are NOT forced into the nearest-looking field.
    Neither is a Source, a URI or a domain, and telling a desk that a file
    hash is a URI is a false statement in the part meant to be machine-read.

    Every field NAME here is a literal in this module and none is derived
    from data. The tempting generalisation, a table from an IOC's own type
    string to a field name, would make a hand-edited manifest able to name
    fields; the set is closed on purpose. Every field VALUE goes through
    _field_value(), at the single point where the pairs are built, for the
    reason the fourth property was learned three times over: validation
    applied per branch gets forgotten on the next branch.
    """
    by_id = {entry["id"]: entry for entry in manifest.get("iocs", [])}
    mine = [by_id[i] for i in destination.get("iocs", []) if i in by_id]

    ips = [e.get("value") for e in mine if e.get("type") in ("ipv4", "ipv6")]
    urls = [e.get("value") for e in mine if e.get("type") == "url"]
    domains = [e.get("value") for e in mine if e.get("type") == "domain"]

    # Source is singular, so the primary indicator fills it and the rest
    # travel in the repeatable fields and in the text part. An IP is the
    # most actionable thing a hosting desk can act on, so it wins when
    # present; a domain beats a URL because a desk suspending a name
    # covers every URL under it.
    primary = (ips or domains or urls or [None])[0]

    # "fraud" over "abuse", and both are checked against the IANA MARF
    # registry rather than a worked example. RFC 5965 defines fraud as
    # "indicates some kind of fraud or phishing activity" and abuse as
    # "unsolicited email or some other kind of email abuse". This tool
    # reports phishing, so fraud is the registered value that says what
    # the report is, and some desks route it separately from bulk spam.
    #
    # There is no Report-Type. It is NOT in the registry (25 registered
    # field names, it is absent), it was carried from the spec's worked
    # example rather than chosen, and Feedback-Type: fraud already states
    # what it was saying. Nothing would have broken by keeping it, since
    # 5965 section 6 makes ignoring unknown fields a MUST, but that same
    # section requires extension fields be registered, and every field
    # emitted here should be one a desk can look up.
    fields = [
        ("Feedback-Type", "fraud"),
        ("User-Agent", f"abusectl/{VERSION}"),
        ("Version", "1"),
    ]

    if primary is not None:
        fields.append(("Source", primary))
    if ips:
        # Once maximum, per RFC 5965. See the docstring.
        fields.append(("Source-IP", ips[0]))
    for domain in domains:
        fields.append(("Reported-Domain", domain))
    for url in urls:
        # "Reported-URI", the spelling in RFC 5965's ABNF and its IANA
        # registration. The RFC's own worked example writes "Reported-Uri"
        # and that inconsistency is where this module's old spelling came
        # from. Field names are case-insensitive per RFC 5322, so nothing
        # was broken; matching the normative spelling costs nothing.
        fields.append(("Reported-URI", url))

    return [(name, _field_value(value)) for name, value in fields]


# The header names the third part may carry, and the reason it is a list here
# rather than a trust in the manifest.
#
# parse.report_headers() already produces a whitelist, so in the normal path
# every name arriving here is one of these and this filter changes nothing.
# The manifest is a FILE THE USER EDITS, though, and the first property's
# structural argument is that report cannot disclose what it was never given.
# A hand-edited manifest is precisely a way it CAN be given something: a user
# pasting a header block back in, or a case written by a future parse.py whose
# whitelist grew, hands this module a "To:" pair and the argument no longer
# holds by construction.
#
# So the same whitelist is applied a second time at the point of publication.
# That is a deliberate exception to "two places to remember is how the fourth
# property leaked", and the trade runs the other way here: this is not a
# second DECISION about what may be disclosed, it is the same decision
# enforced where the disclosure actually happens. parse.py still decides; this
# refuses to publish anything it did not decide for. A duplicated whitelist
# that drifts loses a header from a report, which review sees. A missing one
# publishes a recipient address to the attacker, which nobody sees.
#
# Transcribed from the spec's "Where the headers come from" whitelist, which
# is the same list parse._REPORT_HEADERS holds. Matched case-insensitively,
# because a header name is case-insensitive by RFC 5322 and a hand-edited
# manifest will not have preserved anyone's capitalisation.
_PUBLISHABLE_HEADERS = frozenset({
    "received", "from", "subject", "date", "message-id", "reply-to",
    "return-path", "authentication-results", "received-spf",
    "mime-version", "content-type",
})


def _header_line(name: str, value: str) -> str:
    """One line of the third part, safe to emit even when the value is not.

    A header VALUE here is attacker-supplied free text. The spec keeps
    Subject and the From display name deliberately, because they are what
    lets a desk recognise a campaign, and it says plainly that the whitelist
    governs WHICH headers travel and never what is inside one. A Subject
    carrying a newline therefore reaches this function, and emitting it
    verbatim forges a header line inside a part whose entire content is read
    as headers: "Subject: evil\\nFrom: forged@attacker.invalid" becomes two
    headers, the second of which a desk reads as something the message
    declared. That is the Task 5 injection one part further along, and it is
    worse here, because the forged line is grammatical where a forged x-arf
    field is merely present.

    The answer differs from _field_value()'s percent-encoding, and the
    difference is the audience. This part's content IS rfc822 headers, so the
    encoding a reader of it already knows is RFC 2047, not URL escaping. An
    encoded word neutralises the break by turning it into an RFC 5322 FOLD:
    the value continues on a continuation line, a parser unfolds it back to
    one header whose value is the original text including the character that
    was there, and nothing new appears in the header list. Percent-encoding
    would also be safe and would read as a bug to a mail parser, which is the
    one audience this part has.

    Encoded ONLY when a value actually carries something unsafe. RFC 2047
    encodes indiscriminately, so applying it to every header would render an
    ordinary Subject as "=?utf-8?q?Your_account?=" and cost the desk the
    legibility this part exists for. The condition is the same _UNSAFE set
    the machine part uses, minus "%": "%" is in that set because
    percent-encoding must be a true inverse, and nothing here percent-encodes,
    so a literal "%" in a Subject is just a character.

    Non-strings are coerced for the reason _field_value() coerces them: a
    manifest is hand-edited, JSON has numbers, and a report that raises
    produces nothing at all.
    """
    text = str(value)
    if any(ch in text for ch in _UNSAFE if ch != "%"):
        # maxlinelen leaves room for "Name: " on the first line; the exact
        # number only affects where a fold lands, never what unfolds back.
        return f"{name}: {Header(text, 'utf-8', maxlinelen=64).encode()}"
    return f"{name}: {text}"


def _headers_part(manifest: dict) -> str | None:
    """The third part's body, or None when there is nothing to publish.

    Returns None rather than an empty string so build() can OMIT the part.
    An empty text/rfc822-headers is a positive claim that the message
    declared no headers, which is never true of a real message; absent says
    the report carries none, which is the truth for a case parsed before the
    headers block existed or one a user emptied by hand. Same rule as
    feedback_fields() omitting Source rather than emitting it empty, and the
    same rule as the config's absent-not-empty-string.

    Pairs, not a dict, matching what case.load() hands back: JSON has no
    tuple, so these arrive as lists and both shapes destructure identically.
    """
    lines = [
        _header_line(name, value)
        for name, value in (manifest.get("headers") or [])
        if str(name).lower() in _PUBLISHABLE_HEADERS
    ]
    if not lines:
        return None
    return "\n".join(lines) + "\n"


def build(manifest: dict, destination: dict, identity: dict) -> str:
    """Assemble one destination's report as an RFC 5965 MIME document.

    Three parts: what a human reads, what a parser reads, and the headers.

    THE ORIGINAL MESSAGE IS NOT ATTACHED, and there is no message/rfc822
    part. source.eml carries every identifier the first property exists to
    keep out: To, Cc, Delivered-To, unredacted URLs whose query and path
    segments encode the recipient, the user's own Message-IDs and maildir
    paths. An abuse desk forwards a report to the abused customer, who for a
    phishing domain may be the attacker, and URLhaus is a public feed. RFC
    5965 provides text/rfc822-headers for exactly the case where the full
    message cannot be included, so this is the standard's own answer rather
    than a deviation from it.

    The FROM is built with email.headerregistry.Address rather than by
    formatting a string. A reporting identity legitimately contains a comma,
    "Example Consulting, Ltd" being the obvious one, and a comma is the
    address-list separator: f"{name} <{email}>" then parses back as TWO
    addresses, the first of which is a bogus addr-spec with no domain. A desk
    replying to the report replies to that, and the reply reaches nobody.
    Address quotes the display name when it needs quoting and RFC 2047-encodes
    it when it is not ASCII, which are the two cases a consultant's org name
    actually hits.

    Each subpart's type is set AFTER its content, which is the opposite of
    what looks right and was checked rather than assumed: set_content()
    REPLACES the Content-Type it derived from the payload, so setting the type
    first leaves all three parts as text/plain. Setting it afterwards keeps
    the transfer encoding and charset set_content() chose, which is what makes
    a non-ASCII header value in the third part survive as base64 rather than
    as a malformed 7bit line.

    The multipart boundary is chosen by the generator at SERIALISATION time,
    after it has seen every payload, so a body containing something shaped
    like a delimiter cannot collide with the real one; that is verified
    against a payload carrying a literal "--===============0==" line.
    """
    message = EmailMessage(policy=SMTP)
    message["From"] = Address(str(identity.get("name", "")),
                              addr_spec=str(identity["email"]))
    message["To"] = destination["target"]
    # The case id is generated by case.py from a date and random hex, so it is
    # not attacker-supplied and needs no escaping; it is the string the user
    # greps for when a desk replies.
    message["Subject"] = (
        f"Abuse report: phishing infrastructure, case {manifest['case_id']}"
    )
    message.make_mixed()
    message.set_type("multipart/report")
    message.set_param("report-type", "feedback-report")

    human = EmailMessage(policy=SMTP)
    human.set_content(text_part(manifest, destination, identity))
    message.attach(human)

    machine = EmailMessage(policy=SMTP)
    machine.set_content(
        "\n".join(f"{name}: {value}"
                  for name, value in feedback_fields(manifest, destination))
        + "\n")
    machine.set_type("message/feedback-report")
    message.attach(machine)

    body = _headers_part(manifest)
    if body is not None:
        headers = EmailMessage(policy=SMTP)
        headers.set_content(body)
        headers.set_type("text/rfc822-headers")
        message.attach(headers)

    return message.as_string()


class Frozen(Exception):
    """Raised when a case has already been reported to at least one desk.

    There is no force override. The destinations are not independent
    artifacts, they are one incident reported in parallel: regenerating one
    body after another desk holds the report leaves two desks with
    contradictory accounts of the same case. Worse, the shared parts of
    every body, the identity, the header block and the IOC list, come from
    the manifest, so regenerating one body after the manifest has moved
    produces a case whose bodies were built from two different states.
    """


class Modified(Exception):
    """Raised when a body was edited after generation and --force was not
    given. Silently discarding a review that took twenty minutes is what
    makes a tool untrustworthy.
    """


def body_hash(text: str) -> str:
    """Hash a body's CONTENT, which is what the question actually is.

    An mtime is a poor witness in both directions: a git checkout, an rsync
    or a backup restore all move it with no human having edited anything,
    and an editor that preserves mtime hides a real edit. For a destination
    that has been sent, this hash is also the record of what was disclosed.

    The encoding is PINNED to utf-8 rather than left to the locale, because
    the value is compared against a body read back off disk, possibly by a
    later run under a different LANG. A body that hashed differently on a
    machine set to a non-UTF-8 locale would read as hand-edited and refuse
    to regenerate, and, for a sent destination, would be a false record of
    what was disclosed. Whoever writes the file owes it the same encoding.
    """
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def check_regenerable(manifest: dict, modified: list[str],
                      force: bool = False) -> None:
    """Raise unless this case may be regenerated. Returns None when it may.

    Frozen wins over force, and it is checked FIRST so that an edited body
    on a frozen case raises Frozen rather than Modified: Modified is the one
    --force can clear, and reporting the clearable half of a refusal invites
    the user to force their way past the half that has no override.

    PRESENCE is the test, not truthiness. The marker is write-once and
    monotonic, so `{}` or `""` is a half-written or corrupted record of a
    case a desk may already hold, and the safe direction for an evidence
    record is to refuse. Only an explicit null counts as absent, because
    JSON has no way to omit a key it wrote as null. A marker that is not a
    mapping refuses too, rather than raising AttributeError out of a .get:
    cli turns Frozen into a message and a traceback into a bug report.
    """
    if "frozen" in manifest and manifest["frozen"] is not None:
        frozen = manifest["frozen"]
        details = frozen if isinstance(frozen, dict) else {}
        raise Frozen(
            f"case reported to {details.get('by', 'a destination')} at "
            f"{details.get('at', 'an unknown time')}; bodies cannot be "
            f"regenerated. Edit a body by hand if it must change."
        )

    if modified and not force:
        raise Modified(
            "bodies edited since generation: " + ", ".join(modified) +
            ". Re-run with --force to discard those edits; a timestamped "
            "backup is kept."
        )


def _read_body(path: Path) -> str:
    """Read a body back EXACTLY as it was written.

    Deliberately not read_text(). A body is an RFC 5322 document delimited
    by CRLF, while the feedback part inside it carries bare LFs that
    set_content chose; text mode applies universal newlines and collapses
    every CRLF to LF on the way in. The text that comes back then hashes
    differently from the text that was written even though not one byte on
    disk changed, so an untouched case reads as hand-edited and refuses to
    regenerate, and for a sent destination the recorded hash is a false
    record of what was disclosed.

    The encoding is pinned to utf-8 for the same reason body_hash pins it:
    the locale of the run that reads a body is not the locale of the run
    that wrote it.
    """
    return path.read_bytes().decode("utf-8")


def _write_body(path: Path, text: str) -> None:
    """Write a body byte-exactly, in the encoding body_hash assumes.

    Binary rather than write_text for the mirror of the reason above: text
    mode translates a bare LF to os.linesep, which on a CRLF platform
    would rewrite the bare LFs inside the machine-readable part and
    corrupt the field block a desk's parser reads. The bytes written here
    are the bytes hashed, which is what makes the hash a record of what
    was disclosed rather than of what was intended.
    """
    path.write_bytes(text.encode("utf-8"))


def generate(manifest: dict, case_path: Path, identity: dict,
             force: bool = False, configured: set | None = None) -> dict:
    """Write every body and return the manifest with destinations[] set.

    The manifest is RETURNED rather than saved: case.py is the only writer
    of a case directory's manifest, so the caller saves. The bodies are
    this module's to write because they are not the manifest.

    The case directory must already exist. `bodies` is created if it is
    missing, but its parents are NOT, because a missing case directory
    means a wrong path rather than a first run and creating the tree would
    scatter bodies into an empty directory the user never made.

    ORDER MATTERS AND IS THE POINT. The whole manifest is scanned for
    edited bodies first, then check_regenerable decides, and only then does
    anything on disk change. A refusal therefore leaves the bodies
    directory byte-for-byte as it was: a frozen case is an evidence record
    a desk already holds, and a half-rewritten one is worse than a stale
    one. Scanning every destination before refusing is also what makes the
    error name every edited body rather than the first.

    `configured` is the set of destinations the config carries keys for, an
    ARGUMENT rather than a config read, so this module stays pure and
    testable with no files on disk. Defaulting to nothing configured is the
    safe direction: a caller that forgets it writes no vendor row rather
    than promising a submission nobody set up.

    destinations[] is rebuilt WHOLESALE every run, vendor rows included.
    That is safe only while the freeze rule holds: once a destination has
    landed the case is frozen and this function refuses to run at all. The
    freeze marker is written by submit, so this is a constraint on that
    spec, not merely a description of this one.

    Vendor rows are appended AFTER the body-writing loop, and that ordering
    is load-bearing rather than incidental: the loop mutates the dicts it
    walks, so a vendor row inside it would be handed a body it has no
    payload for. A vendor row's body stays null until submit settles each
    endpoint's payload shape.
    """
    case_path = Path(case_path)
    bodies = case_path / "bodies"
    bodies.mkdir(exist_ok=True)

    modified = []
    for existing in manifest.get("destinations") or []:
        recorded = existing.get("body_sha256")
        if not recorded or not existing.get("body"):
            continue
        path = case_path / existing["body"]
        # A deleted body is not a modified one. It regenerates silently,
        # because deleting a body is how a user asks for a fresh one.
        if path.exists() and body_hash(_read_body(path)) != recorded:
            modified.append(existing["id"])

    check_regenerable(manifest, modified, force=force)

    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    for destination_id in modified:
        existing = next(d for d in manifest["destinations"]
                        if d["id"] == destination_id)
        path = case_path / existing["body"]
        # The stamp goes AFTER the full name rather than replacing the
        # extension, so the destination id stays legible in a directory
        # listing and two forced runs in different seconds keep both
        # edits. with_suffix would be wrong here: it replaces the last
        # suffix, so it depends on the name having exactly the one this
        # writer gave it.
        path.rename(path.with_name(f"{path.name}.{stamp}.orig"))

    destinations = email_destinations(manifest.get("contacts", []))
    for destination in destinations:
        text = build(manifest, destination, identity)
        relative = f"bodies/{destination['id']}.xarf"
        _write_body(case_path / relative, text)
        destination["body"] = relative
        destination["body_sha256"] = body_hash(text)

    destinations.extend(
        vendor_destinations(manifest.get("iocs", []), configured or set())
    )
    manifest["destinations"] = destinations
    manifest["unreportable"] = unreportable(manifest.get("contacts", []))
    return manifest