From 86530e16b1efc40584fa48a00973b77253b2d648 Mon Sep 17 00:00:00 2001 From: interplains Date: Tue, 20 Oct 2026 14:41:44 +0300 Subject: [PATCH] Harden report rendering, indicator loading and browser parsing Report: sanitise artifact-derived text before rendering. Control characters, line separators and bidirectional overrides are replaced and newlines collapse to spaces, so a crafted profile name or registry value cannot forge report lines. The privacy note now appears on every report, not only on ones with findings. Indicators: reject values that are not hostnames, IP addresses or hex digests, including empty, unspecified and over-long values, and cap the indicator count. Browser: treat deeply nested JSON as a malformed artifact instead of aborting the module, and refuse oversized databases before copying them. CLI: refuse --iocs and --json paths that are not on a local disk. Remove the unused write-mode module and an unused quoting helper. --- build/bundle.py | 7 +- scan/adint_scan/arm.py | 554 ----------------------------- scan/adint_scan/cli.py | 47 ++- scan/adint_scan/iocs.py | 44 +++ scan/adint_scan/modules/browser.py | 30 +- scan/adint_scan/modules/host.py | 8 +- scan/adint_scan/report.py | 50 ++- scan/tests/test_arm_undo.py | 392 -------------------- 8 files changed, 154 insertions(+), 978 deletions(-) delete mode 100644 scan/adint_scan/arm.py delete mode 100644 scan/tests/test_arm_undo.py diff --git a/build/bundle.py b/build/bundle.py index 4a1eed0..efead9f 100644 --- a/build/bundle.py +++ b/build/bundle.py @@ -1,9 +1,10 @@ """Build a single-file, self-contained copy of the scanner. One .py file someone can be emailed, read, and run. It embeds the indicator data -so there is no second file to lose, and it deliberately OMITS arm.py, so the -result cannot change anything on the machine it runs on - which is a much better -thing to be able to promise someone receiving a script from a friend. +so there is no second file to lose, and the scanner contains no code that writes +to the machine, so the result cannot change anything on the machine it runs on - +which is a much better thing to be able to promise someone receiving a script from +a friend. The only transformation applied to the module sources is renaming the two names that collide across the three scan modules (NAME and run), plus dropping the diff --git a/scan/adint_scan/arm.py b/scan/adint_scan/arm.py deleted file mode 100644 index dc78aef..0000000 --- a/scan/adint_scan/arm.py +++ /dev/null @@ -1,554 +0,0 @@ -"""Tier 5 — `--arm`: make this machine answerable next time. - -Why this is the most valuable thing in the tool ------------------------------------------------ -The reference run found that `Microsoft-Windows-DNS-Client/Operational` is -DISABLED by default on Windows. On a normal machine there is therefore no DNS -history at all, which is why the 2024-2025 window in which the Al Jazeera devices -were hit cannot be checked retrospectively from the host. We cannot fix that -backwards. We can make the *next* campaign answerable — and a machine that is a -tripwire is worth more to an at-risk person than one that searches an empty attic. - -Doctrine --------- -D5 Nothing here runs without an explicit flag. `--arm` on its own is a DRY RUN: - it prints precisely what each step would change, whether it needs - administrator, and how to undo it — and changes nothing. You opt into each - step by name (or `--all`). -D2 Still no network. Note what is deliberately NOT here: we do not download - Sysmon. An auto-updating security tool with this adversary is itself a - target, and fetching a binary at arm time would be exactly the supply-chain - exposure DESIGN.md warns about. We detect Sysmon and print the manual step. -D3 Every change is journalled with its PRIOR value, so `--undo` restores the - previous state rather than guessing a default. -""" - -from __future__ import annotations - -import ctypes -import datetime -import hashlib -import json -import os -import platform -import subprocess -import sys -from typing import Any, Dict, List, Optional, Tuple - -STATE_DIRNAME = "adint-scan" -STATE_FILE = "arm-state.json" -BASELINE_FILE = "persistence-baseline.json" - -DNS_LOG = "Microsoft-Windows-DNS-Client/Operational" -DNS_LOG_SIZE_BYTES = 64 * 1024 * 1024 # 64 MB of retention, ~months of queries -RESCAN_TASK_NAME = "adint-scan-periodic" - -# --- small helpers ---------------------------------------------------------- - - -def state_dir() -> str: - base = os.environ.get("LOCALAPPDATA") or os.path.expanduser("~") - return os.path.join(base, STATE_DIRNAME) - - -def _state_path() -> str: - return os.path.join(state_dir(), STATE_FILE) - - -def _baseline_path() -> str: - return os.path.join(state_dir(), BASELINE_FILE) - - -def is_admin() -> bool: - if platform.system() != "Windows": - return os.geteuid() == 0 if hasattr(os, "geteuid") else False - try: - return bool(ctypes.windll.shell32.IsUserAnAdmin()) - except Exception: - return False - - -def _ps(cmd: str, timeout: int = 120) -> Tuple[int, str, str]: - """Run a PowerShell command. Used for local system state only — never for - anything that would resolve or contact an indicator. - - encoding/errors are explicit because PowerShell and the Windows console - tools write in the console code page, not UTF-8; strict decoding raises - inside subprocess's reader thread on any non-ASCII byte. - """ - cmd = "[Console]::OutputEncoding=[System.Text.Encoding]::UTF8; " + cmd - proc = subprocess.run( - ["powershell", "-NoProfile", "-NonInteractive", - "-ExecutionPolicy", "Bypass", "-Command", cmd], - capture_output=True, text=True, - encoding="utf-8", errors="replace", timeout=timeout) - return proc.returncode, (proc.stdout or ""), (proc.stderr or "") - - -def _run(argv: List[str], timeout: int = 120) -> Tuple[int, str, str]: - # wevtutil and schtasks emit the console code page; decode leniently so a - # localised Windows cannot crash an undo. - proc = subprocess.run(argv, capture_output=True, text=True, - encoding="utf-8", errors="replace", timeout=timeout) - return proc.returncode, (proc.stdout or ""), (proc.stderr or "") - - -def _load_state() -> Dict[str, Any]: - try: - with open(_state_path(), encoding="utf-8") as fh: - return json.load(fh) - except (OSError, ValueError): - return {"actions": []} - - -def _save_state(state: Dict[str, Any]) -> None: - os.makedirs(state_dir(), exist_ok=True) - with open(_state_path(), "w", encoding="utf-8", newline="\n") as fh: - json.dump(state, fh, indent=2) - fh.write("\n") - - -def _journal(state: Dict[str, Any], action: Dict[str, Any]) -> None: - action["at"] = datetime.datetime.now().replace(microsecond=0).isoformat() - state.setdefault("actions", []).append(action) - _save_state(state) - - -# --- step 1: DNS-Client operational logging --------------------------------- - - -def dns_log_state() -> Dict[str, Any]: - """Read the log's current config with wevtutil, so --undo can restore it.""" - rc, out, err = _run(["wevtutil", "gl", DNS_LOG]) - if rc != 0: - return {"present": False, "error": (err or out).strip()} - cfg: Dict[str, Any] = {"present": True, "enabled": None, "maxsize": None} - for line in out.splitlines(): - line = line.strip() - if line.startswith("enabled:"): - cfg["enabled"] = line.split(":", 1)[1].strip().lower() == "true" - elif line.startswith("maxSize:"): - try: - cfg["maxsize"] = int(line.split(":", 1)[1].strip()) - except ValueError: - pass - return cfg - - -def enable_dns_log(state: Dict[str, Any], apply: bool) -> List[str]: - msgs: List[str] = [] - before = dns_log_state() - if not before.get("present"): - msgs.append(" [dns-log] log not present on this system: {}" - .format(before.get("error", "unknown"))) - return msgs - - msgs.append(" [dns-log] WILL CHANGE: enable {}".format(DNS_LOG)) - msgs.append(" current: enabled={} maxSize={}" - .format(before.get("enabled"), before.get("maxsize"))) - msgs.append(" after : enabled=True maxSize={}".format(DNS_LOG_SIZE_BYTES)) - msgs.append(" needs administrator: YES") - msgs.append(" undo : restores the two values printed above") - if not apply: - return msgs - - if not is_admin(): - msgs.append(" [dns-log] SKIPPED — needs an elevated shell. Re-run the same " - "command from an Administrator prompt.") - return msgs - - rc, out, err = _run(["wevtutil", "sl", DNS_LOG, "/e:true", - "/ms:{}".format(DNS_LOG_SIZE_BYTES)]) - if rc != 0: - msgs.append(" [dns-log] FAILED: {}".format((err or out).strip())) - return msgs - _journal(state, {"action": "dns_log", "prior": before}) - after = dns_log_state() - msgs.append(" [dns-log] DONE — enabled={} maxSize={}" - .format(after.get("enabled"), after.get("maxsize"))) - msgs.append(" From now on this machine keeps DNS query history. That " - "is what makes a future check possible at all.") - return msgs - - -def undo_dns_log(prior: Dict[str, Any]) -> str: - if not is_admin(): - return " [dns-log] SKIPPED — needs an elevated shell." - enabled = "true" if prior.get("enabled") else "false" - argv = ["wevtutil", "sl", DNS_LOG, "/e:{}".format(enabled)] - if prior.get("maxsize"): - argv.append("/ms:{}".format(prior["maxsize"])) - rc, out, err = _run(argv) - if rc != 0: - return " [dns-log] undo FAILED: {}".format((err or out).strip()) - return " [dns-log] restored enabled={} maxSize={}".format( - enabled, prior.get("maxsize")) - - -# --- step 2: persistence baseline ------------------------------------------- -# -# A list of persistence entries is noise (D4: the reference machine had 37 -# per-user COM registrations, every one legitimate). A DIFF against a known-good -# baseline is signal. This records the inventory now so a later run can say -# "these three things appeared since", which is a question worth answering. - - -def _sha256_file(path: str) -> Optional[str]: - try: - h = hashlib.sha256() - with open(path, "rb") as fh: - while True: - chunk = fh.read(1024 * 1024) - if not chunk: - break - h.update(chunk) - return h.hexdigest() - except OSError: - return None - - -def _iter_com_registrations() -> List[Dict[str, Any]]: - """Per-user COM servers — the documented Candiru persistence technique.""" - if platform.system() != "Windows": - return [] - import winreg - - found: List[Dict[str, Any]] = [] - for sub in (r"Software\Classes\CLSID", r"Software\Classes\WOW6432Node\CLSID"): - try: - root = winreg.OpenKey(winreg.HKEY_CURRENT_USER, sub) - except OSError: - continue - with root: - i = 0 - while True: - try: - clsid = winreg.EnumKey(root, i) - except OSError: - break - i += 1 - for kind in ("InprocServer32", "LocalServer32", "InprocHandler32"): - try: - with winreg.OpenKey(root, clsid + "\\" + kind) as k: - val, _ = winreg.QueryValueEx(k, "") - except OSError: - continue - if not val: - continue - target = os.path.expandvars(str(val)).strip('"') - found.append({ - "kind": "com", - "id": "{}\\{}[{}]".format(sub, clsid, kind), - "target": target, - "sha256": _sha256_file(target) if os.path.isfile(target) else None, - }) - return found - - -def _iter_run_keys() -> List[Dict[str, Any]]: - if platform.system() != "Windows": - return [] - import winreg - - hives = [ - (winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Run"), - (winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\RunOnce"), - (winreg.HKEY_LOCAL_MACHINE, r"Software\Microsoft\Windows\CurrentVersion\Run"), - (winreg.HKEY_LOCAL_MACHINE, r"Software\Microsoft\Windows\CurrentVersion\RunOnce"), - ] - out: List[Dict[str, Any]] = [] - for hive, sub in hives: - try: - key = winreg.OpenKey(hive, sub) - except OSError: - continue - with key: - i = 0 - while True: - try: - name, val, _ = winreg.EnumValue(key, i) - except OSError: - break - i += 1 - out.append({ - "kind": "run", - "id": "{}\\{}::{}".format( - "HKCU" if hive == winreg.HKEY_CURRENT_USER else "HKLM", - sub, name), - "target": str(val), - "sha256": None, - }) - return out - - -def _iter_scheduled_tasks() -> List[Dict[str, Any]]: - if platform.system() != "Windows": - return [] - rc, out, _err = _run(["schtasks", "/query", "/fo", "csv", "/nh"], timeout=180) - if rc != 0: - return [] - tasks: List[Dict[str, Any]] = [] - for line in out.splitlines(): - line = line.strip() - if not line: - continue - parts = [p.strip('"') for p in line.split('","')] - if not parts: - continue - name = parts[0].strip('"') - if not name or name.lower() == "taskname": - continue - tasks.append({"kind": "task", "id": name, "target": None, "sha256": None}) - return tasks - - -def collect_inventory() -> Dict[str, Any]: - entries = _iter_com_registrations() + _iter_run_keys() + _iter_scheduled_tasks() - entries.sort(key=lambda e: (e["kind"], e["id"])) - return { - "taken_at": datetime.datetime.now().replace(microsecond=0).isoformat(), - "host": platform.node(), - "counts": { - "com": sum(1 for e in entries if e["kind"] == "com"), - "run": sum(1 for e in entries if e["kind"] == "run"), - "task": sum(1 for e in entries if e["kind"] == "task"), - }, - "entries": entries, - } - - -def write_baseline(state: Dict[str, Any], apply: bool) -> List[str]: - msgs: List[str] = [] - path = _baseline_path() - existed = os.path.exists(path) - msgs.append(" [baseline] WILL CHANGE: write a persistence inventory to") - msgs.append(" {}".format(path)) - msgs.append(" (this writes only inside the tool's own state dir — " - "nothing on the system is modified)") - msgs.append(" needs administrator: NO") - msgs.append(" undo : {}".format( - "restores the previous baseline" if existed else "deletes that file")) - if not apply: - return msgs - - prior = None - if existed: - try: - with open(path, encoding="utf-8") as fh: - prior = fh.read() - except OSError: - prior = None - - inv = collect_inventory() - os.makedirs(state_dir(), exist_ok=True) - with open(path, "w", encoding="utf-8", newline="\n") as fh: - json.dump(inv, fh, indent=2) - fh.write("\n") - _journal(state, {"action": "baseline", "path": path, "prior_contents": prior}) - msgs.append(" [baseline] DONE — {} COM, {} Run, {} scheduled tasks recorded" - .format(inv["counts"]["com"], inv["counts"]["run"], - inv["counts"]["task"])) - msgs.append(" A later `--arm --baseline-diff` now reports what " - "APPEARED since, instead of dumping a list you cannot judge.") - return msgs - - -def baseline_diff() -> List[str]: - path = _baseline_path() - try: - with open(path, encoding="utf-8") as fh: - old = json.load(fh) - except (OSError, ValueError): - return [" [baseline-diff] no baseline recorded yet — run `--arm --baseline` " - "first. Without one there is nothing to compare against, and this " - "tool does not guess."] - new = collect_inventory() - old_map = {e["id"]: e for e in old.get("entries", [])} - new_map = {e["id"]: e for e in new.get("entries", [])} - - added = [i for i in new_map if i not in old_map] - removed = [i for i in old_map if i not in new_map] - changed = [i for i in new_map - if i in old_map and new_map[i].get("sha256") != old_map[i].get("sha256")] - - msgs = [" [baseline-diff] baseline taken {}".format(old.get("taken_at", "?"))] - if not (added or removed or changed): - msgs.append(" [baseline-diff] no change in the persistence set since then.") - return msgs - for i in sorted(added): - msgs.append(" !! APPEARED : {} -> {}".format(i, new_map[i].get("target"))) - for i in sorted(changed): - msgs.append(" !! REPLACED : {} (binary hash changed)".format(i)) - for i in sorted(removed): - msgs.append(" gone : {}".format(i)) - msgs.append(" Appeared/replaced entries are worth READING, not panicking about: " - "software updates legitimately change these.") - return msgs - - -# --- step 3: periodic re-scan ---------------------------------------------- - - -def register_rescan(state: Dict[str, Any], apply: bool) -> List[str]: - msgs: List[str] = [] - launcher = os.path.join(os.path.dirname(os.path.dirname( - os.path.abspath(__file__))), "adint-scan.py") - cmd = '"{}" -X utf8 "{}"'.format(sys.executable, launcher) - msgs.append(" [rescan] WILL CHANGE: create a weekly Scheduled Task named " - "'{}'".format(RESCAN_TASK_NAME)) - msgs.append(" running: {}".format(cmd)) - msgs.append(" needs administrator: NO (per-user task)") - msgs.append(" undo : deletes that scheduled task") - if not apply: - return msgs - - rc, out, err = _run(["schtasks", "/create", "/tn", RESCAN_TASK_NAME, - "/tr", cmd, "/sc", "weekly", "/f"]) - if rc != 0: - msgs.append(" [rescan] FAILED: {}".format((err or out).strip())) - return msgs - _journal(state, {"action": "rescan_task", "name": RESCAN_TASK_NAME}) - msgs.append(" [rescan] DONE — re-runs weekly against whatever indicator set is " - "on disk at that time.") - return msgs - - -def undo_rescan(name: str) -> str: - rc, out, err = _run(["schtasks", "/delete", "/tn", name, "/f"]) - if rc != 0: - return " [rescan] undo FAILED: {}".format((err or out).strip()) - return " [rescan] scheduled task '{}' deleted".format(name) - - -# --- step 4: Sysmon, detected but deliberately NOT installed ---------------- - - -def sysmon_status() -> List[str]: - msgs: List[str] = [] - rc, out, _err = _run(["sc", "query", "Sysmon64"]) - if rc != 0: - rc, out, _err = _run(["sc", "query", "Sysmon"]) - if rc == 0 and "RUNNING" in out.upper(): - msgs.append(" [sysmon] already installed and running — good. Its process and " - "network events are the best retrospective record you can have.") - else: - msgs.append(" [sysmon] not installed. This tool deliberately does NOT download " - "it: fetching a binary at arm time is the supply-chain exposure") - msgs.append(" DESIGN.md warns about, and we will not add a network " - "fetch to a tool used by targeted people.") - msgs.append(" To do it yourself: get Sysmon from Microsoft " - "Sysinternals and a vetted config (e.g. SwiftOnSecurity's or") - msgs.append(" olafhartong's), then `Sysmon64.exe -accepteula -i " - "sysmonconfig.xml` from an elevated prompt.") - return msgs - - -# --- dispatch --------------------------------------------------------------- - - -def _print(lines: List[str]) -> None: - for line in lines: - print(line) - - -def dispatch(args) -> int: - if platform.system() != "Windows": - print("--arm is Windows-only for now (the confirmed sample is a Windows DLL " - "and the documented persistence is a Windows COM hijack).", - file=sys.stderr) - return 1 - - if getattr(args, "undo", False): - return _do_undo() - - if getattr(args, "baseline_diff", False): - _print(baseline_diff()) - return 0 - - want_dns = getattr(args, "enable_dns_log", False) or getattr(args, "all", False) - want_base = getattr(args, "baseline", False) or getattr(args, "all", False) - want_rescan = getattr(args, "rescan", False) or getattr(args, "all", False) - apply = want_dns or want_base or want_rescan - - print("=" * 72) - print(" adint-scan --arm {}".format( - "APPLYING the steps selected below" if apply else "DRY RUN — nothing will change")) - print("=" * 72) - print("Administrator: {}".format("yes" if is_admin() else "no")) - print("State/undo journal: {}".format(_state_path())) - print("") - - state = _load_state() - - print("Step 1 — DNS query logging (the one that matters most)") - _print(enable_dns_log(state, apply and want_dns)) - print("") - print("Step 2 — persistence baseline, so later runs can show a diff") - _print(write_baseline(state, apply and want_base)) - print("") - print("Step 3 — periodic re-scan") - _print(register_rescan(state, apply and want_rescan)) - print("") - print("Step 4 — Sysmon (informational only)") - _print(sysmon_status()) - print("") - - if not apply: - print("-" * 72) - print("Nothing was changed. Opt into steps explicitly, e.g.:") - print(" adint-scan --arm --baseline (no admin needed)") - print(" adint-scan --arm --enable-dns-log (needs admin)") - print(" adint-scan --arm --all (all of the above)") - print("Reverse everything later with: adint-scan --undo") - else: - print("-" * 72) - print("Applied. Undo everything with: adint-scan --undo") - print("Enabling DNS logging does not protect you — it only means that next " - "time,\nthere is something to look at. That is the whole point.") - print("=" * 72) - return 0 - - -def _do_undo() -> int: - state = _load_state() - actions = state.get("actions", []) - if not actions: - print("Nothing to undo — no recorded --arm actions in {}" - .format(_state_path())) - return 0 - - print("Reversing {} recorded action(s), most recent first:".format(len(actions))) - remaining: List[Dict[str, Any]] = [] - for action in reversed(actions): - kind = action.get("action") - if kind == "dns_log": - print(undo_dns_log(action.get("prior", {}))) - elif kind == "baseline": - path = action.get("path", _baseline_path()) - prior = action.get("prior_contents") - try: - if prior is None: - if os.path.exists(path): - os.remove(path) - print(" [baseline] {} deleted".format(path)) - else: - with open(path, "w", encoding="utf-8", newline="\n") as fh: - fh.write(prior) - print(" [baseline] previous baseline restored") - except OSError as exc: - print(" [baseline] undo FAILED: {}".format(exc)) - remaining.append(action) - elif kind == "rescan_task": - print(undo_rescan(action.get("name", RESCAN_TASK_NAME))) - else: - print(" (unknown action {!r}, left alone)".format(kind)) - remaining.append(action) - - state["actions"] = list(reversed(remaining)) - _save_state(state) - if remaining: - print("\n{} action(s) could not be reversed and remain journalled — most " - "likely they need an elevated shell.".format(len(remaining))) - return 1 - print("\nAll recorded changes reversed.") - return 0 diff --git a/scan/adint_scan/cli.py b/scan/adint_scan/cli.py index 1445427..3d33bed 100644 --- a/scan/adint_scan/cli.py +++ b/scan/adint_scan/cli.py @@ -5,14 +5,14 @@ Default behaviour (no flags): read-only tier-1 IOC match across the browser, hash and host modules, under the no-network guard. Requires no administrator (D7). -The system-writing tier-5 feature (--arm / --undo) is HELD BACK from this release. -Its code is in arm.py but is not wired to the CLI: a review found its --undo path -could, from a journal file any program can write, delete every scheduled task on -the machine and still report success, and could shrink the DNS log and destroy the -history it had collected. None of that is acceptable in a tool at-risk people run, -and the one thing it was for - turning on DNS history - is a single manual command -the report already prints. It returns once the undo path has been rewritten with -the write-ahead journal and per-step safety the review specifies. +There is no system-writing mode. An earlier tier-5 feature (--arm / --undo) was +removed: a review found its --undo path could, from a journal file any program can +write, delete every scheduled task on the machine and still report success, and +could shrink the DNS log and destroy the history it had collected. None of that is +acceptable in a tool at-risk people run, and the one thing it was for - turning on +DNS history - is a single manual command the report already prints. The code is +gone rather than merely unwired, so a reviewer reading this read-only tool finds +no write path at all. Exit codes: 0 = no indicators found (CLEAN or with gaps); 2 = indicators matched; 1 = the scan could not be set up (bad IOC file, etc.). @@ -30,6 +30,7 @@ from . import __version__ from .context import ScanContext from .iocs import IocError, load +from .localpaths import path_problem from .netguard import no_network from .report import V_HITS, build_document, render_human, write_json from .result import ModuleResult @@ -155,8 +156,7 @@ def _run_modules(ctx: ScanContext) -> List[ModuleResult]: r.could_not_run( "persistence-sweep", "tier 2 (--deep) is not implemented yet — no full " - "persistence sweep was performed. The --arm commands are " - "also disabled in this release. A full persistence review " + "persistence sweep was performed. A full persistence review " "requires a separate forensic examination") results.append(r) else: @@ -170,10 +170,37 @@ def _run_modules(ctx: ScanContext) -> List[ModuleResult]: return results +def _refuse_remote(label: str, value: str) -> bool: + """Refuse a user-typed path that is not on a local disk. + + --iocs and --json are paths people type, often copied from an instruction + somebody else wrote ("run it against our updated list at ..."). Opening or + creating a UNC path or a mapped network drive makes Windows itself open an + SMB session to that host - sending this account's network credentials and, + for --json, the complete report. The socket guard cannot see that. Every + other path the scanner opens is already checked this way; these two were + the gap. abspath first: path_problem refuses relative paths outright, and + "--json report.json" is the documented usage.""" + problem = path_problem(os.path.abspath(value)) + if problem: + print("error: {} {}: {} - use a path on a local disk".format( + label, value, problem), file=sys.stderr) + return True + return False + + def main(argv: List[str] = None) -> int: _force_utf8_io() args = _parse(sys.argv[1:] if argv is None else argv) + # Checked before anything is opened, and before a 15-minute scan rather than + # after it. The bundled default is never attacker-chosen, so only a path the + # user supplied is checked. + if args.iocs != _default_ioc_path() and _refuse_remote("--iocs", args.iocs): + return 1 + if args.json and _refuse_remote("--json", args.json): + return 1 + try: iocs = load(args.iocs) except IocError as exc: diff --git a/scan/adint_scan/iocs.py b/scan/adint_scan/iocs.py index f5e3985..3347121 100644 --- a/scan/adint_scan/iocs.py +++ b/scan/adint_scan/iocs.py @@ -22,6 +22,7 @@ import io import os +import ipaddress import re import datetime from dataclasses import dataclass @@ -377,6 +378,41 @@ class IocError(Exception): pass +# Indicator values drive substring searches over every artifact and are printed +# into reports people forward, but only the file's schema was checked. A hostile +# or corrupted feed could pass: a lone "." normalises to the empty string, which +# matches between every pair of bytes (measured at ~31 MB of RAM per MB scanned); +# a lone ":" parses as an IP that matches every URL; "0.0.0.0" would match every +# BLOCKED lookup; and a value carrying control characters went straight into the +# report. Real indicators are hostnames, IP addresses and hex digests. Anything +# else is refused, loudly, rather than scanned for. +_DOMAIN_LABEL_RE = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$") +_MAX_INDICATORS = 20000 + + +def _value_problem(value, kind): + """Why an indicator value is unsafe to scan for, or an empty string.""" + if kind == TYPE_IP: + try: + addr = ipaddress.ip_address(value) + except ValueError: + return "not a valid IP address" + if addr.is_unspecified or len(value) < 7: + return "would match far more than the intended host" + return "" + if kind == TYPE_DOMAIN: + if not 4 <= len(value) <= 253: + return "not the length of a hostname" + labels = value.split(".") + if len(labels) < 2: + return "a hostname needs at least one dot" + for label in labels: + if not 1 <= len(label) <= 63 or not _DOMAIN_LABEL_RE.match(label): + return "label {!r} is not a valid hostname label".format(label[:70]) + return "" + return "" # hashes are already pinned to 64 hex digits by infer_type + + def load(path: str) -> IocSet: if not os.path.exists(path): raise IocError("IOC file not found: {}".format(path)) @@ -427,7 +463,15 @@ def load(path: str) -> IocSet: except UnicodeError: raise IocError("line {}: {!r} is not a valid domain " "name".format(lineno, value)) + problem = _value_problem(value, kind) + if problem: + raise IocError("line {}: {!r}: {}".format( + lineno, value[:80], problem)) seen[value] = Indicator(value, tier, confidence, kind, source) + if len(seen) > _MAX_INDICATORS: + raise IocError("more than {:,} indicators - refusing what looks " + "like a corrupted or hostile feed".format( + _MAX_INDICATORS)) if not seen: # An empty list must never produce a report. A truncated download or an # emptied file would otherwise run a full scan, search for nothing, and diff --git a/scan/adint_scan/modules/browser.py b/scan/adint_scan/modules/browser.py index 36f94e0..935e820 100644 --- a/scan/adint_scan/modules/browser.py +++ b/scan/adint_scan/modules/browser.py @@ -147,6 +147,11 @@ def _firefox_roots(): return [] +# Largest browser database we will snapshot, siblings included. Real History / +# Cookies files are tens of MB; 1 GiB leaves generous headroom. +_MAX_SNAPSHOT_BYTES = 1024 * 1024 * 1024 + + def copy_locked(path): """Copy a possibly-locked SQLite db (plus -wal/-shm/-journal) to a temp dir. @@ -162,6 +167,21 @@ def copy_locked(path): problem = path_problem(path) if problem: raise PermissionError(13, problem, path) + # Refuse an implausibly large database before copying anything. A profile + # database the attacker can write could otherwise fill the temp drive - an + # unbounded write by a tool that promises it only reads - and then exhaust + # memory when its rows are read. Checked before mkdtemp so a refusal never + # leaves a temp directory behind; the caller records it as a gap. + try: + total = os.path.getsize(path) + sum( + os.path.getsize(path + sfx) for sfx in ("-wal", "-shm", "-journal") + if os.path.exists(path + sfx)) + except OSError: + total = 0 # let the copy below surface the real error + if total > _MAX_SNAPSHOT_BYTES: + raise OSError(27, "database is {:,} bytes, over the {:,}-byte snapshot " + "limit, so it was not copied".format( + total, _MAX_SNAPSHOT_BYTES), path) tmpdir = tempfile.mkdtemp(prefix="bscan_", dir=local_temp_dir()) try: base = os.path.join(tmpdir, os.path.basename(path)) @@ -236,7 +256,11 @@ def _chromium_profile_names(user_data): try: with io.open(path, encoding="utf-8", errors="replace") as fh: data = json.load(fh) - except (OSError, ValueError): + # RecursionError is not a ValueError. A Local State file of 200,000 nested + # brackets - about 400 KB, writable without admin - made json.load raise it + # before a single profile was read, and the whole browser module was lost. + # Malformed is malformed: fall back to folder names, keep scanning. + except (OSError, ValueError, RecursionError, MemoryError): return out profile = data.get("profile", {}) if isinstance(data, dict) else {} if not isinstance(profile, dict): @@ -1027,7 +1051,7 @@ def scan_connection_records(label, prof): "browser received an HSTS header from it over HTTPS" + (" (observed {})".format(when[:10]) if when else ""), EV_CONTACTED, when) - except (OSError, ValueError) as exc: + except (OSError, ValueError, RecursionError, MemoryError) as exc: r.skipped += 1 mark_fail(label, "hsts", _locked_note(ts_path, exc)) @@ -1085,7 +1109,7 @@ def scan_connection_records(label, prof): nps_path, ind, "{}: the browser has a connection record for {}" .format(kind, host), EV_CONTACTED) - except (OSError, ValueError) as exc: + except (OSError, ValueError, RecursionError, MemoryError) as exc: r.skipped += 1 mark_fail(label, "connection-records", _locked_note(nps_path, exc)) diff --git a/scan/adint_scan/modules/host.py b/scan/adint_scan/modules/host.py index f143dd4..7f8e0a1 100644 --- a/scan/adint_scan/modules/host.py +++ b/scan/adint_scan/modules/host.py @@ -34,8 +34,7 @@ address is a finding. Read-only throughout: this module never writes the registry, never enables the -log, and needs no administrator (D7). (A system-writing "arm" helper exists in -arm.py but is deliberately not wired to the CLI in this release.) +log, and needs no administrator (D7). """ from __future__ import annotations @@ -192,11 +191,6 @@ def _ps_json(cmd: str, timeout: int = 60): return [], "unexpected JSON shape from PowerShell: {}".format(type(data)) -def _ps_literal(value: str) -> str: - """Quote a path for a PowerShell single-quoted string literal.""" - return "'{}'".format(value.replace("'", "''")) - - def _str(row, *names) -> str: """Pull the first present, non-null field from a ConvertTo-Json row.""" if not isinstance(row, dict): diff --git a/scan/adint_scan/report.py b/scan/adint_scan/report.py index 14182f6..de55bac 100644 --- a/scan/adint_scan/report.py +++ b/scan/adint_scan/report.py @@ -619,6 +619,29 @@ def _defang_text(text: str, indicator: str) -> str: return out +# Artifact-derived text is attacker-influenced: a browser profile's display name, +# a registry value, a page title. Rendered raw, a newline or a terminal escape +# inside one can forge whole report lines - a fake "VERDICT: no known indicators +# found" printed under a real hit, which survives being pasted into an email to a +# helpline. Control characters, Unicode line separators and bidirectional +# overrides (which can visually reverse a line without any newline) are +# replaced, and newlines collapse to spaces so a field can never begin a line of +# its own. The JSON report keeps the raw evidence. +_UNSAFE_CHARS = re.compile( + "[\x00-\x08\x0b-\x1f\x7f-\x9f\u2028\u2029\u202a-\u202e\u2066-\u2069]") + + +def _safe_text(value, limit=2000): + text = str(value).replace("\r\n", " ").replace("\n", " ").replace("\r", " ") + text = _UNSAFE_CHARS.sub("�", text) + return text if len(text) <= limit else text[:limit] + "…" + + +# Tighter limits for the single-line fields an attacker can fill most freely. +_HIT_FIELD_LIMITS = {"profile": 120, "observed": 64, "location": 500, + "detail": 500} + + def _human_document(document): """Sanitize artifact-derived fields without changing the JSON evidence. @@ -627,13 +650,13 @@ def _human_document(document): """ doc = copy.deepcopy(document) - def scrub(value): + def scrub(value, limit=2000): if isinstance(value, str): - return _defang_text(value, "") + return _defang_text(_safe_text(value, limit), "") if isinstance(value, list): - return [scrub(item) for item in value] + return [scrub(item, limit) for item in value] if isinstance(value, dict): - return {key: scrub(item) for key, item in value.items()} + return {key: scrub(item, limit) for key, item in value.items()} return value for module in doc["modules"].values(): @@ -643,7 +666,7 @@ def scrub(value): for hit in module["hits"]: for key in ("location", "detail", "observed", "profile"): if key in hit: - hit[key] = scrub(hit[key]) + hit[key] = scrub(hit[key], _HIT_FIELD_LIMITS[key]) doc["verdict"]["gaps"] = scrub(doc["verdict"].get("gaps", [])) return doc @@ -769,10 +792,7 @@ def _next_steps(w, hits): "have to complete these steps first.") w("") _advice_paragraph(w, "Do NOT open, visit or look up indicator addresses. Keep " - "the [.] spelling when sharing them. Reports can contain " - "usernames, profile names, file paths and browsing details. " - "Ask a responder how to share evidence securely; do not " - "post the full report publicly.") + "the [.] spelling when sharing them.") def _support_options(w): @@ -904,11 +924,23 @@ def render_summary(document: Dict[str, Any]) -> str: w(" See \"HOW TO TURN ON DNS HISTORY\" further down. It tells you") w(" nothing about the past, but it means a future check can") w(" actually answer the question.") + # Every report, not only one with findings. A clean or reference-only result + # is the one people are most relaxed about posting publicly to say "I'm + # fine" - and it carries the same usernames, profile names and paths. + w("") + for line in _wrap(_PRIVACY_NOTE, 68): + w(" " + line) w(_RULE) w("") return _defang_report("\n".join(out), document) +_PRIVACY_NOTE = ("Before you share this report: it can contain your Windows " + "username, browser profile names, file paths and browsing " + "details. Ask a responder how to share evidence securely; " + "do not post the full report publicly.") + + def _dedup_serious(hits: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Collapse repeats of the same indicator at the same grade, strongest grade first, so one domain found in three cache files is one line, not three.""" diff --git a/scan/tests/test_arm_undo.py b/scan/tests/test_arm_undo.py deleted file mode 100644 index b0b6ceb..0000000 --- a/scan/tests/test_arm_undo.py +++ /dev/null @@ -1,392 +0,0 @@ -"""Tier 5 — verify `--undo` restores every change `--arm` made. - -SPEC-scanner.md, testing requirements: "Verify --undo restores every change ---arm made." Doctrine under test: - - D5 Read-only by default. `--arm` with no step selected is a DRY RUN: it - states exactly what WOULD change, whether administrator is needed, and - how to undo it, and it changes nothing at all. - D3 Every change is journalled with its PRIOR value, so undo restores the - previous state instead of guessing a default. A missing baseline is not - invented; it is reported as "run --arm --baseline first". - -SAFETY — this test suite must not change the real machine. - * `enable_dns_log(..., apply=True)` and `register_rescan(..., apply=True)` are - NEVER called: they need administrator / create a real Scheduled Task. Only - the dry-run paths of those two steps are exercised. - * The one step applied for real is the baseline, and only after - `LOCALAPPDATA` has been redirected to a temp dir, so every file written - lands under that temp dir. `arm.state_dir()` reads the variable at call - time (asserted below), so patching os.environ is sufficient and no module - attribute caches the path. - * The dry-run tests additionally replace `arm._run` with a tripwire that - fails the test if anything shells out, so a dry run provably cannot touch - wevtutil or schtasks. - * The inventory shape check supplies synthetic collector inputs. No test - enumerates the real registry, hashes registered targets, or invokes schtasks. -""" - -from __future__ import annotations - -import contextlib -import io -import json -import os -import shutil -import sys -import tempfile -import unittest -from unittest.mock import patch - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from adint_scan import arm # noqa: E402 - -_SENTINEL = object() - - -def stub_inventory(tag="one"): - return { - "taken_at": "2025-01-01T00:00:0" + ("0" if tag == "one" else "1"), - "host": "test-host", - "counts": {"com": 1, "run": 1, "task": 1}, - "entries": [ - {"kind": "com", "id": "CLSID/{test}[InprocServer32]", - "target": r"C:\fake\benign.dll", "sha256": "ab" * 32}, - {"kind": "run", "id": "HKCU\\...\\Run::thing", - "target": r"C:\fake\thing.exe", "sha256": None}, - {"kind": "task", "id": "\\test\\task", "target": None, - "sha256": None}, - ], - } - - -class ArmSandbox(unittest.TestCase): - """Redirects the tool's state dir into a temp dir for every test.""" - - def setUp(self): - self.tmp = tempfile.mkdtemp(prefix="adint_arm_") - self._saved_local = os.environ.get("LOCALAPPDATA", _SENTINEL) - os.environ["LOCALAPPDATA"] = self.tmp - # Prove the redirection actually took effect before anything is run: - # if state_dir() were resolved at import time this would fail loudly - # instead of writing to the user's real AppData. - self.assertTrue( - os.path.abspath(arm.state_dir()).startswith(os.path.abspath(self.tmp)), - "arm.state_dir() must honour LOCALAPPDATA at call time; refusing to " - "run a test that would write to the real machine") - self.state_path = arm._state_path() - self.baseline_path = arm._baseline_path() - - def tearDown(self): - if self._saved_local is _SENTINEL: - os.environ.pop("LOCALAPPDATA", None) - else: - os.environ["LOCALAPPDATA"] = self._saved_local - shutil.rmtree(self.tmp, ignore_errors=True) - - # --- helpers ----------------------------------------------------------- - - @contextlib.contextmanager - def no_subprocess(self): - """Any shell-out during the block fails the test.""" - def tripwire(*a, **k): - self.fail("a dry run must not shell out; got {!r}".format(a)) - saved_run, saved_ps = arm._run, arm._ps - arm._run, arm._ps = tripwire, tripwire - try: - yield - finally: - arm._run, arm._ps = saved_run, saved_ps - - @contextlib.contextmanager - def stubbed_inventory(self, tag="one"): - saved = arm.collect_inventory - arm.collect_inventory = lambda: stub_inventory(tag) - try: - yield - finally: - arm.collect_inventory = saved - - def undo(self): - """Run _do_undo() with its printing captured; returns (rc, output).""" - buf = io.StringIO() - with contextlib.redirect_stdout(buf): - rc = arm._do_undo() - return rc, buf.getvalue() - - def files_under_state_dir(self): - d = arm.state_dir() - if not os.path.isdir(d): - return [] - return sorted(os.listdir(d)) - - -class TestStateDirRedirection(ArmSandbox): - def test_state_dir_is_read_from_the_environment_at_call_time(self): - self.assertEqual(os.path.dirname(arm.state_dir()), - os.path.abspath(self.tmp)) - self.assertTrue(arm.state_dir().endswith(arm.STATE_DIRNAME)) - other = os.path.join(self.tmp, "second") - os.environ["LOCALAPPDATA"] = other - self.assertEqual(arm.state_dir(), - os.path.join(other, arm.STATE_DIRNAME)) - os.environ["LOCALAPPDATA"] = self.tmp - - def test_sandbox_starts_with_no_state(self): - self.assertEqual(self.files_under_state_dir(), []) - self.assertEqual(arm._load_state(), {"actions": []}) - - -class TestDryRunChangesNothing(ArmSandbox): - """D5: `--arm` without an explicit step must change nothing, and must say - what it would change, whether admin is needed, and how to undo it.""" - - def setUp(self): - super().setUp() - self.state = {"actions": []} - # dns_log_state() only *reads* config via wevtutil, but stubbing it - # keeps the message contract deterministic on any Windows build and - # lets the no-subprocess tripwire stay armed. - self._saved_dns_state = arm.dns_log_state - arm.dns_log_state = lambda: {"present": True, "enabled": False, - "maxsize": 1052672} - - def tearDown(self): - arm.dns_log_state = self._saved_dns_state - super().tearDown() - - def _assert_dry_run_contract(self, msgs, label): - text = "\n".join(msgs).lower() - self.assertIn("will change", text, - "{}: a dry run must state what WOULD change".format(label)) - self.assertIn("administrator", text, - "{}: a dry run must say whether admin is needed" - .format(label)) - self.assertIn("undo", text, - "{}: a dry run must say how to undo it".format(label)) - - def _assert_nothing_happened(self): - self.assertEqual(self.state.get("actions", []), [], - "a dry run must journal no action") - self.assertFalse(os.path.exists(self.state_path), - "a dry run must not write the state/journal file") - self.assertFalse(os.path.exists(self.baseline_path), - "a dry run must not write a baseline") - self.assertEqual(self.files_under_state_dir(), [], - "a dry run must create no files at all") - - def test_dns_log_dry_run(self): - with self.no_subprocess(): - msgs = arm.enable_dns_log(self.state, apply=False) - self._assert_dry_run_contract(msgs, "dns-log") - self.assertIn("yes", "\n".join(msgs).lower().split( - "needs administrator:")[1][:8], - "dns-log must say administrator is required") - self.assertIn(arm.DNS_LOG, "\n".join(msgs)) - self._assert_nothing_happened() - - def test_baseline_dry_run(self): - with self.no_subprocess(): - msgs = arm.write_baseline(self.state, apply=False) - self._assert_dry_run_contract(msgs, "baseline") - joined = "\n".join(msgs) - self.assertIn(self.baseline_path, joined, - "the dry run must name the exact file it would write") - self.assertIn("needs administrator: NO", joined) - self._assert_nothing_happened() - - def test_rescan_dry_run(self): - with self.no_subprocess(): - msgs = arm.register_rescan(self.state, apply=False) - self._assert_dry_run_contract(msgs, "rescan") - joined = "\n".join(msgs) - self.assertIn(arm.RESCAN_TASK_NAME, joined, - "the dry run must name the task it would create") - self._assert_nothing_happened() - - def test_all_three_dry_runs_together_still_change_nothing(self): - with self.no_subprocess(): - arm.enable_dns_log(self.state, apply=False) - arm.write_baseline(self.state, apply=False) - arm.register_rescan(self.state, apply=False) - self._assert_nothing_happened() - - -class TestBaselineRoundTrip(ArmSandbox): - def test_apply_then_undo_removes_the_file_and_empties_the_journal(self): - state = {"actions": []} - with self.stubbed_inventory(): - msgs = arm.write_baseline(state, apply=True) - self.assertIn("DONE", "\n".join(msgs)) - self.assertTrue(os.path.exists(self.baseline_path)) - - # Exactly one journalled action, carrying the prior value (none). - self.assertEqual(len(state["actions"]), 1) - action = state["actions"][0] - self.assertEqual(action["action"], "baseline") - self.assertIsNone(action["prior_contents"], - "there was no previous baseline, so the prior value " - "must be recorded as None") - self.assertEqual(action["path"], self.baseline_path) - self.assertIn("at", action, "every action must be timestamped") - self.assertEqual(arm._load_state()["actions"], state["actions"], - "the journal must be persisted, or undo cannot work") - - # baseline_diff against the same inventory reports no change. - with self.stubbed_inventory(): - diff = "\n".join(arm.baseline_diff()) - self.assertIn("no change in the persistence set", diff) - self.assertNotIn("APPEARED", diff) - - rc, out = self.undo() - self.assertEqual(rc, 0) - self.assertIn("deleted", out) - self.assertFalse(os.path.exists(self.baseline_path), - "--undo must remove the baseline it created") - self.assertEqual(arm._load_state()["actions"], [], - "a reversed action must leave the journal empty") - - def test_baseline_diff_reports_what_appeared(self): - state = {"actions": []} - with self.stubbed_inventory("one"): - arm.write_baseline(state, apply=True) - - def grown(): - inv = stub_inventory("two") - inv["entries"].append({"kind": "com", "id": "CLSID/{new}[Inproc]", - "target": r"C:\fake\new.dll", - "sha256": "cd" * 32}) - return inv - - saved = arm.collect_inventory - arm.collect_inventory = grown - try: - diff = "\n".join(arm.baseline_diff()) - finally: - arm.collect_inventory = saved - self.assertIn("APPEARED", diff) - self.assertIn("CLSID/{new}[Inproc]", diff) - - def test_undo_restores_the_previous_baseline_rather_than_deleting_it(self): - # Two armed sessions. Each caller passes its own freshly-loaded journal; - # here the second session's journal holds only its own action, which is - # the case that must RESTORE the prior file instead of removing it. - with self.stubbed_inventory("one"): - arm.write_baseline({"actions": []}, apply=True) - with open(self.baseline_path, encoding="utf-8") as fh: - first = fh.read() - - with self.stubbed_inventory("two"): - second_state = {"actions": []} - arm.write_baseline(second_state, apply=True) - with open(self.baseline_path, encoding="utf-8") as fh: - second = fh.read() - self.assertNotEqual(first, second, "the two baselines must differ, or " - "the restore proves nothing") - self.assertEqual(second_state["actions"][0]["prior_contents"], first, - "the journal must record the PRIOR contents") - - rc, out = self.undo() - self.assertEqual(rc, 0) - self.assertIn("previous baseline restored", out) - self.assertTrue(os.path.exists(self.baseline_path), - "undo of an overwrite must restore, not delete") - with open(self.baseline_path, encoding="utf-8") as fh: - self.assertEqual(fh.read(), first, - "undo must restore the FIRST baseline byte for " - "byte, not just delete the file") - self.assertEqual(json.loads(first)["taken_at"], - stub_inventory("one")["taken_at"]) - - def test_undoing_every_journalled_action_returns_to_the_pristine_state(self): - # Both writes journalled in one session: reversing EVERY recorded action - # (newest first) restores the file to the first baseline and then - # removes it, because before --arm ran there was no baseline at all. - state = {"actions": []} - with self.stubbed_inventory("one"): - arm.write_baseline(state, apply=True) - with self.stubbed_inventory("two"): - arm.write_baseline(state, apply=True) - self.assertEqual(len(state["actions"]), 2) - - rc, out = self.undo() - self.assertEqual(rc, 0) - self.assertIn("previous baseline restored", out) - self.assertIn("deleted", out) - self.assertFalse(os.path.exists(self.baseline_path), - "undoing every change must leave the machine as it was " - "before --arm ran: no baseline file") - self.assertEqual(arm._load_state()["actions"], []) - - -class TestUndoEdgeCases(ArmSandbox): - def test_undo_with_an_empty_journal_is_a_no_op(self): - rc, out = self.undo() - self.assertEqual(rc, 0) - self.assertIn("Nothing to undo", out) - self.assertEqual(self.files_under_state_dir(), [], - "a no-op undo must not create state of its own") - - def test_undo_leaves_an_unknown_action_alone_and_reports_it(self): - arm._save_state({"actions": [{"action": "something-new-and-unknown"}]}) - rc, out = self.undo() - self.assertEqual(rc, 1, "an action that could not be reversed must not " - "be reported as success") - self.assertIn("unknown action", out) - self.assertEqual(len(arm._load_state()["actions"]), 1, - "an unreversed action must stay journalled") - - def test_baseline_diff_without_a_baseline_does_not_invent_a_result(self): - def tripwire(): - self.fail("baseline_diff must not collect an inventory when there " - "is no baseline to compare it against") - saved = arm.collect_inventory - arm.collect_inventory = tripwire - try: - msgs = arm.baseline_diff() - finally: - arm.collect_inventory = saved - text = "\n".join(msgs) - self.assertIn("no baseline recorded yet", text) - self.assertIn("--arm --baseline", text, - "the user must be told the exact command to run") - self.assertNotIn("no change", text, - "absence of a baseline must never be reported as 'no " - "change' — that would be an inferred result (D3)") - self.assertNotIn("APPEARED", text) - - -class TestRealInventoryShape(ArmSandbox): - """The real inventory aggregator receives hermetic collector results.""" - - def test_collect_inventory_returns_the_documented_shape(self): - def row(kind): - return [{"kind": kind, "id": kind + "-synthetic", "target": None, - "sha256": None}] - with patch.object(arm, "_iter_com_registrations", return_value=row("com")), \ - patch.object(arm, "_iter_run_keys", return_value=row("run")), \ - patch.object(arm, "_iter_scheduled_tasks", return_value=row("task")), \ - self.no_subprocess(): - inv = arm.collect_inventory() - self.assertEqual(sorted(inv), ["counts", "entries", "host", "taken_at"]) - self.assertIsInstance(inv["taken_at"], str) - self.assertIsInstance(inv["host"], str) - self.assertEqual(sorted(inv["counts"]), ["com", "run", "task"]) - for key in ("com", "run", "task"): - self.assertIsInstance(inv["counts"][key], int) - self.assertIsInstance(inv["entries"], list) - for e in inv["entries"]: - self.assertIn(e["kind"], ("com", "run", "task")) - self.assertIn("id", e) - self.assertIn("target", e) - self.assertIn("sha256", e) - self.assertEqual( - sum(inv["counts"].values()), len(inv["entries"]), - "the counts must describe the entries actually collected") - # Collecting an inventory is read-only: nothing is written anywhere. - self.assertEqual(self.files_under_state_dir(), []) - - -if __name__ == "__main__": - unittest.main(verbosity=2)