"""Audit what changed in a PKGBUILD.

Two principles, both of which exist to stop this becoming noise the user learns
to ignore:

1. **Audit the delta, not the absolute state.** A `curl | sh` that has been in a
   PKGBUILD for five years is not news. One that appeared in this update is.
   Rules therefore compare old against new wherever they can, and pre-existing
   patterns are reported at most as information.

2. **Every finding carries its evidence.** No opaque risk score. A finding names
   the rule, the severity, and the exact spec, URL, or line that triggered it, so
   the user can judge it themselves.

Nothing here ever executes a PKGBUILD. See srcinfo.py for why.
"""

from __future__ import annotations

import enum
import re
from dataclasses import dataclass, field

from .srcinfo import Pkgbuild, SrcInfo

VCS_SUFFIXES = ("-git", "-cvs", "-svn", "-hg", "-bzr", "-darcs", "-nightly", "-fossil")


class Severity(enum.IntEnum):
    """Ordered, so the report can show the worst finding."""

    INFO = 0  # context, no action implied
    LOOK = 1  # worth a look before building
    BLOCK = 2  # do not auto-apply; needs an explicit decision

    @property
    def glyph(self) -> str:
        return {Severity.INFO: "·", Severity.LOOK: "!", Severity.BLOCK: "!!"}[self]


@dataclass(frozen=True)
class Finding:
    rule: str
    severity: Severity
    title: str
    evidence: str = ""
    where: str = ""

    def render(self) -> str:
        bits = [f"[{self.severity.glyph}] {self.title}"]
        if self.evidence:
            bits.append(f"    {self.evidence}")
        if self.where:
            bits.append(f"    in {self.where}")
        return "\n".join(bits)


@dataclass
class SetDiff:
    """Added/removed members of a list-valued field."""

    added: list[str] = field(default_factory=list)
    removed: list[str] = field(default_factory=list)

    def __bool__(self) -> bool:
        return bool(self.added or self.removed)

    def summary(self) -> str:
        if not self:
            return "unchanged"
        parts = []
        if self.added:
            parts.append(f"+{len(self.added)}")
        if self.removed:
            parts.append(f"-{len(self.removed)}")
        return " ".join(parts)


@dataclass
class SourceMove:
    """A source file that now comes from somewhere else."""

    filename: str
    old_url: str
    new_url: str

    @property
    def old_domain(self) -> str:
        return domain_of(self.old_url)

    @property
    def new_domain(self) -> str:
        return domain_of(self.new_url)


@dataclass
class SemanticDiff:
    """What changed, in terms of PKGBUILD fields rather than lines of text."""

    pkgver: tuple[str, str] | None = None
    pkgrel: tuple[str, str] | None = None
    epoch: tuple[str, str] | None = None
    source: SetDiff = field(default_factory=SetDiff)
    sums: SetDiff = field(default_factory=SetDiff)
    depends: SetDiff = field(default_factory=SetDiff)
    makedepends: SetDiff = field(default_factory=SetDiff)
    optdepends: SetDiff = field(default_factory=SetDiff)
    provides: SetDiff = field(default_factory=SetDiff)
    conflicts: SetDiff = field(default_factory=SetDiff)
    replaces: SetDiff = field(default_factory=SetDiff)
    validpgpkeys: SetDiff = field(default_factory=SetDiff)
    sources_moved: list[SourceMove] = field(default_factory=list)
    funcs_added: list[str] = field(default_factory=list)
    funcs_removed: list[str] = field(default_factory=list)
    funcs_changed: list[str] = field(default_factory=list)
    lines_added: int = 0
    lines_removed: int = 0

    @property
    def is_empty(self) -> bool:
        return not any(
            [
                self.pkgver,
                self.pkgrel,
                self.epoch,
                self.source,
                self.sums,
                self.depends,
                self.makedepends,
                self.optdepends,
                self.provides,
                self.conflicts,
                self.replaces,
                self.validpgpkeys,
                self.sources_moved,
                self.funcs_added,
                self.funcs_removed,
                self.funcs_changed,
                self.lines_added,
                self.lines_removed,
            ]
        )


