CVE-2026-104849 Version Report
-Tinypool worker-options prototype-pollution gadget · generated 2026-10-24 14:45 UTC · rules as of 2026-10-24
Tinypool worker-options prototype-pollution gadget · generated 2026-10-25T12:34:56+00:00 · 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.
Results
| Input | Assessment | Reported version | Upstream fix | Precondition observations | Next step and evidence |
|---|---|---|---|---|---|
| example-package-instance inventory | Fixed version reportedat_or_above_fixed_version | 2.1.2build: unknown | 2.1.2 |
| Verify this package instance is actually deployed and contains the upstream fix. Evidence and limitations
|
version
in_documented_affected_range2.1.0build: unknown
2.1.2caller_supplies_pool_run_options_object: not_reportedoptions_lacks_an_own_filename_property: not_reportedseparate_prototype_pollution_primitive: not_reportedattacker_controlled_worker_module_available: not_reported
Verify the deployed tinypool instance and upgrade to 2.1.2 or a reviewed fixed release; review advisory conditions without executing untrusted code.
Evidence and limitations
- Reported version matches the documented affected range; application preconditions and exploitability were not established.
- 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.
version
source_discrepancy2.1.1build: unknown
—caller_supplies_pool_run_options_object: not_reportedoptions_lacks_an_own_filename_property: not_reportedseparate_prototype_pollution_primitive: not_reportedattacker_controlled_worker_module_available: not_reported
Collect or verify the exact upstream tinypool version and build provenance of the running process, then run this checker again.
Evidence and limitations
- Primary advisory and reviewed database disagree at this boundary; obtain maintainer evidence.
- 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.
version
at_or_above_fixed_version2.1.2build: unknown
2.1.2caller_supplies_pool_run_options_object: not_reportedoptions_lacks_an_own_filename_property: not_reportedseparate_prototype_pollution_primitive: not_reportedattacker_controlled_worker_module_available: not_reported
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. diff --git a/test_report_provenance.py b/test_report_provenance.py new file mode 100644 index 0000000..4a9e54a --- /dev/null +++ b/test_report_provenance.py @@ -0,0 +1,96 @@ +"""Offline regressions for detached exports and interrupted run identity.""" +import contextlib +import csv +from datetime import datetime, timezone +import io +import json +import socket +import unittest +from unittest import mock + +import tinypool_options_rce_check as checker + + +STAMP = datetime(2026, 10, 25, 12, 34, 56, tzinfo=timezone.utc) + + +class ProvenanceTests(unittest.TestCase): + def verify_export(self, text, fmt, *, count, complete): + if fmt == 'csv': + rows = list(csv.DictReader(io.StringIO(text, newline=''))) + self.assertEqual(max(1, count), len(rows)) + for row in rows: + self.assertEqual(checker.TOOL_NAME, row['tool']) + self.assertEqual(checker.TOOL_VERSION, row['tool_version']) + self.assertEqual(STAMP.isoformat(), row['generated_at']) + self.assertEqual(checker.RULES_DATE, row['rules_as_of']) + self.assertEqual(str(complete).lower(), row['report_complete']) + self.assertEqual('not_tested', row['exploitability']) + self.assertEqual('result' if count else 'summary', row['record_type']) + return [r['status'] for r in rows] if count else [] + if fmt == 'json': + report = json.loads(text) + self.assertEqual(checker.TOOL_NAME, report['tool']) + self.assertEqual(checker.TOOL_VERSION, report['tool_version']) + self.assertEqual(STAMP.isoformat(), report['generated_at']) + self.assertEqual(checker.RULES_DATE, report['rules_as_of']) + self.assertEqual(complete, report['complete']) + self.assertEqual(count, len(report['results'])) + return [r['status'] for r in report['results']] + self.assertIn(STAMP.isoformat(), text) + self.assertIn(checker.TOOL_NAME + ' ' + checker.TOOL_VERSION, text) + self.assertIn('rules as of ' + checker.RULES_DATE, text) + if not complete: + self.assertIn('Interrupted run.', text) + + def test_complete_and_empty_exports_share_captured_metadata(self): + results = [checker.assess_entry('synthetic', 'version', v) for v in ('2.1.0', '2.1.1', '2.1.2')] + for rows in (results, []): + for fmt in ('csv', 'json', 'html'): + with self.subTest(count=len(rows), fmt=fmt), mock.patch.object(checker, 'datetime') as clock: + clock.now.return_value = STAMP + out = io.StringIO() + checker.write_report(out, rows, format=fmt) + clock.now.assert_called_once_with(timezone.utc) + statuses = self.verify_export(out.getvalue(), fmt, count=len(rows), complete=True) + if statuses: + self.assertEqual(['potentially_affected', 'unknown', 'fixed_version_reported'], statuses) + + def test_interrupted_cli_preserves_start_time_and_completed_rows(self): + real = checker.assess_entry + for completed in (0, 1): + for fmt in ('csv', 'json', 'html'): + calls = 0 + def assess(*args, **kwargs): + nonlocal calls + calls += 1 + if calls > completed: + raise KeyboardInterrupt + return real(*args, **kwargs) + out = io.StringIO() + with self.subTest(completed=completed, fmt=fmt), mock.patch.object(checker, 'datetime') as clock, mock.patch.object(checker, 'assess_entry', side_effect=assess), mock.patch.object(socket, 'socket', side_effect=AssertionError('offline only')), contextlib.redirect_stdout(out), contextlib.redirect_stderr(io.StringIO()): + clock.now.return_value = STAMP + self.assertEqual(130, checker.main(['--version', '2.1.1', '--version', '2.1.2', '--format', fmt])) + clock.now.assert_called_once_with(timezone.utc) + statuses = self.verify_export(out.getvalue(), fmt, count=completed, complete=False) + if statuses: + self.assertEqual(['unknown'], statuses) + + def test_injected_timestamp_is_shared_without_reading_clock(self): + for fmt in ('csv', 'json', 'html'): + with mock.patch.object(checker, 'datetime') as clock: + out = io.StringIO() + checker.write_report(out, [], format=fmt, generated_at=STAMP.isoformat(), complete=False) + clock.now.assert_not_called() + self.verify_export(out.getvalue(), fmt, count=0, complete=False) + + def test_csv_row_cannot_override_run_identity(self): + row = checker.assess_entry('synthetic', 'version', '2.1.1') + row.update(tool='other', tool_version='other', generated_at='other') + out = io.StringIO() + checker.write_report(out, [row], generated_at=STAMP.isoformat()) + self.verify_export(out.getvalue(), 'csv', count=1, complete=True) + + +if __name__ == '__main__': + unittest.main() diff --git a/tinypool_options_rce_check.py b/tinypool_options_rce_check.py index 3a92080..4d8d0ba 100644 --- a/tinypool_options_rce_check.py +++ b/tinypool_options_rce_check.py @@ -148,7 +148,8 @@ def assess_version(package_version="", *, version_candidates=None, build_source= 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"] + "recommendation", "notes", "sources", "report_schema_version", "report_complete", + "tool", "tool_version", "generated_at"] 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"} @@ -370,7 +371,7 @@ def render_html(rows, *, brand="CompleteTech LLC", generated_at=None, complete=T 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") + generated_at = generated_at or datetime.now(timezone.utc).isoformat() 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"), @@ -416,25 +417,31 @@ def esc(value): """ -def write_report(output, rows, *, format="csv", brand="CompleteTech LLC", complete=True): +def write_report(output, rows, *, format="csv", brand="CompleteTech LLC", complete=True, generated_at=None): + # Capture once, then share the envelope with every format and every CSV row. + metadata = {"tool": TOOL_NAME, "tool_version": TOOL_VERSION, + "generated_at": generated_at or datetime.now(timezone.utc).isoformat()} 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"}) + writer.writerow({**{key: csv_safe(row.get(key)) for key in CSV_COLUMNS}, + **{key: csv_safe(value) for key, value in metadata.items()}, + "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"}) + "cve": CVE_ID, "exploitability": "not_tested", + **{key: csv_safe(value) for key, value in metadata.items()}}) elif format == "json": - json.dump({"tool": TOOL_NAME, "tool_version": TOOL_VERSION, "report_schema_version": REPORT_SCHEMA_VERSION, + json.dump({**metadata, "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}, + "results": rows}, output, ensure_ascii=False, indent=2) output.write("\n") else: - output.write(render_html(rows, brand=brand, complete=complete)) + output.write(render_html(rows, brand=brand, complete=complete, generated_at=metadata["generated_at"])) def render_report(rows, **options): @@ -562,7 +569,7 @@ def main(argv=None): 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("--format", choices=("csv", "json", "html"), default="csv", help="Output format, each with checker identity and UTC generation time (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)") @@ -573,7 +580,8 @@ def main(argv=None): 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} + options = {"force": args.force, "format": args.format, "brand": args.brand, "state": state, + "generated_at": datetime.now(timezone.utc).isoformat()} try: try: _validate_arguments(args)