# brotli-checker

Check whether an app or project needs a Brotli security update for CVE-2026-111497, explain the result in plain language, and give practical next steps. Find bundled copies and investigate whether the running app uses them.

- **Kind:** skill
- **Source:** https://github.com/trailofbits/brotli-checker
- **Page:** https://forefy.com/skills/a546f805-b419-4dd8-a839-2c0321f21dc7
- **API (JSON + files):** https://forefy.com/api/asr/a546f805-b419-4dd8-a839-2c0321f21dc7

## Install

Ask your agent: `Install the brotli-checker skill securely from: https://forefy.com/skills/a546f805-b419-4dd8-a839-2c0321f21dc7 globally to ~/.claude/skills`

Or run the pinned install script: `curl -fsSL https://forefy.com/api/asr/a546f805-b419-4dd8-a839-2c0321f21dc7/install | bash` (`?location=project` for the current project, `&harness=agents` for `.agents/`, `&shell=powershell` with `irm ... | iex`).

---

## LICENSE

```

```

## SKILL.md

---
name: brotli-checker
description: Check whether an app or project needs a Brotli security update for CVE-2026-111497, explain the result in plain language, and give practical next steps. Find bundled copies and investigate whether the running app uses them.
---

# Check whether an app needs a Brotli update

Help someone who is not a security expert answer: "Does my app need an update,
and what should I do next?" They should be able to say "check this project" or
give a website address without knowing library versions, build details, or the
CVE number. Gather technical evidence yourself; translate it into a short,
actionable answer.

Assess CVE-2026-111497 for the requested source tree, application, or deployment.
Finding vulnerable source establishes source containment. Build selection,
shipment, attacker-controlled input reaching that decoder, and exploitability
are separate conclusions. Keep a result for each copy; one fixed copy cannot
clear other copies.

## Gather evidence

For "this project", start with the current project folder and discover its
application, dependencies, and deployment files. Use the supplied folder or
server access when available. Collect versions and package information with
trusted, read-only inspection tools; do not run application code or scripts
from the target. Do not make the user supply technical facts you can inspect.

Identify the target revision, build/platform, deployed artifact or container
digest, and the installation actually serving the application when available.
If only a URL is supplied, explain that headers and Brotli response support
cannot identify the decoder or its patch status. Say "I can't confirm this from
the website address alone" and give a short message the user can send to their
hosting provider asking it to check and update the app's Brotli decoder. If
the conversation already provides local/server access, use it. Ask only for
missing access or a target that cannot be inferred. Do not send malformed
streams or exploit probes.

From this skill's directory, run the offline inventory helper (Python 3.12–3.15).
Use JSON internally for the evidence; the default text output is for people:

```sh
python3 scripts/brotli_inventory.py /absolute/target --json
```

It reads files and bounded ZIP/JAR/WAR/wheel, TAR, gzip, bzip2, and XZ containers
without extracting them or executing target code. Known signatures identify
renamed archives and extensionless layers. Recognized unsupported formats such
as Debian, RPM, and Zstandard produce coverage gaps; provide unpacked files.
Signature-free or unrecognized encodings can be missed. Container traversal
does not reconstruct a final image filesystem. It checks eligible file bytes
against source fingerprints, identifies modified C decoder patterns, and records
dependency, native/Wasm, and JavaScript/TypeScript fetch/Undici candidates.
Fetch references include source locations; they are heuristic leads, not
confirmed Node calls or attacker-controlled inputs. For supported ELF32/ELF64 targets it also
matches compiled decoder text, including stripped and statically linked code.
Dependencies, build outputs, and hidden directories
are included; `.git` is excluded. Read its coverage gaps. It does not scan an
entire host, resolve remote dependencies, follow symlinks, prove whole-application patch
status, or establish an absence of Brotli. Supply extracted image filesystems
or additional artifact roots as needed. Filesystem symlink targets visited
directly do not add a gap; other targets still need checking. If a resource limit
is reached, scan a smaller relevant folder or adjust the limit shown by `--help`.
Regular files have no default per-file or total byte cap. ELF analysis uses
sparse offset reads; full-artifact SHA-256 is streamed. Archive expansion is
separately budgeted at 8 GiB/member or stream and 64 GiB total; override with
`--max-expanded-bytes` and `--max-total-expanded-bytes` as needed. Optional
`--max-file-bytes` and `--max-total-bytes` still impose explicit input caps.

For independently obtained upstream versions, optional flags are:

```sh
python3 scripts/brotli_inventory.py --brotli-version 1.2.0 --json
python3 scripts/brotli_inventory.py --node-version 24.22.0 --json
```

These flags assess a supplied version string, not an installed or running build.
Never pass a wrapper package's version as the embedded Brotli version. Version
warnings allow vendor backports; prereleases, vendor suffixes, and branches
outside the snapshot's rules remain unknown. Exit 1 means affected-source or
version evidence; 3 means review/unknown; 0 means only the reported items meet
a known source fingerprint or upstream fixed-version rule; 2 means CLI error.
The helper never certifies the application safe.

For ELF code findings, inspect `evidence.code_matches`: source/object hashes,
compiler/flags, state, and file offset. `vulnerable_code_match` and
`mixed_code_matches` warrant update review (exit 1). `fixed_code_match` concerns
only matched decoder code, leaves other copies unresolved, and does not by itself
produce exit 0. Version strings/imports do not determine code state. Matching
excludes relocated address fields and does not verify their targets or referenced
data or associate separately matched executable sections. Packaged references cover
x86, x86-64, ARM/Thumb, AArch64, RISC-V64, PowerPC64 little-endian, s390x,
MIPS32/64 in both byte orders, and LoongArch64. They cover `-O2`/`-O3` PIC/non-PIC
builds of three affected/fixed release pairs with GCC 14.2, the original AArch64
GCC 16.2.1 builds, and Clang 19 LoongArch builds. LoongArch uses a documented
upstream header portability change and LP64 declaration headers; inspect
`build_note`, `headers_sha256`, and flags. Other compiler/options require
corresponding trusted references via `--code-references`. No match remains unknown;
a supported target with no references produces an explicit coverage gap.
Do not present this as compiler-independent semantic or CFG patch verification.

The inventory also reads bounded ELF symbol tables without executing the target.
It distinguishes decoder API definitions from imports. For an exactly defined
`BrotliDecoderVersion`, it accepts only supported straight-line constant-return
instruction forms. A reported old supported-branch version is
`potentially_vulnerable` and produces the normal exit 1 unless stronger evidence
already applies. A reported fixed version is metadata only and never clears the
binary. Code-reference matches remain code-identity evidence; metadata does not
upgrade an unmatched binary to `fixed_code_match`.

For an unresolved local ELF with decoder name, symbol, version, or code evidence,
collect bounded disassembly for model review instead of stopping at a generic
unknown. Do not disassemble every unrelated ELF. Run this only with a trusted
local `objdump`; the helper rejects target
symlinks, streams the original artifact hash, reads ELF structures by offset,
and passes only bounded code-fragment snapshots to `objdump`. It never executes
or loads target code and does not copy the full target. Detected input mutation
makes the report incomplete:

```sh
python3 scripts/brotli_binary_triage.py /absolute/target > triage.json
```

The default mode selects defined `Decode*BlockSwitch`,
`SafeDecode*BlockSwitch`, and `ProcessCommands*` symbols, with
`BrotliDecoderDecompressStream` as a public-symbol fallback. It records the target
SHA-256, exact commands, selected scope, limits, symbols, and bounded output.
Fragment hashes and source offsets bind the assembly to selected bytes; raw
objdump output uses `.data`, not original section names. Resolve numeric call
targets using the reported symbol addresses and original ELF scope. Exit
0 means evidence collection completed, not that the binary is fixed. Exit 3 means
the evidence is incomplete. For stripped code, use only a range derived from
trusted link/map or scanner evidence, for example `--file-offset OFFSET --length
BYTES`, or `--start-address START --stop-address STOP [--section .text]`. Do not
guess a range from a string location. Symbol and explicit-range modes both enforce
`--max-range-bytes`; raise that bound deliberately when a trusted symbol is larger.
If a trusted tool or defensible code scope is unavailable, keep the result unknown
and state that missing evidence and the exact next request.

Review triage evidence against the reviewed affected and fixed `decode.c` sources,
not against mnemonic or string presence. Trace each literal, command, and distance
block-switch path. The fixed design returns `BrotliDecoderErrorCode`, produces
`BROTLI_DECODER_ERROR_FORMAT_BLOCK_SWITCH` for the invalid switch, and propagates
that status through `ProcessCommands*`; the old design uses boolean/void handling
that can lose the specific failure. Confirm the safe and non-safe callers and the
error propagation, not merely a helper body. A version constant, import/export,
function name, or nearby error string cannot establish the fix. If optimization,
inlining, missing callers, output limits, multiple copies, or unfamiliar control
flow prevents that trace, preserve `unknown`. One fixed-looking range never clears
other copies or establishes runtime use.

For artifact hashes and rollout inventory, optionally use:

```sh
python3 scripts/brotli_binary_inventory.py /absolute/target
```

Its exit codes differ: 1 means inventory candidates (including fixed matches),
3 means coverage gaps, and 0 means no candidates/gaps, not that an app is safe.

End-of-life Node.js versions get supported-branch upgrade advice, while decoder
patch status remains unknown (exit 3) without build or vendor evidence.

Validation has covered decoder source references, synthetic ELF/packaging
fixtures, an opt-in suite for locally compiled stripped ELF artifacts, and a
benign loopback test of built-in Node fetch response decompression. No deployed
application or production binary's patch state was validated, and no malformed
or exploit payload was executed. The benign fetch test establishes a decoding
route, not CVE triggerability or exploitation. Do not describe source
classification as an end-to-end test of a vulnerable app.

## Establish patch status and reachability

Read [the advisory and detection reference](references/advisory.md) for source
provenance, fixed branches, backports, and detection limits. Verify current
vendor guidance if online research is available; otherwise date the assessment
to the packaged snapshot and flag that advisory freshness is unverified.

Inspect lockfiles, submodules, downloaded build inputs, native libraries inside
archives, linking configuration, SBOMs, and the shipped binary. Check all build
variants relevant to the deployment. Do not equate a dependency declaration or
an inventory entry with compilation or shipment. Stripped/static binaries need
build provenance, vendor evidence, or binary analysis; absence from a dynamic
library listing is insufficient. A fixed OS library does not fix embedded copies.

When relevant source is available, use Trailmark to accelerate discovery of
decoder/fetch calls and their callers. From this skill's directory, run the
packaged helper with uv, which installs its pinned Trailmark 0.7.0 dependency:

```sh
uv run --script scripts/brotli_source_discovery.py /absolute/source > discovery.json
```

Choose the relevant source subtree; initial parsing is not resource-bounded.
The first use may download Trailmark and its parser grammars. Do not substitute
an unverified ambient `trailmark` command or run tools supplied by the target.
The helper reads source without importing or executing the target. Keep the
inventory scan: Trailmark skips dependencies, build outputs, and hidden folders,
and can omit top-level JavaScript/TypeScript calls without a diagnostic.

Use `candidates[].call` and the `neighborhood` nodes, calls, and entrypoint tags
to prioritize source inspection. Preserve callsite locations, unresolved targets,
confidence, deferred callback edges, and `configured_by` provenance. Verify
bindings and callers in source; static confidence does not establish execution.
For a reviewed import alias or wrapper, rerun with `--symbol NAME`. This adds a
search target, not a verified binding. Repository `.trailmark` links and
entrypoint overrides are declarations that require independent review.