@dataclass
class Change:
    """One package's old and new metadata. Anything may be None on first import."""

    package: str
    old_srcinfo: SrcInfo | None = None
    new_srcinfo: SrcInfo | None = None
    old_pkgbuild: Pkgbuild | None = None
    new_pkgbuild: Pkgbuild | None = None
    old_maintainer: str | None = None
    new_maintainer: str | None = None
    old_url: str = ""
    new_url: str = ""
    #: `.install` scriptlet text, keyed by filename, read when the Change was
    #: assembled. Carried as data because these run as root on install, and
    #: resolving the filename against the *process* working directory meant they
    #: were silently never examined.
    old_install: dict[str, str] = field(default_factory=dict)
    new_install: dict[str, str] = field(default_factory=dict)

    @property
    def is_vcs(self) -> bool:
        return any(self.package.endswith(s) for s in VCS_SUFFIXES)


class Basis(enum.IntEnum):
    """What an audit actually had to work with.

    "No findings" and "nothing to examine" look identical in a report and mean
    completely different things. A package paru has never cloned has no PKGBUILD
    to check, so every rule returns nothing -- which rendered as a clean bill of
    health for a package nobody had looked at.
    """

    #: compared an old revision against the new one
    COMPARED = 2
    #: only the new revision was available; pre-existing patterns cannot be told
    #: apart from newly introduced ones
    NEW_ONLY = 1
    #: no PKGBUILD at all: nothing was examined
    NOTHING = 0

    @property
    def label(self) -> str:
        return {
            Basis.COMPARED: "compared with the previous revision",
            Basis.NEW_ONLY: "no baseline: only the current revision was read",
            Basis.NOTHING: "not audited: no PKGBUILD available",
        }[self]


@dataclass
class Report:
    package: str
    diff: SemanticDiff
    findings: list[Finding] = field(default_factory=list)
    #: what could be examined, so a caller can never present an empty report as
    #: a clean result
    basis: Basis = field(default_factory=lambda: Basis.NOTHING)

    @property
    def audited(self) -> bool:
        return self.basis is not Basis.NOTHING

    @property
    def verdict(self) -> Severity:
        return max((f.severity for f in self.findings), default=Severity.INFO)

    @property
    def loud(self) -> list[Finding]:
        return [f for f in self.findings if f.severity >= Severity.LOOK]


# --------------------------------------------------------------------------
# helpers
# --------------------------------------------------------------------------

_URL_RE = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE)
#: a source may carry an optional `filename::` prefix before the URL
_SOURCE_PREFIX_RE = re.compile(r"^[^:/]+::")
#: PKGBUILD VCS schemes: `git+https://...`, `svn+https://...`
_VCS_SCHEMES = ("git+", "svn+", "hg+", "bzr+", "darcs+", "cvs+")


#: a detached signature: its integrity comes from the trusted key, not a hash
_SIGNATURE_RE = re.compile(r"\.(?:sig|asc|gpg|sign)$", re.IGNORECASE)


def is_signature_source(source: str) -> bool:
    """Whether a source entry is a detached signature.

    `SKIP` is the convention for these: the signature is checked against
    `validpgpkeys`, so demanding a checksum as well is redundant. Packages that
    verify with GPG were being reported as though they had left a network
    download unverified.
    """
    if not source:
        return False
    name = filename_of(source)
    return bool(name and _SIGNATURE_RE.search(name))


def strip_source_prefix(url: str) -> str:
    """Drop any `filename::` prefixes, of which there may be more than one."""
    while True:
        prefix = _SOURCE_PREFIX_RE.match(url)
        if not prefix:
            return url
        url = url[prefix.end() :]


def is_vcs_source(url: str) -> bool:
    """Whether a source entry pulls a live repository rather than a fixed file.

    This, not the package name, is what makes a SKIP checksum legitimate. A
    package called `foo-git` whose source is a pinned tarball still needs sums.
    """
    return strip_source_prefix(url).startswith(_VCS_SCHEMES)


def domain_of(url: str) -> str:
    """Host of a source entry, or a marker for local sources."""
    url = strip_source_prefix(url)
    if not _URL_RE.match(url):
        return "(local)"
    rest = url.split("://", 1)[1]
    return rest.split("/", 1)[0].split("@")[-1]


