diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
new file mode 100644
index 0000000..2f66d4c
--- /dev/null
+++ b/.github/workflows/tests.yml
@@ -0,0 +1,33 @@
+name: Tests
+
+on:
+ push:
+ pull_request:
+ workflow_dispatch:
+
+permissions:
+ contents: read
+
+jobs:
+ test:
+ name: Python ${{ matrix.python }} on ${{ matrix.os }}
+ runs-on: ${{ matrix.os }}
+ timeout-minutes: 10
+ strategy:
+ fail-fast: false
+ matrix:
+ os: [ubuntu-latest, windows-latest]
+ python: ['3.10', '3.14']
+ steps:
+ - name: Check out repository
+ uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
+ with:
+ persist-credentials: false
+ - name: Set up Python
+ uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6
+ with:
+ python-version: ${{ matrix.python }}
+ - name: Run local-only tests
+ run: python -m unittest -v
+ - name: Check CLI help
+ run: python tinypool_options_rce_check.py --help
diff --git a/AGENTS.md b/AGENTS.md
index 1a10244..e6babf7 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -8,19 +8,18 @@ before changing this project.
This repository covers only **CVE-2026-104849** (Tinypool worker-options prototype-pollution gadget). Its aliases
refer to the same exploit. Keep unrelated vulnerabilities in separate repositories.
-The repository is an initialized research foundation. No checker, tests, or CI
-workflow exists yet; never imply otherwise or document nonexistent commands.
+The repository contains the offline checker `tinypool_options_rce_check.py`, synthetic tests and CI. Preserve the exact affected interval, disputed boundaries and reviewed fixed line; no network mode or application execution is implemented.
## Project map
-- `advisory.json`: source-preserving research record; detection is disabled.
+- `advisory.json`: enabled offline rule metadata with preserved research origin.
- `README.md`: exact scope, advisory snapshot, primary sources, and current status.
- `docs/IMPLEMENTATION.md`: local implementation and verification checklist.
- `LICENSING.md`, `CONTRIBUTING.md`, `BRAND_ASSETS.md`: licensing and contribution terms.
## Implementation constraints
-Follow the established [node2shell-check](https://github.com/CompleteTech-LLC/node2shell-check) and [php2shell-check](https://github.com/CompleteTech-LLC/php2shell-check) conventions. Prefer a small, auditable,
+Follow the established [node2shell-check](https://github.com/completetech-security/node2shell-check) and [php2shell-check](https://github.com/completetech-security/php2shell-check) conventions. Prefer a small, auditable,
dependency-free checker when appropriate. Do not turn this project into a generic
product scanner or execute exploit payloads.
@@ -42,8 +41,8 @@ disabled until a real implementation is reviewed and tested.
## Validation and delivery
-Use focused synthetic cases and local-only fixtures for meaningful behavior tests
-once code exists. Do not invent passing tests, add prose-only tests, or publish
+Use focused synthetic cases and local-only fixtures for meaningful behavior tests.
+Do not invent passing tests, add prose-only tests, or publish
private assessments. Keep CLI, schemas, methodology, and README aligned. Distinguish
local results from hosted CI on the exact published commit. Preserve unrelated
work and follow the licensing and brand policies linked above.
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 013b82f..8b0355b 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -1,22 +1,6 @@
# Contributing
-This repository is initialized for **CVE-2026-104849**. Checker implementation is
-pending; start with the [implementation checklist](docs/IMPLEMENTATION.md).
-Submit focused pull requests for this exploit. Report flaws requiring private
-disclosure according to [SECURITY.md](SECURITY.md).
-
-## Development expectations
-
-Follow [AGENTS.md](AGENTS.md). Prefer a standalone Python 3.10+ checker with no
-third-party runtime/test dependencies when suitable. Validate primary sources
-before encoding affected or fixed ranges. Preserve evidence provenance and
-`unknown` for missing or conflicting facts. Keep exploitability `not_tested`.
-
-Use offline inventories, synthetic examples, mocks, and local fixtures. Do not
-commit credentials, real target lists, private reports, or exploit payloads.
-Network detection must be justified, optional, ordinary, bounded, and authorized.
-Document implemented behavior and actual commands; no checker commands or test
-suite are available in this foundation yet.
+Read AGENTS.md and README.md. Run `python -m unittest -v`, `python tinypool_options_rce_check.py --help` and `git diff --check`. Use synthetic evidence only, verify primary sources before changing ranges, keep source conflicts unknown, and never run application code or send exploit payloads. Preserve safe publication, bounded reads, output escaping and complete=false after interruption. Review and exact-head hosted CI precede merge.
## Licensing of contributions
diff --git a/README.md b/README.md
index e919fb3..cb96601 100644
--- a/README.md
+++ b/README.md
@@ -1,140 +1,57 @@
-
-
-
+# tinypool-options-rce-check
-
-
-
-
-
+Offline evidence checker for **CVE-2026-104849 — Tinypool worker-options prototype-pollution gadget**. Python 3.10+, standard library only. The executable can be copied alone. It never contacts websites, imports application code, runs a package manager, or executes exploit payloads.
-
- Tinypool worker-options prototype-pollution gadget
- CVE-2026-104849 · GHSA-85c8-ppgw-ccpr · research only, no checker yet
-
+```sh
+python tinypool_options_rce_check.py --version 2.1.2 --format json
+python tinypool_options_rce_check.py --inventory examples/inventory.example.json --format html --output report.html
+python tinypool_options_rce_check.py --csv-list inventory.csv --fail-on-finding
+python tinypool_options_rce_check.py --help
+python -m unittest -v
+```
----
+## Reviewed rules
-> **Repository initialized; checker implementation pending.**
-> This project currently contains research and an implementation checklist.
-> It does not assess systems or provide an executable detection rule.
+Primary advisory and reviewed database retrieved 2026-10-24. A public URL cannot provide installed package inventory. Obtain each deployed package instance's exact version and provenance; declarations and local workstation versions are not deployment evidence.
-`tinypool-options-rce-check` is the planned home of a checker for **CVE-2026-104849**, a Tinypool worker-options prototype-pollution gadget. Today it holds source-backed research and a work plan only. It sends no requests, runs no exploit payloads, and reports no results. When built, it is meant to report uncertainty honestly: missing, conflicting, or vendor-qualified evidence stays `unknown`, never "safe".
-
-By [CompleteTech LLC](https://complete.tech).
-
-## Status and what is planned
-
-There are no checker commands, installation steps, downloadable binaries, test results, or CI claims, so there is no Quick start yet. To follow or contribute to the work:
-
-1. Review [advisory.json](advisory.json) and the [primary sources](#primary-sources).
-2. Work through the exploit-specific [implementation checklist](docs/IMPLEMENTATION.md).
-3. Read [AGENTS.md](AGENTS.md), [CONTRIBUTING.md](CONTRIBUTING.md), and [SECURITY.md](SECURITY.md) before contributing.
-
-## Planned report layout
-
-
-
-
-
-
-
-
-
Illustrative only. A layout sketch with placeholder values on reserved example.* domains. No checker, report format, or result exists yet.
Tinypool worker-options prototype-pollution gadget · generated 2026-10-24 14:45 UTC · rules as of 2026-10-24
+
Version evidence, not a proof of exploitability. Every result has exploitability not_tested.
+Unknown results need investigation. Precondition observations are operator-reported and do not change the version status.
Verify this package instance is actually deployed and contains the upstream fix.
Evidence and limitations
Reported version meets the fix within this reviewed release line; this is not installed-patch attestation.
pool.run(task) alone is not the documented trigger. Surface 2.1.1 source conflict and do not report clean. Maintainer severity High differs from database Critical; retain provenance.
+
Upstream fixed releases
Branch
Fixed in
2.1.x
2.1.2
+
Verify the deployed application
+
Check the actual tinypool binary in each serving process or container. Upgrade the runtime, redeploy, then collect the version again.
+Vendor and custom builds require vendor evidence because fixes may be backported.
+
+
diff --git a/examples/inventory.example.json b/examples/inventory.example.json
new file mode 100644
index 0000000..33044a2
--- /dev/null
+++ b/examples/inventory.example.json
@@ -0,0 +1,11 @@
+{
+ "schema_version": 1,
+ "package": "tinypool",
+ "package_version": "2.1.2",
+ "label": "example-package-instance",
+ "build_source": "unknown",
+ "caller_supplies_pool_run_options_object": null,
+ "options_lacks_an_own_filename_property": null,
+ "separate_prototype_pollution_primitive": null,
+ "attacker_controlled_worker_module_available": null
+}
diff --git a/test_assessment.py b/test_assessment.py
new file mode 100644
index 0000000..0ffdd9a
--- /dev/null
+++ b/test_assessment.py
@@ -0,0 +1,99 @@
+"""Independent advisory golden cases and hostile-input/publication behavior."""
+import contextlib
+import csv
+import io
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest import mock
+import tinypool_options_rce_check as c
+CASES = [('2.0.9', 'potentially_affected'), ('2.1.0', 'potentially_affected'), ('2.1.1', 'unknown'), ('2.1.2', 'fixed_version_reported'), ('2.1.3', 'fixed_version_reported'), ('2.2.0', 'unknown'), ('3.0.0', 'unknown')]
+FIXED = '2.1.2'
+
+class AssessmentTests(unittest.TestCase):
+ def run_cli(self,*args):
+ with contextlib.redirect_stdout(io.StringIO()) as out,contextlib.redirect_stderr(io.StringIO()) as err:
+ try: code=c.main(list(args))
+ except SystemExit as exc: code=exc.code
+ return code,out.getvalue(),err.getvalue()
+ def test_primary_source_boundary_cases(self):
+ for version,status in CASES:
+ with self.subTest(version=version): self.assertEqual(status,c.assess_version(version)['status'])
+ def test_version_conflicts_qualifiers_and_custom_provenance(self):
+ for version in ['', 'latest', '1', '1.2', '01.2.3', '１.2.3', FIXED+'-rc.1', FIXED+'+vendor', '\u00a0'+FIXED, True, None]:
+ self.assertEqual('unknown',c.assess_version(version)['status'])
+ self.assertEqual('unknown',c.assess_version(FIXED,version_candidates=['0.0.1'])['status'])
+ for version,_ in CASES:
+ self.assertEqual('unknown',c.assess_version(version,build_source='custom')['status'])
+ def test_observations_never_change_patch_or_exploitability(self):
+ for flag in [True,False,None]:
+ r=c.assess_entry('example','inventory',FIXED,observations={k:flag for k in c.PRECONDITION_KEYS})
+ self.assertEqual('fixed_version_reported',r['status']);self.assertEqual('not_tested',r['exploitability'])
+ def test_inventory_schema_types_and_duplicate_keys(self):
+ good={'schema_version':1,'package_version':FIXED,'package':'tinypool'}
+ with tempfile.TemporaryDirectory() as d:
+ p=Path(d)/'input.json'
+ for data in [{},[],{**good,'extra':1},{**good,'schema_version':True},{**good,c.PRECONDITION_KEYS[0]:'true'},{**good,'package':'different-package'},{**good,'package_version':4}]:
+ p.write_text(json.dumps(data));self.assertEqual(2,self.run_cli('--inventory',str(p))[0])
+ p.write_text('{"sensitive-key":1,"sensitive-key":2}')
+ code,out,err=self.run_cli('--inventory',str(p));self.assertEqual((2,''),(code,out));self.assertNotIn('sensitive-key',err)
+ p.write_text(json.dumps(good));code,out,_=self.run_cli('--inventory',str(p),'--format','json')
+ self.assertEqual(0,code);self.assertEqual('fixed_version_reported',json.loads(out)['results'][0]['status'])
+ def test_input_budgets_and_special_files(self):
+ self.assertEqual(2,self.run_cli('--version','v'*513)[0])
+ with tempfile.TemporaryDirectory() as d:
+ p=Path(d)/'large.json';p.write_bytes(b' '*(c.MAX_INPUT_BYTES+1));self.assertEqual(2,self.run_cli('--inventory',str(p))[0])
+ if hasattr(os,'mkfifo'):
+ q=Path(d)/'pipe';os.mkfifo(q)
+ with mock.patch.object(Path,'open',side_effect=AssertionError('must not open pipe')):
+ self.assertEqual(2,self.run_cli('--inventory',str(q))[0])
+ with mock.patch.object(c,'MAX_ENTRIES',2):
+ self.assertEqual(2,self.run_cli('--version',FIXED,'--version',FIXED,'--version',FIXED)[0])
+ def test_csv_roundtrips_controls_and_neutralizes_mixed_whitespace(self):
+ labels=['alpha\r=1+1',' \t=1+1',' \r=1+1','\v=1+1','\f@x','\u00a0=1+1','=1+1']
+ out=io.StringIO();c.write_report(out,[c.assess_entry(x,'inventory',FIXED) for x in labels],format='csv')
+ rows=list(csv.DictReader(io.StringIO(out.getvalue(),newline='')))
+ self.assertEqual(len(labels),len(rows));self.assertEqual(labels[0],rows[0]['input_label'])
+ for original,row in zip(labels[1:],rows[1:]): self.assertEqual("'"+original,row['input_label'])
+ def test_html_is_escaped_and_script_free(self):
+ text=c.render_html([c.assess_entry('','inventory',FIXED)])
+ self.assertNotIn(' 128:
+ result["version_status"] = "conflicting_versions"
+ notes.append("Too many version candidates were supplied for one runtime.")
+ return result
+ values = [package_version, *candidates]
+ if any(value is not None and not isinstance(value, str) for value in values):
+ result["version_status"] = "unrecognized_or_partial_version"
+ notes.append("Non-text version metadata cannot be compared safely.")
+ return result
+ # One normalization: strip ASCII whitespace only, then at most one leading "v".
+ versions = sorted({value.strip(ASCII_SPACE).removeprefix("v") for value in values if value and value.strip(ASCII_SPACE)})
+ if not versions:
+ notes.append("No tinypool version was reported. Missing metadata does not establish that the runtime is unaffected.")
+ return result
+ if len(versions) != 1:
+ result["version_status"] = "conflicting_versions"
+ notes.append("Conflicting reported versions: " + ", ".join(value[:128] for value in versions))
+ return result
+
+ version = versions[0]
+ match = None if len(version) > 128 else VERSION_PATTERN.fullmatch(version)
+ if not match:
+ qualified = re.match(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)(.+)", version[:128])
+ result["version_status"] = "qualified_version_unverified" if qualified else "unrecognized_or_partial_version"
+ notes.append(
+ f"Version {version[:128]!r} is not a plain stable release. Prereleases, nightly/rc builds, package suffixes "
+ "and custom builds need vendor verification."
+ )
+ return result
+
+ parsed = tuple(int(part) for part in match.groups())
+ source = ("".join(c.lower() if c.isascii() else c for c in build_source.strip(ASCII_SPACE))
+ if isinstance(build_source, str) else "invalid")
+ if source not in OFFICIAL_BUILD_SOURCES:
+ result["version_status"] = "vendor_build_unverified"
+ notes.append("Custom, vendor, or otherwise unrecognized package provenance needs independent patch evidence.")
+ return result
+ if parsed in DISPUTED:
+ result["version_status"] = "source_discrepancy"
+ notes.append("Primary advisory and reviewed database disagree at this boundary; obtain maintainer evidence.")
+ return result
+ if AFFECTED_MIN <= parsed <= AFFECTED_MAX:
+ result.update(status="potentially_affected", version_status="in_documented_affected_range",
+ fixed_version=_fmt(FIXED), recommendation=RECOMMENDATION)
+ notes.append("Reported version matches the documented affected range; application preconditions and exploitability were not established.")
+ elif parsed >= FIXED and parsed[:len(FIXED_LINE)] == FIXED_LINE:
+ result.update(status="fixed_version_reported", version_status="at_or_above_fixed_version",
+ fixed_version=_fmt(FIXED), recommendation="Verify this package instance is actually deployed and contains the upstream fix.")
+ notes.append("Reported version meets the fix within this reviewed release line; this is not installed-patch attestation.")
+ else:
+ result["version_status"] = "outside_reviewed_range"
+ notes.append("Version is outside the documented affected range and reviewed fixed line. No safety or other-vulnerability conclusion is made.")
+ return result
+# =============================================================================
+# END REUSABLE PACKAGE VERSION ENGINE
+# =============================================================================
+
+PRECONDITION_KEYS = tuple(key for key, _ in PRECONDITIONS)
+OBSERVATIONS = {True: "reported_true", False: "reported_false", None: "not_reported"}
+INVENTORY_FIELDS = {"schema_version", "package", "label", "package_version", "build_source", *PRECONDITION_KEYS}
+CSV_COLUMNS = ["record_type", "input_label", "input_source", "package_version", "build_source", "status", "version_status", "fixed_version",
+ "exploitability", "evidence_type", "evidence_confidence", "cve", "rules_as_of", *[f"precondition_{key}" for key in PRECONDITION_KEYS],
+ "recommendation", "notes", "sources", "report_schema_version", "report_complete"]
+STATUS_ORDER = {"potentially_affected": 0, "unknown": 2, "fixed_version_reported": 3}
+STATUS_LABELS = {"potentially_affected": "Potentially affected", "unknown": "Unknown", "fixed_version_reported": "Fixed version reported"}
+
+
+def assess_entry(label, input_source, package_version, build_source="unknown", observations=None):
+ """Build one report row. version_status, preconditions and exploitability stay separate."""
+ verdict = assess_version(package_version, build_source=build_source)
+ observations = observations or {}
+ notes = list(verdict["notes"]) + list(ADVISORY_NOTES)
+ row = {
+ "input_label": label, "input_source": input_source,
+ "package_version": package_version if isinstance(package_version, str) else "",
+ "build_source": build_source if isinstance(build_source, str) else "",
+ "status": verdict["status"], "version_status": verdict["version_status"],
+ "fixed_version": verdict["fixed_version"], "exploitability": "not_tested",
+ "evidence_type": "operator_reported_version",
+ "evidence_confidence": "insufficient" if verdict["status"] == "unknown" else "low",
+ "cve": CVE_ID, "rules_as_of": RULES_DATE,
+ }
+ for key in PRECONDITION_KEYS:
+ row[f"precondition_{key}"] = OBSERVATIONS[observations.get(key)]
+ if any(observations.get(key) is not None for key in PRECONDITION_KEYS):
+ notes.append("Precondition observations are operator-supplied and unverified; they do not change the version status.")
+ row.update({"recommendation": verdict["recommendation"], "notes": notes, "sources": list(SOURCES),
+ "report_schema_version": REPORT_SCHEMA_VERSION})
+ return row
+
+
+def _tri_state(value, name, *, text=False):
+ """JSON inventories take booleans or null only; CSV cells and CLI flags take the
+ exact lowercase text true or false, and CSV cells may also be empty."""
+ if value is None or (text and value == ""):
+ return None
+ if type(value) is bool and not text:
+ return value
+ if text and value in ("true", "false"):
+ return value == "true"
+ raise ValueError(f"{name} must be " + ("true, false or empty" if text else "true, false or null"))
+
+
+def entry_from_record(record, fallback_label, input_source, *, require_schema=True, text=False):
+ if not isinstance(record, dict) or set(record) - INVENTORY_FIELDS:
+ raise ValueError("Each entry must be an object containing only documented fields")
+ if require_schema and (type(record.get("schema_version")) is not int or record["schema_version"] != 1):
+ raise ValueError("Inventory schema_version must be 1")
+ if record.get("package", 'tinypool') != 'tinypool':
+ raise ValueError("This checker assesses tinypool; other packages need their own assessment")
+ for key in ("label", "package_version", "build_source"):
+ if key in record and (not isinstance(record[key], str) or len(record[key]) > MAX_FIELD):
+ raise ValueError(f"{key} must be a string of at most {MAX_FIELD} characters")
+ if "package_version" not in record:
+ raise ValueError("Entry is missing package_version")
+ observations = {key: _tri_state(record.get(key), key, text=text) for key in PRECONDITION_KEYS}
+ return assess_entry(record.get("label") or fallback_label, input_source, record["package_version"],
+ record.get("build_source") or "unknown", observations)
+
+
+def _require_regular_file(path, what="Input"):
+ if not stat.S_ISREG(os.stat(path).st_mode):
+ raise ValueError(f"{what} must be a regular file")
+
+
+def _bounded_text(value, name):
+ if not isinstance(value, str) or len(value) > MAX_FIELD:
+ raise ValueError(f"{name} must be a string of at most {MAX_FIELD} characters")
+ return value
+
+
+def _read_bounded(path):
+ _require_regular_file(path)
+ with Path(path).open("rb") as stream:
+ raw = stream.read(MAX_INPUT_BYTES + 1)
+ if len(raw) > MAX_INPUT_BYTES:
+ raise ValueError("Input file exceeds the 4 MiB limit")
+ return raw.decode("utf-8-sig")
+
+
+class Budget:
+ """Counts entries and files while loading, so limits hold before rows pile up."""
+
+ def __init__(self):
+ self.entries = 0
+ self.files = 0
+
+ def add_file(self):
+ self.files += 1
+ if self.files > MAX_FILES:
+ raise ValueError(f"A run is limited to {MAX_FILES} input files")
+
+ def add_entries(self, count):
+ self.entries += count
+ if self.entries > MAX_ENTRIES:
+ raise ValueError(f"A run is limited to {MAX_ENTRIES:,} inputs")
+
+
+def load_inventory(path, budget=None, out=None):
+ """Return entry rows; rows are also appended to ``out`` as soon as each is assessed."""
+ budget = budget or Budget()
+ out = [] if out is None else out
+ budget.add_file()
+ def unique_object(pairs):
+ value = {}
+ for key, item in pairs:
+ if key in value:
+ raise ValueError("Duplicate inventory key")
+ value[key] = item
+ return value
+
+ def reject_constant(_):
+ raise ValueError("Non-finite JSON number")
+
+ data = json.loads(_read_bounded(path), object_pairs_hook=unique_object, parse_constant=reject_constant)
+ records = data if isinstance(data, list) else [data]
+ if not records or len(records) > MAX_ENTRIES:
+ raise ValueError(f"Inventory must contain between 1 and {MAX_ENTRIES:,} entries")
+ budget.add_entries(len(records))
+ for index, record in enumerate(records):
+ out.append(entry_from_record(record, f"{Path(path).stem}:{index + 1}", "inventory"))
+ return out
+
+
+def _reject_bare_quotes(text):
+ """csv.reader accepts a quote inside an unquoted cell; this checker does not."""
+ state = "start" # start of cell | unquoted | quoted | after a closing quote
+ index, size = 0, len(text)
+ while index < size:
+ char = text[index]
+ if state == "start":
+ state = "quoted" if char == '"' else ("start" if char in ",\r\n" else "unquoted")
+ elif state == "unquoted":
+ if char == '"':
+ raise ValueError("CSV list has a quote inside an unquoted cell")
+ if char in ",\r\n":
+ state = "start"
+ elif state == "quoted":
+ if char == '"':
+ state = "after"
+ else: # after a closing quote: an escaped quote, or the end of the cell
+ if char == '"':
+ state = "quoted"
+ elif char in ",\r\n":
+ state = "start"
+ else:
+ raise ValueError("CSV list has text after a closing quote")
+ index += 1
+ if state in ("quoted",):
+ raise ValueError("CSV list has an unterminated quote")
+
+
+def load_csv_list(path, budget=None, out=None):
+ budget = budget or Budget()
+ out = [] if out is None else out
+ budget.add_file()
+ csv.field_size_limit(MAX_FIELD * 4)
+ text = _read_bounded(path)
+ if "\x00" in text:
+ raise ValueError("CSV list contains a NUL character")
+ _reject_bare_quotes(text)
+ reader = csv.DictReader(io.StringIO(text, newline=""), strict=True)
+ if not reader.fieldnames or "package_version" not in reader.fieldnames:
+ raise ValueError("CSV list needs a header row with a package_version column")
+ if len(set(reader.fieldnames)) != len(reader.fieldnames):
+ raise ValueError("CSV list header has duplicate columns")
+ if set(reader.fieldnames) - (INVENTORY_FIELDS - {"schema_version", "package"}):
+ raise ValueError("CSV list contains undocumented columns")
+ count = 0
+ for index, record in enumerate(reader):
+ if None in record or None in record.values():
+ raise ValueError(f"CSV row {index + 2} has a different number of fields than the header")
+ budget.add_entries(1)
+ out.append(entry_from_record(record, f"{Path(path).stem}:{index + 1}", "csv", require_schema=False, text=True))
+ count += 1
+ if not count:
+ raise ValueError("CSV list contains no rows")
+ return out
+
+
+def csv_safe(value):
+ if isinstance(value, list):
+ value = " | ".join(value)
+ if isinstance(value, str) and (value[:1] in ("\t", "\r", "\n") or value.lstrip().startswith(("=", "+", "-", "@"))):
+ return "'" + value
+ return value
+
+
+REPORT_CSS = """
+:root{--bg:#F8FAFC;--surface:#FFFFFF;--surface-2:#EEF2FF;--ink:#0F172A;--ink-2:#1E293B;--ink-3:#64748B;--line:#E2E8F0;
+--accent:#1E3A8A;--accent-text:#1E3A8A;--on-accent:#FFFFFF;--bad:#B42318;--bad-bg:#FEE4E2;--ok:#067647;--ok-bg:#DCFAE6;--unk:#475467;--unk-bg:#EAECF0}
+@media (prefers-color-scheme:dark){:root{--bg:#0F172A;--surface:#1E293B;--surface-2:#273449;--ink:#F1F5F9;--ink-2:#CBD5E1;--ink-3:#94A3B8;
+--line:#334155;--accent-text:#93C5FD;--bad:#FDA29B;--bad-bg:#55160C;--ok:#75E0A7;--ok-bg:#053321;--unk:#D0D5DD;--unk-bg:#344054}}
+*{box-sizing:border-box}body{margin:0;background:var(--bg);color:var(--ink);font:15px/1.5 Arial,system-ui,sans-serif}
+main{max-width:1240px;margin:0 auto;padding:24px 16px 48px}
+header{padding:20px 0;border-bottom:3px solid var(--accent);margin-bottom:20px}
+.eyebrow{font-size:12px;letter-spacing:.12em;font-weight:700;color:var(--accent-text);text-transform:uppercase}
+h1{margin:2px 0;font-size:26px}.sub{color:var(--ink-3);margin:0}h2{margin:32px 0 10px;font-size:18px;color:var(--ink-2)}
+.tiles{display:grid;grid-template-columns:repeat(auto-fit,minmax(160px,1fr));gap:12px}
+.tile{background:var(--surface);border:1px solid var(--line);border-radius:10px;padding:14px}
+.tile b{display:block;font-size:28px;color:var(--accent-text)}.tile span{color:var(--ink-3);font-size:13px}
+.tile.bad b{color:var(--bad)}.tile.ok b{color:var(--ok)}
+.wrap{overflow-x:auto}table{width:100%;border-collapse:collapse;background:var(--surface);border:1px solid var(--line)}
+table.narrow{max-width:360px}
+th{background:var(--accent);color:var(--on-accent);text-align:left;padding:8px 10px;font-size:13px;white-space:nowrap}
+td{padding:8px 10px;border-top:1px solid var(--line);vertical-align:top;font-size:14px}tr:nth-child(even) td{background:var(--surface-2)}
+td.site{overflow-wrap:anywhere;min-width:160px}td.notes{color:var(--ink-3);font-size:13px;min-width:240px}
+code{font:13px Consolas,monospace}ul.obs{list-style:none;margin:0;padding:0;font-size:12px;overflow-wrap:anywhere}
+.pill{display:inline-block;padding:2px 10px;border-radius:999px;font-size:12px;font-weight:700;white-space:nowrap}
+.pill.potentially_affected{color:var(--bad);background:var(--bad-bg)}.pill.fixed_version_reported{color:var(--ok);background:var(--ok-bg)}
+.pill.unknown{color:var(--unk);background:var(--unk-bg)}
+.callout{background:var(--surface);border:1px solid var(--line);border-left:4px solid var(--accent);border-radius:8px;padding:12px 16px;color:var(--ink-2)}
+footer{margin-top:40px;padding-top:12px;border-top:1px solid var(--line);color:var(--ink-3);font-size:12px}
+a{color:var(--accent-text)}details{max-width:430px}summary{cursor:pointer;color:var(--accent-text)}details ul{padding-left:18px}
+@media print{body{background:#fff}tr,.tile{break-inside:avoid}details>summary{display:none}}
+"""
+
+
+def render_html(rows, *, brand="CompleteTech LLC", generated_at=None, complete=True):
+ def esc(value):
+ return html.escape("" if value is None else str(value), quote=True)
+
+ generated_at = generated_at or datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
+ rows = sorted(rows, key=lambda row: STATUS_ORDER.get(row.get("status"), 2))
+ counts = {key: sum(row.get("status") == key for row in rows) for key in STATUS_ORDER}
+ tiles = [("", len(rows), "Inputs assessed"), ("bad", counts["potentially_affected"], "Potentially affected"),
+ ("", counts["unknown"], "Unknown"),
+ ("ok", counts["fixed_version_reported"], "Fixed version reported")]
+ body = []
+ for row in rows:
+ status = row.get("status") if row.get("status") in STATUS_ORDER else "unknown"
+ observed = "".join(
+ f"
{esc(key)}: {esc(row.get('precondition_' + key))}
" for key, description in PRECONDITIONS)
+ notes = "".join(f"
{esc(note)}
" for note in dict.fromkeys(row.get("notes", [])))
+ body.append(
+ "
{esc(row.get('package_version') or chr(8212))} build: {esc(row.get('build_source') or chr(8212))}
"
+ f"
{esc(row.get('fixed_version') or chr(8212))}
"
+ f"
{observed}
"
+ f"
{esc(row.get('recommendation'))}
"
+ f"Evidence and limitations
{notes}
")
+ fixes = "".join(f"
{major}.x
{_fmt(version)}
" for major, version in FIXES.items())
+ tile_html = "".join(f"
{value}{esc(label)}
" for cls, value, label in tiles)
+ sources = " · ".join(f"{esc(url.rsplit('/', 1)[-1])}" for url in SOURCES)
+ incomplete = "" if complete else "
Interrupted run. Only completed inputs are shown; assess the rest again.
"
+ return f"""
+
+
+{esc(CVE_ID)} Version Report
+
{esc(brand)}
{esc(CVE_ID)} Version Report
+
{esc(ADVISORY_TITLE)} · generated {esc(generated_at)} · rules as of {esc(RULES_DATE)}
+
Version evidence, not a proof of exploitability. Every result has exploitability not_tested.
+Unknown results need investigation. Precondition observations are operator-reported and do not change the version status.
+{incomplete}
{tile_html}
+
Results
Input
Assessment
Reported version
+
Upstream fix
Precondition observations
Next step and evidence
+{''.join(body) or '
No results.
'}
+
Upstream fixed releases
Branch
Fixed in
{fixes}
+
Verify the deployed application
+
Check the actual tinypool binary in each serving process or container. Upgrade the runtime, redeploy, then collect the version again.
+Vendor and custom builds require vendor evidence because fixes may be backported.
+
+
+"""
+
+
+def write_report(output, rows, *, format="csv", brand="CompleteTech LLC", complete=True):
+ if format == "csv":
+ # QUOTE_ALL keeps a bare CR/LF/TAB inside a quoted cell, so it can never start a new row.
+ writer = csv.DictWriter(output, fieldnames=CSV_COLUMNS, lineterminator="\n", quoting=csv.QUOTE_ALL)
+ writer.writeheader()
+ for row in rows:
+ writer.writerow({**{key: csv_safe(row.get(key)) for key in CSV_COLUMNS}, "record_type": "result", "report_complete": "true" if complete else "false"})
+ if not rows:
+ writer.writerow({"record_type": "summary", "report_complete": "true" if complete else "false",
+ "report_schema_version": REPORT_SCHEMA_VERSION, "rules_as_of": RULES_DATE,
+ "cve": CVE_ID, "exploitability": "not_tested"})
+ elif format == "json":
+ json.dump({"tool": TOOL_NAME, "tool_version": TOOL_VERSION, "report_schema_version": REPORT_SCHEMA_VERSION,
+ "cve": CVE_ID, "rules_as_of": RULES_DATE, "sources": list(SOURCES), "complete": complete,
+ "generated_at": datetime.now(timezone.utc).isoformat(), "results": rows},
+ output, ensure_ascii=False, indent=2)
+ output.write("\n")
+ else:
+ output.write(render_html(rows, brand=brand, complete=complete))
+
+
+def render_report(rows, **options):
+ """Render fully in memory and encode, so an encoding error happens before any output."""
+ buffer = io.StringIO(newline="")
+ write_report(buffer, rows, **options)
+ text = buffer.getvalue()
+ return text.encode("utf-8") # raises UnicodeEncodeError for lone surrogates
+
+
+def save_report(path, rows, *, force=False, state=None, **options):
+ """Publish a complete file atomically, refusing an existing path unless forced.
+
+ ``state`` records progress for interrupt handling: ``stdout_started`` is set before the single
+ buffered stdout write, and ``published`` right after the file is published or stdout is written.
+ """
+ state = {} if state is None else state
+ data = render_report(rows, **options)
+ if path is None:
+ out = getattr(sys.stdout, "buffer", None)
+ state["stdout_started"] = True
+ if out is None:
+ sys.stdout.write(data.decode("utf-8"))
+ else:
+ out.write(data)
+ out.flush()
+ state["published"] = True
+ return
+ path = Path(path)
+ temporary = None
+ try:
+ with tempfile.NamedTemporaryFile("wb", dir=path.parent, prefix=".report-", delete=False) as stream:
+ temporary = Path(stream.name)
+ stream.write(data)
+ stream.flush()
+ os.fsync(stream.fileno())
+ if force:
+ os.replace(temporary, path)
+ state["published"] = True
+ else:
+ try:
+ os.link(temporary, path) # atomic no-clobber publication; never fall back to a rename
+ except FileExistsError:
+ raise ValueError("Output exists; use a new path or explicitly pass --force") from None
+ except OSError as exc:
+ raise ValueError("Cannot publish without overwriting: hard links are unavailable here "
+ f"({exc.strerror or type(exc).__name__}); use a new path on a filesystem with hard links, "
+ "or pass --force to replace explicitly") from None
+ state["published"] = True
+ finally:
+ if temporary is not None:
+ temporary.unlink(missing_ok=True)
+
+
+def _message(exc):
+ """A short, single-line, ASCII-safe description of an operational error."""
+ if isinstance(exc, UnicodeError):
+ return "text could not be encoded or decoded as UTF-8"
+ if isinstance(exc, RecursionError):
+ return "input is nested too deeply"
+ text = str(exc) or type(exc).__name__
+ return text[:300].encode("unicode_escape", "backslashreplace").decode("ascii")[:300]
+
+
+def _validate_arguments(args):
+ """Checks that need no input data; run before any file is loaded."""
+ _bounded_text(args.build_source, "--build-source")
+ _bounded_text(args.brand, "--brand")
+ for version in args.package_version:
+ _bounded_text(version, "--version")
+ if len(args.package_version) > MAX_ENTRIES:
+ raise ValueError(f"A run is limited to {MAX_ENTRIES:,} inputs")
+ if len(args.inventory) + len(args.csv_list) > MAX_FILES:
+ raise ValueError(f"A run is limited to {MAX_FILES} input files")
+ for path in (*args.inventory, *args.csv_list):
+ _require_regular_file(path)
+ if args.output is not None:
+ output = args.output
+ parent = output.parent
+ if not parent.is_dir():
+ raise ValueError("Output directory must already exist")
+ if os.path.lexists(output):
+ if not args.force:
+ raise ValueError("Output exists; use a new path or explicitly pass --force")
+ if os.path.islink(output) and not output.exists():
+ raise ValueError("Output is a dangling or looping symbolic link")
+ if output.is_dir() or (output.exists() and not stat.S_ISREG(os.stat(output).st_mode)):
+ raise ValueError("Output must be a regular file path")
+ for path in (*args.inventory, *args.csv_list):
+ if os.path.samefile(output, path):
+ raise ValueError("--output must differ from every input file, even with --force")
+ for path in (*args.inventory, *args.csv_list):
+ if os.path.realpath(output) == os.path.realpath(path):
+ raise ValueError("--output must differ from every input file, even with --force")
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser(
+ description=__doc__, allow_abbrev=False, formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="Exit status: 0 success; 1 findings (only with --fail-on-finding: potentially_affected); 2 usage or operational error (nothing assessed or report not written); "
+ "130 interrupted (completed results are written with complete=false).")
+ parser.add_argument("--tool-version", action="version", version=f"{TOOL_NAME} {TOOL_VERSION}")
+ parser.add_argument("--version", "--package-version", dest="package_version", action="append", default=[], metavar="VERSION",
+ help="Assess an explicitly supplied tinypool version offline (leading v accepted); repeatable")
+ parser.add_argument("--build-source", default="unknown", metavar="SOURCE",
+ help="Provenance of the --version builds, at most 512 characters: unknown (default, compared as upstream), "
+ "upstream, official, or a vendor name")
+ for key, description in PRECONDITIONS:
+ parser.add_argument("--" + key.replace("_", "-"), choices=("true", "false"), dest=key,
+ help=f"Observation applied to --version inputs: {description}")
+ parser.add_argument("--inventory", action="append", default=[], type=Path, metavar="FILE",
+ help="JSON object or array of entries (see docs/REFERENCE.md); repeatable")
+ parser.add_argument("--csv-list", action="append", default=[], type=Path, metavar="FILE",
+ help="CSV with a package_version column and optional label, build_source and precondition columns; repeatable")
+ parser.add_argument("--format", choices=("csv", "json", "html"), default="csv", help="Output format (default: csv)")
+ parser.add_argument("--output", type=Path, help="Save report to a new file; default: stdout")
+ parser.add_argument("--force", action="store_true", help="Replace an existing output file explicitly")
+ parser.add_argument("--brand", default="CompleteTech LLC", help="Organization name for the HTML report (at most 512 characters)")
+ parser.add_argument("--fail-on-finding", action="store_true",
+ help="Exit 1 for potentially_affected results (errors still exit 2)")
+ args = parser.parse_args(argv)
+ if not (args.package_version or args.inventory or args.csv_list):
+ parser.error("provide --version, --inventory or --csv-list")
+ rows = []
+ state = {"validated": False, "published": False, "stdout_started": False}
+ options = {"force": args.force, "format": args.format, "brand": args.brand, "state": state}
+ try:
+ try:
+ _validate_arguments(args)
+ state["validated"] = True
+ observations = {key: _tri_state(getattr(args, key), key, text=True) for key in PRECONDITION_KEYS}
+ budget = Budget()
+ budget.add_entries(len(args.package_version))
+ for version in args.package_version:
+ rows.append(assess_entry("tinypool " + version.strip(ASCII_SPACE), "version", version, args.build_source, observations))
+ for path in args.inventory:
+ load_inventory(path, budget, rows)
+ for path in args.csv_list:
+ load_csv_list(path, budget, rows)
+ save_report(args.output, rows, complete=True, **options)
+ except KeyboardInterrupt:
+ if not state["validated"]:
+ print("Interrupted before the output path was validated; nothing was written.", file=sys.stderr)
+ return 130
+ if state["published"]:
+ print("Interrupted after the complete report was written; not retrying.", file=sys.stderr)
+ return 130
+ if state["stdout_started"]:
+ print("Interrupted while writing to stdout; output may be truncated and no second report is written.", file=sys.stderr)
+ return 130
+ print("Interrupted; writing completed results.", file=sys.stderr)
+ save_report(args.output, rows, complete=False, **options)
+ return 130
+ except KeyboardInterrupt:
+ print("Interrupted again; the partial report may not have been written.", file=sys.stderr)
+ return 130
+ except (OSError, ValueError, RecursionError, RuntimeError, csv.Error) as exc: # json errors are ValueError
+ print(f"{TOOL_NAME}: error: {_message(exc)}", file=sys.stderr)
+ return 2
+ if args.fail_on_finding and any(row["status"] in {"potentially_affected"} for row in rows):
+ return 1
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
-
- 
- 
-