diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 1d472b2..1f93817 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -25,8 +25,12 @@ jobs: - name: Run the test suite (standard library only, no install step) working-directory: scan run: python -m unittest discover -s tests -v + - name: Check source and download with no optional dependencies + if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.8' + run: python tools/release_check.py --e2e-only smoke-scan: + timeout-minutes: 10 # Prove the scan itself runs end to end on real hardware for each OS, not # just the unit tests. The runners have no real browser data, so the result # is an honest "nothing found / with gaps" - the point is that it executes @@ -61,6 +65,9 @@ jobs: run: python -m pip install -r requirements-patterns.txt - name: Test optional patterns and default checks using harmless fixtures run: python -B -m unittest discover -s scan/tests -v + - name: End-to-end source and download release fixtures + if: matrix.python-version == '3.12' + run: python tools/release_check.py --e2e-only --require-patterns reproducible-build: # The build must be deterministic and the committed artifacts up to date, so diff --git a/.gitignore b/.gitignore index da1f5a1..5efab5a 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,7 @@ __pycache__/ *.pyc *.pyo .pytest_cache/ +.venv/ .DS_Store # The single-file build is a release artifact, regenerated by build/bundle.py. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 30ae9ab..aa124e0 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -13,6 +13,12 @@ cd scan && python -m unittest discover -s tests Every change must keep the suite green. New behaviour needs a test. +Before publishing, run `python tools/release_check.py --require-patterns` in +the virtual environment with the pinned optional engine installed. This builds +and tests the download against controlled fixtures without scanning your own +browser profiles. See [TESTING.md](TESTING.md) for setup, website checks, scope +and the short release checklist. + ## Rebuilding the generated files The STIX/MISP feeds, the blocklists, the indicator checksum and the single-file diff --git a/README.md b/README.md index c8b66b0..091ff86 100644 --- a/README.md +++ b/README.md @@ -336,6 +336,11 @@ file shipped with the signed release, not later development-branch updates; see ## Tests +Maintainers: run `python tools/release_check.py --require-patterns` before +publishing. It builds the download and exercises source/download fixture scans, +reports, optional patterns and the webpage when supplied. Setup and the short +release checklist are in [TESTING.md](TESTING.md). + ``` cd scan && python -m unittest discover -s tests ``` diff --git a/TESTING.md b/TESTING.md new file mode 100644 index 0000000..a2fb7bb --- /dev/null +++ b/TESTING.md @@ -0,0 +1,98 @@ +# Before publishing + +Run the release check from the repository root. It uses temporary, harmless +fixtures rather than your browser profiles, registry or real DNS history. +No VM, malware samples, administrator account or test service is needed. + +## One-time optional-engine setup + +Use Python 3.9+ for the pinned optional YARA-X engine (Python 3.12 is used in CI): + +```text +python -m venv .venv +``` + +On Windows: + +```powershell +.venv\Scripts\python.exe -m pip install -r requirements-patterns.txt +.venv\Scripts\python.exe tools/release_check.py --require-patterns +``` + +On macOS/Linux: + +```sh +.venv/bin/python -m pip install -r requirements-patterns.txt +.venv/bin/python tools/release_check.py --require-patterns +``` + +Use the same interpreter for installation and tests. The runner never installs +anything. For ordinary standard-library development, `python +tools/release_check.py` works without YARA-X; its summary explicitly records the +untested optional positives as a skip. Use `--require-patterns` before release. + +## What the command checks + +- The existing unit/regression suite. +- Two byte-identical builds of the downloadable script, plus its help/version. +- Eleven fixture scenarios, each in fresh source-CLI and single-file processes: + clean data, contacted domain, harmless research reference, exact hash hit, + corrupt database, a hit alongside a corrupt database, `--quick` coverage gap, + a DNS hit past the old 20,000-event limit, unread older DNS events without a + hit, invalid indicator input, and COM-referenced optional file patterns. +- Real browser-shaped SQLite files and harmless rule-matching bytes pass through + the real scanner modules, evidence grading, exit codes and text/JSON reports. + Source and download must agree. Cookie values must stay out of reports; + full finding paths must remain visible. Fixtures must remain byte-identical + and temporary snapshots must be removed. +- During fixture scans, an audit hook rejects socket activity, subprocesses and + file opens/listings outside fixture, project-code and Python-runtime roots. + The scanner's network-guard recorder is checked too, so caught network + attempts cannot pass silently. Each unit-suite run has its own temp directory. + Each fixture process has a 45-second timeout; timeouts or assertion failures + make the command fail. Results/logs stay under ignored + `findings/release-checks/summary.json`. `PASS_WITH_SKIPS` names missing coverage. + +The host's registry and PowerShell collectors are replaced by fixed inputs; +the actual OS enumeration is **not** tested by these fixtures. This is an +end-to-end check of the application flow, not an OS sandbox or a measurement of +spyware detection rates. CI retains its real `--quick` smoke scans on disposable +Windows, macOS and Linux runners. It also runs the fixtures on all three with +YARA-X, and on Python 3.8 without it. Routine local runs normally take seconds; +timings are recorded rather than used as a hardware benchmark. + +## Include the website and staged download + +The Scan website is maintained separately from this public repository. When +its local source is available, include it explicitly: + +```text +python tools/release_check.py --require-patterns --web-dir PATH_TO_SCAN_DIRECTORY --scanner PATH_TO_STAGED_ZERO_BREACH_PY +``` + +`--web-dir` requires Node, `index.html`, `scan.js`, and +`ops/test_scan_logic.js`. It runs the existing offline page-flow matrix and + requires the verification card's code text and copy value to match the candidate script. These +are DOM/logic fixtures; actual rendering and touch behaviour still need the +browser smoke check. `--scanner` requires the staged download to match the +fresh build byte for byte. Omit it to test the freshly built download. +If `web/scan/` exists in this checkout, it is checked automatically; otherwise +the website is explicitly listed as not tested. A requested website check must +pass; it is never silently skipped. Use `--output-dir PATH` to keep separate runs. + +## Short release checklist + +1. Run the command above on the frozen candidate, including optional patterns, + staged scanner and prepared website when those are being published. +2. Require all CI jobs for that exact commit. Run the existing live-browser + smoke check for website changes, including a small mobile viewport. For OS + collector changes, also use a disposable Windows VM with controlled data; + do not substitute a scan of your everyday laptop. +3. Review remaining findings; get the requested independent Astra final review. + Verify release signatures and checksums following [SIGNING.md](SIGNING.md). +4. Publish the tested bytes with a saved rollback. Verify the public download, + page checksum/links and a basic user flow after deployment. + +No soak test, load-test infrastructure or live malware is required for this +routine gate. A timed-out real scan is an unresolved performance observation, +even when these bounded fixtures pass. diff --git a/scan/tests/test_release_check.py b/scan/tests/test_release_check.py new file mode 100644 index 0000000..76d0417 --- /dev/null +++ b/scan/tests/test_release_check.py @@ -0,0 +1,60 @@ +"""Regressions for false-green failures in the offline release gate itself.""" +import importlib.util +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[2] +spec = importlib.util.spec_from_file_location("release_gate", ROOT / "tools/release_check.py") +gate = importlib.util.module_from_spec(spec) +spec.loader.exec_module(gate) + + +class TestWebDownloadChecksum(unittest.TestCase): + DIGEST = "a" * 64 + + def panel(self, displayed=None, copied=None): + return ('
{}' + '
').format( + displayed or self.DIGEST, copied or self.DIGEST) + + def test_actual_code_and_copy_value_match(self): + gate.check_web_checksum(self.panel(), self.DIGEST) + + def test_correct_hash_in_comment_does_not_mask_wrong_display(self): + html = self.panel(displayed="b" * 64) + "" + with self.assertRaisesRegex(RuntimeError, "displayed/copied"): + gate.check_web_checksum(html, self.DIGEST) + + def test_correct_display_does_not_mask_wrong_copy_value(self): + with self.assertRaisesRegex(RuntimeError, "displayed/copied"): + gate.check_web_checksum(self.panel(copied="b" * 64), self.DIGEST) + + +class TestCaughtNetworkAttempt(unittest.TestCase): + def test_scanner_cannot_hide_an_attempt_by_catching_guard_exception(self): + with tempfile.TemporaryDirectory(prefix="zb_gate_regression_") as tmp: + root = Path(tmp) + candidate = root / "candidate.py" + subprocess.run([sys.executable, "-B", str(ROOT / "build/bundle.py"), + str(ROOT), str(candidate)], check=True, capture_output=True) + source = candidate.read_text(encoding="utf-8") + needle = "def _pattern_engine():" + self.assertIn(needle, source) + # no_network blocks this before any real resolution; the pattern + # module catches it and would normally just report UNKNOWN. + source = source.replace(needle, needle + '\n __import__("socket").' + 'gethostbyname("release-fixture.invalid")') + candidate.write_text(source, encoding="utf-8") + proc = subprocess.run( + [sys.executable, "-B", str(ROOT / "tools/release_fixture.py"), + "bundle", "patterns", str(root / "fixture"), str(candidate)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=45) + self.assertNotEqual(proc.returncode, 0) + self.assertIn(b"blocked external activity occurred", proc.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/release_check.py b/tools/release_check.py new file mode 100644 index 0000000..25343a2 --- /dev/null +++ b/tools/release_check.py @@ -0,0 +1,198 @@ +#!/usr/bin/env python3 +"""Small, offline pre-release gate. Never runs a scan against the real host.""" +import argparse +import hashlib +from html.parser import HTMLParser +import importlib.util +import json +import os +from pathlib import Path +import shutil +import subprocess +import sys +import tempfile +import time + +ROOT = Path(__file__).resolve().parents[1] +CASES = ("clean", "contact", "reference", "hash", "corrupt", "hit-and-gap", "quick", + "dns-cap-hit", "dns-cap-gap", "bad-feed", "patterns") + + +class DownloadPanel(HTMLParser): + """Read code/copy values from the actual verification card, not comments.""" + def __init__(self): + super().__init__() + self.depth = 0 + self.group_depth = None + self.group = None + self.capture = None + self.cards = 0 + self.groups = [] + + def handle_starttag(self, tag, attributes): + attrs = dict(attributes) + if tag == "div": + if attrs.get("id") == "verify-download": + self.cards += 1 + self.depth = 1 + elif self.depth: + self.depth += 1 + if self.depth and "ea-copy" in attrs.get("class", "").split(): + self.group = {"code": "", "label": "", "copy": None} + self.groups.append(self.group) + self.group_depth = self.depth + elif self.group is not None and tag in ("code", "button"): + self.capture = "code" if tag == "code" else "label" + if tag == "button": + self.group["copy"] = attrs.get("data-copy") + + def handle_endtag(self, tag): + if tag in ("code", "button"): + self.capture = None + if tag == "div" and self.depth: + if self.depth == self.group_depth: + self.group = None + self.group_depth = None + self.depth -= 1 + + def handle_data(self, text): + if self.group is not None and self.capture: + self.group[self.capture] += text + + +def check_web_checksum(html, digest): + panel = DownloadPanel() + panel.feed(html) + candidates = [g for g in panel.groups if g["label"].strip() == "Copy SHA-256"] + if panel.cards != 1 or len(candidates) != 1: + raise RuntimeError("Scan page must have one verification card and SHA-256 copy control") + value = candidates[0] + if value["code"].strip() != digest or value["copy"] != digest: + raise RuntimeError("Scan page displayed/copied checksum differs from the candidate download") + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--require-patterns", action="store_true", + help="require the pinned optional engine (use for releases)") + parser.add_argument("--e2e-only", action="store_true", + help="CI: unit suite already ran in the same job") + parser.add_argument("--scanner", type=Path, + help="also require this staged download to match the fresh build") + parser.add_argument("--web-dir", type=Path, + help="Scan-page directory; require its logic test and checksum to pass") + parser.add_argument("--output-dir", type=Path, + default=ROOT / "findings" / "release-checks") + args = parser.parse_args() + output = args.output_dir.resolve() + output.mkdir(parents=True, exist_ok=True) + started = time.monotonic() + steps = [] + summary = {"status": "FAIL", "steps": steps, "skipped": [], + "scope": "Offline fixtures; OS collection is simulated; no personal scan"} + env = dict(os.environ, PYTHONUTF8="1", PYTHONDONTWRITEBYTECODE="1") + + def run(name, command, timeout=60, environment=None): + tick = time.monotonic() + try: + proc = subprocess.run(command, cwd=ROOT, env=environment or env, stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + timeout=timeout) + text = proc.stdout.decode("utf-8", errors="replace") + (output / (name + ".log")).write_text(text, encoding="utf-8") + if proc.returncode: + raise RuntimeError("{} failed (exit {}); see {}".format( + name, proc.returncode, output / (name + ".log"))) + except subprocess.TimeoutExpired: + raise RuntimeError("{} exceeded {}s; release check failed".format(name, timeout)) + steps.append({"name": name, "seconds": round(time.monotonic() - tick, 2)}) + print("PASS {} ({:.1f}s)".format(name, time.monotonic() - tick), flush=True) + + try: + sys.path.insert(0, str(ROOT / "scan")) + from adint_scan.candiru_rules import CANDIRU_ENGINE_VERSION + engine = importlib.util.find_spec("yara_x") is not None + if engine: + from importlib.metadata import version + if version("yara-x") != CANDIRU_ENGINE_VERSION: + raise RuntimeError("optional engine version differs from requirements-patterns.txt") + elif args.require_patterns: + raise RuntimeError("pinned YARA-X missing; install requirements-patterns.txt " + "in this interpreter's virtual environment first") + else: + summary["skipped"].append("Real YARA-X positives: engine not installed") + + if not args.e2e_only: + # Snapshot-cleanup tests inspect the temp directory. Separate + # release runs must not see each other's transient snapshots. + with tempfile.TemporaryDirectory(prefix="zb_unit_") as unit_temp: + unit_env = dict(env, TMP=unit_temp, TEMP=unit_temp, TMPDIR=unit_temp) + run("unit-tests", [sys.executable, "-B", "-m", "unittest", "discover", + "-s", "scan/tests"], timeout=180, environment=unit_env) + + with tempfile.TemporaryDirectory(prefix="zb_release_") as temp: + work = Path(temp) + first, second = work / "scanner.py", work / "scanner-again.py" + for name, path in (("build", first), ("rebuild", second)): + run(name, [sys.executable, "-B", str(ROOT / "build/bundle.py"), + str(ROOT), str(path)]) + if first.read_bytes() != second.read_bytes(): + raise RuntimeError("two single-file builds differ") + digest = hashlib.sha256(first.read_bytes()).hexdigest() + summary["scanner_sha256"] = digest + scanner = first + if args.scanner: + scanner = args.scanner.resolve() + if scanner.read_bytes() != first.read_bytes(): + raise RuntimeError("staged scanner differs from the tested source build") + run("scanner-help", [sys.executable, "-B", str(scanner), "--help"]) + run("scanner-version", [sys.executable, "-B", str(scanner), "--version"]) + + for case in CASES: + pair = [] + for target in ("source", "bundle"): + case_root = work / (target + "-" + case) + result = case_root / "result.json" + run(target + "-" + case, + [sys.executable, "-B", str(ROOT / "tools/release_fixture.py"), + target, case, str(case_root), str(scanner)], timeout=45) + pair.append(json.loads(result.read_text(encoding="utf-8"))) + if pair[0] != pair[1]: + raise RuntimeError("source/download results disagree for " + case) + + web = args.web_dir + if web is None and (ROOT / "web/scan").is_dir(): + web = ROOT / "web/scan" + if web is not None: + web = web.resolve() + test = web / "ops/test_scan_logic.js" + if not all((web / name).is_file() for name in + ("index.html", "scan.js", "ops/test_scan_logic.js")) or not shutil.which("node"): + raise RuntimeError("web check requires Node, index.html, scan.js and ops/test_scan_logic.js") + run("web-logic", ["node", str(test)]) + html = (web / "index.html").read_text(encoding="utf-8") + check_web_checksum(html, digest) + steps.append({"name": "web-download-checksum", "seconds": 0}) + else: + summary["skipped"].append("Website: separate site source not supplied (--web-dir)") + + summary["status"] = "PASS_WITH_SKIPS" if summary["skipped"] else "PASS" + print("\nPASS: source/download fixtures agree; inputs unchanged; " + "no socket/subprocess activity during fixture scans.") + for skip in summary["skipped"]: + print("NOT RUN: " + skip) + print("Scanner SHA-256: " + digest) + return 0 + except (OSError, RuntimeError, ValueError) as exc: + summary["error"] = str(exc) + print("\nFAIL: " + str(exc), file=sys.stderr) + return 1 + finally: + summary["seconds"] = round(time.monotonic() - started, 2) + (output / "summary.json").write_text( + json.dumps(summary, indent=2) + "\n", encoding="utf-8") + print("Results: " + str(output / "summary.json")) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/release_fixture.py b/tools/release_fixture.py new file mode 100644 index 0000000..95c6821 --- /dev/null +++ b/tools/release_fixture.py @@ -0,0 +1,230 @@ +"""Child-process CLI fixtures for release_check.py; not a scanner feature. + +Only discovery roots and OS collectors are substituted. The actual CLI, +browser/hash/host/pattern modules, evidence grading, JSON and text rendering run. +""" +import contextlib +import hashlib +import importlib.util +import io +import json +import os +from pathlib import Path +import sqlite3 +import sys +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +DOMAIN = "release-fixture.invalid" +PAYLOAD = b"Harmless Zero Breach release test bytes; not malware.\n" +DIGEST = hashlib.sha256(PAYLOAD).hexdigest() + + +def require(condition, message): + if not condition: + raise RuntimeError(message) + + +def snapshot(root): + return {str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in root.rglob("*") if p.is_file()} + + +def main(): + target, case, directory, scanner = sys.argv[1:] + root = Path(directory).resolve() + inputs, outputs, scratch = root / "inputs", root / "outputs", root / "tmp" + for p in (inputs, outputs, scratch): + p.mkdir(parents=True) + # Even an accidentally unpatched discovery point sees a synthetic home. + for key in ("HOME", "USERPROFILE", "APPDATA", "LOCALAPPDATA", "PROGRAMDATA", "SystemRoot"): + os.environ[key] = str(inputs) + for key in ("TMP", "TEMP", "TMPDIR"): + os.environ[key] = str(scratch) + import tempfile + tempfile.tempdir = str(scratch) + sys.path[:0] = [str(ROOT / "scan"), str(ROOT / "scan/tests")] + if target == "bundle": + spec = importlib.util.spec_from_file_location("release_candidate", scanner) + cli = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = cli + spec.loader.exec_module(cli) + browser = hashes = host = patterns = reporting = cli + else: + from adint_scan import cli, report as reporting + from adint_scan.modules import browser, hashscan as hashes, host, patternscan as patterns + + profile = inputs / "Browser" / "Default" + profile.mkdir(parents=True) + history = profile / "History" + with sqlite3.connect(str(history)) as con: + con.execute("CREATE TABLE urls (url TEXT, title TEXT)") + con.execute("INSERT INTO urls VALUES (?, ?)", ("https://example.invalid/", "ordinary")) + if case in ("corrupt", "hit-and-gap"): + history.write_bytes(b"not a SQLite database") + if case in ("contact", "hit-and-gap"): + cookie = profile / "Network" / "Cookies" + cookie.parent.mkdir() + with sqlite3.connect(str(cookie)) as con: + con.execute("CREATE TABLE cookies (host_key TEXT, name TEXT, " + "creation_utc INTEGER, encrypted_value BLOB)") + con.execute("INSERT INTO cookies VALUES (?, 'sid', 13300000000000000, ?)", + ("." + DOMAIN, b"PRIVATE_COOKIE_SENTINEL_DO_NOT_REPORT")) + if case == "reference": + blob = profile / "Local Storage" / "leveldb" / "000001.log" + blob.parent.mkdir(parents=True) + blob.write_text("saved article https://" + DOMAIN + "/research", encoding="utf-8") + files = inputs / "Files" + files.mkdir() + (files / "ordinary.dll").write_bytes(PAYLOAD if case in ("hash", "quick") else b"ordinary") + feed = inputs / "iocs.txt" + feed.write_text("# as-of: 2026-10-19\n" + DOMAIN + "\tdelivery\tprimary\tfixture\n" + + DIGEST + "\tsample\tprimary\tfixture\n", encoding="utf-8") + hosts = inputs / "System32" / "drivers" / "etc" / "hosts" + hosts.parent.mkdir(parents=True) + # Protective blocklist entries must never create findings. + hosts.write_text("127.0.0.1 localhost\n0.0.0.0 " + DOMAIN + "\n", encoding="utf-8") + registered = inputs / "Registered" / "server.dll" + registered.parent.mkdir() + engine = importlib.util.find_spec("yara_x") is not None + if case == "patterns" and engine: + from test_patterns import synthetic_mstic_fixture + registered.write_bytes(synthetic_mstic_fixture()) + else: + registered.write_bytes(b"ordinary registered file") + before = snapshot(inputs) + denied_activity = [] + original_network_guard = cli.no_network + + @contextlib.contextmanager + def checked_network_guard(): + # The scanner's socket wrappers reject calls before Python emits an + # audit event. Observe their recorder too, including caught attempts. + with original_network_guard() as recorder: + try: + yield recorder + finally: + denied_activity.extend(recorder.attempts) + + allowed = (root, ROOT, Path(sys.prefix).absolute(), Path(sys.prefix).resolve(), + Path(sys.base_prefix).absolute(), Path(sys.base_prefix).resolve()) + + def within(path): + # Lexical only; no link resolution before the scanner's locality guard. + path = os.path.normcase(os.path.abspath(os.fspath(path))) + return any(path == os.path.normcase(str(base)) + or path.startswith(os.path.normcase(str(base)) + os.sep) for base in allowed) + + def audit(event, args): + if event.startswith("socket.") or event in ("subprocess.Popen", "os.system"): + denied_activity.append(event) + raise RuntimeError("fixture attempted external activity: " + event) + if event in ("open", "os.listdir", "os.scandir") and args: + path = args[0] + if isinstance(path, (str, bytes, os.PathLike)) and not within(os.fsdecode(path)): + denied_activity.append(event) + raise RuntimeError("fixture attempted file access outside test/code/runtime roots") + + events = [{"Time": "2025-06-01T00:00:00", "Message": "ordinary.invalid"}] + if case.startswith("dns-cap-"): + events = events * 50000 + if case == "dns-cap-hit": + events[20000] = {"Time": "2025-06-01T00:00:00", "Message": DOMAIN} + + def ps_json(command, timeout=60): + if "-ListLog " in command: + return [{"IsEnabled": True, "RecordCount": len(events) + int(case.startswith("dns-cap-"))}], "" + if "Get-WinEvent -LogName " in command: + require("-MaxEvents 50000" in command, "unexpected DNS read cap") + return events, "" + if "Get-DnsClientCache" in command: + return [{"Entry": DOMAIN, "Data": "0.0.0.0", "Type": 1}], "" + if "Win32_OperatingSystem" in command: + return [{"Boot": "2026-01-01T00:00:00"}], "" + raise RuntimeError("unexpected OS query in fixture") + + entries = ([{"clsid": "{FIXTURE}", "kind": "InprocServer32", "raw": str(registered), + "view": "native"}] if case == "patterns" else []) + report_path = outputs / "report.json" + argv = ["--iocs", str(feed if case != "bad-feed" else inputs / "missing.txt"), + "--json", str(report_path)] + if target == "bundle": + argv.append("--no-pause") + if case == "quick": + argv.append("--quick") + if case == "patterns": + argv.append("--patterns") + with contextlib.ExitStack() as stack: + stack.enter_context(patch.object(cli, "no_network", checked_network_guard)) + for mod, name, value in ( + (browser, "CHROMIUM_ROOTS", [("fixture", str(profile.parent))]), + (browser, "FIREFOX_ROOTS", []), + (hashes, "_roots", lambda: [str(files)]), + (patterns, "_pattern_roots", lambda: [str(files)]), + (host, "winreg", object()), + (host, "_ps_json", ps_json), + (host, "_enum_com_registrations", lambda r: (entries, [])), + (host, "_network_drive_letters", lambda: (set(), True)), + (host, "_user_writable_dirs", lambda: [str(inputs)]), + (reporting, "_platform_info", lambda: {"system": "Windows", "release": "fixture"})): + stack.enter_context(patch.object(mod, name, value)) + stack.enter_context(patch.object(host.platform, "system", return_value="Windows")) + stdout, stderr = io.StringIO(), io.StringIO() + # Do not let the CLI reconfigure the capture streams. + stack.enter_context(patch.object(cli, "_force_utf8_io", lambda: None)) + sys.addaudithook(audit) + with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + code = cli.main(argv) + text = stdout.getvalue() + (outputs / "report.txt").write_text(text, encoding="utf-8") + require(not denied_activity, "blocked external activity occurred: " + str(denied_activity)) + require(snapshot(inputs) == before, "scanner changed fixture inputs") + require(not list(scratch.iterdir()), "scanner left temporary snapshot files behind") + require("PRIVATE_COOKIE_SENTINEL_DO_NOT_REPORT" not in text, "cookie value leaked to text") + if case == "bad-feed": + require(code == 1 and not report_path.exists(), "invalid feed must fail setup without a report") + result = {"exit": code, "setup_error": True} + else: + doc = json.loads(report_path.read_text(encoding="utf-8")) + require("PRIVATE_COOKIE_SENTINEL_DO_NOT_REPORT" not in report_path.read_text(encoding="utf-8"), + "cookie value leaked to JSON") + modules, verdict = doc["modules"], doc["verdict"] + hits = sorted((name, h["indicator"], h["evidence"]) + for name, mod in modules.items() for h in mod["hits"]) + expected = {"contact": ("browser", DOMAIN, "contacted"), + "hit-and-gap": ("browser", DOMAIN, "contacted"), + "reference": ("browser", DOMAIN, "referenced"), + "hash": ("hash", DIGEST, "on-disk"), + "dns-cap-hit": ("host", DOMAIN, "dns-record")} + if case == "patterns" and engine: + expected[case] = ("patterns", "yara:MAL_DevilsTongue_HijackDll", "file-pattern") + require(hits == ([expected[case]] if case in expected else []), + "wrong evidence/hits: " + str(hits)) + require(code == (2 if hits else 0), "incorrect CLI exit code") + require(verdict["hits_total"] == len(hits), "incorrect total finding count") + expected_verdict = ("INDICATORS_FOUND" if hits else + "NO_KNOWN_INDICATORS_WITH_GAPS" if case in + ("corrupt", "quick", "dns-cap-gap", "patterns") else "NO_KNOWN_INDICATORS") + require(verdict["status"] == expected_verdict, "incorrect overall verdict: " + str(verdict)) + if case.startswith("dns-cap-"): + require(any("dns-log-older-events" in g for g in verdict["gaps"]), "missing DNS cap gap") + require(modules["host"]["submetrics"]["dns_log_events_scanned"] == 50000, "wrong DNS count") + if case == "quick": + require(modules["hash"]["status"] == "UNKNOWN", "quick scan falsely cleared file hashes") + if case in ("corrupt", "hit-and-gap"): + require(any("history" in g for g in verdict["gaps"]), "corrupt database gap missing") + if case in ("contact", "hit-and-gap"): + require(str(cookie) in text, "human report lost the full cookie file path") + if case == "patterns": + require(modules["patterns"]["status"] == ("HITS" if engine else "UNKNOWN"), + "optional engine status is dishonest") + require("HOW FAR BACK THIS SCAN COULD SEE" in text, "human report did not render") + result = {"exit": code, "verdict": verdict["status"], "hits": hits, + "modules": {n: m["status"] for n, m in modules.items()}, + "gaps": sorted(g.split(":", 1)[0] for g in verdict["gaps"])} + (root / "result.json").write_text(json.dumps(result), encoding="utf-8") + print("PASS {} {}: inputs unchanged, expected report/exit, no external activity".format(target, case)) + + +if __name__ == "__main__": + main()