def filename_of(url: str) -> str:
    """The filename a source will land as, honouring an explicit `::` prefix."""
    prefix = _SOURCE_PREFIX_RE.match(url)
    if prefix:
        return prefix.group(0)[:-2]
    base = url.split("://", 1)[-1]
    base = base.split("?", 1)[0].split("#", 1)[0].rstrip("/")
    return base.rsplit("/", 1)[-1]


def _setdiff(old: list[str], new: list[str]) -> SetDiff:
    o, n = set(old), set(new)
    return SetDiff(added=sorted(n - o), removed=sorted(o - n))


def _pair(old: str | None, new: str | None) -> tuple[str, str] | None:
    if old is None or new is None or old == new:
        return None
    return (old, new)


def _lines_changed(old: str, new: str) -> tuple[int, int]:
    import difflib

    added = removed = 0
    for line in difflib.unified_diff(old.splitlines(), new.splitlines(), n=0):
        if line.startswith(("+++", "---")):
            continue
        if line.startswith("+"):
            added += 1
        elif line.startswith("-"):
            removed += 1
    return added, removed


# --------------------------------------------------------------------------
# suspicious patterns
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class Pattern:
    rule: str
    severity: Severity
    title: str
    regex: re.Pattern[str]
    #: restrict the scan to these PKGBUILD functions; None means the whole file
    funcs: tuple[str, ...] | None = None
    #: only report when the pattern is newly introduced
    new_only: bool = True
    #: an optional veto for a match the regex alone cannot judge, such as text
    #: that is merely inside a quoted string
    guard: object = None