Exit 0 means evidence was collected, not that a decoder is fixed or reachable.
Exit 3 means collection is incomplete; inspect diagnostics, truncation, and gaps.
Missing paths or entrypoint tags never establish unreachability. If uv, parsers,
or supported source are unavailable, state the gap and continue manual review.
See [source discovery](references/reachability.md#trailmark-source-discovery)
for scope and review steps.

For each relevant surface, use [the reachability guide](references/reachability.md).
Trace the actual input entry point through middleware/wrappers to the selected
native decoder. Record configuration, authentication before decompression,
input ownership, and process boundaries. Consider inbound requests, outbound
responses, uploads and WOFF2/fonts, Java JNI stacks, workerd, and negotiated TLS
certificate compression. Pure Java/Rust implementations require their own
evidence; a language or algorithm name alone does not establish this C bug.

For Node.js apps, explicitly investigate built-in `fetch()`, `globalThis.fetch`,
and Undici wrappers, including attacker-controlled URLs and redirects. Automatic
Brotli response decoding can reach the native decoder without an explicit
Brotli dependency or `node:zlib` import. Establish the actual runtime's patch
state. The offline inventory records source-reference and dependency leads. Use
Trailmark's call evidence to find wrappers and callers, then verify fetch
bindings, URL ownership, and the running Node version yourself. Review source
locations and continue with the Node/fetch section of the reachability guide.
A fixed supplied version or standalone Brotli library does not clear an
unresolved runtime lead.

For Electron apps, use the guide's Electron section. Separate Chromium's
renderer/net/session fetch paths from Node fetch, and verify which decoder the
embedder links. A standalone Node version rule cannot classify a vendor Electron
build. Mach-O and ASAR packages require additional inspection. Benign Electron
and Slack renderer measurements confirmed local response decoding, while Slack's
shipped decoder patch status and attacker-controlled product paths remain unknown.
The offline scanner does not launch either application.

Report reachability as `confirmed`, `not_reachable_in_reviewed_configuration`,
or `unknown`, with paths/lines or deployment evidence. A negative conclusion
must name the exact configuration and reviewed paths. Missing evidence remains
unknown. Do not infer exploitability from reachability, CVSS, a crash, or a
third-party PoC.

## Explain the result and next action

Lead with a short answer that matches the evidence:

- "Update needed" when the app is confirmed to use an affected copy. State
  separately whether outside input can reach it.
- "Vulnerable Brotli source found" when affected code is present but its use by
  the running app is unknown. Say "We still need to check whether your app uses
  this copy." Do not label the whole app vulnerable from source presence alone.
- "Fix confirmed for the checked copy" when supported by verified evidence.
  State the scope; other copies and the running app may still need checking.
- "More information needed" when the result is inconclusive. Explain the
  specific missing information and who can obtain it.

Follow with what to do next. Name the affected component and the applicable
update when established, who should make it, and how to confirm the running app
has it. For a bundled copy, explain that the app must be rebuilt and deployed
again; updating the computer's shared library may leave that copy unchanged.
If action needs a maintainer or hosting provider, provide a short, ready-to-send
message with the specific finding and requested update/confirmation. Do not
send it yourself.

For several copies, use a small table: location, plain-language result, next
action. A fixed copy must not hide a vulnerable copy. Prioritize affected
shipped copies with confirmed untrusted input, then resolve other candidates.
State important unchecked files or missing running-app evidence in ordinary
language. An empty scan is inconclusive. Do not label it "safe" or "unaffected".

Keep the initial answer to a few sentences plus any useful short table. Avoid
raw status codes, hashes, long evidence columns, and unexplained terms such as
"shipment", "reachability", "JNI", or "provenance" in that answer. Preserve the
full per-copy evidence (patch confidence, build selection, shipped artifact,
input path, paths/lines, advisory date, and source/vendor links) for a maintainer
or a request for technical details. Do not make the user interpret a security
audit table to decide what to do.

Where an update cannot be deployed immediately, identify the specific receiving
decoder path that could be disabled and verify that boundary. Response encoding
settings alone do not establish that request, font, or certificate decoding is
disabled. Never present the community source inventory as vulnerable deployments.

## agents

```

```

## agents/openai.yaml

```yaml
interface:
  display_name: "Brotli Security Checker"
  short_description: "Check whether your app needs a Brotli security update"
```

## references

```

```

## references/BROTLI-LICENSE

```

```

## references/THIRD_PARTY_NOTICES.md

# Third-party notices

`binary-fingerprints.json.gz` contains compiled decoder text derived from Google
Brotli. The original copyright and MIT permission notice are retained in
[BROTLI-LICENSE](BROTLI-LICENSE). Keep this notice and license with the catalog
when redistributing the skill. Each reference records its source URL and hash,
compiler, build flags, ELF class/byte order, and object hash. New references also
record a hash of local header paths and contents. Source hashes may identify
identical decoder files shared by several release trees; they do not authenticate
surrounding headers or the full build environment.

## references/advisory.md

# Advisory and detection evidence

Snapshot verified **2026-10-23**. Refresh vendor guidance when assessing a live
deployment; the scanner is offline and does not update its rules automatically.

## Primary sources

- [Google Brotli fix, 0ed4268a4401abcb4472096340477cf4af7d1a2b](https://github.com/google/brotli/commit/0ed4268a4401abcb4472096340477cf4af7d1a2b).
- [Brotli 1.0.10](https://github.com/google/brotli/releases/tag/v1.0.10),
  [1.1.1](https://github.com/google/brotli/releases/tag/v1.1.1), and
  [1.2.1](https://github.com/google/brotli/releases/tag/v1.2.1) security releases.
- [CNA record](https://www.cve.org/CVERecord?id=CVE-2026-111497), also available as
  [machine-readable CVE JSON](https://github.com/CVEProject/cvelistV5/blob/main/cves/2026/111xxx/CVE-2026-111497.json).
- [Node.js October 21 security advisory](https://nodejs.org/en/blog/vulnerability/october-2026-security-releases).

The decoder conflated an invalid block switch with insufficient input. The fix
returns a format error and propagates status through the block-switch wrappers
and command processing. Node's advisory describes repeated input buffering into
an eight-byte internal buffer, causing memory corruption with possible crash,
disclosure, or execution.

| Component/branch | Published affected versions / scope | First fixed release |
| --- | --- | --- |
| Brotli 1.0 | 1.0.3 through 1.0.9 | 1.0.10 |
| Brotli 1.1 | 1.1.0 | 1.1.1 |
| Brotli 1.2 | 1.2.0 | 1.2.1 |
| Node.js 22 | Advisory covers 22.x before the security release | 22.23.4 |
| Node.js 24 | Advisory covers 24.x before the security release | 24.22.1 |
| Node.js 26 | Advisory covers 26.x before the security release | 26.12.1 |

The CNA record has disjoint Brotli version ranges. A single comparison against
1.2.1 would incorrectly warn on fixed 1.0.10 and 1.1.1. Other Node branches,
forks, prereleases, and custom builds need decoder-specific evidence. The helper
does not assert that every historical Node release contains an affected decoder.
The [Node.js release schedule](https://github.com/nodejs/Release/blob/main/schedule.json)
marks Node 20 and the historical odd-numbered branches through 25 as end of life
on this snapshot date. The helper recommends a supported fixed branch for these
versions while retaining unknown decoder patch status and exit 3. Future branches
outside the snapshot are not assumed to be end of life.

An older version can contain a vendor backport. Establish the OS release,
provider, architecture, and complete package revision, then use that provider's
advisory and native package version ordering. Do not compare distro revisions as
upstream SemVer or transfer one provider's fixed revision to another. Verify the
file/package actually loaded by the serving process and whether it restarted.
An updated host package does not establish the state of a container, native
extension, or statically linked executable.

## Fingerprint provenance

[fingerprints.json](fingerprints.json) contains 20 distinct decoder-file SHA-256
fingerprints: 17 vulnerable and three fixed. For upstream 1.0.3–1.0.10,
1.1.0–1.1.1, and 1.2.0–1.2.1, release files were downloaded and hashed. Several
releases share identical files. The remaining vulnerable variants came from
immutable repository revisions in the community inventory linked below. Their bytes were
downloaded, their Git blob hashes verified, and their old block-switch handling
compared with the upstream fix. Each entry includes its retrieval URL and hash.
An exact fixed match concerns only that decoder file, not an entire backport,
header set, selected build, or running application.

The [community inventory](https://gist.github.com/N3mes1s/ed33893da270decbb0b4128807fb2141)
at revision `6f34a3ae446cbba851214fd6a0b40d3a6e3c786f` is a positive set of source
containment findings. Its default-branch matches and verified pinned dependency
chains are discovery leads. They do not prove compilation, shipment, input
reachability, or exploitation, and omission from the inventory is not evidence
of absence. Do not use a changing repository count as a deployment count.

## Detection limits

- Exact hashes work independently of filenames but change with whitespace,
  line endings, or patches. Modified C-like source uses conservative pattern
  checks; an old invalid-switch guard plus old call sites is a warning, while
  unfamiliar or fixed-looking source remains unknown until reviewed.
- The pattern checks are not a C parser or preprocessor. Conditional code,
  partial backports, macros, amalgamations, and alternate fixes need review.
- Library names, decoder strings, imports/exports, and manifest dependencies
  establish candidates, not vulnerable implementation identity. Static, stripped,
  or renamed binaries can evade these clues. Compiled ELF decoder references can
  identify matching executable code without symbols, but only within the
  documented architecture/compiler corpus. Relocation targets/data and other
  copies remain unverified. On unmatched ELF builds, defined/imported decoder
  symbols and statically decoded constant-return version functions provide
  additional triage evidence. An old reported version warrants update/backport
  review, never a definitive vulnerable-code claim; a fixed reported version
  never clears a binary. The inventory does not load libraries or execute target
  tools. The separate opt-in triage helper invokes trusted host objdump to collect
  bounded disassembly for model inspection; it never executes target code.
- JavaScript/TypeScript fetch references and Undici imports/dependencies identify
  possible Node HTTP response-decoding routes even without a Brotli dependency.
  They do not establish the fetch binding, runtime, patch state, or input owner.
  The bounded heuristic skips comments/ordinary strings and preserves templates;
  literal template text can produce extra leads. Computed names, dynamic loading,
  bundler rewrites, and unavailable source can hide calls. See the
  [Node/fetch guide](reachability.md#nodes-built-in-fetch).
- ZIP-family archives, TAR, gzip, bzip2, and XZ are identified by known content
  signatures as well as filename hints, including extensionless container layers.
  Expansion, member reads, nesting, and file counts are bounded. Recognized
  unsupported formats (including AR/Debian, RPM, Zstandard, 7z, and RAR), encrypted
  content, unreadable files, unchecked symlink targets, and limits produce gaps.
  Signature-free or unrecognized encodings can be missed. No member is extracted;
  hostile names and error messages are escaped in text and JSON output. Container
  traversal inventories layers without reconstructing a final image filesystem.
- Filesystem symlinks are never opened. When a resolved target was visited
  directly among the supplied scan roots, the alias adds no gap.
  Broken/external/unchecked targets and archive links remain gaps. Python bytecode
  symbol strings are not native-decoder evidence; Python project metadata is
  checked in dependency fields rather than its project description.
- Exit 0 applies only to reported items that match known fixed source or a
  supported upstream version rule. Exit 3 includes no matches. Neither clears
  uninspected copies or establishes application reachability.

See [contributing](https://github.com/trailofbits/brotli-checker/blob/main/CONTRIBUTING.md)
for regression tests, source fingerprint verification, local ELF build checks,
and benign runtime fixtures. These checks exercise detection and decoding routes;
they do not establish deployment reachability or exploitability.

## Escalating an unresolved result

For a maintainer investigating exploitation, establish attacker control,
authentication order, the actual decoder/error path, buffering and chunking,
allocator/platform, mitigations, process privileges/sandbox, repeatability, and
demonstrated impact. A reachable decoder or a lab-specific PoC alone does not
settle an application's exploitability. Keep this analysis in the technical
evidence; give non-experts a specific update or a clear request for missing
information.

## references/binary-fingerprints.json.gz

```

```

## references/elf-relocations.json

```json
{
  "schema_version": 1,
  "sources": ["https://github.com/llvm/llvm-project/tree/llvmorg-18.1.8/llvm/include/llvm/BinaryFormat/ELFRelocs", "https://github.com/llvm/llvm-project/tree/llvmorg-18.1.8/lld/ELF/Arch"],
  "machines": {
    "3": {"0": "", "1": "00000000", "2": "00000000", "3": "00000000", "4": "00000000", "9": "00000000", "10": "00000000", "43": "00000000"},
    "8": {"0": "", "2": "00000000", "3": "00000000", "4": "fc000000", "5": "ffff0000", "6": "ffff0000", "7": "ffff0000", "9": "ffff0000", "10": "ffff0000", "11": "ffff0000", "18": "0000000000000000", "19": "ffff0000", "20": "ffff0000", "21": "ffff0000", "22": "ffff0000", "23": "ffff0000", "28": "ffff0000", "29": "ffff0000", "30": "ffff0000", "31": "ffff0000", "37": "00000000"},
    "21": {"0": "", "1": "00000000", "3": "0000", "4": "0000", "5": "0000", "6": "0000", "10": "fc000003", "17": "0000", "26": "00000000", "38": "0000000000000000", "44": "0000000000000000", "47": "0000", "48": "0000", "49": "0000", "50": "0000", "56": "0003", "57": "0003", "58": "0003", "59": "0003", "63": "0003", "64": "0003", "116": "fc000003", "249": "0000", "250": "0000", "251": "0000", "252": "0000"},
    "22": {"0": "", "3": "0000", "4": "00000000", "5": "00000000", "7": "00000000", "8": "00000000", "14": "00000000", "15": "0000", "16": "0000", "17": "0000", "18": "0000", "19": "00000000", "20": "00000000", "21": "00000000", "22": "0000000000000000", "23": "0000000000000000", "24": "0000000000000000", "25": "0000000000000000", "26": "00000000"},
    "40": {"0": "", "1": "ff000000", "2": "00000000", "3": "00000000", "10": "c000f800", "24": "00000000", "25": "00000000", "26": "00000000", "28": "0e000000", "29": "ff000000", "30": "d000f800", "43": "fff0f000", "44": "fff0f000", "45": "fff0f000", "46": "fff0f000", "47": "8f00fbf0", "48": "8f00fbf0", "49": "8f00fbf0", "50": "8f00fbf0"},
    "62": {"0": "", "1": "0000000000000000", "2": "00000000", "4": "00000000", "9": "00000000", "10": "00000000", "11": "00000000", "41": "00000000", "42": "00000000"},
    "183": {"0": "", "274": "9f00001f", "275": "9f00001f", "276": "9f00001f", "277": "ffc003ff", "278": "ffc003ff", "282": "fc000000", "283": "fc000000", "284": "ffc003ff", "285": "ffc003ff", "286": "ffc003ff", "299": "ffc003ff", "311": "9f00001f", "312": "ffc003ff"},
    "243": {"0": "", "16": "01fff07f", "17": "00000fff", "18": "000fffff00000fff", "19": "000fffff00000fff", "20": "00000fff", "23": "00000fff", "24": "000fffff", "25": "01fff07f", "26": "00000fff", "27": "000fffff", "28": "01fff07f", "44": "e383", "45": "e003"},
    "258": {"0": "", "1": "00000000", "2": "0000000000000000", "64": "fc0003ff", "65": "fc0003e0", "66": "fc000000", "67": "fe00001f", "68": "ffc003ff", "69": "fe00001f", "70": "ffc003ff", "71": "fe00001f", "72": "ffc003ff", "73": "fe00001f", "74": "ffc003ff", "75": "fe00001f", "76": "ffc003ff", "77": "fe00001f", "78": "ffc003ff", "99": "00000000"}
  }
}
```

## references/fingerprints.json

```json
{
  "schema_version": 1,
  "cve": "CVE-2026-111497",
  "verified_on": "2026-10-23",
  "upstream_fix": "https://github.com/google/brotli/commit/0ed4268a4401abcb4472096340477cf4af7d1a2b",
  "inventory_revision": "6f34a3ae446cbba851214fd6a0b40d3a6e3c786f",
  "verification": "Fetched upstream releases and pinned inventory source; verified Git blob and SHA-256, and compared block-switch handling with upstream fix. Source-file evidence only.",
  "fingerprints": [
    {
      "sha256": "26281b8a8dbe1f85fe96dcb16eeae68620cd106876b7e006ae29398afd0537ba",
      "git_blob": "08bd76ca16e2a8d6a45ec48a18c95c5fa118561f",
      "size": 89681,
      "state": "vulnerable",
      "sources": [
        {
          "label": "upstream-1.0.7",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.7/c/dec/decode.c"
        },
        {
          "label": "inventory-08bd76ca16e2a8d6a45ec48a18c95c5fa118561f",
          "url": "https://raw.githubusercontent.com/pmq20/node-packer/12c46c6e44fbc14d9ee645ebd17d5296b324f7e0/current/deps/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "2026366b7511e77ca16bafcd708299ffe33579c8c2bc7b8af66c2902a858cbd5",
      "git_blob": "114c5057d7d91f5a634167e00d84a8f4f9112c53",
      "size": 92978,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-114c5057d7d91f5a634167e00d84a8f4f9112c53",
          "url": "https://raw.githubusercontent.com/hexhive/el3xir/af0e3573a3f4317d089dd6fbad25bfadc7ba0c54/src/aflvatarplusplus/qemu_mode/qemuafl/roms/edk2/BaseTools/Source/C/BrotliCompress/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "4fd20d5f0c6d8afb007ddb90ce4b7015762ff881658798aad83a46766e589268",
      "git_blob": "220c7e85c6331bff7a0a44c55f951823b143fd56",
      "size": 103991,
      "state": "vulnerable",
      "sources": [
        {
          "label": "upstream-1.1.0",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.1.0/c/dec/decode.c"
        },
        {
          "label": "inventory-220c7e85c6331bff7a0a44c55f951823b143fd56",
          "url": "https://raw.githubusercontent.com/mozilla/rr/c145f2f80e63e9627b79deed6e0d3eae0d658e38/third-party/brotli/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "06320e9f33a9d49a1880ba0709ae609407e7f8fe9dea97ee435b1d4ad60e6faf",
      "git_blob": "278b62255e260f066208ca3e36cbaed73abc0bab",
      "size": 88563,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-278b62255e260f066208ca3e36cbaed73abc0bab",
          "url": "https://raw.githubusercontent.com/xingyun86/ppshuai_brotli/9898e334a9a749fc387048433cdef9643657852a/src/brotli/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "79db5a1879540eb7c915ce948e8fb94adc1ff49be87dca01529b9283c6ba343e",
      "git_blob": "2b45765563464d49c91ac480d6ba1e868d97f113",
      "size": 105535,
      "state": "fixed",
      "sources": [
        {
          "label": "upstream-1.1.1",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.1.1/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "48591f9bafcf49390f5d44c1a201e5121860ffdfdf1066715c315e4e56090159",
      "git_blob": "2fe58a7b2f89f76bc0f9e304ce28cc8794f74586",
      "size": 101068,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-2fe58a7b2f89f76bc0f9e304ce28cc8794f74586",
          "url": "https://raw.githubusercontent.com/moatlab/femu/2d50161f24c4cb86d58185b0efd44457412eadce/roms/edk2/BaseTools/Source/C/BrotliCompress/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "4b24ab7b9278acd6930b28965347650367c2e37e41086fe136ca54298de1c253",
      "git_blob": "3ee1963a8509674d1f4c4698592659c3e485b613",
      "size": 103111,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-3ee1963a8509674d1f4c4698592659c3e485b613",
          "url": "https://raw.githubusercontent.com/bleedingxiko/voxelbones-4.1.3/e77cdce21ceb31f02af46f29f67aed73d58d5f68/thirdparty/brotli/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "8a2f9cb02e281bcee227d9adc1484ca67b059409161826e796ff846272e22502",
      "git_blob": "630edeb73e0b93cd89189a52e90ca4fc90d1ce29",
      "size": 89078,
      "state": "vulnerable",
      "sources": [
        {
          "label": "upstream-1.0.3",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.3/c/dec/decode.c"
        },
        {
          "label": "inventory-630edeb73e0b93cd89189a52e90ca4fc90d1ce29",
          "url": "https://raw.githubusercontent.com/interactiveplus/easycrossplatform_src/3e1f33c0cf11ec46e7a7cb88802f99f4c94c7e4c/deps/brotli/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "be092f655322dbf42a4e116864f18358649f79a501fc368a17b8e225ee15a344",
      "git_blob": "674b829a32e1cc004387cc03f98a238dc58a74bc",
      "size": 88595,
      "state": "vulnerable",
      "sources": [
        {
          "label": "upstream-1.0.5",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.5/c/dec/decode.c"
        },
        {
          "label": "upstream-1.0.6",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.6/c/dec/decode.c"
        },
        {
          "label": "inventory-674b829a32e1cc004387cc03f98a238dc58a74bc",
          "url": "https://raw.githubusercontent.com/samueltulach/rainbow/43e461adb22baa6318b7c7355af98fc545ce8997/edk2/BaseTools/Source/C/BrotliCompress/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "7327740d587020895a9fbcb8c49c4c9b5f9b233f7987d9ac9ede9e0415767633",
      "git_blob": "70d594bf44875762a3f42ae6665daa2f2ab6edf8",
      "size": 107672,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-70d594bf44875762a3f42ae6665daa2f2ab6edf8",
          "url": "https://raw.githubusercontent.com/jamaika1/plugins_ffmpeg_vvc_evc_htj2k/3f96189722b72bab48945c69ae1760ff155d604b/libavcodec/libjxl/brotli/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "8d7beeafed5ff9f3f00f6e528e92ad6a36059b3549356bed5a9ca3ee16938438",
      "git_blob": "74db1a292b95b535e5a0ed2e7f7ab8ed61b9b55a",
      "size": 109324,
      "state": "fixed",
      "sources": [
        {
          "label": "upstream-1.2.1",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.2.1/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "d8ae8a954051f317ac26fe633f5ecaa0ec537296cc175a3afdb3bdb6a8dc2979",
      "git_blob": "845f556c0436b65da796a62dec7ca47b8731d9a4",
      "size": 101330,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-845f556c0436b65da796a62dec7ca47b8731d9a4",
          "url": "https://raw.githubusercontent.com/telegrammessenger/telegram-ios/fb0e62ba8a147f0821da6aa63c09da7d79a10111/third-party/libjxl/libjxl/third_party/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "77b3b1d24dd6ecba71b35d8bf10db835fd47c0f68ad0afdb51b405b832ec8a57",
      "git_blob": "86dcd5fb79f5b1a25b09ff228e75a2037a349ac5",
      "size": 88557,
      "state": "vulnerable",
      "sources": [
        {
          "label": "upstream-1.0.4",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.4/c/dec/decode.c"
        },
        {
          "label": "inventory-86dcd5fb79f5b1a25b09ff228e75a2037a349ac5",
          "url": "https://raw.githubusercontent.com/superk589/princessguide/4e2c467f265f009dc760ee299ce3790dec0106ec/Pods/BrotliKit/Dependencies/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "9ac66fb95903c2b1bcbfc298bed8472075f6a0e0f0079df4dc7155d65b3b1bfc",
      "git_blob": "9c10f50fbfacdd34e25dbd838054c40658c9da66",
      "size": 92968,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-9c10f50fbfacdd34e25dbd838054c40658c9da66",
          "url": "https://raw.githubusercontent.com/schnaader/precomp-cpp/a11b077425fb6f6c22fc39c3265a2c1af7333665/contrib/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "3c6209b73b0b41526448d2315d3a104bf58a3e46d3096574158ba93ef10480c8",
      "git_blob": "ae5a3d3fa359e53c5d641ac79b793f746e4d03b7",
      "size": 93022,
      "state": "vulnerable",
      "sources": [
        {
          "label": "upstream-1.0.8",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.8/c/dec/decode.c"
        },
        {
          "label": "upstream-1.0.9",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.9/c/dec/decode.c"
        },
        {
          "label": "inventory-ae5a3d3fa359e53c5d641ac79b793f746e4d03b7",
          "url": "https://raw.githubusercontent.com/szcnick/p7zip-hw/6819e2dc1917e1267babddc6391cea56ead7123d/C/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "44a4c1973553725c693dc38136cd32a5ec97a471bba8bf5c25bd23fa5b486542",
      "git_blob": "b23d9fb2208887eb79faaf10bd00fad745bffcfd",
      "size": 94479,
      "state": "fixed",
      "sources": [
        {
          "label": "upstream-1.0.10",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.0.10/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "5a993957e39a73857c19dd2cd12932c55c24ef3681a6f115102c014e05ea3293",
      "git_blob": "de46eb43f0f3aa43c976cc21f9e005dd449982d5",
      "size": 103973,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-de46eb43f0f3aa43c976cc21f9e005dd449982d5",
          "url": "https://raw.githubusercontent.com/inotfound/magic/f49b432737d8fe5c4745cd2a87241a00b7353f3c/ThirdParty/Brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "3ff07cde137f680a3e7081314d2e65bf2b88f05546cbd40670aee8ad0692fa0d",
      "git_blob": "e03f7b3dc9294c303eb30841e3654b877b07d7e3",
      "size": 107782,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-e03f7b3dc9294c303eb30841e3654b877b07d7e3",
          "url": "https://raw.githubusercontent.com/s1986q/mybrotli/eee8e875785904e7fa8557ebda069c1184d73703/jni/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "f7749b1296128c0add4eaa454b9d82ee1c7b0451c1c9ed47d96733e6de6f2d7b",
      "git_blob": "e87e7f5ea0f9d06acdb52fe0c08d298e7ee85357",
      "size": 107780,
      "state": "vulnerable",
      "sources": [
        {
          "label": "upstream-1.2.0",
          "url": "https://raw.githubusercontent.com/google/brotli/v1.2.0/c/dec/decode.c"
        },
        {
          "label": "inventory-e87e7f5ea0f9d06acdb52fe0c08d298e7ee85357",
          "url": "https://raw.githubusercontent.com/bunkerity/bunkerweb/b09e418447718dc9a7eb40158e3e12ce719bfd5f/src/deps/src/brotli/c/dec/decode.c"
        }
      ]
    },
    {
      "sha256": "6468095399eb044ac27e65f0612ed61a6fee0874d1e4cf0046c5cedfcd9211a3",
      "git_blob": "fc7d535a3cbdffc33fd1ed2f01c608cadde798d2",
      "size": 107666,
      "state": "vulnerable",
      "sources": [
        {
          "label": "inventory-fc7d535a3cbdffc33fd1ed2f01c608cadde798d2",
          "url": "https://raw.githubusercontent.com/utelle/wxpdfdoc/1aa2154023d7da66d9c19665564f334ecf8bfbf0/thirdparty/woff2/brotli/dec/decode.c"
        }
      ]
    }
  ]
}
```

## references/reachability.md

# Reachability guide

Use this guide for surfaces present in the target. These are investigation
routes, not assertions that a listed project or deployment is vulnerable.

For each copy, collect a concrete chain:

`artifact/revision -> selected decoder -> caller -> input entry point -> input owner`

Build selection needs a manifest, link/build evidence, or corresponding artifact
provenance. Shipment needs the deployed image/package/binary identity. Reachability
needs the call path and active configuration, including attacker influence and
authentication before decoding. A benign controlled request can establish a
decoding path when authorized; do not send malformed or exploit payloads.

## Trailmark source discovery

Use the packaged helper to find relevant callsites and callers before tracing
the actual input path:

```sh
uv run --script scripts/brotli_source_discovery.py /absolute/source > discovery.json
```

Run from the installed skill directory with trusted uv. This pins Trailmark
0.7.0; first use can download its dependencies and parser grammars. It reads
source without executing the target. Select a focused source subtree because
initial parsing and indexing are not bounded by the output limits.

For patch review in a Brotli source tree, add `--language c` and repeated
`--symbol` options for `DecodeLiteralBlockSwitchInternal`,
`DecodeCommandBlockSwitchInternal`, `DecodeDistanceBlockSwitchInternal`, and
`ProcessCommands`. This traces source callers independently of the compiled
architecture; verify build selection and unresolved edges separately.

Review `candidates[].call` for Brotli decoder calls (including zlib's Brotli
APIs), or fetch/Undici. The `neighborhood` joins callers by full node IDs and
records source spans, unresolved proxy targets, confidence, deferred edges, and relevant
entrypoint tags. Use those locations to prioritize source review, then check:

1. Resolve the actual binding. A fetch name can refer to a browser API, local
   function, or parameter. Review import aliases and wrappers; `--symbol NAME`
   adds an exact search target without asserting what it calls.
2. Follow callers to the input owner. Deferred callback edges do not prove the
   callback runs. Entry tags are heuristic or repository configured, not proof
   of attacker control or active routing.
3. Verify build selection, shipped artifact, runtime, and configuration using
   the same evidence required for a manual trace.

Keep `configured_by` on edges: `.trailmark/links.toml` can add calls with no
extracted callsite. `.trailmark/entrypoints.toml` can supply entrypoint tags.
Treat both as target-supplied declarations and independently verify them.

Trailmark's directory parser skips `vendor`, `node_modules`, `build`, `dist`,
hidden directories, common environment/cache directories, and directory
symlinks; it can follow file symlinks. The broader inventory covers different
inputs and must still run. Top-level JavaScript/TypeScript calls, ESM aliases,
dynamic calls, unsupported frameworks, and missing dependencies can be absent
without a parser diagnostic. No paths, no tags, or no diagnostics never prove
complete coverage or unreachability.

The helper bounds candidates, caller nodes/edges, traversal depth, and reported
diagnostics. Exit 3 marks parsing errors, empty source, and truncation as
incomplete. Exit 0 means collection completed within those limits. Neither exit
is a patch verdict or a runtime reachability conclusion. When tooling or source
is unavailable, state that gap and continue with direct source inspection.

## Node.js and HTTP services/clients

Search for `fetch`, `globalThis.fetch`, `undici`, `brotliDecompress`,
`createBrotliDecompress`, `node:zlib`,
`Content-Encoding`, `body-parser`, and request decoding middleware. Follow the
actual middleware order: authentication in a handler can run after automatic
decompression. Check body parsers and proxies at every receiving boundary.
Ordinary response encoding or an `Accept-Encoding: br` advertisement does not
prove vulnerable request decoding.

Inspect outbound fetch/HTTP response decompression too. Identify whether an
attacker can control an origin, redirect target, fetched document, or intermediary
input. Do not assume all clients decode automatically. Tie the executing Node
binary and its bundled decoder to the deployment; a shell's default `node`
can differ from a service, container, Electron embedder, or packaged executable.

### Node's built-in fetch

Node's [built-in `fetch()` uses Undici](https://nodejs.org/api/globals.html#fetch).
The [bundled implementation in Node v24.22.0](https://github.com/nodejs/node/blob/v24.22.0/deps/undici/src/lib/web/fetch/index.js#L2028-L2088)
pipes Brotli-encoded response bodies through `zlib.createBrotliDecompress()`.
An attacker-controlled response can therefore reach the affected decoder on an
unpatched runtime even when the app never imports Brotli or `node:zlib` itself.

Trace URL inputs, redirects, wrappers, and the actual fetch implementation.
Establish the deployed Node build and vendor patch state. A `fetch()` reference
alone does not prove an attacker-controlled URL or an unpatched runtime.
Changing `Accept-Encoding` alone does not disable response decompression: the
reviewed implementation selects its decoder from response `Content-Encoding`.
Its default HTTPS request advertises Brotli; its HTTP request does not, but a
server can still return a Brotli-encoded body.

The offline helper records JavaScript/TypeScript fetch references, global fetch
property references, Undici imports, and Undici dependency declarations. Its
JSON evidence includes source lines and columns, with at most 20 source matches
per file and an explicit truncation flag. Each lead keeps patch status and
reachability unknown until reviewed. Browser/custom bindings and literal template
text can produce extra leads. Computed names, dynamic loading, bundler rewrites,
and unavailable source can hide calls; the heuristic is not a JavaScript parser.

The inventory does not resolve bindings, trace URL ownership, or discover the running
Node version. Its `--node-version` option assesses only a separately obtained
version, and does not associate that version with an unresolved source lead.
Identify the real binary/build and follow the call path through wrappers and
redirects. An Undici import alone does not establish use of its fetch API or
automatic response decoding.

The [benign loopback probe](../scripts/verify_node_fetch.cjs) checks built-in
fetch's invocation of Node's Brotli decoder for valid HTTP response bodies,
including with an identity request header and after a redirect. It does not test
malformed streams, deployment reachability, or exploitation. See
[runtime fixtures](https://github.com/trailofbits/brotli-checker/blob/main/CONTRIBUTING.md#runtime-fixtures)
for maintainer instructions.

## Electron applications

Resolve the API and process before applying Node's version rules. Renderer
fetch, Electron `net.fetch`, and `session.fetch` use Chromium networking;
Node's main-process fetch uses Undici. Electron's
[networking documentation](https://www.electronjs.org/docs/latest/api/net#netfetchinput-init)
distinguishes these routes. Chromium also decodes Brotli inside WOFF2 fonts.

Electron can replace Node's bundled decoder with Chromium's copy. For example,
[Electron 43.1.1's build patch](https://github.com/electron/electron/blob/v43.1.1/patches/node/build_add_gn_build_files.patch#L323-L338)
links Node to Chromium's Brotli targets. Trace the target release's DEPS, patches,
selected Chromium decoder, and shipped artifact. An affected standalone Node
version or a Brotli version string alone does not classify an Electron build.
Vendor builds with the same Electron version can contain different native code.

Do not transfer Node's header behavior to every Electron API, or treat a
response-decoding error as proof that no decoder was entered. The
[runtime fixtures](https://github.com/trailofbits/brotli-checker/blob/main/CONTRIBUTING.md#runtime-fixtures)
can compare valid-response handling in a separately obtained Electron runtime
or an explicitly authorized installed Slack test instance.

For a real app, establish whether a message, preview, external URL, attachment,
font, or redirect actually delivers attacker-controlled bytes to the selected
decoder. Account for proxying, re-encoding, allowlists, authentication, and
process boundaries. A trusted debugger opening a local fixture confirms only
that tested runtime route; it does not prove an attacker-controlled product path.
Mach-O matching and ASAR parsing are outside the offline helper's supported
classifiers.

Keep app inspection read-only unless runtime testing is explicitly authorized.
For an authorized runtime test, use a fresh profile and quit the normal app
first. The maintainer Slack fixture refuses to launch over an existing main
process. Normal decoding tests must not be described as CVE trigger tests.

## Java HTTP stacks and native JNI providers

Look for `com.aayushatharva.brotli4j`, `DecoderJNI`, `BrotliDecoder`, native
classifier JARs, shaded/nested JARs, and extracted `.so`/`.dll`/`.dylib` files.
Resolve the platform-specific native provider selected at runtime; the wrapper
version is not the embedded C library version. A JVM application can call this
C decoder through JNI, while a pure Java implementation requires separate review.

The reviewed Netty source provides a concrete route:
[HttpContentDecompressor](https://github.com/netty/netty/blob/4.1/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecompressor.java)
selects a Brotli decoder for `br` when the provider is available, and
[BrotliDecoder](https://github.com/netty/netty/blob/4.1/codec/src/main/java/io/netty/handler/codec/compression/BrotliDecoder.java)
uses Brotli4j's JNI decoder. Presence of Netty or Brotli4j alone does not show
that this handler is installed, this branch/version is shipped, or native bytes
are vulnerable. Follow server and client pipelines separately.

## WOFF2, fonts, document and image processing

[WOFF2's decoder](https://github.com/google/woff2/blob/master/src/woff2_dec.cc)
calls `BrotliDecoderDecompress` while expanding font data. Trace font uploads,
document conversion, preview rendering, browser/font sanitizers, and asynchronous
jobs to the decoder actually linked by the worker. Encoding fonts is a different
operation. A pinned submodule in the inventory does not establish that a product
ships that submodule instead of a patched system library.

Record one-shot versus streaming calls, input segmentation, length checks, and
the process sandbox. Reaching a public decompression API does not itself prove
the malformed-input/error-buffer path or a usable corruption primitive. Do not
assert that a one-shot call is safe or exploitable without implementation evidence.

## workerd and other embedders

[workerd's system streams](https://github.com/cloudflare/workerd/blob/main/src/workerd/api/system-streams.c++)
include KJ Brotli input streams and content-encoding selection under a compatibility
flag. Follow the target revision through KJ/Cap'n Proto, Bazel/module downloads,
selected Brotli inputs, and the deployed workerd binary. Inspect the active
compatibility flags and the code which supplies each stream. Source support
does not establish a cloud service's deployed version or tenant isolation.

## TLS certificate compression — investigate per implementation

[RFC 8879](https://www.rfc-editor.org/rfc/rfc8879.html) defines Brotli as one TLS
certificate compression algorithm. The receiver decompresses the certificate
message before normal certificate parsing and verification. As an inference,
certificate validation cannot protect a decoder invoked earlier in that path;
the actual implementation and negotiation must still be established.

Check both client-side decoding of server certificates and, where implemented
and negotiated, server-side decoding of client certificates. Identify TLS 1.3
support, advertised/selected compression algorithms, registration of a Brotli
decompression callback, its linked library, and attacker control of the peer.
BoringSSL's
[certificate compression extension logic](https://github.com/google/boringssl/blob/main/ssl/extensions.cc)
uses registered compression/decompression callbacks; merely linking BoringSSL
does not establish that a Brotli decoder is registered. Search for
`SSL_CTX_add_cert_compression_alg`, `TLSEXT_cert_compression_brotli`,
`CompressedCertificate`, and `BrotliDecoder` at the call sites and build inputs.

Distinguish certificate compression from HTTP body compression and historical
TLS record compression. A successful TLS handshake or a trusted-certificate
policy alone does not establish this path's patch status or eliminate the
pre-verification input surface. Do not claim universal unauthenticated server
exposure: protocol role, negotiation, provider, buffering, and configuration matter.

## Bounded binary patch triage

When the inventory leaves a local ELF copy unknown, the skill can collect a
bounded, read-only disassembly packet:

Before disassembly, inspect `evidence.decoder_metadata`. The inventory reads ELF
symbols and supported constant-return `BrotliDecoderVersion` definitions directly,
without invoking target code. Defined decoder APIs establish embedded symbol
evidence; imports point to another provider. An old supported-branch version is a
`potentially_vulnerable` update lead. A fixed reported version, symbol definition,
or import never establishes fixed code and does not clear other copies.

```sh
python3 scripts/brotli_binary_triage.py /absolute/target > triage.json
```

Use a trusted local `objdump`; never use a tool or script from the target. The
helper rejects target symlinks and streams the original artifact SHA-256 while
using sparse reads for ELF structures. Only bounded selected code fragments are
copied for raw-binary objdump, with original virtual addresses preserved. It
records each fragment's hash/source offset, exact command, scope, and limits.
The original regular file is not copied or memory-mapped. Changes detected on the
open input invalidate collection. It neither loads nor executes the target and
removes all fragments afterward. Raw output's `.data` label is not the source ELF
section; use recorded scopes and symbol addresses for numeric call targets.
Treat exit 0 as successful
evidence collection only. Treat exit 3, a timeout/output cap, unsupported ELF,
missing relevant symbols, or truncated symbol selection as explicitly incomplete.

Prefer the selected internal block-switch and `ProcessCommands*` symbols. The
public `BrotliDecoderDecompressStream` fallback can orient a stripped or partially
symbolized build but may be too large or may hide inlined callees. If relevant
symbols are absent, request an address range from trusted link/map evidence, or
translate a scanner-reported executable file offset with `--file-offset OFFSET
--length BYTES`. Use `--start-address` and `--stop-address`, optionally constrained
by `--section`, only when those addresses have trusted provenance. A string's file
offset is not code provenance.

Compare the packet with both reviewed sides of the decoder patch. Trace all three
block-switch classes (literal, command, distance), including safe variants, from
the invalid `max_block_type` condition to `ProcessCommands*` and its caller. The
fixed path preserves `BROTLI_DECODER_ERROR_FORMAT_BLOCK_SWITCH` as a
`BrotliDecoderErrorCode`; the affected path's boolean/void handling can lose that
specific status. Check return-value tests and propagation in control flow rather
than looking only for a constant, call target, symbol, or string. Compiler
optimization can inline, clone, merge, or remove these functions.

Keep patch status `unknown` when the packet omits a relevant caller/callee, the
control-flow mapping is ambiguous, output was bounded early, or multiple decoder
copies are not all accounted for. Report an affected conclusion only when the old
status-loss behavior is affirmatively traced. Do not report a fixed binary merely
from a version-return function, fixed-version string, import/export, or one
fixed-looking range. Patch evidence, shipment, runtime selection, and reachability
remain separate conclusions.

## Reporting unresolved cases

If only source is present, leave build selection and shipment unknown. If an
artifact is known but native provenance is missing, leave patch status unknown.
If the caller or active configuration is missing, leave reachability unknown.
Record the next missing evidence for each row so another reviewer can close the
specific gap. Avoid calling an entire application fixed based on one source file,
one package, or one reviewed route.

## scripts

```

```

## scripts/brotli_binary_data.py

```python
"""Sparse, read-only access to binary inputs with explicit temporary lifetimes."""

from __future__ import annotations

import errno
import hashlib
import io
import os
import stat
import tempfile
from contextlib import contextmanager
from dataclasses import dataclass
from typing import TYPE_CHECKING, BinaryIO, overload

if TYPE_CHECKING:
    from collections.abc import Generator

SMALL_BYTES_LIMIT = 8 * 1024 * 1024
COPY_CHUNK_BYTES = 1024 * 1024
SEARCH_CHUNK_BYTES = 4 * 1024 * 1024
MAX_BYTE = 0xFF


class DataLimitError(ValueError):
    """A stream or regular file exceeded an explicit byte limit."""

    def __init__(self, limit: int, bytes_read: int) -> None:
        super().__init__(f"Binary input exceeds the explicit {limit}-byte limit")
        self.limit = limit
        self.bytes_read = bytes_read


class DataChangedError(OSError):
    """An open input changed while it was being inspected."""


@dataclass(frozen=True)
class _FileSignature:
    device: int
    inode: int
    size: int
    modified_ns: int
    changed_ns: int

    @classmethod
    def from_stat(cls, metadata: os.stat_result) -> _FileSignature:
        return cls(
            device=metadata.st_dev,
            inode=metadata.st_ino,
            size=metadata.st_size,
            modified_ns=metadata.st_mtime_ns,
            changed_ns=metadata.st_ctime_ns,
        )


class FileData:
    """Read-only, pread-backed bytes and file access over one stable regular-file fd."""

    def __init__(self, descriptor: int, signature: _FileSignature) -> None:
        self._descriptor = descriptor
        self._signature = signature
        self._position = 0

    @classmethod
    def from_stream(cls, stream: BinaryIO, max_bytes: int | None = None) -> FileData:
        _validate_limit(max_bytes)
        metadata = os.fstat(stream.fileno())
        if not stat.S_ISREG(metadata.st_mode):
            msg = "Binary input is not a regular file"
            raise ValueError(msg)
        if max_bytes is not None and metadata.st_size > max_bytes:
            raise DataLimitError(max_bytes, 0)
        descriptor = os.dup(stream.fileno())
        try:
            signature = _FileSignature.from_stat(os.fstat(descriptor))
        except BaseException:
            os.close(descriptor)
            raise
        if max_bytes is not None and signature.size > max_bytes:
            os.close(descriptor)
            raise DataLimitError(max_bytes, 0)
        return cls(descriptor, signature)

    def __len__(self) -> int:
        return self._signature.size

    @overload
    def __getitem__(self, key: int) -> int: ...

    @overload
    def __getitem__(self, key: slice) -> bytes: ...

    def __getitem__(self, key: int | slice) -> int | bytes:
        if isinstance(key, int):
            offset = key + len(self) if key < 0 else key
            if not 0 <= offset < len(self):
                raise IndexError(key)
            return self.read_at(offset, 1)[0]
        start, stop, step = key.indices(len(self))
        indexes = range(start, stop, step)
        if not indexes:
            return b""
        if step > 0:
            content = self.read_at(start, stop - start)
            return content if step == 1 else content[::step]
        lowest = indexes[-1]
        content = self.read_at(lowest, start - lowest + 1)
        return bytes(content[index - lowest] for index in indexes)

    def __contains__(self, needle: object) -> bool:
        if isinstance(needle, int):
            if not 0 <= needle <= MAX_BYTE:
                msg = "byte must be in range(0, 256)"
                raise ValueError(msg)
            return self.find(bytes((needle,))) >= 0
        if isinstance(needle, (bytes, bytearray, memoryview)):
            return self.find(bytes(needle)) >= 0
        msg = f"a bytes-like object is required, not {type(needle).__name__}"
        raise TypeError(msg)

    def read_at(self, offset: int, size: int) -> bytes:
        if offset < 0 or size < 0 or offset > len(self) or size > len(self) - offset:
            msg = "Binary range outside file"
            raise ValueError(msg)
        chunks = []
        remaining = size
        position = offset
        while remaining:
            chunk = os.pread(self.fileno(), min(COPY_CHUNK_BYTES, remaining), position)
            if not chunk:
                self.check_unchanged()
                msg = "Binary input ended while a validated range was being read"
                raise DataChangedError(msg)
            chunks.append(chunk)
            position += len(chunk)
            remaining -= len(chunk)
        return b"".join(chunks)

    def find(  # noqa: C901 -- Bytes-compatible bounds plus chunk-overlap search.
        self, sub: bytes, start: int = 0, end: int | None = None
    ) -> int:
        needle = bytes(sub)
        length = len(self)
        if start < 0:
            start = max(0, length + start)
        elif start > length:
            return -1
        if end is None:
            end = length
        elif end < 0:
            end = max(0, length + end)
        else:
            end = min(end, length)
        if end < start:
            return -1
        if not needle:
            return start
        overlap = len(needle) - 1
        tail = b""
        position = start
        while position < end:
            chunk = os.pread(self.fileno(), min(SEARCH_CHUNK_BYTES, end - position), position)
            if not chunk:
                self.check_unchanged()
                return -1
            candidate = tail + chunk
            found = candidate.find(needle)
            if found >= 0:
                offset = position - len(tail) + found
                if offset + len(needle) <= end:
                    return offset
            tail = candidate[-overlap:] if overlap else b""
            position += len(chunk)
        return -1

    def startswith(
        self, prefix: bytes | tuple[bytes, ...], start: int = 0, end: int | None = None
    ) -> bool:
        if isinstance(prefix, tuple):
            return any(self.startswith(item, start, end) for item in prefix)
        if start > len(self):
            return False
        normalized_start, normalized_stop, _ = slice(start, end).indices(len(self))
        if len(prefix) > normalized_stop - normalized_start:
            return False
        return (
            self.find(prefix, normalized_start, normalized_start + len(prefix)) == normalized_start
        )

    def read(self, size: int = -1) -> bytes:
        if self._position >= len(self):
            return b""
        if size is None or size < 0:
            size = len(self) - self._position
        else:
            size = max(0, min(size, len(self) - self._position))
        content = self.read_at(self._position, size)
        self._position += len(content)
        return content

    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
        if whence == io.SEEK_SET:
            position = offset
        elif whence == io.SEEK_CUR:
            position = self._position + offset
        elif whence == io.SEEK_END:
            position = len(self) + offset
        else:
            msg = f"Invalid whence: {whence}"
            raise ValueError(msg)
        if position < 0:
            raise OSError(errno.EINVAL, "Invalid argument")
        self._position = position
        return position

    def tell(self) -> int:
        return self._position

    def fileno(self) -> int:
        if self._descriptor < 0:
            msg = "I/O operation on closed binary data"
            raise ValueError(msg)
        return self._descriptor

    def readable(self) -> bool:
        return True

    def seekable(self) -> bool:
        return True

    def writable(self) -> bool:
        return False

    @property
    def closed(self) -> bool:
        return self._descriptor < 0

    def check_unchanged(self) -> None:
        current = _FileSignature.from_stat(os.fstat(self.fileno()))
        if current != self._signature:
            msg = "Binary input changed while it was being inspected"
            raise DataChangedError(msg)

    def close(self) -> None:
        if self._descriptor >= 0:
            os.close(self._descriptor)
            self._descriptor = -1


type Buffer = bytes | FileData


def _validate_limit(max_bytes: int | None) -> None:
    if max_bytes is not None and max_bytes < 0:
        msg = "max_bytes must be nonnegative or None"
        raise ValueError(msg)


def _read_size(total: int, max_bytes: int | None) -> int:
    if max_bytes is None:
        return COPY_CHUNK_BYTES
    return min(COPY_CHUNK_BYTES, max_bytes - total + 1)


@contextmanager
def file_data(stream: BinaryIO, max_bytes: int | None = None) -> Generator[FileData, None, None]:
    """Yield sparse access to an already-open regular file without reading it eagerly."""
    data = FileData.from_stream(stream, max_bytes)
    completed = False
    try:
        yield data
        completed = True
    finally:
        try:
            if completed:
                data.check_unchanged()
        finally:
            data.close()


@contextmanager
def read_data(stream: BinaryIO, max_bytes: int | None = None) -> Generator[Buffer, None, None]:
    """Read a sequential stream as small bytes or a temporary pread-backed file."""
    _validate_limit(max_bytes)
    buffered = bytearray()
    temporary: BinaryIO | None = None
    total = 0
    try:
        while True:
            chunk = stream.read(_read_size(total, max_bytes))
            if not chunk:
                break
            total += len(chunk)
            if max_bytes is not None and total > max_bytes:
                raise DataLimitError(max_bytes, total)
            if temporary is None and total <= SMALL_BYTES_LIMIT:
                buffered.extend(chunk)
                continue
            if temporary is None:
                temporary = tempfile.TemporaryFile(  # noqa: SIM115 -- Closed after yielded data.
                    prefix="brotli-binary-"
                )
                temporary.write(buffered)
                buffered.clear()
            temporary.write(chunk)
        if temporary is None:
            yield bytes(buffered)
            return
        temporary.flush()
        with file_data(temporary) as data:
            yield data
    finally:
        if temporary is not None:
            temporary.close()


def sha256_data(data: Buffer) -> str:
    """Hash bytes or an open regular file in fixed-size chunks."""
    digest = hashlib.sha256()
    if isinstance(data, FileData):
        data.check_unchanged()
        for offset in range(0, len(data), COPY_CHUNK_BYTES):
            digest.update(data.read_at(offset, min(COPY_CHUNK_BYTES, len(data) - offset)))
        data.check_unchanged()
        return digest.hexdigest()
    view = memoryview(data)
    try:
        for offset in range(0, len(view), COPY_CHUNK_BYTES):
            digest.update(view[offset : offset + COPY_CHUNK_BYTES])
    finally:
        view.release()
    return digest.hexdigest()
```

## scripts/brotli_binary_inventory.py

```python
#!/usr/bin/env python3

"""Read-only binary inventory with compiled ELF decoder code matching."""

from __future__ import annotations

import argparse
import hashlib
import json
import sys
from pathlib import Path

from brotli_binary_data import Buffer, sha256_data
from brotli_binary_match import (
    REFERENCES,
    CodeReference,
    load_references,
)
from brotli_inventory import (
    DEFAULT_MAX_EXPANDED_BYTES,
    DEFAULT_MAX_FILE_BYTES,
    DEFAULT_MAX_TOTAL_BYTES,
    DEFAULT_MAX_TOTAL_EXPANDED_BYTES,
    Inventory,
    archive_kind,
    assess_binary,
    is_binary_name_candidate,
    positive,
)

MAGICS = {
    b"\x7fELF": "ELF",
    b"MZ": "PE candidate",
    b"\xfe\xed\xfa\xce": "Mach-O",
    b"\xce\xfa\xed\xfe": "Mach-O",
    b"\xfe\xed\xfa\xcf": "Mach-O",
    b"\xcf\xfa\xed\xfe": "Mach-O",
    b"\xca\xfe\xba\xbe": "Mach-O fat or Java class",
    b"\xbe\xba\xfe\xca": "Mach-O fat",
    b"\xca\xfe\xba\xbf": "Mach-O fat",
    b"\xbf\xba\xfe\xca": "Mach-O fat",
    b"\0asm": "WebAssembly",
    b"!<arch>\n": "ar static library",
}


class BinaryInventory(Inventory):
    def __init__(  # noqa: PLR0913 -- Traversal and expansion limits are public CLI controls.
        self,
        catalog: dict[str, str],
        references: list[CodeReference],
        *,
        max_file_bytes: int | None,
        max_total_bytes: int | None,
        max_files: int,
        archive_depth: int,
        max_expanded_bytes: int = DEFAULT_MAX_EXPANDED_BYTES,
        max_total_expanded_bytes: int = DEFAULT_MAX_TOTAL_EXPANDED_BYTES,
    ) -> None:
        super().__init__(
            {"fingerprints": []},
            max_file_bytes,
            max_total_bytes,
            max_files,
            archive_depth,
            max_expanded_bytes=max_expanded_bytes,
            max_total_expanded_bytes=max_total_expanded_bytes,
            code_references=None,
        )
        self.catalog = catalog
        self.references = references

    def record_name_fallback(self, label: str, size_bytes: int | None = None) -> None:
        if not is_binary_name_candidate(label):
            return
        self.findings.append(
            self._finding(
                label,
                kind="artifact_name_candidate",
                size_bytes=size_bytes,
                code_assessment="Content was not inspected; filename only",
            )
        )

    @staticmethod
    def _finding(  # noqa: PLR0913 -- Stable schema has independent evidence fields.
        path: str,
        *,
        kind: str,
        sha256: str | None = None,
        size_bytes: int | None = None,
        format_hint: str | None = None,
        indicators: list[str] | None = None,
        catalog_label: str | None = None,
        patch_status: str = "not_assessed",
        linkage: str = "unknown",
        code_matches: list[dict] | None = None,
        decoder_metadata: dict | None = None,
        code_assessment: str,
    ) -> dict:
        return {
            "path": path,
            "kind": kind,
            "sha256": sha256,
            "size_bytes": size_bytes,
            "format_hint": format_hint,
            "indicators": indicators or [],
            "catalog_label": catalog_label,
            "patch_status": patch_status,
            "linkage": linkage,
            "code_matches": code_matches or [],
            "decoder_metadata": decoder_metadata or {},
            "code_assessment": code_assessment,
        }

    def inspect(self, data: Buffer, label: str, depth: int) -> None:
        digest = sha256_data(data)
        file_format = next((name for magic, name in MAGICS.items() if data.startswith(magic)), None)
        catalog_label = self.catalog.get(digest)
        assessment = assess_binary(data, self.references)
        for reason in assessment["gaps"]:
            self.gap(label, reason)
        named = is_binary_name_candidate(label)
        if file_format or named or catalog_label is not None:
            self.findings.append(
                self._finding(
                    label,
                    kind="exact_artifact_match"
                    if catalog_label is not None
                    else assessment["kind"],
                    sha256=digest,
                    size_bytes=len(data),
                    format_hint=file_format,
                    indicators=assessment["indicators"],
                    catalog_label=catalog_label,
                    patch_status="not_assessed"
                    if assessment["patch_status"] == "unknown"
                    else assessment["patch_status"],
                    linkage=assessment["linkage"],
                    code_matches=assessment["code_matches"],
                    decoder_metadata=assessment["decoder_metadata"],
                    code_assessment=assessment["reason"],
                )
            )
        kind = archive_kind(data, label)
        if kind is not None:
            self._inspect_container(data, label, depth, kind, digest)


def load_catalog(path: Path | None) -> dict[str, str]:
    if path is None:
        return {}
    catalog = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(catalog, dict) or any(
        not isinstance(key, str)
        or len(key) != hashlib.sha256().digest_size * 2
        or any(character not in "0123456789abcdef" for character in key)
        or not isinstance(value, str)
        or not value.strip()
        for key, value in catalog.items()
    ):
        message = "Catalog must map lowercase SHA-256 hex digests to nonempty artifact labels."
        raise ValueError(message)
    return catalog


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("roots", nargs="+", type=Path)
    parser.add_argument(
        "--catalog", type=Path, help="JSON mapping of SHA-256 to known artifact labels"
    )
    parser.add_argument(
        "--code-references",
        type=Path,
        default=REFERENCES,
        help="Compiled decoder reference catalog (defaults to packaged references)",
    )
    parser.add_argument(
        "--max-file-bytes",
        type=positive,
        default=DEFAULT_MAX_FILE_BYTES,
        help="optional maximum bytes read from one file/member (default: no cap)",
    )
    parser.add_argument(
        "--max-total-bytes",
        type=positive,
        default=DEFAULT_MAX_TOTAL_BYTES,
        help="optional maximum cumulative bytes read (default: no cap)",
    )
    parser.add_argument(
        "--max-expanded-bytes",
        type=positive,
        default=DEFAULT_MAX_EXPANDED_BYTES,
        help="maximum expanded bytes read from one archive member/stream",
    )
    parser.add_argument(
        "--max-total-expanded-bytes",
        type=positive,
        default=DEFAULT_MAX_TOTAL_EXPANDED_BYTES,
        help="maximum cumulative expanded archive/member bytes",
    )
    parser.add_argument(
        "--max-files",
        type=positive,
        default=100000,
        help="maximum files/archive members considered (default: 100000)",
    )
    parser.add_argument(
        "--archive-depth",
        type=int,
        choices=range(6),
        default=2,
        help="maximum nested archive levels, from 0 to 5 (default: 2)",
    )
    args = parser.parse_args(argv)
    try:
        catalog = load_catalog(args.catalog)
        references = load_references(args.code_references)
    except (OSError, ValueError) as error:
        parser.error(str(error))
    inventory = BinaryInventory(
        catalog,
        references,
        max_file_bytes=args.max_file_bytes,
        max_total_bytes=args.max_total_bytes,
        max_files=args.max_files,
        archive_depth=args.archive_depth,
        max_expanded_bytes=args.max_expanded_bytes,
        max_total_expanded_bytes=args.max_total_expanded_bytes,
    )
    roots = [root.expanduser().absolute() for root in args.roots]
    inventory.scan_roots(roots)
    report = {
        "schema_version": 2,
        "scope": [str(root) for root in roots],
        "findings": inventory.findings,
        "coverage": {
            "files_read": inventory.files_read,
            "files_seen": inventory.files_seen,
            "bytes_read": inventory.bytes_read,
            "expanded_items_read": inventory.expanded_items_read,
            "expanded_bytes_read": inventory.expanded_bytes_read,
            "archives_attempted": inventory.archives_attempted,
            "archives_opened": inventory.archives_opened,
            "gap_count": inventory.gap_count,
            "gap_details_retained": len(inventory.gaps),
            "gap_details_omitted": inventory.gap_count - len(inventory.gaps),
            "gaps": inventory.gaps,
            "excluded": [".git"],
            "symlinks_followed": False,
            "limits": {
                key: getattr(args, key)
                for key in (
                    "max_file_bytes",
                    "max_total_bytes",
                    "max_expanded_bytes",
                    "max_total_expanded_bytes",
                    "max_files",
                    "archive_depth",
                )
            },
        },
        "limitations": [
            "Old reported versions support warnings, not proof of vulnerable code or a fix.",
            "Strings and artifact catalog labels do not establish code patch status.",
            "Code matches cover reference decoder text only, not all copies or runtime use.",
            "Relocation targets and referenced data are not verified by masked code matching.",
            "Unmatched code remains unidentified; decoder metadata may still support warnings.",
            "Code reference provenance/state is trusted input; custom references require review.",
            "No matches does not establish absence of Brotli, especially in stripped binaries.",
            "Catalog matches identify exact bytes only; supplied labels are not verified.",
            "Archive member paths use ! notation and must not be passed to filesystem commands.",
            "No files are executed, loaded, extracted, moved, or quarantined.",
        ],
    }
    print(json.dumps(report, indent=2, ensure_ascii=True))
    return 3 if inventory.gap_count else (1 if inventory.findings else 0)


# gremlin: pardon[out_of_scope] CLI bootstrap guard; imports must not launch argparse.
if __name__ == "__main__":
    sys.exit(main())
```

## scripts/brotli_binary_match.py

```python
"""Conservative ELF code identity matching; never executes the inspected artifact."""

from __future__ import annotations

import gzip
import hashlib
import json
import re
import struct
import zlib
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Literal

from brotli_binary_data import Buffer, sha256_data

if TYPE_CHECKING:
    from collections.abc import Iterator

REFERENCES = Path(__file__).resolve().parent.parent / "references/binary-fingerprints.json.gz"
RELOCATION_MASKS = json.loads((REFERENCES.parent / "elf-relocations.json").read_text())["machines"]
ELF_HEADER = struct.Struct("<16sHHIQQQIHHHHHH")
PROGRAM_HEADER = struct.Struct("<IIQQQQQQ")
SECTION_HEADER = struct.Struct("<IIQQQQIIQQ")
RELOCATION = struct.Struct("<QQq")
EM_X86_64 = 62
EM_AARCH64 = 183
EM_386 = 3
EM_MIPS = 8
EM_PPC64 = 21
EM_S390 = 22
EM_ARM = 40
EM_RISCV = 243
EM_LOONGARCH = 258
SHT_RELA = 4
PN_XNUM = 0xFFFF
# ELF machine, word size, and byte order are separate parts of reference identity.
TARGETS = {
    (EM_386, 32, "little"): ("i386", "i386"),
    (EM_X86_64, 64, "little"): ("x86_64", "i386:x86-64"),
    (EM_ARM, 32, "little"): ("arm", "arm"),
    (EM_AARCH64, 64, "little"): ("aarch64", "aarch64"),
    (EM_PPC64, 64, "little"): ("ppc64le", "powerpc:common64"),
    (EM_PPC64, 64, "big"): ("ppc64", "powerpc:common64"),
    (EM_S390, 64, "big"): ("s390x", "s390:64-bit"),
    (EM_RISCV, 32, "little"): ("riscv32", "riscv:rv32"),
    (EM_RISCV, 64, "little"): ("riscv64", "riscv:rv64"),
    (EM_MIPS, 32, "little"): ("mipsel", "mips:isa32r2"),
    (EM_MIPS, 32, "big"): ("mips", "mips:isa32r2"),
    (EM_MIPS, 64, "little"): ("mips64el", "mips:isa64r2"),
    (EM_MIPS, 64, "big"): ("mips64", "mips:isa64r2"),
    (EM_LOONGARCH, 64, "little"): ("loongarch64", "loongarch64"),
}
MIN_CODE_BYTES = 256
MIN_ANCHOR_BYTES = 32
MAX_REFERENCE_BYTES = 64 * 1024 * 1024
MAX_MATCHES = 128
MAX_CANDIDATES = 4096
MAX_CODE_SECTIONS = 16
STRUCT_BATCH_BYTES = 1024 * 1024
ELF32_BITS = 32
ELF64_BITS = 64
GOT_OPCODE_BYTES = 2
X86_GOT_OPCODES = (0x8B, 0x03, 0x0B, 0x13, 0x1B, 0x23, 0x2B, 0x33, 0x3B, 0x85, 0xFF)
X86_GOT_RELOCATIONS = {EM_386: (43,), EM_X86_64: (41, 42)}
R_X86_64_REX_GOTPCRELX = 42
R_390_GOTENT = 26
R_MIPS_GPREL16 = 7
R_PPC64_REL24 = 10
R_PPC64_GOT16_LO_DS = 59
R_PPC64_TOC16_LO_DS = 64
R_PPC64_REL16_HA = 252


@dataclass(frozen=True)
class ELFLayout:
    bits: int
    byteorder: Literal["little", "big"]

    @classmethod
    def of(cls, header: tuple) -> ELFLayout:
        return cls(32 if header[0][4] == 1 else 64, "little" if header[0][5] == 1 else "big")

    def parser(self, kind: str) -> struct.Struct:
        formats = {
            "header": ("16sHHIIIIIHHHHHH", "16sHHIQQQIHHHHHH"),
            "program": ("IIIIIIII", "IIQQQQQQ"),
            "section": ("IIIIIIIIII", "IIQQQQIIQQ"),
            "symbol": ("IIIBBH", "IBBHQQ"),
            "rel": ("II", "QQ"),
            "rela": ("IIi", "QQq"),
        }
        return struct.Struct(
            ("<" if self.byteorder == "little" else ">") + formats[kind][self.bits == ELF64_BITS]
        )


def target_identity(header: tuple) -> tuple[int, int, str]:
    layout = ELFLayout.of(header)
    return header[2], layout.bits, layout.byteorder


def program_headers(data: Buffer, header: tuple) -> Iterator[tuple]:
    layout = ELFLayout.of(header)
    parser = layout.parser("program")
    if header[9] != parser.size or not 0 < header[10] < PN_XNUM:
        msg = "Unsupported ELF program header table"
        raise ValueError(msg)
    for row in iter_structs(data, parser, header[5], header[9] * header[10]):
        if layout.bits == ELF32_BITS:
            kind, offset, virtual, physical, size, memory, flags, alignment = row
            yield (kind, flags, offset, virtual, physical, size, memory, alignment)
        else:
            yield row


def section_headers(data: Buffer, header: tuple) -> list[tuple]:
    parser = ELFLayout.of(header).parser("section")
    if header[11] != parser.size:
        msg = "Unsupported ELF section table"
        raise ValueError(msg)
    return list(iter_structs(data, parser, header[6], header[11] * header[12]))


def symbol_rows(data: Buffer, header: tuple, section: tuple) -> Iterator[tuple]:
    layout = ELFLayout.of(header)
    for row in iter_structs(data, layout.parser("symbol"), section[4], section[5]):
        if layout.bits == ELF32_BITS:
            name, address, size, info, other, index = row
            yield (name, info, other, index, address, size)
        else:
            yield row


def checked_range(data: Buffer, offset: int, size: int) -> tuple[int, int]:
    if offset < 0 or size < 0 or offset > len(data) or size > len(data) - offset:
        msg = "ELF range outside file"
        raise ValueError(msg)
    return offset, offset + size


def checked_slice(data: Buffer, offset: int, size: int) -> bytes:
    start, end = checked_range(data, offset, size)
    return data[start:end]


def iter_structs(data: Buffer, parser: struct.Struct, offset: int, size: int) -> Iterator[tuple]:
    checked_range(data, offset, size)
    if size % parser.size:
        msg = "ELF structure table has a partial entry"
        raise ValueError(msg)
    batch_size = max(parser.size, STRUCT_BATCH_BYTES // parser.size * parser.size)
    end = offset + size
    for start in range(offset, end, batch_size):
        yield from parser.iter_unpack(checked_slice(data, start, min(batch_size, end - start)))


def elf_header(data: Buffer) -> tuple:
    ident = checked_slice(data, 0, 16)
    if ident[:4] != b"\x7fELF" or ident[4] not in (1, 2) or ident[5] not in (1, 2) or ident[6] != 1:
        msg = "Expected ELF32/ELF64 with a supported byte order"
        raise ValueError(msg)
    layout = ELFLayout.of((ident,))
    parser = layout.parser("header")
    header = parser.unpack(checked_slice(data, 0, parser.size))
    ident, _, machine, version = header[:4]
    if (
        (machine, layout.bits, layout.byteorder) not in TARGETS
        or version != 1
        or header[8] != parser.size
    ):
        msg = "Unsupported ELF architecture, word size, byte order, or header version"
        raise ValueError(msg)
    return header


def _file_backed_load_segments(data: Buffer) -> tuple[int, list[tuple[int, int, int]]]:
    header = elf_header(data)
    if header[1] not in (2, 3):
        msg = "Code matching requires a linked ELF executable or shared object"
        raise ValueError(msg)
    segments = []
    for kind, flags, start, _, _, size, memory_size, _ in program_headers(data, header):
        if kind == 1:
            checked_range(data, start, size)
            if size > memory_size:
                msg = "ELF segment file size exceeds memory size"
                raise ValueError(msg)
            if size:
                segments.append((start, size, flags))
    return header[2], sorted(set(segments))


def file_backed_ranges(data: Buffer) -> tuple[int, list[tuple[int, int]]]:
    machine, segments = _file_backed_load_segments(data)
    ranges = sorted({(start, size) for start, size, _ in segments})
    if not ranges:
        msg = "No file-backed ELF load segment"
        raise ValueError(msg)
    return machine, ranges


def executable_ranges(data: Buffer) -> tuple[int, list[tuple[int, int]]]:
    machine, segments = _file_backed_load_segments(data)
    ranges = sorted({(start, size) for start, size, flags in segments if flags & 1})
    if not ranges:
        msg = "No file-backed executable ELF load segment"
        raise ValueError(msg)
    return machine, ranges


def validate_metadata(entry: dict) -> None:
    for field in ("id", "source_sha256", "object_sha256", "source", "compiler", "flags"):
        if not isinstance(entry.get(field), str) or not entry[field].strip():
            msg = f"Invalid code reference {field}"
            raise ValueError(msg)
    for field in ("source_sha256", "object_sha256"):
        value = entry[field]
        if len(value) != hashlib.sha256().digest_size * 2 or any(
            c not in "0123456789abcdef" for c in value
        ):
            msg = f"Invalid code reference {field}"
            raise ValueError(msg)
    if (
        entry.get("state") not in ("fixed", "vulnerable")
        or type(entry.get("machine")) is not int
        or type(entry.get("elf_class", 64)) is not int
        or not isinstance(entry.get("byteorder", "little"), str)
        or (entry["machine"], entry.get("elf_class", 64), entry.get("byteorder", "little"))
        not in TARGETS
    ):
        msg = "Invalid code reference state or machine"
        raise ValueError(msg)


@dataclass(frozen=True)
class CodeReference:
    metadata: dict
    code: bytes
    mask: bytes
    anchor_offset: int
    anchor: bytes
    additional_sections: tuple[CodeReference, ...] = ()

    @classmethod
    def from_json(cls, entry: object, *, fragment: bool = False) -> CodeReference:
        if not isinstance(entry, dict):
            msg = "Code reference must be an object"
            raise ValueError(msg)  # noqa: TRY004 -- Malformed catalog is a CLI value error.
        validate_metadata(entry)
        try:
            code = bytes.fromhex(entry["code"])
            mask = bytes.fromhex(entry["mask"])
        except (KeyError, TypeError, ValueError) as error:
            msg = "Invalid code reference bytes"
            raise ValueError(msg) from error
        if len(code) < (1 if fragment else MIN_CODE_BYTES) or len(code) != len(mask):
            msg = "Code reference must contain equal code/mask lengths of at least 256 bytes"
            raise ValueError(msg)
        # An exact run anchors the search; every other unmasked bit is also checked.
        if not fragment and (
            b"\xff" * MIN_ANCHOR_BYTES not in mask
            or int.from_bytes(mask).bit_count() < len(mask) * 6
        ):
            msg = "Code reference masks too much code or has no 32-byte exact anchor"
            raise ValueError(msg)
        runs = list(re.finditer(b"\xff+", mask))
        if not runs:
            msg = "Code section has no exact anchor"
            raise ValueError(msg)
        run = max(runs, key=lambda match: match.end() - match.start())
        anchor_offset = run.start()
        anchor_size = min(MIN_ANCHOR_BYTES, run.end() - run.start())
        metadata = {
            key: value
            for key, value in entry.items()
            if key not in ("code", "mask", "additional_sections")
        }
        extra = entry.get("additional_sections", [])
        if (
            not isinstance(extra, list)
            or len(extra) >= MAX_CODE_SECTIONS
            or (fragment and extra)
            or any(not isinstance(part, dict) or set(part) != {"code", "mask"} for part in extra)
        ):
            msg = "Invalid additional executable sections"
            raise ValueError(msg)
        sections = tuple(cls.from_json({**metadata, **part}, fragment=True) for part in extra)
        return cls(
            metadata,
            code,
            mask,
            anchor_offset,
            code[anchor_offset : anchor_offset + anchor_size],
            sections,
        )

    def to_json(self) -> dict:
        entry = {**self.metadata, "code": self.code.hex(), "mask": self.mask.hex()}
        if self.additional_sections:
            entry["additional_sections"] = [
                {"code": part.code.hex(), "mask": part.mask.hex()}
                for part in self.additional_sections
            ]
        return entry


def load_references(path: Path) -> list[CodeReference]:
    with path.open("rb") as stream:
        compressed = stream.peek(2)[:2] == b"\x1f\x8b"
        if compressed:
            try:
                with gzip.GzipFile(fileobj=stream) as decoded:
                    raw = decoded.read(MAX_REFERENCE_BYTES + 1)
            except (EOFError, zlib.error) as error:
                msg = "Invalid compressed catalog"
                raise ValueError(msg) from error
        else:
            raw = stream.read(MAX_REFERENCE_BYTES + 1)
    if len(raw) > MAX_REFERENCE_BYTES:
        msg = "Code reference catalog exceeds the uncompressed byte limit"
        raise ValueError(msg)
    catalog = json.loads(raw)
    if (
        not isinstance(catalog, dict)
        or catalog.get("schema_version") != 1
        or not isinstance(catalog.get("references"), list)
        or not catalog["references"]
    ):
        msg = "Expected schema_version 1 and nonempty references list"
        raise ValueError(msg)
    references = [CodeReference.from_json(entry) for entry in catalog["references"]]
    ids = [reference.metadata["id"] for reference in references]
    if len(ids) != len(set(ids)):
        msg = "Duplicate code reference id"
        raise ValueError(msg)
    return references


def candidate_offsets(
    data: Buffer, reference: CodeReference, start: int, size: int
) -> Iterator[int]:
    end = start + size
    cursor = start + reference.anchor_offset
    while cursor < end:
        found = data.find(reference.anchor, cursor, end)
        if found < 0:
            return
        cursor = found + 1
        offset = found - reference.anchor_offset
        if offset + len(reference.code) <= end:
            yield offset


@dataclass(frozen=True)
class CodeAssessment:
    matches: list[dict]
    reason: str
    limited: bool = False

    @property
    def patch_status(self) -> str:
        states = {match["state"] for match in self.matches}
        if len(states) > 1:
            return "mixed_code_matches"
        if states:
            return f"{next(iter(states))}_code_match"
        return "unknown"


class CodeMatchLimitError(ValueError):
    """The shared candidate budget was exhausted."""


def matching_offsets(
    data: Buffer, reference: CodeReference, ranges: list[tuple[int, int]], budget: list[int]
) -> Iterator[int]:
    expected = int.from_bytes(reference.code, "little")
    mask = int.from_bytes(reference.mask, "little")
    for start, size in ranges:
        for offset in candidate_offsets(data, reference, start, size):
            budget[0] -= 1
            if budget[0] < 0:
                msg = "Code candidate limit reached; additional copies not assessed"
                raise CodeMatchLimitError(msg)
            actual = int.from_bytes(data[offset : offset + len(reference.code)], "little")
            if not (actual ^ expected) & mask:
                yield offset


def match_code(  # noqa: C901 -- Bounded reference/segment/candidate search with explicit limits.
    data: Buffer, references: list[CodeReference]
) -> CodeAssessment:
    try:
        header = elf_header(data)
        _, ranges = executable_ranges(data)
    except ValueError as error:
        return CodeAssessment([], str(error))
    matches = []
    budget = [MAX_CANDIDATES]
    identity = target_identity(header)
    compatible = [
        reference
        for reference in references
        if (
            reference.metadata["machine"],
            reference.metadata.get("elf_class", 64),
            reference.metadata.get("byteorder", "little"),
        )
        == identity
    ]
    if not compatible:
        return CodeAssessment(
            [],
            f"No compiled references for {TARGETS[identity][0]}; patch status unknown",
            limited=True,
        )
    try:
        for reference in compatible:
            extra_offsets = None
            for offset in matching_offsets(data, reference, ranges, budget):
                if extra_offsets is None:
                    extra_offsets = [
                        next(matching_offsets(data, part, ranges, budget), None)
                        for part in reference.additional_sections
                    ]
                if None in extra_offsets:
                    break
                matches.append(
                    {
                        **reference.metadata,
                        "file_offset": offset,
                        "size_bytes": len(reference.code),
                        "method": "relocation_masked_decoder_text",
                        "additional_section_offsets": extra_offsets,
                    }
                )
                if len(matches) >= MAX_MATCHES:
                    return CodeAssessment(
                        matches,
                        "Code match limit reached; additional copies not assessed",
                        limited=True,
                    )
    except CodeMatchLimitError as error:
        return CodeAssessment(matches, str(error), limited=True)
    return CodeAssessment(
        matches,
        "Matched reference code only; other copies remain unassessed"
        if matches
        else "No supported compiled decoder reference matched; patch status unknown",
    )


def relocation_mask(
    machine: int, kind: int, byteorder: Literal["little", "big"] = "little"
) -> bytes:
    """Mask only psABI relocation fields; reject types outside the reviewed table."""
    try:
        field = bytes.fromhex(RELOCATION_MASKS[str(machine)][str(kind)])
    except KeyError as error:
        msg = f"Unsupported executable relocation: machine={machine}, type={kind}"
        raise ValueError(msg) from error
    return field if byteorder == "big" else field[::-1]


def apply_relocations(  # noqa: C901, PLR0913 -- Bounded ELF relocation walk and ABI rewrites.
    data: Buffer, sections: list[tuple], index: int, code: bytes, mask: bytearray, *, machine: int
) -> None:
    header = elf_header(data)
    layout = ELFLayout.of(header)
    for relocation in sections:
        if relocation[1] not in (4, 9) or relocation[7] != index:
            continue
        parser = layout.parser("rela" if relocation[1] == SHT_RELA else "rel")
        if relocation[9] != parser.size:
            msg = "Unsupported REL/RELA executable relocation entry size"
            raise ValueError(msg)
        entries = checked_slice(data, relocation[4], relocation[5])
        if len(entries) % parser.size:
            msg = "Malformed relocation table"
            raise ValueError(msg)
        for row in parser.iter_unpack(entries):
            offset, info = row[:2]
            kind = info & (0xFF if layout.bits == ELF32_BITS else 0xFFFFFFFF)
            if machine == EM_MIPS and layout.bits == ELF64_BITS:
                kind = mips64_relocation(info, layout.byteorder)
            field_mask = relocation_mask(machine, kind, layout.byteorder)
            checked_slice(code, offset, len(field_mask))
            if (
                kind in X86_GOT_RELOCATIONS.get(machine, ())
                and offset >= GOT_OPCODE_BYTES
                and code[offset - 2] in X86_GOT_OPCODES
            ):
                # GOT relaxations rewrite opcode/ModRM and sometimes the REX register bits.
                size = 3 if kind == R_X86_64_REX_GOTPCRELX else 2
                checked_slice(code, offset - size, size)
                mask[offset - size : offset] = bytes(size)
            if machine == EM_S390 and kind == R_390_GOTENT and offset >= GOT_OPCODE_BYTES:
                # LGRL C4x8 becomes LARL C0x0; keep register and other opcode bits.
                mask[offset - 2] &= 0xFB
                mask[offset - 1] &= 0xF7
            if machine == EM_PPC64:
                ppc64_linker_masks(code, mask, offset, kind, layout.byteorder)
            for i, value in enumerate(field_mask):
                mask[offset + i] &= value


def ppc64_linker_masks(
    code: bytes, mask: bytearray, offset: int, kind: int, byteorder: Literal["little", "big"]
) -> None:
    """Account for the ELFv2 TOC and call-slot rewrites permitted by the linker."""
    word = offset if byteorder == "little" else offset - 2
    if kind in (17, 50):  # GOT16_HA / TOC16_HA: ADDIS can become NOP.
        checked_slice(code, word, 4)
        mask[word : word + 4] = bytes(4)
    if kind in (R_PPC64_GOT16_LO_DS, R_PPC64_TOC16_LO_DS):
        checked_slice(code, word, 4)
        # A TOC/GOT load can become ADDI: opcode and the full immediate change.
        mask[offset : offset + 2] = bytes(2)
        mask[word + (3 if byteorder == "little" else 0)] &= 3
    if kind in (48, R_PPC64_GOT16_LO_DS, R_PPC64_TOC16_LO_DS, R_PPC64_REL16_HA):
        checked_slice(code, word, 4)
        mask[word + (2 if byteorder == "little" else 1)] &= 0xE0
    if kind == R_PPC64_REL24 and code[offset + 4 : offset + 8] == (0x60000000).to_bytes(
        4, byteorder
    ):
        mask[offset + 4 : offset + 8] = bytes(4)


def mips64_relocation(info: int, byteorder: Literal["little", "big"]) -> int:
    # MIPS64 uses a symbol word followed by ssym, type3, type2, type1 bytes.
    packed = info.to_bytes(8, byteorder)
    special, third, second, first = packed[4:]
    if (
        special
        or (second, third) not in ((0, 0), (24, 5), (24, 6))
        or (second and first != R_MIPS_GPREL16)
    ):
        msg = "Unsupported compound MIPS64 relocation"
        raise ValueError(msg)
    return third or first


def reference_from_object(data: Buffer, metadata: dict) -> dict:
    """Build a complete decoder .text signature, rejecting unsupported object layouts."""
    header = elf_header(data)
    if header[1] != 1 or not header[12]:
        msg = "Expected ELF relocatable object with section headers"
        raise ValueError(msg)
    sections = section_headers(data, header)
    code_sections = [
        (i, section) for i, section in enumerate(sections) if section[2] & 4 and section[5]
    ]
    if not code_sections or len(code_sections) > MAX_CODE_SECTIONS:
        msg = (
            "Expected one complete executable section or bounded code sections; LTO is unsupported"
        )
        raise ValueError(msg)
    code_sections.sort(key=lambda pair: pair[1][5], reverse=True)
    index, section = code_sections[0]
    code = checked_slice(data, section[4], section[5])
    mask = bytearray(b"\xff" * len(code))
    apply_relocations(data, sections, index, code, mask, machine=header[2])
    entry = {
        **metadata,
        "machine": header[2],
        "elf_class": ELFLayout.of(header).bits,
        "byteorder": ELFLayout.of(header).byteorder,
        "object_sha256": sha256_data(data),
        "code": code.hex(),
        "mask": mask.hex(),
    }
    if len(code_sections) > 1:
        entry["additional_sections"] = []
        for index, section in code_sections[1:]:
            part = checked_slice(data, section[4], section[5])
            part_mask = bytearray(b"\xff" * len(part))
            apply_relocations(data, sections, index, part, part_mask, machine=header[2])
            entry["additional_sections"].append({"code": part.hex(), "mask": part_mask.hex()})
    CodeReference.from_json(entry)
    return entry
```

## scripts/brotli_binary_metadata.py

```python
"""Bounded ELF decoder symbols and constant version returns, without target execution."""

from __future__ import annotations

import struct
from typing import TYPE_CHECKING

from brotli_binary_match import (
    EM_386,
    EM_AARCH64,
    EM_ARM,
    EM_X86_64,
    ELFLayout,
    checked_range,
    checked_slice,
    elf_header,
    executable_ranges,
    program_headers,
    section_headers,
    symbol_rows,
)

if TYPE_CHECKING:
    from brotli_binary_data import Buffer

SYMBOL = struct.Struct("<IBBHQQ")
MAX_SYMBOLS = 1_000_000
MAX_SYMBOL_TABLES = 16
MAX_MAPPING_WORK = 1_000_000
MAX_FUNCTION_QUERIES = 4096
MAX_SYMBOL_NAME_BYTES = 64
MAX_VERSION_FUNCTION_BYTES = 64
SHT_STRTAB = 3
STT_FUNC = 2
SHF_ALLOC = 2
SHF_EXECINSTR = 4
X86_CONST_RETURN_BYTES = 6
X86_MOV_EAX_IMMEDIATE = 0xB8
X86_RET = 0xC3
AARCH64_BTI_C = 0xD503245F
AARCH64_RET = 0xD65F03C0
AARCH64_MOVZ_W0 = 0x52800000
AARCH64_MOVK_W0 = 0x72800000
AARCH64_MIN_RETURN_WORDS = 2
AARCH64_MAX_RETURN_WORDS = 3
DECODER_APIS = frozenset(
    name.encode("ascii")
    for name in (
        "BrotliDecoderVersion",
        "BrotliDecoderDecompress",
        "BrotliDecoderDecompressStream",
        "BrotliDecoderCreateInstance",
        "BrotliDecoderDestroyInstance",
    )
)


def constant_version(  # noqa: C901, PLR0911 -- Strict instruction-form decoder.
    code: bytes, machine: int
) -> int | None:
    """Recognize only complete, straight-line constant-return function forms."""
    if machine in (EM_X86_64, EM_386):
        code = code.removeprefix(bytes.fromhex("f30f1efa" if machine == EM_X86_64 else "f30f1efb"))
        if code.startswith(bytes.fromhex("554889e5")) and code.endswith(b"\x5d\xc3"):
            code = code[4:-2] + b"\xc3"
        if (
            len(code) == X86_CONST_RETURN_BYTES
            and code[0] == X86_MOV_EAX_IMMEDIATE
            and code[-1] == X86_RET
        ):
            return int.from_bytes(code[1:5], "little")
        return None
    if machine != EM_AARCH64 or len(code) % 4:
        return None
    words = list(struct.unpack(f"<{len(code) // 4}I", code))
    if words and words[0] == AARCH64_BTI_C:
        words.pop(0)
    # GCC branch-protected frame, with no arbitrary intervening instructions.
    if words[:2] == [0xD503233F, 0xA9BF7BFD] and words[-4:] == [
        0x910003FD,
        0xA8C17BFD,
        0xD50323BF,
        0xD65F03C0,
    ]:
        words = [*words[2:-4], AARCH64_RET]
    if (
        not words
        or words[-1] != AARCH64_RET
        or not AARCH64_MIN_RETURN_WORDS <= len(words) <= AARCH64_MAX_RETURN_WORDS
    ):
        return None
    first = words[0]
    if first & 0xFFC0001F != AARCH64_MOVZ_W0:  # MOVZ W0, imm16, LSL {0,16}.
        return None
    value = ((first >> 5) & 0xFFFF) << (((first >> 21) & 1) * 16)
    if len(words) == AARCH64_MAX_RETURN_WORDS:
        second = words[1]
        if second & 0xFFC0001F != AARCH64_MOVK_W0:
            return None
        shift = ((second >> 21) & 1) * 16
        value = (value & ~(0xFFFF << shift)) | (((second >> 5) & 0xFFFF) << shift)
    return value


class MetadataLimitError(ValueError):
    """Decoder metadata exceeded a combined CPU or retained-query budget."""


class FunctionOffsets:
    """Parse mappings once and bound work across all distinct function lookups."""

    def __init__(self, data: Buffer, header: tuple) -> None:
        self.segments = [
            (start, virtual, file_size)
            for kind, flags, start, virtual, _, file_size, _, _ in program_headers(data, header)
            if kind == 1 and flags & 1
        ]
        self.cache: dict[tuple[int, int], int | None] = {}
        self.work_left = MAX_MAPPING_WORK

    def resolve(self, address: int, size: int) -> int | None:
        key = (address, size)
        if key in self.cache:
            return self.cache[key]
        if len(self.cache) >= MAX_FUNCTION_QUERIES:
            msg = "Decoder function query limit reached"
            raise MetadataLimitError(msg)
        for start, virtual, file_size in self.segments:
            if self.work_left <= 0:
                msg = "Decoder mapping work limit reached"
                raise MetadataLimitError(msg)
            self.work_left -= 1
            if virtual <= address and address + size <= virtual + file_size:
                offset = start + address - virtual
                self.cache[key] = offset
                return offset
        self.cache[key] = None
        return None


def decoder_metadata(data: Buffer) -> dict:
    result = {
        "defined_symbols": [],
        "imported_symbols": [],
        "versions": [],
        "reason": "No decoder API symbols found; stripped copies remain possible",
        "limited": False,
        "limits": {
            "max_symbols": MAX_SYMBOLS,
            "max_symbol_tables": MAX_SYMBOL_TABLES,
            "max_mapping_work": MAX_MAPPING_WORK,
            "max_function_queries": MAX_FUNCTION_QUERIES,
        },
    }
    try:
        _read_symbols(data, result)
    except MetadataLimitError as error:
        result.update(reason=str(error), limited=True)
    except (ValueError, struct.error) as error:
        result["reason"] = f"Decoder metadata unavailable: {error}"
    return result


def _read_symbols(  # noqa: C901, PLR0912, PLR0915 -- Bounded ELF table walk.
    data: Buffer, result: dict
) -> None:
    header = elf_header(data)
    executable_ranges(data)
    if not header[12]:
        result["reason"] = "No ELF section table; symbol-based decoder evidence unavailable"
        return
    sections = section_headers(data, header)
    symbol = ELFLayout.of(header).parser("symbol")
    seen = 0
    tables_seen = 0
    offsets = FunctionOffsets(data, header)
    version_locations: set[tuple[int, int]] = set()
    for section in sections:
        if section[1] not in (2, 11):  # SHT_SYMTAB, SHT_DYNSYM.
            continue
        tables_seen += 1
        if tables_seen > MAX_SYMBOL_TABLES:
            result.update(reason="Decoder symbol-table limit reached", limited=True)
            return
        if section[9] != symbol.size or section[5] % symbol.size or section[6] >= len(sections):
            msg = "Malformed ELF symbol table"
            raise ValueError(msg)
        seen += section[5] // symbol.size
        if seen > MAX_SYMBOLS:
            result.update(reason="Decoder symbol limit reached", limited=True)
            return
        string_section = sections[section[6]]
        if string_section[1] != SHT_STRTAB:
            msg = "ELF symbol names do not reference a string table"
            raise ValueError(msg)
        string_start, string_end = checked_range(data, string_section[4], string_section[5])
        for name_offset, info, _, index, symbol_address, size in symbol_rows(data, header, section):
            if name_offset >= string_section[5]:
                msg = "ELF symbol name outside string table"
                raise ValueError(msg)
            name_start = string_start + name_offset
            end = data.find(b"\0", name_start, min(name_start + MAX_SYMBOL_NAME_BYTES, string_end))
            name = data[name_start:end] if end >= 0 else b""
            if name not in DECODER_APIS or info & 0xF != STT_FUNC:
                continue
            offset = None
            if index == 0:
                bucket = result["imported_symbols"]
            else:
                if index >= len(sections):
                    continue
                owner = sections[index]
                # ARM function pointers mark Thumb state in bit zero.
                address = symbol_address & (~1 if header[2] == EM_ARM else -1)
                offset = offsets.resolve(address, max(size, 1))
                if (
                    owner[1] != 1
                    or owner[2] & (SHF_ALLOC | SHF_EXECINSTR) != (SHF_ALLOC | SHF_EXECINSTR)
                    or address < owner[3]
                    or address + max(size, 1) > owner[3] + owner[5]
                    or offset != owner[4] + address - owner[3]
                ):
                    continue
                bucket = result["defined_symbols"]
            text = name.decode("ascii")
            if text not in bucket:
                bucket.append(text)
            if (
                name != b"BrotliDecoderVersion"
                or index == 0
                or not 0 < size <= MAX_VERSION_FUNCTION_BYTES
                or offset is None
            ):
                continue
            location = (offset, size)
            if location in version_locations:
                continue
            version_locations.add(location)
            value = constant_version(checked_slice(data, offset, size), header[2])
            if value is not None:
                version = f"{value >> 24}.{(value >> 12) & 0xFFF}.{value & 0xFFF}"
                evidence = {
                    "version": version,
                    "file_offset": offset,
                    "method": "constant_decoder_version_return",
                }
                result["versions"].append(evidence)
    if result["defined_symbols"] or result["imported_symbols"]:
        result["reason"] = (
            "Defined/imported decoder symbols; reported versions do not establish patch status"
        )
```

## scripts/brotli_binary_triage.py

```python
#!/usr/bin/env python3

"""Collect bounded objdump evidence for model review of one unresolved ELF."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import selectors
import shutil
import stat
import subprocess
import sys
import tempfile
import time
from contextlib import contextmanager
from pathlib import Path
from typing import TYPE_CHECKING

from brotli_binary_data import Buffer, FileData, file_data, sha256_data
from brotli_binary_match import (
    EM_ARM,
    TARGETS,
    ELFLayout,
    elf_header,
    executable_ranges,
    program_headers,
    section_headers,
    symbol_rows,
    target_identity,
)
from brotli_binary_metadata import MAX_SYMBOL_TABLES, MAX_SYMBOLS

if TYPE_CHECKING:
    from collections.abc import Generator

DEFAULT_MAX_FILE_BYTES = None
DEFAULT_MAX_OUTPUT_BYTES = 1024 * 1024
DEFAULT_MAX_RANGE_BYTES = 64 * 1024
DEFAULT_TIMEOUT_SECONDS = 15
DEFAULT_MAX_SYMBOLS = 12
CHUNK_BYTES = 64 * 1024
PT_LOAD = 1
PF_X = 1
SHT_STRTAB = 3
STT_FUNC = 2
SHF_EXECINSTR = 4
SHF_ALLOC = 2
MAX_TRIAGE_SYMBOL_NAME_BYTES = 128
PATCH_SYMBOL = re.compile(
    r"^(?:Safe)?Decode(Literal|Command|Distance)BlockSwitch(?:Internal)?(?:[.$].*)?$"
)
PROCESS_SYMBOL = re.compile(r"^(?:Safe)?ProcessCommands(?:Internal)?(?:[.$].*)?$")
PUBLIC_FALLBACK = "BrotliDecoderDecompressStream"


class TriageError(ValueError):
    """An input cannot be inspected safely within the requested scope."""


def positive(value: str) -> int:
    number = int(value, 0)
    if number <= 0:
        message = "must be positive"
        raise argparse.ArgumentTypeError(message)
    return number


def nonnegative(value: str) -> int:
    number = int(value, 0)
    if number < 0:
        message = "must be nonnegative"
        raise argparse.ArgumentTypeError(message)
    return number


def trusted_objdump(value: str | None) -> Path:
    located = value or shutil.which("objdump")
    if not located:
        message = "objdump was not found; provide an absolute trusted --objdump path"
        raise TriageError(message)
    candidate = Path(located).expanduser()
    if value and not candidate.is_absolute():
        message = "--objdump must be an absolute trusted path"
        raise TriageError(message)
    try:
        resolved = candidate.resolve(strict=True)
        metadata = resolved.stat()
    except OSError as error:
        message = f"Cannot inspect objdump: {error}"
        raise TriageError(message) from error
    if (
        not resolved.is_absolute()
        or not stat.S_ISREG(metadata.st_mode)
        or not os.access(resolved, os.X_OK)
    ):
        message = "objdump must resolve to an executable regular file"
        raise TriageError(message)
    return resolved


@contextmanager
def read_target(path: Path, max_file_bytes: int | None) -> Generator[FileData]:
    try:
        metadata = path.lstat()
        if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
            message = "Target symlinks and nonregular files are not inspected"
            raise TriageError(message)
        descriptor = os.open(
            path,
            os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0),
        )
        with os.fdopen(descriptor, "rb") as stream:
            metadata = os.fstat(stream.fileno())
            if not stat.S_ISREG(metadata.st_mode):
                message = "Target is not a regular file"
                raise TriageError(message)
            if max_file_bytes is not None and metadata.st_size > max_file_bytes:
                message = f"Target exceeds --max-file-bytes ({metadata.st_size} > {max_file_bytes})"
                raise TriageError(message)
            with file_data(stream, max_file_bytes) as data:
                yield data
    except (OSError, ValueError) as error:
        message = f"Cannot read stable target: {error}"
        raise TriageError(message) from error


def run_bounded(  # noqa: C901 -- Process/pipe/time bounds share one lifecycle.
    command: list[str], timeout_seconds: float, max_output_bytes: int
) -> dict:
    """Run one trusted tool command with wall-clock and combined-output bounds."""
    started = time.monotonic()
    try:
        process = subprocess.Popen(  # noqa: S603 -- Caller supplies a validated objdump path.
            command,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )
    except OSError as error:
        return {
            "returncode": None,
            "output": "",
            "limited": True,
            "reason": f"Cannot start objdump: {error}",
            "elapsed_seconds": time.monotonic() - started,
        }
    assert process.stdout is not None  # noqa: S101 -- PIPE guarantees this internal invariant.
    output = bytearray()
    reason = None
    selector = selectors.DefaultSelector()
    selector.register(process.stdout, selectors.EVENT_READ)
    deadline = started + timeout_seconds
    try:
        while selector.get_map():
            remaining_time = deadline - time.monotonic()
            if remaining_time <= 0:
                reason = "objdump timeout reached"
                process.kill()
                break
            for key, _ in selector.select(min(remaining_time, 0.1)):
                chunk = os.read(key.fd, min(CHUNK_BYTES, max_output_bytes - len(output) + 1))
                if not chunk:
                    selector.unregister(key.fileobj)
                    continue
                output.extend(chunk)
                if len(output) > max_output_bytes:
                    del output[max_output_bytes:]
                    reason = "objdump output limit reached"
                    process.kill()
                    break
            if reason is not None:
                break
        remaining_time = max(0.0, deadline - time.monotonic())
        try:
            returncode = process.wait(timeout=remaining_time)
        except subprocess.TimeoutExpired:
            reason = reason or "objdump timeout reached"
            process.kill()
            returncode = process.wait()
    finally:
        selector.close()
        process.stdout.close()
    return {
        "returncode": returncode,
        "output": output.decode("utf-8", errors="replace"),
        "limited": reason is not None,
        "reason": reason,
        "elapsed_seconds": time.monotonic() - started,
    }


def checked_range(data: Buffer, offset: int, size: int) -> tuple[int, int]:
    if offset < 0 or size < 0 or offset > len(data) or size > len(data) - offset:
        message = "ELF range outside file"
        raise ValueError(message)
    return offset, offset + size


def function_offset(data: Buffer, header: tuple, address: int, size: int) -> int | None:
    for kind, flags, start, virtual, _, file_size, _, _ in program_headers(data, header):
        if (
            kind == PT_LOAD
            and flags & PF_X
            and virtual <= address < address + size <= virtual + file_size
        ):
            return start + address - virtual
    return None


def relevant_symbols(  # noqa: C901, PLR0911, PLR0912, PLR0915 -- Bounded ELF validation.
    data: Buffer, deadline: float, candidate_limit: int = DEFAULT_MAX_SYMBOLS + 1
) -> tuple[list[dict], str | None]:
    """Select relevant defined/imported functions from bounded in-file ELF tables."""
    header = elf_header(data)
    if not header[12]:
        return [], "No ELF section table; symbol-based triage is unavailable"
    if header[11] != ELFLayout.of(header).parser("section").size:
        return [], "Unsupported ELF section table"
    sections = section_headers(data, header)
    symbol = ELFLayout.of(header).parser("symbol")
    defined_symbols = []
    imported_symbols = []
    candidate_overflow = False
    scanned = 0
    tables = 0
    for table in sections:
        if table[1] not in (2, 11):  # SHT_SYMTAB, SHT_DYNSYM.
            continue
        tables += 1
        if tables > MAX_SYMBOL_TABLES:
            return [*defined_symbols, *imported_symbols], "ELF symbol-table limit reached"
        if table[9] != symbol.size or table[5] % symbol.size or table[6] >= len(sections):
            return [*defined_symbols, *imported_symbols], "Malformed ELF symbol table"
        scanned += table[5] // symbol.size
        if scanned > MAX_SYMBOLS:
            return [*defined_symbols, *imported_symbols], "ELF symbol limit reached"
        names = sections[table[6]]
        if names[1] != SHT_STRTAB:
            return [*defined_symbols, *imported_symbols], (
                "ELF symbols do not reference a string table"
            )
        name_start, name_stop = checked_range(data, names[4], names[5])
        for name_offset, info, _, index, symbol_address, size in symbol_rows(data, header, table):
            address = symbol_address & (~1 if header[2] == EM_ARM else -1)
            if time.monotonic() >= deadline:
                return [*defined_symbols, *imported_symbols], "ELF symbol scan timeout reached"
            if info & 0xF != STT_FUNC or name_offset >= names[5]:
                continue
            absolute_name = name_start + name_offset
            end = data.find(
                b"\0",
                absolute_name,
                min(absolute_name + MAX_TRIAGE_SYMBOL_NAME_BYTES, name_stop),
            )
            if end < 0:
                continue
            try:
                name = data[absolute_name:end].decode("ascii")
            except UnicodeDecodeError:
                continue
            base = name.split("@", 1)[0]
            if not (
                PATCH_SYMBOL.match(base) or PROCESS_SYMBOL.match(base) or base == PUBLIC_FALLBACK
            ):
                continue
            if index == 0:
                section = "*UND*"
                defined = False
            else:
                if index >= len(sections):
                    continue
                owner = sections[index]
                offset = function_offset(data, header, address, max(size, 1))
                if (
                    owner[1] != 1
                    or owner[2] & (SHF_ALLOC | SHF_EXECINSTR) != (SHF_ALLOC | SHF_EXECINSTR)
                    or address < owner[3]
                    or address + max(size, 1) > owner[3] + owner[5]
                    or offset != owner[4] + address - owner[3]
                ):
                    continue
                section = f"section[{index}]"
                defined = True
            item = {
                "name": name,
                "base_name": base,
                "address": address,
                "size_bytes": size,
                "section": section,
                "defined": defined,
            }
            if header[2] == EM_ARM:
                item["thumb"] = bool(symbol_address & 1)
            bucket = defined_symbols if defined else imported_symbols
            if item in bucket:
                continue
            if len(bucket) < candidate_limit:
                bucket.append(item)
            else:
                candidate_overflow = True
    symbols = [*defined_symbols, *imported_symbols]
    if len(symbols) > candidate_limit:
        candidate_overflow = True
        symbols = symbols[:candidate_limit]
    return symbols, "Relevant ELF symbol candidate limit reached" if candidate_overflow else None


def symbol_priority(symbol: dict) -> tuple[int, int, int, str]:
    name = symbol["base_name"]
    priority = 0 if PATCH_SYMBOL.match(name) else 1 if PROCESS_SYMBOL.match(name) else 2
    return 0 if symbol["defined"] else 1, priority, symbol["address"], symbol["name"]


def file_offset_address(data: Buffer, file_offset: int, length: int) -> int:
    header = elf_header(data)
    for kind, flags, start, virtual, _, file_size, _, _ in program_headers(data, header):
        if (
            kind == PT_LOAD
            and flags & PF_X
            and start <= file_offset
            and file_offset + length <= start + file_size
        ):
            return virtual + file_offset - start
    message = "File-offset range is not inside one file-backed executable PT_LOAD segment"
    raise TriageError(message)


def validate_address_range(data: Buffer, start: int, stop: int) -> None:
    header = elf_header(data)
    for kind, flags, _, virtual, _, file_size, _, _ in program_headers(data, header):
        if kind == PT_LOAD and flags & PF_X and virtual <= start < stop <= virtual + file_size:
            return
    message = "Address range is not inside one file-backed executable PT_LOAD segment"
    raise TriageError(message)


def base_report(path: Path, args: argparse.Namespace) -> dict:
    return {
        "schema_version": 2,
        "target": {"path": str(path), "sha256": None, "size_bytes": None},
        "tool": {"path": None, "input_mode": "code_fragments", "fragments": [], "commands": []},
        "scope": {"mode": "symbols"},
        "limits": {
            "max_file_bytes": args.max_file_bytes,
            "max_output_bytes": args.max_output_bytes,
            "max_range_bytes": args.max_range_bytes,
            "timeout_seconds": args.timeout_seconds,
            "max_symbols": args.max_symbols,
        },
        "collection": {"status": "incomplete", "incomplete": True, "reasons": []},
        "classification": {
            "patch_status": "unknown",
            "reason": (
                "Disassembly requires model comparison with reviewed source; no automatic verdict"
            ),
        },
        "evidence": {"symbols": [], "disassembly": []},
        "limitations": [
            "SHA-256 requires one streamed full-file pass; ELF analysis uses bounded offset reads.",
            (
                "The target is never executed or loaded; only bounded code fragments are copied "
                "for objdump and removed after use."
            ),
            "Disassembly is bounded and may omit inlined, stripped, split, or optimized code.",
            "Strings and version evidence cannot establish that the fix is present.",
            "One reviewed range cannot clear other embedded decoder copies.",
        ],
    }


def command_evidence(
    report: dict, command: list[str], scope: dict, deadline: float, remaining_output: int
) -> int:
    remaining_time = deadline - time.monotonic()
    if remaining_time <= 0 or remaining_output <= 0:
        report["collection"]["reasons"].append("Global objdump time/output budget exhausted")
        return 0
    report["tool"]["commands"].append(command)
    result = run_bounded(command, remaining_time, remaining_output)
    report["evidence"]["disassembly"].append(
        {
            "scope": scope,
            "returncode": result["returncode"],
            "limited": result["limited"],
            "reason": result["reason"],
            "text": result["output"],
        }
    )
    if result["limited"] or result["returncode"] != 0:
        report["collection"]["reasons"].append(
            result["reason"] or f"objdump exited {result['returncode']}"
        )
    return len(result["output"].encode("utf-8"))


def validate_scope(parser: argparse.ArgumentParser, args: argparse.Namespace) -> str:
    address_range = args.start_address is not None or args.stop_address is not None
    offset_range = args.file_offset is not None or args.length is not None
    if (args.start_address is None) != (args.stop_address is None):
        parser.error("--start-address and --stop-address must be supplied together")
    if (args.file_offset is None) != (args.length is None):
        parser.error("--file-offset and --length must be supplied together")
    if address_range and offset_range:
        parser.error("address and file-offset ranges are mutually exclusive")
    if args.section is not None and not address_range:
        parser.error("--section requires an explicit address range")
    if args.thumb and not (address_range or offset_range):
        parser.error("--thumb requires an explicit ARM range")
    return "address_range" if address_range else "file_offset_range" if offset_range else "symbols"


def explicit_range(data: Buffer, args: argparse.Namespace, mode: str) -> tuple[int, int]:
    if mode == "address_range":
        start, stop = args.start_address, args.stop_address
    else:
        start = file_offset_address(data, args.file_offset, args.length)
        stop = start + args.length
    if stop <= start or stop - start > args.max_range_bytes:
        message = "Requested range must be nonempty and within --max-range-bytes"
        raise TriageError(message)
    validate_address_range(data, start, stop)
    return start, stop


def require_section(data: Buffer, name: str, start: int, stop: int) -> None:
    header = elf_header(data)
    if header[11] == ELFLayout.of(header).parser("section").size and 0 < header[13] < header[12]:
        sections = section_headers(data, header)
        strings = sections[header[13]]
        checked_range(data, strings[4], strings[5])
        for section in sections:
            if section[1] != 1 or not section[2] & SHF_EXECINSTR:
                continue
            if section[0] >= strings[5]:
                continue
            offset = strings[4] + section[0]
            value = data[
                offset : min(offset + len(name.encode("utf-8")) + 1, strings[4] + strings[5])
            ]
            if (
                value == name.encode("utf-8") + b"\0"
                and section[3] <= start < stop <= section[3] + section[5]
            ):
                return
    message = "Requested section does not contain the executable address range"
    raise TriageError(message)


def fragment_command(  # noqa: PLR0913, PLR0917 -- Bound original addresses to a small tool input.
    report: dict,
    data: Buffer,
    directory: Path,
    objdump: Path,
    start: int,
    stop: int,
    *,
    thumb: bool = False,
) -> list[str]:
    header = elf_header(data)
    if thumb and header[2] != EM_ARM:
        message = "--thumb requires an ARM ELF"
        raise TriageError(message)
    offset = function_offset(data, header, start, stop - start)
    if offset is None:
        message = "Code fragment is outside file-backed executable data"
        raise TriageError(message)
    code = data[offset : offset + stop - start]
    path = directory / f"code-{len(report['tool']['fragments'])}.bin"
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o400)
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(code)
    report["tool"]["fragments"].append(
        {
            "path": str(path),
            "sha256": hashlib.sha256(code).hexdigest(),
            "file_offset": offset,
            "start_address": start,
            "stop_address": stop,
            "size_bytes": len(code),
            "ephemeral": True,
        }
    )
    architecture = TARGETS[target_identity(header)][1]
    return [
        str(objdump),
        "--disassemble-all",
        "--wide",
        "--target=binary",
        f"--architecture={architecture}",
        f"--endian={ELFLayout.of(header).byteorder}",
        *(["--disassembler-options=force-thumb"] if thumb else []),
        f"--adjust-vma=0x{start:x}",
        str(path),
    ]


def collect_ranges(  # noqa: C901, PLR0912, PLR0913, PLR0915, PLR0917 -- Bounded orchestration.
    report: dict,
    args: argparse.Namespace,
    mode: str,
    data: Buffer,
    objdump: Path,
    directory: Path,
    deadline: float,
) -> tuple[int, dict]:
    output_left = args.max_output_bytes
    if mode != "symbols":
        try:
            start, stop = explicit_range(data, args, mode)
        except TriageError as error:
            report["collection"]["reasons"].append(str(error))
            return 3, report
        report["scope"].update(start_address=start, stop_address=stop)
        if mode == "file_offset_range":
            report["scope"].update(file_offset=args.file_offset, length=args.length)
        if args.section:
            require_section(data, args.section, start, stop)
            report["scope"]["section"] = args.section
        command = fragment_command(report, data, directory, objdump, start, stop, thumb=args.thumb)
        used = command_evidence(report, command, dict(report["scope"]), deadline, output_left)
        output_left -= used
    else:
        try:
            relevant, symbol_reason = relevant_symbols(data, deadline, args.max_symbols + 1)
        except ValueError as error:
            relevant, symbol_reason = [], f"Cannot inspect ELF symbols: {error}"
        relevant = sorted(relevant, key=symbol_priority)
        report["evidence"]["symbols"] = relevant[: args.max_symbols]
        if symbol_reason is not None:
            report["collection"]["reasons"].append(symbol_reason)
        if len(relevant) > args.max_symbols:
            report["collection"]["reasons"].append("Relevant symbol limit reached")
        defined = [symbol for symbol in relevant[: args.max_symbols] if symbol["defined"]]
        if not defined:
            report["collection"]["reasons"].append(
                "No defined patch-bearing symbols; use an explicit trusted address "
                "or file-offset range"
            )
        block_classes = {
            match.group(1)
            for symbol in defined
            if (match := PATCH_SYMBOL.match(symbol["base_name"])) is not None
        }
        missing_classes = sorted({"Literal", "Command", "Distance"} - block_classes)
        if missing_classes:
            report["collection"]["reasons"].append(
                "Missing defined block-switch classes: " + ", ".join(missing_classes)
            )
        if not any(PROCESS_SYMBOL.match(symbol["base_name"]) for symbol in defined):
            report["collection"]["reasons"].append(
                "No defined ProcessCommands symbol; patch-status propagation scope is incomplete"
            )
        for index, symbol in enumerate(defined):
            if output_left <= 0 or time.monotonic() >= deadline:
                report["collection"]["reasons"].append(
                    "Global objdump time/output budget exhausted; selected symbols omitted"
                )
                report["collection"]["omitted_symbols"] = [item["name"] for item in defined[index:]]
                break
            size = symbol["size_bytes"]
            if not 0 < size <= args.max_range_bytes:
                report["collection"]["reasons"].append(
                    f"Symbol {symbol['name']} has no bounded size within --max-range-bytes"
                )
                continue
            start, stop = symbol["address"], symbol["address"] + size
            try:
                validate_address_range(data, start, stop)
            except TriageError as error:
                report["collection"]["reasons"].append(f"Symbol {symbol['name']}: {error}")
                continue
            command = fragment_command(
                report, data, directory, objdump, start, stop, thumb=symbol.get("thumb", False)
            )
            used = command_evidence(
                report,
                command,
                {
                    "mode": "symbol",
                    "name": symbol["name"],
                    "start_address": start,
                    "stop_address": stop,
                    "size_bytes": size,
                },
                deadline,
                output_left,
            )
            output_left -= used
    if report["evidence"]["disassembly"] and not report["collection"]["reasons"]:
        report["collection"].update(status="evidence_collected", incomplete=False)
        return 0, report
    report["collection"]["next_request"] = (
        "Provide a trusted linker/map-derived --start-address/--stop-address, or an executable "
        "--file-offset/--length; keep the range within the reported limits."
    )
    return 3, report


def collect(args: argparse.Namespace, parser: argparse.ArgumentParser) -> tuple[int, dict]:
    path = args.target.expanduser().absolute()
    report = base_report(path, args)
    mode = validate_scope(parser, args)
    report["scope"]["mode"] = mode
    try:
        objdump = trusted_objdump(args.objdump)
        report["tool"]["path"] = str(objdump)
        with read_target(path, args.max_file_bytes) as data:
            report["target"].update(sha256=sha256_data(data), size_bytes=len(data))
            try:
                executable_ranges(data)
            except ValueError as error:
                report["collection"]["reasons"].append(f"Unsupported or malformed ELF: {error}")
                return 3, report
            deadline = time.monotonic() + args.timeout_seconds
            with tempfile.TemporaryDirectory(prefix="brotli-triage-") as directory:
                code, report = collect_ranges(
                    report, args, mode, data, objdump, Path(directory), deadline
                )
    except (TriageError, OSError) as error:
        report["collection"]["reasons"].append(str(error))
        report["collection"].update(status="incomplete", incomplete=True)
        return 3, report
    else:
        return code, report


def write_new_output(path: Path, rendered: str, forbidden: list[str | None]) -> None:
    destination = path.expanduser().absolute()
    if any(
        item is not None and destination == Path(item).expanduser().absolute() for item in forbidden
    ):
        message = "--output cannot alias the target or objdump"
        raise TriageError(message)
    try:
        descriptor = os.open(
            destination,
            os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
            0o600,
        )
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            stream.write(rendered)
    except OSError as error:
        message = f"--output must be a new file and cannot alias the target or objdump: {error}"
        raise TriageError(message) from error


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("target", type=Path, help="one local ELF regular file (symlinks rejected)")
    parser.add_argument("--objdump", help="absolute path to trusted objdump (default: PATH lookup)")
    parser.add_argument("--output", type=Path, help="write JSON to this file instead of stdout")
    parser.add_argument("--start-address", type=nonnegative)
    parser.add_argument("--stop-address", type=nonnegative)
    parser.add_argument("--section", help="section constraint for an explicit address range")
    parser.add_argument("--file-offset", type=nonnegative)
    parser.add_argument("--length", type=positive)
    parser.add_argument(
        "--max-file-bytes",
        type=positive,
        default=DEFAULT_MAX_FILE_BYTES,
        help="optional file byte limit; no default size cap",
    )
    parser.add_argument("--max-output-bytes", type=positive, default=DEFAULT_MAX_OUTPUT_BYTES)
    parser.add_argument("--max-range-bytes", type=positive, default=DEFAULT_MAX_RANGE_BYTES)
    parser.add_argument("--timeout-seconds", type=positive, default=DEFAULT_TIMEOUT_SECONDS)
    parser.add_argument("--max-symbols", type=positive, default=DEFAULT_MAX_SYMBOLS)
    parser.add_argument(
        "--thumb", action="store_true", help="decode an explicit ARM range as Thumb code"
    )
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        code, report = collect(args, parser)
    except TriageError as error:
        report = base_report(args.target.expanduser().absolute(), args)
        report["collection"]["reasons"].append(str(error))
        code = 3
    rendered = json.dumps(report, indent=2, ensure_ascii=True) + "\n"
    if args.output is None:
        print(rendered, end="")
    else:
        try:
            write_new_output(
                args.output,
                rendered,
                [report["target"]["path"], report["tool"]["path"]],
            )
        except TriageError as error:
            parser.error(str(error))
    return code


# gremlin: pardon[out_of_scope] CLI bootstrap guard; imports must not launch argparse.
if __name__ == "__main__":
    sys.exit(main())
```

## scripts/brotli_inventory.py

```python
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.12,<3.16"
# dependencies = []
# [tool.ty.environment]
# extra-paths = ["."]
# ///

"""Offline, non-executing evidence collection for CVE-2026-111497."""

from __future__ import annotations

import argparse
import bz2
import gzip
import io
import json
import lzma
import os
import re
import stat
import sys
import tarfile
import tomllib
import zipfile
import zlib
from contextlib import contextmanager
from itertools import islice
from pathlib import Path
from typing import TYPE_CHECKING, BinaryIO, cast

from brotli_binary_data import (
    Buffer,
    DataChangedError,
    DataLimitError,
    file_data,
    read_data,
    sha256_data,
)
from brotli_binary_match import (
    REFERENCES,
    CodeAssessment,
    CodeReference,
    file_backed_ranges,
    load_references,
    match_code,
)
from brotli_binary_metadata import decoder_metadata

if TYPE_CHECKING:
    from collections.abc import Generator

REGISTRY = Path(__file__).resolve().parent.parent / "references" / "fingerprints.json"
SOURCE_SUFFIXES = {"", ".c", ".cc", ".cpp", ".cxx", ".h", ".inc", ".inl", ".unbr"}
JS_SUFFIXES = {".js", ".mjs", ".cjs", ".jsx", ".ts", ".mts", ".cts", ".tsx"}
NATIVE_SUFFIXES = {".so", ".dylib", ".dll", ".node", ".a", ".lib", ".wasm", ".exe"}
ZIP_SUFFIXES = {".zip", ".jar", ".war", ".ear", ".whl", ".nupkg", ".apk"}
BINARY_SYMBOLS = (
    b"BrotliDecoderDecompress",
    b"BrotliDecoderCreateInstance",
    b"BrotliDecoderVersion",
    b"BrotliDecoderDestroyInstance",
    b"BrotliEncoderCompress",
    b"BrotliEncoderVersion",
    b"libbrotlidec",
    b"libbrotlicommon",
    b"libbrotlienc",
    b"org/brotli/dec/",
    b"com/aayushatharva/brotli4j/",
    b"DecoderJNI",
)
BINARY_INDICATOR_PATTERN = re.compile(
    b"|".join(re.escape(symbol) for symbol in sorted(BINARY_SYMBOLS, key=len, reverse=True))
)
INDICATOR_CHUNK_BYTES = 4 * 1024 * 1024
TEXT_PROBE_BYTES = 4096
COMPRESSED_SUFFIXES = {".gz": "gzip", ".bz2": "bzip2", ".xz": "xz"}
TAR_SUFFIXES = (".tar", ".tgz", ".txz", ".tbz", ".tbz2")
UNSUPPORTED_SUFFIXES = {
    ".deb",
    ".rpm",
    ".zst",
    ".zstd",
    ".tzst",
    ".7z",
    ".rar",
    ".br",
    ".lz",
    ".lz4",
    ".lzma",
    ".z",
    ".cab",
    ".iso",
    ".dmg",
    ".cpio",
    ".ar",
}
ARCHIVE_SIGNATURES = (
    (b"PK\x03\x04", "zip"),
    (b"PK\x05\x06", "zip"),
    (b"PK\x07\x08", "zip"),
    (b"\x1f\x8b", "gzip"),
    (b"BZh", "bzip2"),
    (b"\xfd7zXZ\0", "xz"),
    (b"!<arch>\n", "unsupported"),
    (b"!<thin>\n", "unsupported"),
    (b"\xed\xab\xee\xdb", "unsupported"),
    (b"\x28\xb5\x2f\xfd", "unsupported"),
    (b"7z\xbc\xaf\x27\x1c", "unsupported"),
    (b"Rar!\x1a\x07", "unsupported"),
    (b"\x04\x22\x4d\x18", "unsupported"),
    (b"LZIP", "unsupported"),
    (b"\x1f\x9d", "unsupported"),
    (b"MSCF", "unsupported"),
    (b"xar!", "unsupported"),
    (b"070701", "unsupported"),
    (b"070702", "unsupported"),
    (b"070707", "unsupported"),
)
NODE_FIXES = {22: (22, 23, 4), 24: (24, 22, 1), 26: (26, 12, 1)}
# Frozen from the Node.js release schedule on the advisory snapshot date.
NODE_EOL_BRANCHES = {0, *range(4, 22), 23, 25}
SKIPPABLE_FRAMES = tuple(bytes((first, 0x2A, 0x4D, 0x18)) for first in range(0x50, 0x60))
MANIFEST_NAMES = {
    "package.json",
    "package-lock.json",
    "npm-shrinkwrap.json",
    "yarn.lock",
    "pnpm-lock.yaml",
    "pom.xml",
    "build.gradle",
    "build.gradle.kts",
    "gradle.lockfile",
    "cargo.toml",
    "cargo.lock",
    "go.mod",
    "go.sum",
    "pyproject.toml",
    "requirements.txt",
    "uv.lock",
    "poetry.lock",
    "cmakelists.txt",
    "cmakecache.txt",
    "module.bazel",
    "module.bazel.lock",
    "workspace",
    "workspace.bazel",
    "deps",
    ".gitmodules",
    "vcpkg.json",
    "conanfile.txt",
    "conanfile.py",
    "bom.json",
    "metadata",
    "pkg-info",
}
UNDICI = re.compile(r"(?<![\w-])undici(?![\w-])", re.IGNORECASE)
DEPENDENCY = re.compile(
    r"brotli(?:4j|dec|common|enc)?|woff2|workerd|(?<![\w-])undici(?![\w-])",
    re.IGNORECASE,
)
JS_NON_CODE = re.compile(
    r"(?P<comment>//[^\n]*|/\*[\s\S]*?\*/)"
    r"""|(?P<string>'(?:\\.|[^'\\])*'|"(?:\\.|[^"\\])*")"""
    r"|(?P<template>\x60(?:\\.|[^\x60\\])*\x60)"
)
JS_FETCH_LEAD = re.compile(
    r"""(?P<fetch_reference>(?<![\w$'"])fetch\b)"""
    r"""|(?P<global_fetch_property>\b(?:globalThis|global)\s*\[\s*['"]fetch['"]\s*\])"""
    r"""|(?P<undici_import>\b(?:from\s+|(?:require|import)\s*(?:\(\s*)?)"""
    r"""['"]undici(?:/[^'"\\]*)?['"])"""
)
SWITCH_CALL = r"Decode(?:Literal|Command|Distance)BlockSwitch\s*\("
OLD_CALL = re.compile(r"\bBROTLI_SAFE\s*\(\s*(" + SWITCH_CALL + r")")
NEW_CALL = re.compile(r"\bBROTLI_SAFE_WITH_STATUS\s*\(\s*(" + SWITCH_CALL + r")")
BAD_GUARD = re.compile(r"if\s*\(\s*max_block_type\s*<=\s*1\s*\)\s*\{\s*return\s+BROTLI_FALSE\s*;")
COMMENTS = re.compile(r"/\*.*?\*/|//[^\n]*", re.DOTALL)
GOOD = {"fixed_source", "upstream_fixed_version"}
WARNING = {
    "vulnerable_source",
    "potentially_vulnerable",
    "vulnerable_code_match",
    "mixed_code_matches",
}
MAX_RECORDED_GAPS = 200
MAX_DISPLAY_COPIES = 10
MAX_DISPLAY_GAPS = 5
DEFAULT_MAX_FILE_BYTES = None
DEFAULT_MAX_TOTAL_BYTES = None
DEFAULT_MAX_EXPANDED_BYTES = 8 * 1024 * 1024 * 1024
DEFAULT_MAX_TOTAL_EXPANDED_BYTES = 64 * 1024 * 1024 * 1024
DEFAULT_MAX_TEXT_BYTES = 128 * 1024 * 1024
MAX_SOURCE_MATCHES = 20
PYC_HEADER_BYTES = 16
DECODER_SYMBOLS = (
    b"BrotliDecoderDecompress",
    b"BrotliDecoderCreateInstance",
    b"DecoderJNI",
)


def leaf_name(label: str) -> str:
    return label.rsplit("!", 1)[-1].replace("\\", "/").rsplit("/", 1)[-1].lower()


def tar_content(data: Buffer) -> bool:
    if data[257:262] == b"ustar":
        return True
    try:
        tarfile.TarInfo.frombuf(data[: tarfile.BLOCKSIZE], "utf-8", "surrogateescape")
    except tarfile.HeaderError:
        return False
    return True


@contextmanager
def buffer_stream(data: Buffer) -> Generator[BinaryIO]:
    if isinstance(data, bytes):
        yield io.BytesIO(data)
        return
    data.seek(0)
    try:
        yield cast("BinaryIO", data)
    finally:
        data.seek(0)


def archive_kind(data: Buffer, label: str) -> str | None:
    """Recognize containers by bytes, then use names to catch damaged/opaque ones."""
    for signature, kind in ARCHIVE_SIGNATURES:
        if data.startswith(signature):
            return kind
    if data.startswith(SKIPPABLE_FRAMES):
        return "unsupported"  # Zstandard/LZ4 skippable frames.
    if tar_content(data):
        return "tar"
    with buffer_stream(data) as stream:
        try:
            if zipfile.is_zipfile(stream):
                return "zip"  # Also recognizes ZIPs with a prepended executable stub.
        except (OSError, ValueError):
            pass
    return named_archive_kind(label)


def named_archive_kind(label: str) -> str | None:
    name = leaf_name(label)
    suffix = Path(name).suffix
    if suffix in ZIP_SUFFIXES:
        return "zip"
    if suffix in COMPRESSED_SUFFIXES:
        return COMPRESSED_SUFFIXES[suffix]
    if name.endswith(TAR_SUFFIXES):
        return {".tgz": "gzip", ".txz": "xz", ".tbz": "bzip2", ".tbz2": "bzip2"}.get(suffix, "tar")
    return "unsupported" if suffix in UNSUPPORTED_SUFFIXES else None


def safe_text(value: str) -> str:
    """Quote untrusted display fields, escaping controls, bidi marks, and surrogates."""
    return json.dumps(value, ensure_ascii=True)


def python_bytecode(data: Buffer, name: str) -> bool:
    return (
        Path(name).suffix in {".pyc", ".pyo"}
        and len(data) >= PYC_HEADER_BYTES
        and data[2:4] == b"\r\n"
        and int.from_bytes(data[4:8], "little") in {0, 1, 3}
    )


def mask_js_non_code(match: re.Match[str]) -> str:
    token = match.group()
    # Keep templates so references in template expressions remain visible.
    # Literal template text can produce extra leads; this is not a JS parser.
    if match.lastgroup == "template":
        return token
    if match.lastgroup == "string" and (
        token[1:-1] in {"fetch", "undici"} or token[1:-1].startswith("undici/")
    ):
        return token
    return re.sub(r"[^\r\n]", " ", token)


def fetch_references(text: str) -> tuple[list[dict], bool]:
    code = JS_NON_CODE.sub(mask_js_non_code, text)
    found = list(islice(JS_FETCH_LEAD.finditer(code), MAX_SOURCE_MATCHES + 1))
    matches = []
    line = 1
    previous = 0
    for match in found[:MAX_SOURCE_MATCHES]:
        offset = match.start()
        line += text.count("\n", previous, offset)
        start = text.rfind("\n", 0, offset) + 1
        end = text.find("\n", offset)
        matches.append(
            {
                "line": line,
                "column": offset - start + 1,
                "signal": match.lastgroup,
                "text": text[start : min(end if end >= 0 else len(text), start + 300)],
            }
        )
        previous = offset
    return matches, len(found) > MAX_SOURCE_MATCHES


def dependency_matches(text: str, name: str) -> list[dict]:
    if name != "pyproject.toml":
        return [
            {"line": i, "text": line[:300]}
            for i, line in enumerate(text.splitlines(), 1)
            if DEPENDENCY.search(line)
        ]
    try:
        metadata = tomllib.loads(text)
    except tomllib.TOMLDecodeError:
        # Preserve a lead when malformed metadata cannot be parsed.
        return (
            [{"text": "Unparseable Python metadata mentioning Brotli-related components."}]
            if DEPENDENCY.search(text)
            else []
        )
    sections = (
        "project.dependencies",
        "project.optional-dependencies",
        "dependency-groups",
        "build-system.requires",
        "tool.poetry.dependencies",
        "tool.poetry.dev-dependencies",
        "tool.poetry.group",
        "tool.pdm.dev-dependencies",
        "tool.uv.sources",
    )
    matches = []
    for field in sections:
        value = metadata
        for key in field.split("."):
            value = value.get(key, {}) if isinstance(value, dict) else {}
        rendered = json.dumps(value, ensure_ascii=True)
        if DEPENDENCY.search(rendered):
            matches.append({"field": field, "text": rendered[:300]})
    return matches


def finding(path: str, kind: str, patch_status: str, reason: str, **evidence: object) -> dict:
    return {
        "path": path,
        "kind": kind,
        "patch_status": patch_status,
        "reason": reason,
        "evidence": evidence,
        "build_selection": "unknown",
        "shipment": "unknown",
        "reachability": "unknown",
        "exploitability": "not_assessed",
    }


def is_binary_name_candidate(label: str) -> bool:
    name = leaf_name(label)
    return "brotli" in name and (
        Path(name).suffix in NATIVE_SUFFIXES | ZIP_SUFFIXES or ".so." in name
    )


def binary_indicators(data: Buffer) -> list[str]:
    if data.startswith(b"\x7fELF"):
        try:
            _, ranges = file_backed_ranges(data)
        except ValueError:
            ranges = []
    else:
        ranges = [(0, len(data))]
    found = set()
    overlap = max(map(len, BINARY_SYMBOLS)) - 1
    for start, size in ranges:
        tail = b""
        position = start
        stop = start + size
        while position < stop:
            chunk = data[position : min(position + INDICATOR_CHUNK_BYTES, stop)]
            candidate = tail + chunk
            found.update(match.group() for match in BINARY_INDICATOR_PATTERN.finditer(candidate))
            tail = candidate[-overlap:]
            position += len(chunk)
    return [symbol.decode("ascii") for symbol in BINARY_SYMBOLS if symbol in found]


def assess_version(component: str, value: str) -> dict:
    match = re.fullmatch(r"v?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)", value.strip())
    path = f"supplied:{component}:{value}"
    evidence = {"component": component, "supplied_version": value}
    if not match:
        return finding(
            path,
            "supplied_version",
            "unknown",
            "Not an unambiguous upstream stable version.",
            **evidence,
        )
    version = tuple(map(int, match.groups()))
    if component == "node":
        return assess_node_version(path, version, evidence)
    branch = (version[0], version[1])
    fixes = {(1, 0): 10, (1, 1): 1, (1, 2): 1}
    if branch not in fixes or version < (1, 0, 3):
        return finding(
            path,
            "supplied_version",
            "unknown",
            "Outside snapshot patch rules; inspect the actual decoder.",
            **evidence,
        )
    minimum = f"{branch[0]}.{branch[1]}.{fixes[branch]}"
    return version_finding(path, minimum, evidence, fixed=version[2] >= fixes[branch])


def assess_node_version(path: str, version: tuple[int, ...], evidence: dict) -> dict:
    if version[0] not in NODE_FIXES:
        return finding(
            path,
            "supplied_version",
            "unknown",
            "No upstream patch rule for this Node branch; "
            "inspect embedded Brotli or vendor guidance.",
            support_status="end_of_life" if version[0] in NODE_EOL_BRANCHES else "outside_snapshot",
            **evidence,
        )
    minimum = ".".join(map(str, NODE_FIXES[version[0]]))
    return version_finding(path, minimum, evidence, fixed=version >= NODE_FIXES[version[0]])


def version_finding(path: str, minimum: str, evidence: dict, *, fixed: bool) -> dict:
    return finding(
        path,
        "supplied_version",
        "upstream_fixed_version" if fixed else "potentially_vulnerable",
        "Supplied upstream version meets the branch fix; installed build unverified."
        if fixed
        else "Supplied upstream version precedes the branch fix; vendor backport status unknown.",
        fixed_branch_version=minimum,
        **evidence,
    )


def assess_binary(  # noqa: C901 -- Ordered evidence precedence is intentionally explicit.
    data: Buffer, references: list[CodeReference]
) -> dict:
    is_elf = data.startswith(b"\x7fELF")
    code = (
        match_code(data, references)
        if is_elf
        else CodeAssessment([], "Code matching requires a supported linked ELF32/ELF64 binary")
    )
    metadata = decoder_metadata(data) if is_elf else {}
    versions = metadata.get("versions", [])
    for version in versions:
        rule = assess_version("brotli", version["version"])
        version["upstream_status"] = rule["patch_status"]
        version["fixed_branch_version"] = rule["evidence"].get("fixed_branch_version")
    old_version = any(v["upstream_status"] == "potentially_vulnerable" for v in versions)
    indicators = binary_indicators(data)
    for symbol in (*metadata.get("defined_symbols", []), *metadata.get("imported_symbols", [])):
        if symbol not in indicators:
            indicators.append(symbol)
    status, reason = code.patch_status, code.reason
    kind = "decoder_code_match" if code.matches else "binary_candidate"
    linkage = "embedded_code" if code.matches else "unknown"
    if not code.matches:
        if metadata.get("defined_symbols"):
            kind, linkage = "decoder_symbol_candidate", "defined_decoder_symbols"
            reason = "Decoder API definitions found; compiled reference did not match."
        elif metadata.get("imported_symbols"):
            kind, linkage = "decoder_import_candidate", "decoder_import"
            reason = "Decoder API imports found; inspect the library providing these imports."
        elif {symbol.decode("ascii") for symbol in DECODER_SYMBOLS}.intersection(indicators):
            kind = "decoder_symbol_candidate"
            reason = "Decoder symbol/string found; implementation and patch status unresolved."
        if versions:
            reason += " Reported decoder version is metadata, not proof of a fix."
    if old_version and status not in {"vulnerable_code_match", "mixed_code_matches"}:
        status, kind = "potentially_vulnerable", "binary_version_candidate"
        reason = (
            "Statically read decoder version precedes the upstream fix. Update or verify a "
            "vendor backport; version evidence does not prove the code is vulnerable."
        )
    return {
        "kind": kind,
        "patch_status": status,
        "reason": reason,
        "linkage": linkage,
        "indicators": indicators,
        "decoder_metadata": metadata,
        "code_matches": code.matches,
        "code_assessment": code.reason,
        "gaps": ([code.reason] if code.limited else [])
        + ([metadata["reason"]] if metadata.get("limited") else []),
    }


class Inventory:
    def __init__(  # noqa: PLR0913 -- Traversal limits plus optional code references.
        self,
        registry: dict,
        max_file_bytes: int | None,
        max_total_bytes: int | None,
        max_files: int,
        archive_depth: int,
        *,
        max_expanded_bytes: int = DEFAULT_MAX_EXPANDED_BYTES,
        max_total_expanded_bytes: int = DEFAULT_MAX_TOTAL_EXPANDED_BYTES,
        max_text_bytes: int = DEFAULT_MAX_TEXT_BYTES,
        code_references: list[CodeReference] | None = None,
    ) -> None:
        self.fingerprints = {entry["sha256"]: entry for entry in registry["fingerprints"]}
        self.max_file_bytes = max_file_bytes
        self.max_total_bytes = max_total_bytes
        self.max_files = max_files
        self.archive_depth = archive_depth
        self.max_expanded_bytes = max_expanded_bytes
        self.max_total_expanded_bytes = max_total_expanded_bytes
        self.max_text_bytes = max_text_bytes
        self.findings: list[dict] = []
        self.gaps: list[dict] = []
        self.gap_count = 0
        self.files_seen = 0
        self.files_read = 0
        self.bytes_read = 0
        self.expanded_items_read = 0
        self.expanded_bytes_read = 0
        self.archives_attempted = 0
        self.archives_opened = 0
        self.code_references = code_references
        self.checked_paths: set[Path] = set()
        self.pending_symlinks: list[Path] = []
        self.covered_symlinks = 0

    def gap(self, path: str, reason: str) -> None:
        self.gap_count += 1
        if len(self.gaps) < MAX_RECORDED_GAPS:
            self.gaps.append({"path": path, "reason": reason})

    def exhausted(self) -> bool:
        return self.files_seen >= self.max_files or (
            self.max_total_bytes is not None and self.bytes_read >= self.max_total_bytes
        )

    def record_name_fallback(self, label: str, size_bytes: int | None = None) -> None:
        if is_binary_name_candidate(label):
            self.findings.append(
                finding(
                    label,
                    "artifact_name_candidate",
                    "unknown",
                    (
                        "Artifact name suggests a Brotli component; contents, linkage, and "
                        "patch status need review."
                    ),
                    size_bytes=size_bytes,
                )
            )

    def _read_limit(self, *, expanded: bool) -> int | None:
        limits = []
        if self.max_file_bytes is not None:
            limits.append(self.max_file_bytes)
        if self.max_total_bytes is not None:
            limits.append(max(0, self.max_total_bytes - self.bytes_read))
        if expanded:
            limits.extend(
                (
                    self.max_expanded_bytes,
                    max(0, self.max_total_expanded_bytes - self.expanded_bytes_read),
                )
            )
        return min(limits) if limits else None

    def _account_read(self, size: int, *, expanded: bool) -> None:
        self.bytes_read += size
        if expanded:
            self.expanded_bytes_read += size

    def _limit_gap(self, label: str, limit: int | None, *, expanded: bool) -> None:
        scope = "Expanded member/stream" if expanded else "File/member"
        rendered = "active byte limits" if limit is None else f"the active {limit}-byte limit"
        self.gap(label, f"{scope} exceeds {rendered}.")

    @contextmanager
    def _load_data(
        self,
        stream: BinaryIO,
        size: int,
        label: str,
        *,
        expanded: bool,
        record_fallback: bool = True,
    ) -> Generator[Buffer | None]:
        limit = self._read_limit(expanded=expanded)
        if size >= 0 and limit is not None and size > limit:
            if record_fallback:
                self.record_name_fallback(label, size)
            self._limit_gap(label, limit, expanded=expanded)
            yield None
            return
        entered = False
        try:
            with read_data(stream, limit) as data:
                entered = True
                self._account_read(len(data), expanded=expanded)
                if expanded:
                    self.expanded_items_read += 1
                yield data
        except DataLimitError as error:
            if entered:
                raise
            self._account_read(error.bytes_read, expanded=expanded)
            if record_fallback:
                self.record_name_fallback(label, size if size >= 0 else None)
            self._limit_gap(label, error.limit, expanded=expanded)
            yield None

    def read(
        self,
        stream: BinaryIO,
        size: int,
        label: str,
        depth: int,
        *,
        expanded: bool = False,
    ) -> None:
        self.files_seen += 1
        if self.files_seen > self.max_files:
            self.record_name_fallback(label, size)
            self.gap(label, "File/member count limit reached.")
            return
        with self._load_data(stream, size, label, expanded=expanded) as data:
            if data is not None:
                self.files_read += 1
                self.inspect(data, label, depth)

    def inspect(self, data: Buffer, label: str, depth: int) -> None:
        digest = sha256_data(data)
        if data.startswith(b"\x7fELF") and self.code_references is not None:
            self._inspect_elf(data, label, digest)
            kind = archive_kind(data, label)
            if kind is not None:
                self._inspect_container(data, label, depth, kind, digest)
            return
        if digest in self.fingerprints:
            entry = self.fingerprints[digest]
            self.findings.append(
                finding(
                    label,
                    "exact_source_match",
                    "fixed_source" if entry["state"] == "fixed" else "vulnerable_source",
                    "Bytes match a reviewed decoder file; surrounding tree/build "
                    "and runtime remain unverified.",
                    sha256=digest,
                    git_blob=entry["git_blob"],
                    sources=entry["sources"],
                )
            )
            return
        kind = archive_kind(data, label)
        if kind is not None:
            first_finding = len(self.findings)
            self._inspect_container(data, label, depth, kind, digest)
            if len(self.findings) == first_finding:
                self.record_name_fallback(label, len(data))
            return
        self._inspect_noncontainer(data, label, digest)

    def _text(self, data: Buffer, label: str) -> str:
        if b"\0" in data[: min(TEXT_PROBE_BYTES, len(data))]:
            return ""
        if len(data) > self.max_text_bytes:
            self.gap(
                label,
                f"Text classification exceeds the {self.max_text_bytes}-byte text limit; "
                "binary analysis still completed.",
            )
            return ""
        raw = data if isinstance(data, bytes) else data[:]
        if b"\0" in raw:
            return ""
        return raw.decode("utf-8", errors="replace")

    def _inspect_noncontainer(self, data: Buffer, label: str, digest: str) -> None:
        name = leaf_name(label)
        text = self._text(data, label)
        if self._inspect_source(text, label, digest):
            return
        first_finding = len(self.findings)
        self._inspect_fetch(text, label)
        indicators = []
        if text and (
            name in MANIFEST_NAMES or name.endswith((".cmake", ".spdx.json", ".cdx.json"))
        ):
            self._inspect_manifest(text, label)
        elif not python_bytecode(data, name):
            indicators = binary_indicators(data)
            self._inspect_symbols(indicators, text, label, digest)
        if is_binary_name_candidate(label) and len(self.findings) == first_finding:
            self.findings.append(
                finding(
                    label,
                    "artifact_name_candidate",
                    "unknown",
                    "Filename suggests Brotli; implementation and patch status unresolved.",
                    sha256=digest,
                    indicators=indicators,
                )
            )

    def _inspect_elf(self, data: Buffer, label: str, digest: str) -> None:
        references = self.code_references
        if references is None:
            return
        assessment = assess_binary(data, references)
        for reason in assessment["gaps"]:
            self.gap(label, reason)
        self.findings.append(
            finding(
                label,
                assessment["kind"],
                assessment["patch_status"],
                assessment["reason"],
                sha256=digest,
                code_matches=assessment["code_matches"],
                code_assessment=assessment["code_assessment"],
                indicators=assessment["indicators"],
                decoder_metadata=assessment["decoder_metadata"],
                linkage=assessment["linkage"],
            )
        )

    def _inspect_source(self, text: str, label: str, digest: str) -> bool:
        if not text or Path(leaf_name(label)).suffix not in SOURCE_SUFFIXES:
            return False
        code = COMMENTS.sub("", text)
        if not all(
            marker in code
            for marker in (
                "DecodeBlockTypeAndLength",
                "ProcessCommandsInternal",
                "BrotliDecoderDecompressStream",
            )
        ):
            return False
        old = len(OLD_CALL.findall(code))
        new = len(NEW_CALL.findall(code))
        bad_guard = bool(BAD_GUARD.search(code))
        potential = bool(old and bad_guard)
        self.findings.append(
            finding(
                label,
                "modified_decoder_source",
                "potentially_vulnerable" if potential else "unknown",
                "Old block-switch error handling detected; review the source against the fix."
                if potential
                else "Unrecognized decoder source; fix markers alone do not validate a backport.",
                sha256=digest,
                old_switch_calls=old,
                status_switch_calls=new,
                false_on_invalid_switch=bad_guard,
                format_error_marker="BROTLI_DECODER_ERROR_FORMAT_BLOCK_SWITCH" in code,
            )
        )
        return True

    def _inspect_fetch(self, text: str, label: str) -> None:
        if not text or Path(leaf_name(label)).suffix not in JS_SUFFIXES:
            return
        matches, truncated = fetch_references(text)
        if matches:
            self.findings.append(
                finding(
                    label,
                    "node_fetch_candidate",
                    "unknown",
                    "Fetch/Undici source reference; execution runtime, fetch binding, "
                    "Node patch status, and attacker control of responses remain unverified.",
                    matches=matches,
                    matches_truncated=truncated,
                    detection="javascript_typescript_reference_heuristic",
                    required_checks=[
                        "Identify the fetch implementation and executing Node.js build.",
                        "Verify that runtime's security fix or vendor backport.",
                        "Trace URL inputs, redirects, and who controls response bytes.",
                    ],
                )
            )

    def _inspect_manifest(self, text: str, label: str) -> None:
        matches = dependency_matches(text, leaf_name(label))
        if matches:
            self.findings.append(
                finding(
                    label,
                    "dependency_candidate",
                    "unknown",
                    "Declaration or metadata only; resolve the selected implementation "
                    "and embedded version.",
                    matches=matches[:20],
                    matching_lines=len(matches),
                    possible_node_http_client=any(UNDICI.search(row["text"]) for row in matches),
                )
            )

    def _inspect_symbols(self, indicators: list[str], text: str, label: str, digest: str) -> None:
        name = leaf_name(label)
        if text and Path(name).suffix not in NATIVE_SUFFIXES and ".so." not in name:
            return
        if {symbol.decode("ascii") for symbol in DECODER_SYMBOLS}.intersection(indicators):
            self.findings.append(
                finding(
                    label,
                    "decoder_symbol_candidate",
                    "unknown",
                    "Decoder symbol/string may identify imports, exports, or embedded code; "
                    "patch status unknown.",
                    sha256=digest,
                )
            )

    def _inspect_container(
        self, data: Buffer, label: str, depth: int, kind: str, digest: str
    ) -> None:
        previous_gaps = self.gap_count
        if kind == "unsupported":
            self.gap(label, "Unsupported archive or compression format; provide unpacked files.")
        elif depth >= self.archive_depth:
            self.gap(label, "Archive nesting limit reached; members not inspected.")
        else:
            self.inspect_archive(data, label, depth, kind)
        indicators = binary_indicators(data) if self.gap_count > previous_gaps else []
        if {symbol.decode("ascii") for symbol in DECODER_SYMBOLS}.intersection(indicators):
            self.findings.append(
                finding(
                    label,
                    "container_symbol_candidate",
                    "unknown",
                    "Decoder marker in container bytes with incomplete member coverage; "
                    "it may come from a member already reported, not an independent decoder copy.",
                    sha256=digest,
                    indicators=indicators,
                    evidence_scope="container",
                    member_coverage="incomplete",
                )
            )

    def inspect_archive(self, data: Buffer, label: str, depth: int, kind: str) -> None:
        self.archives_attempted += 1
        try:
            if kind == "zip":
                self._inspect_zip(data, label, depth)
            elif kind == "tar":
                self._inspect_tar(data, label, depth)
            else:
                self._inspect_compressed(data, label, depth, kind)
        except (
            OSError,
            EOFError,
            ValueError,
            tarfile.TarError,
            zipfile.BadZipFile,
            zlib.error,
            lzma.LZMAError,
        ) as error:
            self.gap(label, f"Cannot inspect archive: {error}")

    def _inspect_zip(self, data: Buffer, label: str, depth: int) -> None:
        with buffer_stream(data) as stream, zipfile.ZipFile(stream) as archive:
            self.archives_opened += 1
            for info in archive.infolist():
                if info.is_dir():
                    continue
                member = f"{label}!{info.filename}"
                if self.exhausted():
                    self.gap(label, "Budget exhausted; remaining archive members not inspected.")
                    break
                if stat.S_ISLNK(info.external_attr >> 16):
                    self.record_name_fallback(member, info.file_size)
                    self.gap(member, "Archive symlink not followed.")
                    continue
                try:
                    with archive.open(info) as member_stream:
                        self.read(
                            cast("BinaryIO", member_stream),
                            info.file_size,
                            member,
                            depth + 1,
                            expanded=True,
                        )
                except (
                    OSError,
                    RuntimeError,
                    ValueError,
                    zipfile.BadZipFile,
                    NotImplementedError,
                    zlib.error,
                    lzma.LZMAError,
                ) as error:
                    self.record_name_fallback(member, info.file_size)
                    self.gap(member, f"Cannot inspect archive member: {error}")

    def _inspect_compressed(self, data: Buffer, label: str, depth: int, kind: str) -> None:
        open_compressed = {"gzip": gzip.open, "bzip2": bz2.open, "xz": lzma.open}[kind]
        with (
            buffer_stream(data) as source,
            open_compressed(source, "rb") as expanded_stream,
            self._load_data(
                cast("BinaryIO", expanded_stream),
                -1,
                label,
                expanded=True,
                record_fallback=False,
            ) as expanded,
        ):
            if expanded is None:
                return
            self.archives_opened += 1
            name = leaf_name(label)
            if tar_content(expanded) or name.endswith(
                (
                    ".tar.gz",
                    ".tar.bz2",
                    ".tar.xz",
                    ".tgz",
                    ".txz",
                    ".tbz",
                    ".tbz2",
                )
            ):
                self._inspect_tar(expanded, label, depth, count_open=False)
                return
            member = f"{label}!{Path(name).stem or 'decompressed'}"
            self.files_seen += 1
            if self.files_seen > self.max_files:
                self.record_name_fallback(member, len(expanded))
                self.gap(member, "File/member count limit reached.")
                return
            self.files_read += 1
            self.inspect(expanded, member, depth + 1)

    def _inspect_tar(
        self, data: Buffer, label: str, depth: int, *, count_open: bool = True
    ) -> None:
        with buffer_stream(data) as stream, tarfile.open(fileobj=stream, mode="r|") as archive:
            if count_open:
                self.archives_opened += 1
            for info in archive:
                if info.isdir():
                    continue
                member = f"{label}!{info.name}"
                if self.exhausted():
                    self.gap(label, "Budget exhausted; remaining archive members not inspected.")
                    break
                if not info.isfile():
                    self.record_name_fallback(member, info.size)
                    self.gap(member, "Nonregular archive member not followed.")
                    continue
                member_stream = archive.extractfile(info)
                if member_stream is not None:
                    with member_stream:
                        self.read(
                            cast("BinaryIO", member_stream),
                            info.size,
                            member,
                            depth + 1,
                            expanded=True,
                        )

    def scan_file(self, path: Path) -> None:  # noqa: C901 -- Stable-fd lifecycle is one operation.
        label = str(path)
        try:
            metadata = path.lstat()
            if stat.S_ISLNK(metadata.st_mode):
                self._record_symlink(path)
                return
            if not stat.S_ISREG(metadata.st_mode):
                self.record_name_fallback(label)
                self.gap(label, "Nonregular file not read.")
                return
            descriptor = os.open(
                path,
                os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0),
            )
            with os.fdopen(descriptor, "rb") as stream:
                metadata = os.fstat(stream.fileno())
                if not stat.S_ISREG(metadata.st_mode):
                    self.record_name_fallback(label)
                    self.gap(label, "Nonregular file not read.")
                    return
                checkpoint = (
                    len(self.findings),
                    len(self.gaps),
                    self.gap_count,
                    self.files_seen,
                    self.files_read,
                    self.bytes_read,
                    self.expanded_items_read,
                    self.expanded_bytes_read,
                    self.archives_attempted,
                    self.archives_opened,
                )
                self.files_seen += 1
                if self.files_seen > self.max_files:
                    self.record_name_fallback(label, metadata.st_size)
                    self.gap(label, "File/member count limit reached.")
                    return
                limit = self._read_limit(expanded=False)
                if limit is not None and metadata.st_size > limit:
                    self.record_name_fallback(label, metadata.st_size)
                    self._limit_gap(label, limit, expanded=False)
                    return
                try:
                    with file_data(stream, limit) as data:
                        self._account_read(len(data), expanded=False)
                        self.files_read += 1
                        self.inspect(data, label, 0)
                except (DataChangedError, DataLimitError) as error:
                    (
                        findings,
                        gaps,
                        self.gap_count,
                        seen,
                        self.files_read,
                        self.bytes_read,
                        self.expanded_items_read,
                        self.expanded_bytes_read,
                        self.archives_attempted,
                        self.archives_opened,
                    ) = checkpoint
                    del self.findings[findings:]
                    del self.gaps[gaps:]
                    self.files_seen = seen + 1
                    self.record_name_fallback(label, metadata.st_size)
                    if isinstance(error, DataLimitError):
                        self._limit_gap(label, error.limit, expanded=False)
                    else:
                        self.gap(label, str(error))
                    return
                self.checked_paths.add(path.resolve())
        except OSError as error:
            self.record_name_fallback(label)
            self.gap(label, f"Cannot read file: {error}")

    def scan_root(self, root: Path) -> None:
        self.scan_roots([root])

    def scan_roots(self, roots: list[Path]) -> None:
        try:
            for root in roots:
                if self.exhausted():
                    self.gap(str(root), "Budget exhausted; root not inspected.")
                else:
                    self._scan_root(root)
        finally:
            self._check_symlinks()

    def _record_symlink(self, path: Path) -> None:
        self.files_seen += 1
        if self.files_seen > self.max_files:
            self.gap(str(path), "File count limit reached; symlink target not checked.")
        else:
            self.pending_symlinks.append(path)

    def _check_symlinks(self) -> None:
        for path in self.pending_symlinks:
            try:
                target = path.resolve(strict=True)
            except (OSError, RuntimeError):
                target = None
            if target in self.checked_paths:
                self.covered_symlinks += 1
            else:
                self.gap(
                    str(path),
                    "Symlink target was not checked directly; scan the real target separately.",
                )
        self.pending_symlinks.clear()

    def _scan_root(self, root: Path) -> None:
        if root.is_symlink() or not root.is_dir():
            self.scan_file(root)
            return

        def walk_error(error: OSError) -> None:
            self.gap(str(error.filename), f"Cannot walk directory: {error}")

        for directory, dirs, files in os.walk(root, followlinks=False, onerror=walk_error):
            if self.exhausted():
                self.gap(
                    str(root),
                    "Budget exhausted; remaining filesystem paths not inspected.",
                )
                break
            self.checked_paths.add(Path(directory).resolve())
            dirs[:] = self._filter_directories(Path(directory), dirs)
            for name in sorted(files):
                if name == ".git":
                    continue
                if self.exhausted():
                    self.gap(
                        str(root),
                        "Budget exhausted; remaining filesystem paths not inspected.",
                    )
                    return
                self.scan_file(Path(directory) / name)

    def _filter_directories(self, directory: Path, names: list[str]) -> list[str]:
        kept = []
        for name in sorted(names):
            path = directory / name
            if name == ".git":
                continue
            if path.is_symlink():
                self._record_symlink(path)
            else:
                kept.append(name)
        return kept


def positive(value: str) -> int:
    number = int(value)
    if number <= 0:
        message = "must be positive"
        raise argparse.ArgumentTypeError(message)
    return number


def explain_possible(item: dict) -> dict:
    evidence = item["evidence"]
    if item["kind"] == "binary_version_candidate":
        versions = evidence["decoder_metadata"]["versions"]
        affected = [v for v in versions if v["upstream_status"] == "potentially_vulnerable"]
        labels = ", ".join(sorted({v["version"] for v in affected}))
        fixes = ", ".join(sorted({v["fixed_branch_version"] for v in affected}))
        return {
            "result": f"Embedded decoder reports Brotli {labels}; update review needed",
            "next_action": (
                f"Update to the corresponding fixed branch release ({fixes}) or verify a "
                "vendor backport for this exact build. The version was read statically, "
                "not by running the binary; it does not prove vulnerable behavior."
            ),
        }
    if item["kind"] == "supplied_version":
        component = "Node.js" if evidence["component"] == "node" else "Brotli"
        minimum = evidence["fixed_branch_version"]
        return {
            "result": f"The supplied {component} version may need an update",
            "next_action": (
                f"Ask the maintainer to update {component} to {minimum} or a later "
                "fixed release on that branch, or confirm a vendor-supplied fix. "
                "Check the version actually used by the running app."
            ),
        }
    return {
        "result": "Possible vulnerable Brotli source",
        "next_action": (
            "Ask the maintainer to check the changes in this copy against the "
            "Brotli fix and update it if needed. Confirm whether the running app "
            "uses this copy."
        ),
    }


def explain_finding(item: dict) -> dict:  # noqa: PLR0911 -- Explicit per-evidence guidance.
    status = item["patch_status"]
    if status in {"vulnerable_code_match", "mixed_code_matches", "fixed_code_match"}:
        return {
            "result": "Affected decoder code matched in this binary"
            if status != "fixed_code_match"
            else "Fixed decoder code matched in this binary",
            "next_action": (
                "Review the matched build evidence and rebuild affected embedded copies. "
                "Confirm runtime use and check other copies; a fixed match does not clear "
                "the whole binary."
            ),
        }
    if status == "vulnerable_source":
        return {
            "result": "Vulnerable Brotli source found",
            "next_action": (
                "Have the app's maintainer check whether this copy is used. If it "
                "is, update its Brotli dependency, rebuild and deploy the app, and "
                "confirm the running app has the fix."
            ),
        }
    if status == "potentially_vulnerable":
        return explain_possible(item)
    if status == "fixed_source":
        return {
            "result": "Fix identified in this source file",
            "next_action": (
                "Confirm that the running app uses this fixed copy. Other copies "
                "listed here need their own checks."
            ),
        }
    if status == "upstream_fixed_version":
        return {
            "result": "Supplied version meets a known fix",
            "next_action": (
                "Confirm that this version belongs to the running app, and check any "
                "additional Brotli copies bundled inside it."
            ),
        }
    if item["kind"] == "decoder_import_candidate":
        return {
            "result": "Brotli decoder imported from another library",
            "next_action": (
                "Scan the actual library providing this decoder in the deployment; "
                "an import does not identify its patch state."
            ),
        }
    if item["evidence"].get("decoder_metadata", {}).get("defined_symbols"):
        return {
            "result": "Brotli decoder definitions found; patch state unresolved",
            "next_action": (
                "Verify a vendor fix for this exact build or provide matching code references. "
                "A reported fixed version alone does not clear embedded copies."
            ),
        }
    return (
        end_of_life_advice()
        if item["evidence"].get("support_status") == "end_of_life"
        else explain_lead(item)
    )


def end_of_life_advice() -> dict:
    return {
        "result": "This Node.js release is end of life",
        "next_action": (
            "Upgrade to a supported Node.js branch: 22.23.4, 24.22.1, or 26.12.1 "
            "or a later fixed release on that branch. Confirm the running app uses "
            "the new build. This version alone does not identify its embedded Brotli."
        ),
    }


def fetch_related(item: dict) -> bool:
    return item["kind"] == "node_fetch_candidate" or item["evidence"].get(
        "possible_node_http_client", False
    )


def fetch_advice() -> dict:
    return {
        "result": "Possible Node.js fetch response-decoding path",
        "next_action": (
            "Confirm whether this uses Node's built-in fetch or Undici, a browser, or a "
            "custom implementation. Identify the running app's Node.js build and check it "
            "with --node-version and vendor patch records. Trace URL inputs and redirects "
            "to establish who can supply response bytes. Node's built-in fetch and Undici's "
            "fetch can decode Brotli responses through node:zlib without a Brotli package "
            "declaration. "
            "Update the actual Node.js runtime if affected; adding a Brotli npm package "
            "does not patch that runtime. Review any other dependencies listed in this file."
        ),
    }


def explain_lead(item: dict) -> dict:
    if fetch_related(item):
        return fetch_advice()
    if item["kind"] == "supplied_version":
        return {
            "result": "Supplied version needs verification",
            "next_action": (
                f"{item['reason']} Obtain the running app's exact component build and vendor "
                "security-patch records. A vendor suffix or prerelease cannot be assessed "
                "as an upstream stable release."
            ),
        }
    advice = {
        "binary_candidate": (
            "Unclassified native binary",
            (
                "No supported decoder code matched. Use build records or scoped disassembly "
                "if this binary may embed Brotli; this is not evidence of absence."
            ),
        ),
        "dependency_candidate": (
            "Dependency or metadata mention",
            (
                "Check the matching declaration to see whether it selects a Brotli decoder. "
                "Descriptions and package names alone do not establish an installed decoder. "
                "If selected, identify its implementation and the build used by the running app."
            ),
        ),
        "decoder_symbol_candidate": (
            "Possible decoder in a binary",
            (
                "Have the maintainer identify the library that provides this decoder symbol "
                "and its patch status, using build records or binary analysis. Confirm whether "
                "the running app loads or embeds it."
            ),
        ),
        "container_symbol_candidate": (
            "Decoder marker in an incompletely checked container",
            (
                "Resolve this container's scan gaps or inspect its unpacked contents. "
                "The marker may belong to members already listed; do not count the container "
                "as another decoder copy. Identify the decoder and patch status in its members."
            ),
        ),
        "artifact_name_candidate": (
            "Possible Brotli component by filename",
            (
                "Check this package or library's contents and build records to identify its "
                "decoder and patch status. The filename alone does not establish its version."
            ),
        ),
        "modified_decoder_source": (
            "Modified decoder source needs review",
            (
                "Compare this source and its surrounding headers against the upstream fix. "
                "Fix markers alone do not validate a backport. Confirm whether the app builds it."
            ),
        ),
    }
    result, action = advice.get(
        item["kind"],
        (
            "More information needed",
            (
                "Ask the maintainer to identify the decoder used by this component "
                "and its patch status."
            ),
        ),
    )
    return {
        "result": result,
        "next_action": action,
    }


def end_of_life_summary() -> dict:
    return {
        "status": "needs_attention",
        "title": "The supplied Node.js release is end of life",
        "meaning": (
            "This Node.js branch no longer receives upstream security fixes. Its version "
            "alone does not establish the patch status of the embedded Brotli decoder."
        ),
        "next_steps": [
            (
                "Upgrade to Node.js 22.23.4, 24.22.1, or 26.12.1 or a later fixed release "
                "on that branch, and confirm the running app uses it."
            ),
            "Resolve any unchecked files and review any other Brotli copies listed below.",
        ],
    }


def attention_summary(statuses: set[str]) -> dict:
    if statuses & {"vulnerable_code_match", "mixed_code_matches"}:
        return {
            "status": "needs_attention",
            "title": "Affected Brotli decoder code matched in a binary",
            "meaning": (
                "Executable code matches an affected decoder reference. This is code "
                "identity evidence, not proof of runtime use or exploitability."
            ),
            "next_steps": [
                (
                    "Review the reference build evidence, update affected copies, "
                    "rebuild and redeploy."
                ),
                "Check all other copies, source findings, and scan gaps separately.",
            ],
        }
    if "vulnerable_source" in statuses:
        return {
            "status": "needs_attention",
            "title": "Vulnerable Brotli source found",
            "meaning": (
                "Affected Brotli code is present. We still need to check whether "
                "your running app uses it to process data supplied by other people."
            ),
            "next_steps": [
                (
                    "Ask the app's maintainer to check the flagged copies below and "
                    "update any that the app uses."
                ),
                (
                    "If the copy is included inside the app, rebuild and deploy the app "
                    "after updating it. Updating the computer's Brotli package alone may "
                    "leave the copy inside the app unchanged."
                ),
                (
                    "Confirm the updated copy is in the running app. The fixed copies "
                    "listed here do not clear other copies."
                ),
            ],
        }
    if "potentially_vulnerable" in statuses:
        return {
            "status": "needs_attention",
            "title": "A Brotli update may be needed",
            "meaning": (
                "Some checked code or decoder versions may be affected. A "
                "maintainer needs to confirm which copy the app uses and whether its "
                "vendor has already fixed it."
            ),
            "next_steps": [
                "Follow the update or verification step for each flagged item below.",
                "Confirm the running app uses the fixed copy after the update.",
            ],
        }
    return {}


def fetch_summary() -> dict:
    return {
        "status": "needs_more_information",
        "title": "Check the runtime used for fetch responses",
        "meaning": (
            "Fetch or Undici references were found. If they run through Node.js, Brotli "
            "responses can reach the runtime's native decoder. These references alone "
            "do not identify the runtime, its patch status, or who controls responses."
        ),
        "next_steps": [
            (
                "Identify the actual fetch implementation and running Node.js build. "
                "Supply that build's version with --node-version and check vendor patch records."
            ),
            "Trace URL inputs and redirects to determine who can supply response bytes.",
            (
                "If affected, update and redeploy the Node.js runtime, then confirm the "
                "app uses it. A fixed standalone Brotli package does not clear this path."
            ),
        ],
    }


def summarize(report: dict) -> dict:
    findings = report["findings"]
    statuses = {item["patch_status"] for item in findings}
    attention = attention_summary(statuses)
    if attention:
        return attention
    if any(item["evidence"].get("support_status") == "end_of_life" for item in findings):
        return end_of_life_summary()
    if report["coverage"]["gap_count"]:
        return {
            "status": "needs_more_information",
            "title": "Some files could not be checked",
            "meaning": (
                "This check is incomplete. Any fixes identified in other files do "
                "not tell us what is inside the skipped files."
            ),
            "next_steps": [
                (
                    "For byte, file-count, or depth limits, rerun a smaller relevant folder "
                    "or raise the named limit shown in --help."
                ),
                "For unsupported packages, provide unpacked files and scan them again.",
                "For symlinks, scan the real target separately if it was not checked directly.",
                "Ask the app's maintainer for any files or build records you cannot obtain.",
            ],
        }
    if findings and statuses <= GOOD:
        return {
            "status": "checked_items_include_fix",
            "title": "Fix identified in the checked items",
            "meaning": (
                "The checked source files or supplied versions match a known fix. "
                "This does not yet confirm what your running app uses or cover "
                "copies outside this check."
            ),
            "next_steps": [
                "Confirm the running app uses these fixed copies or versions.",
                "Check any additional copies bundled inside the app or its packages.",
            ],
        }
    if any(fetch_related(item) for item in findings):
        return fetch_summary()
    return {
        "status": "needs_more_information",
        "title": "More information needed",
        "meaning": "This check could not confirm whether the app's Brotli decoder includes the fix."
        if findings
        else (
            "No matching Brotli decoder was identified in the files checked. It "
            "may be inside a built application or outside the folders checked."
        ),
        "next_steps": [
            (
                "Ask the app's maintainer or hosting provider to identify the Brotli "
                "copies used by the running app and confirm whether they include the "
                "fix. Share the findings below if available."
            )
        ],
    }


def display_copies(findings: list[dict]) -> list[dict]:
    # Merge evidence about the same file for display; full evidence stays in JSON.
    priorities = {
        "vulnerable_code_match": 0,
        "mixed_code_matches": 0,
        "fixed_code_match": 2,
        "vulnerable_source": 0,
        "potentially_vulnerable": 1,
        "unknown": 2,
        "fixed_source": 3,
        "upstream_fixed_version": 3,
    }
    copies = {}
    for item in findings:
        previous = copies.get(item["path"])
        if (
            previous is None
            or priorities[item["patch_status"]] < priorities[previous["patch_status"]]
        ):
            copies[item["path"]] = item
    return sorted(copies.values(), key=lambda item: priorities[item["patch_status"]])


def display_location(item: dict) -> str:
    if item["kind"] == "supplied_version" and "supplied_version" in item["evidence"]:
        evidence = item["evidence"]
        component = "Node.js" if evidence["component"] == "node" else "Brotli"
        return safe_text(f"{component} {evidence['supplied_version']} (version supplied)")
    return safe_text(item["path"])


def render_copies(findings: list[dict], *, containers: bool = False) -> list[str]:
    ordered = display_copies(
        [item for item in findings if (item["kind"] == "container_symbol_candidate") == containers]
    )
    if not ordered:
        return []
    lines = ["", "Containers needing inspection:" if containers else "Files and versions checked:"]
    for item in ordered[:MAX_DISPLAY_COPIES]:
        explanation = explain_finding(item)
        lines.extend(
            [
                f"- {explanation['result']}: {display_location(item)}",
                f"  Next: {explanation['next_action']}",
            ]
        )
    if len(ordered) > MAX_DISPLAY_COPIES:
        lines.append(
            f"- {len(ordered) - MAX_DISPLAY_COPIES} more items are included "
            "in the full report (--json)."
        )
    return lines


def render_gaps(coverage: dict) -> list[str]:
    gaps = coverage["gaps"]
    if not gaps:
        return []
    lines = ["", "Could not check:"]
    lines.extend(
        f"- {safe_text(gap['path'])}: {safe_text(gap['reason'])}" for gap in gaps[:MAX_DISPLAY_GAPS]
    )
    remaining = len(gaps) - min(MAX_DISPLAY_GAPS, len(gaps))
    if remaining:
        lines.append(f"- {remaining} more recorded gap details are available with --json.")
    omitted = coverage["gap_count"] - len(gaps)
    if omitted:
        lines.append(
            f"- {omitted} additional gaps have no retained details "
            f"({MAX_RECORDED_GAPS}-record limit). Rerun a smaller scope to inspect them."
        )
    return lines


def render_text(report: dict) -> str:
    summary = report["user_summary"]
    lines = [
        f"Brotli check: {summary['title']}",
        "",
        summary["meaning"],
        "",
        "What to do next:",
    ]
    lines.extend(f"- {step}" for step in summary["next_steps"])
    lines.extend(render_copies(report["findings"]))
    lines.extend(render_copies(report["findings"], containers=True))
    lines.extend(render_gaps(report["coverage"]))
    lines.extend(
        [
            "",
            (
                f"Checked {report['coverage']['files_read']} files and archive entries. "
                f"{report['cve']}; advisory data: {report['snapshot_date']}."
            ),
            "Full technical evidence for a maintainer: run with --json.",
        ]
    )
    return "\n".join(lines)


def parse_args(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Check local files for Brotli copies that may need a security update.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument(
        "roots",
        nargs="*",
        type=Path,
        help="project folders or app packages to check; defaults to the current folder",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="show full technical evidence for maintainers and automation",
    )
    parser.add_argument(
        "--brotli-version",
        action="append",
        default=[],
        help="upstream Brotli version obtained separately; repeat for multiple copies",
    )
    parser.add_argument(
        "--node-version",
        action="append",
        default=[],
        help="upstream Node.js version obtained separately; repeat for multiple copies",
    )
    parser.add_argument(
        "--code-references",
        type=Path,
        default=REFERENCES,
        help="compiled ELF decoder references; unmatched code remains unidentified",
    )
    parser.add_argument(
        "--max-file-bytes",
        type=positive,
        default=DEFAULT_MAX_FILE_BYTES,
        help="optional maximum bytes read from one file/member (default: no cap)",
    )
    parser.add_argument(
        "--max-total-bytes",
        type=positive,
        default=DEFAULT_MAX_TOTAL_BYTES,
        help="optional maximum cumulative bytes read (default: no cap)",
    )
    parser.add_argument(
        "--max-expanded-bytes",
        type=positive,
        default=DEFAULT_MAX_EXPANDED_BYTES,
        help="maximum expanded bytes read from one archive member/stream",
    )
    parser.add_argument(
        "--max-total-expanded-bytes",
        type=positive,
        default=DEFAULT_MAX_TOTAL_EXPANDED_BYTES,
        help="maximum cumulative expanded archive/member bytes",
    )
    parser.add_argument(
        "--max-text-bytes",
        type=positive,
        default=DEFAULT_MAX_TEXT_BYTES,
        help="maximum bytes decoded for source/manifest text classification",
    )
    parser.add_argument(
        "--max-files",
        type=positive,
        default=50000,
        help="maximum filesystem files, symlinks, and archive members checked",
    )
    parser.add_argument(
        "--archive-depth",
        type=int,
        choices=range(6),
        default=2,
        help="maximum nested containers opened; 0 disables archive traversal",
    )
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    roots = args.roots or ([] if args.brotli_version or args.node_version else [Path.cwd()])
    registry = json.loads(REGISTRY.read_text(encoding="utf-8"))
    try:
        code_references = load_references(args.code_references)
    except (OSError, ValueError) as error:
        print(f"Invalid code references: {error}", file=sys.stderr)
        return 2
    inventory = Inventory(
        registry,
        args.max_file_bytes,
        args.max_total_bytes,
        args.max_files,
        args.archive_depth,
        max_expanded_bytes=args.max_expanded_bytes,
        max_total_expanded_bytes=args.max_total_expanded_bytes,
        max_text_bytes=args.max_text_bytes,
        code_references=code_references,
    )
    inventory.scan_roots([root.expanduser().absolute() for root in roots])
    for component, versions in (
        ("brotli", args.brotli_version),
        ("node", args.node_version),
    ):
        inventory.findings.extend(assess_version(component, value) for value in versions)
    statuses = {item["patch_status"] for item in inventory.findings}
    if statuses & WARNING:
        result, code = "affected_evidence_found", 1
    elif inventory.gap_count or not statuses or statuses - GOOD:
        result, code = "review_required", 3
    else:
        result, code = "reported_items_meet_fix_rules", 0
    report = make_report(inventory, roots, args, registry, result)
    report["user_summary"] = summarize(report)
    if args.json:
        print(json.dumps(report, indent=2, ensure_ascii=True))
    else:
        print(render_text(report))
    return code


def make_report(
    inventory: Inventory,
    roots: list[Path],
    args: argparse.Namespace,
    registry: dict,
    result: str,
) -> dict:
    return {
        "schema_version": 1,
        "cve": "CVE-2026-111497",
        "result": result,
        "snapshot_date": registry["verified_on"],
        "advisory_freshness": "not_checked_offline",
        "scope": [str(root.expanduser().absolute()) for root in roots],
        "findings": inventory.findings,
        "coverage": {
            "files_seen": inventory.files_seen,
            "files_read": inventory.files_read,
            "bytes_read": inventory.bytes_read,
            "expanded_items_read": inventory.expanded_items_read,
            "expanded_bytes_read": inventory.expanded_bytes_read,
            "archives_attempted": inventory.archives_attempted,
            "archives_opened": inventory.archives_opened,
            "gap_count": inventory.gap_count,
            "gap_details_retained": len(inventory.gaps),
            "gap_details_omitted": inventory.gap_count - len(inventory.gaps),
            "gaps": inventory.gaps,
            "excluded": [".git"],
            "symlinks_followed": False,
            "symlink_targets_checked_directly": inventory.covered_symlinks,
            "limits": {
                key: getattr(args, key)
                for key in (
                    "max_file_bytes",
                    "max_total_bytes",
                    "max_expanded_bytes",
                    "max_total_expanded_bytes",
                    "max_text_bytes",
                    "max_files",
                    "archive_depth",
                )
            },
            "limitations": [
                "No absence-of-Brotli guarantee.",
                (
                    "Compiled code matches exclude relocation targets/data "
                    "and do not clear other copies."
                ),
                "Unrecognized or signature-free container/compression formats can be missed.",
                "Container layers are inspected without reconstructing the final image filesystem.",
                (
                    "Static/stripped binaries, encrypted archives, remote build inputs, "
                    "and unprovided artifacts require review."
                ),
                (
                    "Build selection, shipment, runtime identity, reachability, and "
                    "exploitability are not established."
                ),
            ],
        },
    }


# gremlin: pardon[out_of_scope] CLI bootstrap guard; imports must not launch argparse.
if __name__ == "__main__":
    sys.exit(main())
```

## scripts/brotli_source_discovery.py

```python
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.12,<3.16"
# dependencies = ["trailmark==0.7.0"]
# ///
"""Collect Trailmark source-call evidence for an agent's Brotli investigation."""

from __future__ import annotations

import argparse
import json
import sys
from collections import defaultdict, deque
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from trailmark.models.edges import CodeEdge
    from trailmark.models.graph import CodeGraph
    from trailmark.models.nodes import CodeUnit

DECODER_SYMBOLS = frozenset(
    {
        "BrotliDecoderDecompress",
        "BrotliDecoderDecompressStream",
        "brotliDecompress",
        "brotliDecompressSync",
        "createBrotliDecompress",
    }
)
LIMITATIONS = [
    (
        "Source-call evidence does not establish build selection, runtime reachability, "
        "patch status, or exploitability. No graph path does not mean unreachable."
    ),
    "Bounds limit reported evidence and caller traversal, not initial source parsing or indexing.",
    (
        "Trailmark does not model every call; top-level JavaScript/TypeScript calls may be absent "
        "without a diagnostic."
    ),
    (
        "ESM import aliases and dynamic dispatch may remain unresolved. Use --symbol to add "
        "reviewed aliases or wrappers; no import binding is invented by this helper."
    ),
    (
        "Directory parsing skips vendor, node_modules, build, dist, dot-directories and common "
        "environment/cache directories; directory symlinks are not followed. "
        "File symlinks may be read."
    ),
    (
        "Entrypoints are heuristic or repository configured. Missing entrypoint tags do not prove "
        "that a function is unused. .trailmark/entrypoints.toml override tags are repository "
        "declarations, not proof of runtime exposure."
    ),
    (
        "Edge confidence describes static inference. Deferred callback edges do not establish "
        "callback execution; fetch names do not identify Node.js or its actual runtime version."
    ),
    (
        "Repository .trailmark/links.toml edges are declarations, not extracted source callsites. "
        "Empty parse diagnostics do not establish complete source coverage."
    ),
]


@dataclass(frozen=True)
class Limits:
    max_candidates: int = 32
    max_caller_nodes: int = 128
    max_call_edges: int = 256
    max_depth: int = 6
    max_diagnostics: int = 100


def node_row(node: CodeUnit) -> dict:
    return {
        "id": node.id,
        "name": node.name,
        "kind": node.kind.value,
        "origin": node.origin.value,
        "location": asdict(node.location),
        "attributes": dict(node.attributes),
    }


def edge_row(index: int, edge: CodeEdge, graph: CodeGraph) -> dict:
    target = graph.nodes.get(edge.target_id)
    attributes = dict(target.attributes) if target is not None else {}
    return {
        "index": index,
        "source": edge.source_id,
        "target": edge.target_id,
        "kind": edge.kind.value,
        "confidence": edge.confidence.value,
        "location": asdict(edge.location) if edge.location is not None else None,
        "attributes": dict(edge.attributes),
        "resolution": "unresolved"
        if target is None or attributes.get("proxy_kind") == "unresolved"
        else "source"
        if target.origin.value == "source"
        else "external_or_synthetic",
    }


def candidate_symbol(edge: CodeEdge, graph: CodeGraph, symbols: tuple[str, ...]) -> str | None:
    target = graph.nodes.get(edge.target_id)
    attributes = dict(edge.attributes)
    raw = str(attributes.get("raw_target", target.name if target else edge.target_id))
    label = raw.rsplit(":", 1)[-1].rsplit(".", 1)[-1]
    names = {edge.target_id, raw, label}
    if target is not None:
        names.add(target.name)
    requested = next((symbol for symbol in symbols if symbol in names), None)
    if requested is not None:
        return requested
    if label in DECODER_SYMBOLS:
        return label
    if label == "decompress" and "brotli" in raw.lower():
        return raw
    return fetch_symbol(label, raw, target, attributes)


def fetch_symbol(label: str, raw: str, target: CodeUnit | None, attributes: dict) -> str | None:
    if attributes.get("binding_blocked") or (target and target.origin.value == "source"):
        return None
    if label == "fetch":
        return "fetch"
    if raw.startswith(("undici:", "undici.")) and label in {"request", "stream", "pipeline"}:
        return raw
    return None


def empty_report(root: str, limits: Limits) -> dict:
    return {
        "schema_version": 1,
        "tool": "brotli_source_discovery",
        "trailmark_version": None,
        "root": root,
        "status": "incomplete",
        "coverage": {
            "source_file_counts": {},
            "source_files": None,
            "initial_parsing_bounded": False,
        },
        "limits": asdict(limits),
        "candidates": [],
        "candidate_count": 0,
        "neighborhood": {"nodes": [], "caller_depths": {}, "calls": [], "entrypoints": []},
        "parse_diagnostics": [],
        "parse_diagnostic_count": 0,
        "gaps": [],
        "truncation": dict.fromkeys(
            ("candidates", "caller_nodes", "call_edges", "depth", "diagnostics"), False
        ),
        "limitations": list(LIMITATIONS),
    }


@dataclass
class Neighborhood:
    graph: CodeGraph
    limits: Limits
    truncation: dict
    edges: dict[int, CodeEdge] = field(default_factory=dict)
    depths: dict[str, int] = field(default_factory=dict)
    queue: deque[str] = field(default_factory=deque)

    def add_caller(self, node_id: str, depth: int) -> bool:
        if node_id in self.depths:
            return True
        if len(self.depths) >= self.limits.max_caller_nodes:
            self.truncation["caller_nodes"] = True
            return False
        self.depths[node_id] = depth
        self.queue.append(node_id)
        return True

    def add_edge(self, index: int, edge: CodeEdge) -> bool:
        if index in self.edges:
            return True
        if len(self.edges) >= self.limits.max_call_edges:
            self.truncation["call_edges"] = True
            return False
        self.edges[index] = edge
        return True

    def callers(self, incoming: dict[str, list[tuple[int, CodeEdge]]]) -> None:
        while self.queue:
            node_id = self.queue.popleft()
            depth = self.depths[node_id]
            if depth >= self.limits.max_depth:
                if incoming.get(node_id):
                    self.truncation["depth"] = True
                continue
            if not self.expand(incoming.get(node_id, []), depth):
                break

    def expand(self, incoming: list[tuple[int, CodeEdge]], depth: int) -> bool:
        for index, edge in incoming:
            if not self.add_edge(index, edge):
                return False
            if not self.add_caller(edge.source_id, depth + 1):
                self.edges.pop(index, None)
                return False
        return True

    def unresolved(self, calls: list[tuple[int, CodeEdge]]) -> None:
        unresolved = set(self.graph.unresolved_calls())
        for index, edge in calls:
            if (
                edge.source_id in self.depths
                and edge in unresolved
                and not self.add_edge(index, edge)
            ):
                break

    def report(self) -> dict:
        node_ids = set(self.depths)
        for edge in self.edges.values():
            node_ids.update((edge.source_id, edge.target_id))
        return {
            "nodes": [
                node_row(self.graph.nodes[key])
                for key in sorted(node_ids)
                if key in self.graph.nodes
            ],
            "caller_depths": dict(self.depths),
            "calls": [edge_row(index, edge, self.graph) for index, edge in self.edges.items()],
            "entrypoints": [
                {
                    "node_id": key,
                    "kind": tag.kind.value,
                    "trust_level": tag.trust_level.value,
                    "asset_value": tag.asset_value.value,
                    "description": tag.description,
                }
                for key, tag in self.graph.entrypoints.items()
                if key in self.depths
            ],
        }


def discover(
    graph: CodeGraph,
    version: str,
    limits: Limits,
    symbols: tuple[str, ...] = (),
) -> dict:
    report = empty_report(graph.root_path, limits)
    report["trailmark_version"] = version
    report["coverage"] = {
        "language": graph.language,
        "source_file_counts": dict(graph.source_file_counts),
        "source_files": sum(graph.source_file_counts.values()),
        "initial_parsing_bounded": False,
    }
    neighborhood = Neighborhood(graph, limits, report["truncation"])
    calls = [(index, edge) for index, edge in enumerate(graph.edges) if edge.kind.value == "calls"]
    incoming: dict[str, list[tuple[int, CodeEdge]]] = defaultdict(list)
    for index, edge in calls:
        incoming[edge.target_id].append((index, edge))
        symbol = candidate_symbol(edge, graph, symbols)
        if symbol is not None:
            add_candidate(report, neighborhood, index, edge, symbol)
    neighborhood.callers(incoming)
    neighborhood.unresolved(calls)
    report["neighborhood"] = neighborhood.report()
    finish_report(report, graph, limits)
    return report


def add_candidate(
    report: dict, neighborhood: Neighborhood, index: int, edge: CodeEdge, symbol: str
) -> None:
    report["candidate_count"] += 1
    if len(report["candidates"]) >= neighborhood.limits.max_candidates:
        report["truncation"]["candidates"] = True
        return
    if not neighborhood.add_edge(index, edge):
        return
    neighborhood.add_caller(edge.source_id, 0)
    report["candidates"].append(
        {"symbol": symbol, "call": edge_row(index, edge, neighborhood.graph)}
    )


def finish_report(report: dict, graph: CodeGraph, limits: Limits) -> None:
    report["parse_diagnostics"] = [
        asdict(diagnostic) for diagnostic in graph.parse_diagnostics[: limits.max_diagnostics]
    ]
    report["parse_diagnostic_count"] = len(graph.parse_diagnostics)
    report["truncation"]["diagnostics"] = len(graph.parse_diagnostics) > limits.max_diagnostics
    if not report["coverage"]["source_files"]:
        report["gaps"].append({"kind": "empty_source", "message": "No source files inspected."})
    if graph.parse_diagnostics:
        report["gaps"].append({"kind": "parse_diagnostics", "message": "Source parsing has gaps."})
    if any(report["truncation"].values()):
        report["gaps"].append({"kind": "bounded_output", "message": "Evidence was truncated."})
    report["status"] = "incomplete" if report["gaps"] else "evidence_collected"


def runtime_version() -> str:
    from trailmark import __version__  # noqa: PLC0415 -- Optional runtime dependency.

    if tuple(int(part) for part in __version__.split(".")[:2]) < (0, 7):
        message = "Trailmark >=0.7 is required; use the pinned uv script invocation."
        raise ValueError(message)
    return __version__


def load_graph(root: str, language: str) -> CodeGraph:
    from trailmark.analysis.entrypoints import detect_entrypoints  # noqa: PLC0415
    from trailmark.parse import parse_directory  # noqa: PLC0415

    graph = parse_directory(root, language=language, allow_empty=True, strict=False)
    graph.entrypoints.update(detect_entrypoints(graph, root))
    return graph


def positive(value: str) -> int:
    number = int(value)
    if number <= 0:
        message = "must be positive"
        raise argparse.ArgumentTypeError(message)
    return number


def nonnegative(value: str) -> int:
    number = int(value)
    if number < 0:
        message = "must be nonnegative"
        raise argparse.ArgumentTypeError(message)
    return number


def arguments(argv: list[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("root", type=Path, help="Source directory; target code is never executed.")
    parser.add_argument(
        "--language", default="auto", help="Trailmark language or comma-separated set."
    )
    parser.add_argument(
        "--symbol", action="append", default=[], help="Also match this exact alias/wrapper name."
    )
    for name in ("max_candidates", "max_caller_nodes", "max_call_edges", "max_diagnostics"):
        parser.add_argument(
            "--" + name.replace("_", "-"), type=positive, default=getattr(Limits(), name)
        )
    parser.add_argument("--max-depth", type=nonnegative, default=Limits().max_depth)
    args = parser.parse_args(argv)
    if any(not symbol.strip() for symbol in args.symbol):
        parser.error("--symbol must not be empty")
    return args


def main(argv: list[str] | None = None) -> int:
    args = arguments(argv)
    limits = Limits(**{name: getattr(args, name) for name in asdict(Limits())})
    root = str(args.root)
    report = empty_report(root, limits)
    try:
        root = str(args.root.expanduser().resolve())
        report["root"] = root
        report["trailmark_version"] = runtime_version()
        graph = load_graph(root, args.language)
        report = discover(graph, report["trailmark_version"], limits, tuple(args.symbol))
    except (ImportError, OSError, ValueError, RuntimeError) as error:
        report["gaps"].append({"kind": "discovery_error", "message": str(error)})
    print(json.dumps(report, indent=2))
    return 0 if report["status"] == "evidence_collected" else 3


# gremlin: pardon[out_of_scope] CLI bootstrap guard; imports must not launch argparse.
if __name__ == "__main__":
    sys.exit(main())
```

## scripts/verify_electron_fetch.cjs

```js
"use strict";

// Trusted, benign fixture for a separately obtained Electron runtime.
// Never launch a target app or use its profile to run this fixture.
const assert = require("node:assert/strict");
const { execFileSync } = require("node:child_process");
const { X509Certificate } = require("node:crypto");
const { once } = require("node:events");
const fs = require("node:fs");
const http = require("node:http");
const https = require("node:https");
const os = require("node:os");
const path = require("node:path");
const tls = require("node:tls");
const zlib = require("node:zlib");
const { app, BrowserWindow, net, session } = require("electron");

assert.ok(process.versions.electron, "Run with an Electron executable");
const profile = fs.mkdtempSync(
  path.join(os.tmpdir(), "brotli-electron-profile-"),
);
app.setPath("userData", profile);
app.setPath("sessionData", profile);
app.setPath("logs", path.join(profile, "logs"));
app.commandLine.appendSwitch("disable-background-networking");
app.commandLine.appendSwitch("disable-component-update");
app.commandLine.appendSwitch("disable-sync");
app.commandLine.appendSwitch("no-first-run");

const expected = "brotli-checker benign Electron response fixture\n";
const compressed = zlib.brotliCompressSync(Buffer.from(expected));
assert.notEqual(compressed.toString("utf8"), expected);
const requests = [];
const blockedRequests = [];
const certificatePath = path.join(profile, "localhost-cert.pem");
const keyPath = path.join(profile, "localhost-key.pem");
execFileSync(
  "openssl",
  [
    "req",
    "-x509",
    "-newkey",
    "rsa:2048",
    "-nodes",
    "-days",
    "1",
    "-keyout",
    keyPath,
    "-out",
    certificatePath,
    "-subj",
    "/CN=127.0.0.1",
    "-addext",
    "subjectAltName=IP:127.0.0.1",
  ],
  { stdio: "ignore" },
);
const certificate = fs.readFileSync(certificatePath, "utf8");
const fingerprint = new X509Certificate(certificate).fingerprint256;
assert.equal(
  typeof tls.setDefaultCACertificates,
  "function",
  "Electron must embed Node >=24.5",
);
const originalCAs = tls.getCACertificates("default");
tls.setDefaultCACertificates([...originalCAs, certificate]);

function handleRequest(request, response) {
  requests.push({
    path: request.url,
    accept_encoding: request.headers["accept-encoding"] ?? null,
  });
  const url = new URL(request.url, "http://127.0.0.1");
  const headers = { "Cache-Control": "no-store", Connection: "close" };
  if (url.pathname === "/page") {
    response.writeHead(200, {
      ...headers,
      "Content-Type": "text/html",
      "Content-Security-Policy": "default-src 'none'; connect-src 'self'",
    });
    response.end("<!doctype html><title>Local Brotli measurement</title>");
  } else if (url.pathname === "/redirect") {
    response.writeHead(302, {
      ...headers,
      Location: "/body" + url.search,
    });
    response.end();
  } else if (url.pathname === "/body") {
    response.writeHead(200, {
      ...headers,
      "Content-Type": "text/plain",
      "Content-Encoding": "br",
      "Content-Length": compressed.length,
    });
    response.end(compressed);
  } else if (url.pathname === "/plain") {
    response.writeHead(200, { ...headers, "Content-Type": "text/plain" });
    response.end(expected);
  } else {
    response.writeHead(404, headers);
    response.end();
  }
}
const servers = [
  http.createServer(handleRequest),
  https.createServer(
    { key: fs.readFileSync(keyPath), cert: certificate },
    handleRequest,
  ),
];

const descriptor = Object.getOwnPropertyDescriptor(
  zlib,
  "createBrotliDecompress",
);
let decoderCalls = 0;
Object.defineProperty(zlib, "createBrotliDecompress", {
  ...descriptor,
  value: function (...args) {
    decoderCalls += 1;
    return Reflect.apply(descriptor.value, this, args);
  },
});

async function readResponse(fetcher, url, headers) {
  const response = await fetcher(url, {
    headers,
    signal: AbortSignal.timeout(5000),
  });
  return {
    status: response.status,
    content_encoding: response.headers.get("content-encoding"),
    redirected: response.redirected,
    body: await response.text(),
  };
}

function restrictSession(ses, origins) {
  ses.setPermissionRequestHandler((_contents, _permission, callback) =>
    callback(false),
  );
  ses.setPermissionCheckHandler(() => false);
  ses.webRequest.onBeforeRequest((details, callback) => {
    const allowed =
      origins.some((origin) => details.url.startsWith(origin + "/")) ||
      details.url === "about:blank";
    if (!allowed) blockedRequests.push(details.url);
    callback({ cancel: !allowed });
  });
  // Trust this ephemeral certificate only, in these isolated probe sessions.
  ses.setCertificateVerifyProc((request, callback) => {
    const matches =
      request.hostname === "127.0.0.1" &&
      new X509Certificate(request.certificate.data).fingerprint256 ===
        fingerprint;
    callback(matches ? 0 : -2);
  });
}

async function main() {
  await app.whenReady();
  const origins = [];
  for (const [index, server] of servers.entries()) {
    server.listen(0, "127.0.0.1");
    await once(server, "listening");
    origins.push(
      (index === 0 ? "http" : "https") +
        "://127.0.0.1:" +
        server.address().port,
    );
  }
  restrictSession(session.defaultSession, origins);
  const isolated = session.fromPartition("brotli-checker-measurement", {
    cache: false,
  });
  restrictSession(isolated, origins);
  await session.defaultSession.setProxy({ mode: "direct" });
  await isolated.setProxy({ mode: "direct" });
  const window = new BrowserWindow({
    show: false,
    webPreferences: {
      session: isolated,
      sandbox: true,
      nodeIntegration: false,
      contextIsolation: true,
      spellcheck: false,
    },
  });
  try {
    await window.loadURL(origins[0] + "/page");
    const renderer = await window.webContents.executeJavaScript(`({
      node_require_available: typeof require !== "undefined",
      node_process_available: typeof process !== "undefined",
      user_agent: navigator.userAgent,
      secure_context: isSecureContext,
    })`);
    assert.equal(renderer.node_require_available, false);
    assert.equal(renderer.node_process_available, false);
    const routes = [
      [
        "main_node_fetch",
        (url, headers) => readResponse(globalThis.fetch, url, headers),
      ],
      [
        "main_net_fetch",
        (url, headers) => readResponse(net.fetch.bind(net), url, headers),
      ],
      [
        "main_session_fetch",
        (url, headers) =>
          readResponse(isolated.fetch.bind(isolated), url, headers),
      ],
      [
        "sandboxed_renderer_fetch",
        (url, headers) =>
          window.webContents.executeJavaScript(
            "(" +
              readResponse.toString() +
              ")(globalThis.fetch," +
              JSON.stringify(url) +
              "," +
              JSON.stringify(headers) +
              ")",
          ),
      ],
    ];
    const cases = [
      { name: "plain_control", path: "/plain", headers: {}, compressed: false },
      { name: "default", path: "/body", headers: {}, compressed: true },
      {
        name: "identity",
        path: "/body",
        headers: { "Accept-Encoding": "identity" },
        compressed: true,
      },
      {
        name: "redirect_identity",
        path: "/redirect",
        headers: { "Accept-Encoding": "identity" },
        compressed: true,
      },
    ];
    const results = [];
    const violations = [];
    for (const origin of origins) {
      await window.loadURL(origin + "/page");
      for (const [route, fetcher] of routes) {
        for (const test of cases) {
          const start = requests.length;
          const before = decoderCalls;
          let result;
          let error = null;
          try {
            result = await fetcher(
              origin + test.path + "?case=" + route + "_" + test.name,
              test.headers,
            );
          } catch (failure) {
            error = String(failure);
          }
          const observed = requests.slice(start);
          // Instrumentation measures Node's JS/native wrapper, not Chromium calls.
          const calls = decoderCalls - before;
          const decoded = result?.status === 200 && result.body === expected;
          if (
            (!test.compressed ||
              route === "main_node_fetch" ||
              test.name === "default") &&
            !decoded
          ) {
            violations.push(
              origin.split(":")[0] +
                ":" +
                route +
                ":" +
                test.name +
                ": " +
                error,
            );
          }
          if (
            decoded &&
            calls !== (route === "main_node_fetch" && test.compressed ? 1 : 0)
          ) {
            violations.push(
              route +
                ":" +
                test.name +
                ": unexpected Node wrapper count " +
                calls,
            );
          }
          if (
            decoded &&
            result.content_encoding !== (test.compressed ? "br" : null)
          ) {
            violations.push(
              route + ":" + test.name + ": unexpected Content-Encoding",
            );
          }
          results.push({
            transport: origin.split(":")[0],
            route,
            case: test.name,
            requested_accept_encoding: test.headers["Accept-Encoding"] ?? null,
            requests: observed,
            content_encoding: result?.content_encoding ?? null,
            decoded_body_matches: decoded,
            node_zlib_wrapper_calls: calls,
            redirected: result?.redirected ?? null,
            error,
          });
        }
      }
    }
    const report = {
      schema_version: 1,
      platform: process.platform,
      architecture: process.arch,
      versions: process.versions,
      scope:
        "Trusted fixture in a separate Electron runtime; valid loopback HTTP/HTTPS responses only.",
      certificate_trust:
        "Ephemeral loopback certificate pinned for Chromium sessions and added to Node trust for this process only.",
      renderer,
      chromium_wrapper_instrumentation:
        "not_performed; decoded bytes establish response decoding",
      patch_status: "not_assessed_by_probe",
      target_application_input_path: "not_assessed",
      exploitability: "not_assessed",
      blocked_requests: blockedRequests,
      validation_errors: violations,
      results,
    };
    const json = JSON.stringify(report, null, 2) + "\n";
    if (process.env.BROTLI_CHECKER_OUTPUT)
      fs.writeFileSync(process.env.BROTLI_CHECKER_OUTPUT, json);
    else process.stdout.write(json);
    assert.deepEqual(
      violations,
      [],
      "Required decoding/control cases failed; inspect the report",
    );
  } finally {
    window.destroy();
  }
}

const timeout = setTimeout(() => {
  console.error("Electron measurement exceeded 45 seconds");
  app.exit(1);
}, 45000);

main().then(
  () => cleanup(0),
  (error) => {
    console.error(error);
    cleanup(1);
  },
);

async function cleanup(code) {
  clearTimeout(timeout);
  Object.defineProperty(zlib, "createBrotliDecompress", descriptor);
  tls.setDefaultCACertificates(originalCAs);
  for (const server of servers) {
    if (!server.listening) continue;
    await new Promise((resolve) => {
      server.close(resolve);
      server.closeAllConnections();
    });
  }
  if (code !== 0) {
    fs.rmSync(profile, { recursive: true, force: true });
    app.exit(code);
    return;
  }
  app.once("will-quit", () =>
    fs.rmSync(profile, { recursive: true, force: true }),
  );
  process.exitCode = code;
  app.quit();
}
```

## scripts/verify_node_fetch.cjs

```js
"use strict";

// Benign loopback-only check of Node's built-in fetch -> node:zlib route.
// Run this trusted fixture, never application code supplied to the inventory.
const assert = require("node:assert/strict");
const { once } = require("node:events");
const http = require("node:http");
const zlib = require("node:zlib");

async function main() {
  assert.equal(typeof globalThis.fetch, "function", "Node must provide built-in fetch");
  const expected = "brotli-checker benign response fixture\n";
  const compressed = zlib.brotliCompressSync(Buffer.from(expected));
  const requests = [];
  const server = http.createServer((request, response) => {
    requests.push({
      path: request.url,
      accept_encoding: request.headers["accept-encoding"] ?? null,
    });
    if (request.url === "/redirect") {
      response.writeHead(302, { Location: "/body", Connection: "close" });
      response.end();
      return;
    }
    // Deliberately send br even when the request advertises identity.
    response.writeHead(200, {
      "Content-Type": "text/plain",
      "Content-Encoding": "br",
      "Content-Length": compressed.length,
      Connection: "close",
    });
    response.end(compressed);
  });

  const descriptor = Object.getOwnPropertyDescriptor(zlib, "createBrotliDecompress");
  let decoderCalls = 0;
  Object.defineProperty(zlib, "createBrotliDecompress", {
    ...descriptor,
    value: function (...args) {
      decoderCalls += 1;
      return Reflect.apply(descriptor.value, this, args);
    },
  });

  try {
    server.listen(0, "127.0.0.1");
    await once(server, "listening");
    const origin = "http://127.0.0.1:" + server.address().port;
    const cases = [
      { name: "default_http", path: "/body", headers: {} },
      { name: "identity_http", path: "/body", headers: { "Accept-Encoding": "identity" } },
      { name: "redirect_identity_http", path: "/redirect", headers: { "Accept-Encoding": "identity" } },
    ];
    const results = [];
    for (const test of cases) {
      const previous = decoderCalls;
      const requestStart = requests.length;
      const response = await globalThis.fetch(origin + test.path, {
        headers: test.headers,
        signal: AbortSignal.timeout(5000),
      });
      const decoded = await response.text();
      assert.equal(response.status, 200);
      assert.equal(response.headers.get("content-encoding"), "br");
      assert.equal(decoded, expected, test.name + " must decode the response");
      assert.ok(decoderCalls > previous, test.name + " must call the native Brotli wrapper");
      assert.equal(response.redirected, test.path === "/redirect");
      const observedRequests = requests.slice(requestStart);
      assert.equal(observedRequests.length, test.path === "/redirect" ? 2 : 1);
      if (test.headers["Accept-Encoding"] === "identity") {
        for (const request of observedRequests) {
          assert.equal(request.accept_encoding, "identity");
        }
      }
      results.push({
        case: test.name,
        requests: observedRequests,
        decoded_body_matches: decoded === expected,
        native_brotli_decoder_calls: decoderCalls - previous,
        redirected: response.redirected,
      });
    }
    console.log(JSON.stringify({
      node: process.version,
      brotli: process.versions.brotli,
      undici: process.versions.undici ?? null,
      scope: "Valid Brotli responses from a loopback HTTP server; no exploit payload.",
      patch_status: "not_assessed",
      exploitability: "not_assessed",
      results,
    }, null, 2));
  } finally {
    Object.defineProperty(zlib, "createBrotliDecompress", descriptor);
    if (server.listening) {
      await new Promise((resolve, reject) => {
        server.close((error) => error ? reject(error) : resolve());
        server.closeAllConnections();
      });
    }
  }
}

main().catch((error) => {
  console.error(error);
  process.exitCode = 1;
});
```