PATTERNS: tuple[Pattern, ...] = (
    Pattern(
        "pipe-to-shell",
        Severity.BLOCK,
        "pipes a download straight into a shell",
        # A shell is reached by more than its bare name: `| /bin/bash`,
        # `| /usr/bin/env bash`, `| dash`, `| zsh`. Matching only the bare
        # name meant the most explicit form of the same attack was not a
        # BLOCK, only a generic "network in build" LOOK.
        re.compile(
            r"(?:curl|wget|fetch)\b[^|;\n]*+\|\s*+(?:sudo\s+)?"
            r"(?:[\w./-]*/)?(?:env\s+)?(?:ba|z|k|da)?sh\b"
        ),
    ),
    Pattern(
        "decode-to-shell",
        Severity.BLOCK,
        "decodes a blob and pipes it into a shell",
        # One `[^|;\n]*` either side of a literal is cubic: both classes can
        # match the literal, so every start position backtracks over every
        # split. An 8 KB line took 4.5s and a 20 KB line over a minute, and
        # the text is whatever the maintainer wrote. The gap before `-d` is
        # bounded and the rest are possessive, so each start position does a
        # fixed amount of work.
        # Both gaps are bounded. A possessive `*` here still scans to the
        # end of the file on every attempt, which is what made it quadratic;
        # 120 characters is far more than a real `base64 -d <file> | sh`.
        re.compile(
            r"(?:base64|xxd|openssl\s+enc)\b[^|;\n]{0,30}-d"
            r"[^|;\n]{0,120}\|[^|;\n]{0,120}sh\b"
        ),
    ),
    Pattern(
        "sudo-in-pkgbuild",
        Severity.BLOCK,
        "invokes sudo during the build",
        # At a command position and followed by a command, so that the
        # dependency `depends=('sudo')` is not reported as running sudo.
        # `sudo -v` is still an invocation, hence the dash.
        re.compile(r"(?:^|[;&|(`\n]|\s)\s*sudo\s+[a-zA-Z-]"),
        # and not merely mentioned inside a message: scriptlets routinely
        # echo "now run sudo whatever" to the user
        guard=lambda text, match: not inside_quotes(text, match.start()),
    ),
    Pattern(
        "rm-outside-builddir",
        Severity.BLOCK,
        "deletes an absolute path or the user's home",
        # `~` and `$HOME` may be a prefix of the path being deleted, so only
        # an absolute path is anchored. /tmp is exempt: wiping scratch space
        # is normal, wiping /etc is not.
        re.compile(
            r"""\brm\s+-[a-zA-Z]*[rf][a-zA-Z]*\s+["']?"""
            r"""(?:/(?!tmp(?:/|\s|$))|\*|~/|\$HOME|\$\{HOME\})"""
        ),
    ),
    Pattern(
        "write-system-path",
        Severity.BLOCK,
        "writes into a system path instead of $pkgdir",
        # The `>` must not close a quote or a tag, or XML inside a heredoc
        # (`path">/usr/bin/thing</annotate>`) reads as a redirect.
        # A redirect, or the far more common `install`/`cp` into a system
        # path. Two things keep this honest:
        #   - the system path must be the *last* argument, or
        #     `ln -s /opt/X "$pkgdir/etc/Y"` matches on the source it reads
        #   - `$pkgdir/usr` and `$DESTDIR/usr` have no whitespace before the
        #     slash, which is exactly what makes them correct
        # MULTILINE so `$` means end of line: without it the system path
        # had to be the last thing in the whole file.
        re.compile(
            r"(?<![\x22\x27/>])>>?\s*/(?:etc|usr|boot|var|opt)/"
            r"|\b(?:install|cp|mv|ln|tee)\b[^\n]*[ \t]"
            r"/(?:etc|usr|boot|var|opt)/[^\s\x22\x27]*[\x22\x27]?"
            r"[ \t]*[;&|)}\]]*[ \t]*$",
            re.MULTILINE,
        ),
    ),
    Pattern(
        "setuid",
        Severity.BLOCK,
        "sets a setuid or setgid bit",
        # `chmod +s` has no `u`/`g`, and a five-digit octal (`04755`) puts
        # the setuid digit second, so both spellings were missed.
        re.compile(r"\bchmod\s+(?:-R\s+)?(?:[ug]?\+s\b|[0-7]?[24][0-7]{3}\b)"),
    ),
    Pattern(
        "network-in-build",
        Severity.LOOK,
        "fetches from the network while building",
        # `[ \t]`, not `\s`: `\s` matches a newline, so a variable
        # assignment ending in the word (`_tool=curl`) read as a fetch.
        re.compile(r"\b(?:curl|wget)[ \t]"),
        # Every function, not the four that usually build. A download in
        # `pkgver()` or in a helper that `build()` calls is the same act and
        # was simply not looked at.
        funcs=("*",),
    ),
    Pattern(
        "git-clone-in-build",
        Severity.LOOK,
        "clones a git repository while building",
        re.compile(r"\bgit\s+clone\b"),
        funcs=("*",),
    ),
    Pattern("eval", Severity.LOOK, "uses eval", re.compile(r"\beval\b")),
    Pattern(
        "chmod-777",
        Severity.LOOK,
        "sets world-writable permissions",
        re.compile(r"\bchmod\s+(?:-R\s+)?777\b"),
    ),
    Pattern(
        "skippgpcheck",
        Severity.LOOK,
        "disables PGP signature checking",
        re.compile(r"\b(?:--skippgpcheck|!?skippgpcheck|!?pgpcheck)\b"),
    ),
    Pattern(
        "weak-checksum",
        Severity.LOOK,
        "uses a weak checksum algorithm",
        re.compile(r"\b(?:md5sums|sha1sums)\s*="),
    ),
    Pattern(
        "curl-pipe-no-fail",
        Severity.INFO,
        "pipes a download without failing on HTTP errors",
        # Both quantifiers are bounded and the last is possessive. With an
        # unbounded lookahead, one long line containing `curl` but no pipe
        # made every start position scan to the end of the file.
        re.compile(r"\bcurl\b(?![^\n|]{0,120}(?:-f|--fail))[^\n|]{0,400}+\|"),
        funcs=("*",),
    ),
)


#: a PKGBUILD that creates its own functions cannot be attributed per-function
_DYNAMIC_FUNCS = re.compile(r"\beval\b[^\n]*\(\s*\)", re.IGNORECASE)


def functions_generated(pkgbuild) -> bool:
    """Whether this PKGBUILD builds functions at runtime.

    `eval "package_${pkgbase}() { ... }"` produces function bodies that static
    parsing cannot attribute to a name. Scoped rules would then find no body and
    quietly report nothing, which is the worst possible outcome for a security
    check: it looks like a clean result.
    """
    return bool(pkgbuild is not None and _DYNAMIC_FUNCS.search(pkgbuild.raw))


#: Patterns are run against at most this much text.
#:
#: The input is whatever the maintainer wrote, and a regex that backtracks turns
#: one long line into a hung review. The largest real PKGBUILD across this
#: machine's 92 clones is 30 KB (median 1.5 KB), so 64 KB cannot truncate a real
#: package while bounding the worst case. Truncation is reported, never silent.
MAX_SCAN = 64 * 1024


def inside_quotes(text: str, position: int) -> bool:
    """Whether `position` falls inside a double- or single-quoted run on its line.

    `echo "run sudo thing"` is a message, not an invocation, and a regex cannot
    tell the two apart. Counting the quotes earlier on the line can.
    """
    start = text.rfind("\n", 0, position) + 1
    quote = ""
    i = start
    while i < position:
        char = text[i]
        if char == "\\" and quote != "'":
            i += 2
            continue
        if quote:
            if char == quote:
                quote = ""
        elif char in "'\"":
            quote = char
        i += 1
    return bool(quote)


def scan_patterns(change: Change) -> list[Finding]:
    """Scan the new PKGBUILD, suppressing patterns that were already there."""
    new = change.new_pkgbuild
    if new is None:
        return []
    old = change.old_pkgbuild
    findings: list[Finding] = []
    truncated = len(new.raw) > MAX_SCAN
    if truncated:
        findings.append(
            Finding(
                "scan-truncated",
                Severity.INFO,
                f"the PKGBUILD is larger than {MAX_SCAN // 1024} KiB; only the "
                "first part was pattern-scanned",
            )
        )

    dynamic = functions_generated(new)

    for pat in PATTERNS:
        scoped_to_raw = False
        if pat.funcs == ("*",) and not dynamic:
            scope_names = {fn: body for fn, body in new.funcs.items() if body}
        elif pat.funcs and not dynamic:
            scopes = {fn: new.funcs.get(fn, "") for fn in pat.funcs}
            scope_names = {fn: text for fn, text in scopes.items() if text}
        else:
            # Either the rule is file-wide, or the file generates its own
            # functions and per-function attribution is impossible. Scanning the
            # whole file is noisy, but reporting nothing is a false negative.
            scope_names = {"PKGBUILD": new.code}
            scoped_to_raw = bool(pat.funcs)

        for scope, text in scope_names.items():
            match = pat.regex.search(text[:MAX_SCAN])
            if not match:
                continue
            if pat.guard is not None and not pat.guard(text, match):
                continue
            if pat.new_only and old is not None:
                old_text = (
                    old.funcs.get(scope, "")
                    if pat.funcs and not scoped_to_raw and not dynamic
                    else old.code
                )
                if pat.regex.search(old_text[:MAX_SCAN]):
                    continue  # pre-existing; not this update's problem
            line = _line_of(text, match.start())
            findings.append(
                Finding(
                    rule=pat.rule,
                    severity=pat.severity,
                    title=pat.title,
                    evidence=line.strip()[:120],
                    where=scope if pat.funcs else "PKGBUILD",
                )
            )
    return findings


def _line_of(text: str, offset: int) -> str:
    start = text.rfind("\n", 0, offset) + 1
    end = text.find("\n", offset)
    return text[start : end if end != -1 else len(text)]


# --------------------------------------------------------------------------
# field-level rules
# --------------------------------------------------------------------------


def scan_sums(change: Change, diff: SemanticDiff) -> list[Finding]:
    """SKIP is judged per source entry, against that entry's scheme.

    The checksum list is parallel to `source`, so the question is never "is this
    a -git package" but "is this particular entry a live repository". A tarball
    source with SKIP is unverified no matter what the package is called.

    Reads `primary_sums`, not `sha256sums`: a package may checksum with b2sums
    or sha512sums instead, and looking only at sha256sums made those packages
    appear to declare no checksums at all -- so a SKIP newly introduced on a
    plain tarball source produced no finding.
    """
    new = change.new_srcinfo
    if new is None:
        return []
    sums = new.primary_sums
    if not sums:
        return []

    old = change.old_srcinfo
    old_sums = old.primary_sums if old else []
    # A package that verifies with GPG and ships a detached signature covers its
    # sibling metadata through that signature: a repository `Release` file is
    # signed, and the `Packages` index is hashed inside it. SKIP on those is the
    # convention, not an unverified download.
    gpg_verified = bool(new.validpgpkeys) and any(
        is_signature_source(entry) for entry in new.source
    )
    out: list[Finding] = []
    for index, checksum in enumerate(sums):
        if checksum.strip().upper() != "SKIP":
            continue
        source = new.source[index] if index < len(new.source) else ""
        if source and is_signature_source(source):
            out.append(
                Finding(
                    "sums-skip-signature",
                    Severity.INFO,
                    "SKIP checksum for a detached signature (verified against validpgpkeys)",
                    evidence=source[:120],
                    where=f"source[{index}]",
                )
            )
            continue
        if source and domain_of(source) == "(local)":
            # Nothing is downloaded, so there is no supply chain to trust. It
            # removes integrity checking on a file shipped inside the package,
            # which is worth knowing and not worth blocking on.
            out.append(
                Finding(
                    "sums-skip-local",
                    Severity.INFO,
                    "SKIP checksum for a file shipped in the package",
                    evidence=source[:120],
                    where=f"source[{index}]",
                )
            )
            continue
        if source and is_vcs_source(source):
            out.append(
                Finding(
                    "sums-skip-vcs",
                    Severity.INFO,
                    "SKIP checksum for a live VCS source (expected)",
                    evidence=source[:120],
                    where=f"source[{index}]",
                )
            )
            continue

        was_skip = bool(index < len(old_sums) and old_sums[index].strip().upper() == "SKIP")
        if was_skip or gpg_verified:
            severity = Severity.INFO
        else:
            severity = Severity.BLOCK
        out.append(
            Finding(
                "sums-skip-existing" if was_skip else "sums-skip-introduced",
                severity,
                "SKIP checksum on a source that is not a live repository"
                + ("" if was_skip else " (newly introduced)"),
                evidence=source[:120] or f"checksum entry {index}",
                where=f"source[{index}]" if source else "",
            )
        )
    return out


def scan_pgp(change: Change, diff: SemanticDiff) -> list[Finding]:
    """Removing a trusted key matters only if nothing replaced it.

    A maintainer rotating an upstream signing key removes the old one and adds
    the new one in the same commit, which is routine. Reporting that as a BLOCK
    trains the reader to ignore the one that is not routine.
    """
    removed = diff.validpgpkeys.removed
    if not removed:
        return []
    if diff.validpgpkeys.added:
        return [
            Finding(
                "pgpkey-rotated",
                Severity.INFO,
                "a trusted PGP key was replaced",
                evidence=", ".join(removed)[:120],
                where="the new key is in this revision",
            )
        ]
    return [
        Finding(
            "pgpkey-removed",
            Severity.BLOCK,
            "a trusted PGP key was removed and not replaced",
            evidence=", ".join(removed)[:120],
        )
    ]


def scan_deps(change: Change, diff: SemanticDiff) -> list[Finding]:
    """New AUR trust edges: a new dep means trusting another maintainer."""
    out: list[Finding] = []
    if diff.depends.added:
        out.append(
            Finding(
                "new-runtime-dep",
                Severity.INFO,
                f"{len(diff.depends.added)} new runtime dependency(ies)",
                evidence=", ".join(diff.depends.added)[:120],
            )
        )
    if diff.makedepends.added:
        out.append(
            Finding(
                "new-build-dep",
                Severity.INFO,
                f"{len(diff.makedepends.added)} new build dependency(ies)",
                evidence=", ".join(diff.makedepends.added)[:120],
            )
        )
    if diff.replaces.added:
        out.append(
            Finding(
                "new-replaces",
                Severity.LOOK,
                "now declares that it replaces another package",
                evidence=", ".join(diff.replaces.added)[:120],
            )
        )
    if diff.conflicts.added:
        out.append(
            Finding(
                "new-conflicts",
                Severity.LOOK,
                "now declares a conflict",
                evidence=", ".join(diff.conflicts.added)[:120],
            )
        )
    return out


def scan_install_scriptlet(change: Change, diff: SemanticDiff) -> list[Finding]:
    """A `.install` scriptlet runs as root on the user's machine.

    The text arrives on the Change rather than being read here. Reading it with
    `Path(name)` resolved against the process working directory, and the CLI
    never chdirs, so from a normal invocation the file was never found and the
    scriptlet was never looked at. The old side was read from the same working
    tree rather than from the baseline commit, so a scriptlet that changed
    reported "already present".
    """
    if change.new_pkgbuild is None:
        return []

    dangerous = re.compile(
        r"\b(?:systemctl|sysusers|useradd|groupadd|udevadm|update-desktop-database)\b"
        r"|(?<![\x22\x27/>])>>?\s*/(?:etc|usr|boot|var)/"
        r"|\bcrontab\b|\bat\s+now\b"
    )

    out: list[Finding] = []
    for name, text in sorted(change.new_install.items()):
        match = dangerous.search(text)
        if not match:
            continue
        old_text = change.old_install.get(name, "")
        newly = not old_text or not dangerous.search(old_text)
        out.append(
            Finding(
                "install-scriptlet-system",
                Severity.LOOK if newly else Severity.INFO,
                "an install scriptlet touches the system"
                + ("" if newly else " (unchanged from the previous revision)"),
                evidence=match.group(0)[:120],
                where=name,
            )
        )

    # A scriptlet is shell like any other, so the same patterns apply. Nothing
    # was scanning it before, which meant `curl | bash` inside one was invisible.
    for name, text in sorted(change.new_install.items()):
        old_text = change.old_install.get(name, "")
        for pat in PATTERNS:
            found = pat.regex.search(text[:MAX_SCAN])
            if not found:
                continue
            if pat.guard is not None and not pat.guard(text, found):
                continue
            if old_text and pat.regex.search(old_text[:MAX_SCAN]):
                continue
            out.append(
                Finding(
                    pat.rule,
                    pat.severity,
                    pat.title + " (in an install scriptlet)",
                    evidence=found.group(0)[:120],
                    where=name,
                )
            )
    return out


def scan_maintainer(change: Change) -> list[Finding]:
    """Who maintains the package now, against who maintained it before.

    `None` is "no record" and `""` is "orphaned". Adoption -- an orphaned
    package acquiring a maintainer -- is the AUR account-takeover vector, so it
    is reported even though nothing else about the update looks unusual.
    Orphaning is not suspicious on its own and is only context.
    """
    old, new = change.old_maintainer, change.new_maintainer

    if new is None:
        # the package is orphaned now
        if old:
            return [
                Finding(
                    "orphaned",
                    Severity.INFO,
                    "package has been orphaned",
                    evidence=f"was maintained by {old}",
                )
            ]
        return []

    if not new.strip():
        if old:
            return [
                Finding(
                    "orphaned",
                    Severity.INFO,
                    "package has been orphaned",
                    evidence=f"was maintained by {old}",
                )
            ]
        return []

    if old is None:
        # no record of who had it: a first audit cannot tell adoption from
        # continuity, and reporting every package would be pure noise
        return []

    if not old:
        return [
            Finding(
                "maintainer-adopted",
                Severity.LOOK,
                "an orphaned package has been adopted by a new maintainer",
                evidence=f"was orphaned -> {new}",
            )
        ]

    if old == new:
        return []
    return [
        Finding(
            "maintainer-changed",
            Severity.LOOK,
            "package changed maintainer",
            evidence=f"{old} -> {new}",
        )
    ]


def scan_sources(change: Change, diff: SemanticDiff) -> list[Finding]:
    out: list[Finding] = []
    for move in diff.sources_moved:
        if move.old_domain == move.new_domain:
            continue
        out.append(
            Finding(
                "source-domain-changed",
                Severity.LOOK,
                f"{move.filename} now comes from a different host",
                evidence=f"{move.old_domain} -> {move.new_domain}",
            )
        )
    # A version bump renames the tarball, so `added` and `removed` are both
    # non-empty on almost every commit; requiring `added and not removed`
    # reports nothing at all and hides the shape that matters, which is slipping
    # an extra source in during a routine update. The signal is the list getting
    # *longer*, not an entry being unfamiliar.
    # Without an old side there is nothing to compare: every source looks newly
    # added, which is how "was 0, now 1" appears on packages that have no
    # baseline. Not knowing is not evidence of a change.
    if change.old_srcinfo is None or change.new_srcinfo is None:
        return out

    old_sources = change.old_srcinfo.source
    new_sources = change.new_srcinfo.source
    grew = len(new_sources) > len(old_sources)
    if not grew:
        return out

    known_hosts = {domain_of(entry) for entry in old_sources}
    for entry in diff.source.added:
        host = domain_of(entry)
        if host and host != "(local)" and host not in known_hosts:
            out.append(
                Finding(
                    "source-host-added",
                    Severity.LOOK,
                    f"a source from {host} was added, a host this package has not used before",
                    evidence=entry[:120],
                )
            )
    out.append(
        Finding(
            "source-added",
            Severity.INFO,
            f"the package now has {len(new_sources)} source(s), was {len(old_sources)}",
            evidence=", ".join(diff.source.added)[:120],
        )
    )
    return out


RULES = (
    scan_sums,
    scan_pgp,
    scan_deps,
    scan_install_scriptlet,
    scan_maintainer,
    scan_sources,
    scan_patterns,
)


# --------------------------------------------------------------------------
# entry point
# --------------------------------------------------------------------------


def semantic_diff(change: Change) -> SemanticDiff:
    old, new = change.old_srcinfo, change.new_srcinfo
    d = SemanticDiff()
    if old and new:
        d.pkgver = _pair(old.pkgver, new.pkgver)
        d.pkgrel = _pair(old.pkgrel, new.pkgrel)
        d.epoch = _pair(old.epoch, new.epoch)
        d.source = _setdiff(old.source, new.source)
        d.sums = _setdiff(old.primary_sums, new.primary_sums)
        d.depends = _setdiff(old.depends, new.depends)
        d.makedepends = _setdiff(old.makedepends, new.makedepends)
        d.optdepends = _setdiff(old.optdepends, new.optdepends)
        d.provides = _setdiff(old.provides, new.provides)
        d.conflicts = _setdiff(old.conflicts, new.conflicts)
        d.replaces = _setdiff(old.replaces, new.replaces)
        d.validpgpkeys = _setdiff(old.validpgpkeys, new.validpgpkeys)

        # "the source moved" detection. Pairing by index handles the common case
        # where a version bump renames the tarball; pairing by filename is the
        # fallback when the source list itself changed shape.
        pairs: list[tuple[str, str]]
        if len(old.source) == len(new.source):
            pairs = list(zip(old.source, new.source))
        else:
            before = {filename_of(u): u for u in old.source}
            after = {filename_of(u): u for u in new.source}
            pairs = [(before[k], after[k]) for k in sorted(set(before) & set(after))]
        for old_url, new_url in pairs:
            if old_url != new_url:
                d.sources_moved.append(SourceMove(filename_of(new_url), old_url, new_url))

    if change.old_pkgbuild and change.new_pkgbuild:
        o, n = change.old_pkgbuild, change.new_pkgbuild
        d.funcs_added = sorted(set(n.funcs) - set(o.funcs))
        d.funcs_removed = sorted(set(o.funcs) - set(n.funcs))
        d.funcs_changed = sorted(
            fn for fn in set(o.funcs) & set(n.funcs) if o.funcs[fn] != n.funcs[fn]
        )
        d.lines_added, d.lines_removed = _lines_changed(o.raw, n.raw)

    return d


def audit(change: Change) -> Report:
    diff = semantic_diff(change)
    findings: list[Finding] = []
    for rule in RULES:
        try:
            findings.extend(rule(change, diff) if _takes_diff(rule) else rule(change))
        except Exception as exc:  # noqa: BLE001 - a broken rule must not kill the run
            findings.append(
                Finding(rule.__name__, Severity.INFO, f"rule failed: {type(exc).__name__}: {exc}")
            )
    order = {Severity.BLOCK: 0, Severity.LOOK: 1, Severity.INFO: 2}
    findings.sort(key=lambda f: (order[f.severity], f.rule))
    if change.new_pkgbuild is None:
        basis = Basis.NOTHING
    elif change.old_pkgbuild is None:
        basis = Basis.NEW_ONLY
    else:
        basis = Basis.COMPARED
    return Report(package=change.package, diff=diff, findings=findings, basis=basis)


def _takes_diff(rule) -> bool:
    return rule.__name__ in {
        "scan_sums",
        "scan_pgp",
        "scan_deps",
        "scan_sources",
        "scan_install_scriptlet",
    }