Capstone: an IOC sweep tool

[build-system]
requires = ["hatchling==1.32.4"]
build-backend = "hatchling.build"
 
[project]
name = "ioc-sweep"
version = "0.1.0"
description = "Sweep a directory for known-bad files: hashes, threat intel, a scanner, quarantine."
requires-python = ">=3.14"
dependencies = []  # standard library only
 
[project.scripts]
ioc-sweep = "ioc_sweep.cli:main"
 
[tool.hatch.build.targets.sdist]
include = ["/src"]

# ioc-sweep settings. Every key is required and an unknown key is an error.
# Paths are absolute: cron and systemd do not start in this directory.
 
[api]
url = "https://localhost:18160/v1"
ca_file = "/home/deploy/ps-capstone/lab/ca.pem"
token_file = "/home/deploy/ps-capstone/api-token"
timeout = 5           # seconds for the connect, and for each read
batch_size = 3        # hashes per request
max_pages = 5         # per batch; more means the API is broken
attempts = 3          # per request, the first one included
deadline = 15         # seconds for one batch: all its pages and retries
max_retry_after = 5   # a 429 that asks for longer ends the batch
 
[scan]
max_size = 1048576    # bytes; a bigger file is not checked and the run is partial
scanner = "/home/deploy/ps-capstone/lab/fakescan"
scanner_timeout = 10  # seconds per file
 
[output]
summary_csv = "/home/deploy/ps-capstone/out/summary.csv"
quarantine_dir = "/home/deploy/ps-capstone/quarantine"

"""Load sweep.toml and check every value once, so the rest of the tool can trust them."""
import tomllib
from pathlib import Path
 
# Every table, every key in it, and the type its value must have. Nothing else is accepted.
SCHEMA = {
    "api": {"url": str, "ca_file": Path, "token_file": Path, "timeout": float,
            "batch_size": int, "max_pages": int, "attempts": int, "deadline": float,
            "max_retry_after": float},
    "scan": {"max_size": int, "scanner": Path, "scanner_timeout": float},
    "output": {"summary_csv": Path, "quarantine_dir": Path},
}
 
 
class ConfigError(Exception):
    """sweep.toml is missing, is not TOML, or has a wrong, missing or unknown setting."""
 
 
def load_config(path: Path) -> dict:
    """Return {key: value} for every key in SCHEMA, or raise ConfigError with one line."""
    try:
        with open(path, "rb") as f:  # tomllib reads bytes and decodes UTF-8 itself
            data = tomllib.load(f)
    except OSError as err:
        raise ConfigError(f"{path}: {err.strerror}") from err
    except tomllib.TOMLDecodeError as err:
        raise ConfigError(f"{path}: {err}") from err
    unknown = sorted(set(data) - set(SCHEMA))  # a typo is reported, never ignored
    if unknown:
        raise ConfigError(f"{path}: unknown table [{unknown[0]}]")
    settings = {}
    for table, keys in SCHEMA.items():
        values = data.get(table)
        if not isinstance(values, dict):
            raise ConfigError(f"{path}: no [{table}] table")
        unknown = sorted(set(values) - set(keys))
        if unknown:
            raise ConfigError(f"{path}: unknown key {table}.{unknown[0]}")
        for key, kind in keys.items():
            settings[key] = check(f"{path}: {table}.{key}", values.get(key), kind)
    if not settings["url"].startswith("https://"):
        raise ConfigError(f"{path}: api.url must start with https://")
    return settings
 
 
def check(name: str, value, kind: type):
    if value is None:
        raise ConfigError(f"{name} is missing")
    if kind is Path:  # a path is written as a string and must be absolute
        if type(value) is not str or not Path(value).is_absolute():
            raise ConfigError(f"{name} must be an absolute path, not {value!r}")
        return Path(value)
    if kind is float and type(value) is int:
        value = float(value)  # timeout = 5 means 5.0
    # type(), not isinstance(): True is an instance of int, and "attempts = true" is a mistake.
    if type(value) is not kind:
        raise ConfigError(f"{name} must be {kind.__name__}, not {type(value).__name__}")
    if kind in (int, float) and value <= 0:
        raise ConfigError(f"{name} must be greater than 0, not {value}")
    return value

"""The API token: read from a file only its owner can read, and kept out of every log line."""
import logging
import os
import stat
from pathlib import Path
 
 
class SecretError(Exception):
    """The token file is missing, empty, or readable by other users."""
 
 
def read_token(path: Path) -> str:
    """The token: systemd's credential api-token if the service has one, else the file at path."""
    cred_dir = os.environ.get("CREDENTIALS_DIRECTORY")  # set by LoadCredential=api-token:...
    systemd = bool(cred_dir) and Path(cred_dir, "api-token").is_file()
    if systemd:
        path = Path(cred_dir, "api-token")
    try:
        with open(path, encoding="utf-8") as f:
            info = os.fstat(f.fileno())  # the file we opened, not whatever the name points to later
            # systemd's copy is root-owned and readable by this service only (an ACL): the owner
            # and mode checks are for a file we manage ourselves
            if not systemd and info.st_uid != os.getuid():
                raise SecretError(f"{path}: owned by uid {info.st_uid}, not by uid {os.getuid()}")
            if not systemd and info.st_mode & 0o077:
                raise SecretError(f"{path}: mode {stat.filemode(info.st_mode)} lets other users read it")
            token = f.read().strip()
    except OSError as err:
        raise SecretError(f"{path}: {err.strerror}") from err
    if not token:
        raise SecretError(f"{path}: empty")
    return token
 
 
class RedactSecrets(logging.Filter):
    """Replace known secret values in every record before the handler writes it."""
 
    def __init__(self, secrets: list[str]):
        super().__init__()
        self.secrets = [s for s in secrets if s]
 
    def scrub(self, text: str) -> str:
        for secret in self.secrets:
            text = text.replace(secret, "[REDACTED]")
        return text
 
    def filter(self, record: logging.LogRecord) -> bool:
        record.msg = self.scrub(record.getMessage())
        record.args = None
        if record.exc_info:  # a traceback can hold the secret too
            record.exc_text = self.scrub(logging.Formatter().formatException(record.exc_info))
        return True

"""List the regular files under a directory with their SHA-256, never following a symbolic link."""
import hashlib
import logging
import stat
from pathlib import Path
 
log = logging.getLogger(__name__)
 
 
def inventory(root: Path, max_size: int) -> list[dict]:
    """One record per regular file: its path and SHA-256, or the problems that stopped the hash."""
    records = []
 
    def cannot_list(err: OSError) -> None:  # walk() calls this for a directory it cannot read
        records.append({"path": Path(err.filename), "sha256": "", "problems": [err.strerror]})
 
    for dirpath, dirnames, filenames in root.walk(on_error=cannot_list):
        dirnames.sort()  # walk() visits the subdirectories in this list's order
        for name in sorted(filenames):
            path = dirpath / name
            record = {"path": path, "sha256": "", "problems": []}
            try:
                info = path.lstat()  # the entry itself: a link is seen as a link
                if not stat.S_ISREG(info.st_mode):
                    log.info("skipped %s: not a regular file", path)
                    continue
                if info.st_size > max_size:
                    record["problems"].append(f"{info.st_size} bytes, over scan.max_size")
                else:
                    with open(path, "rb") as f:
                        record["sha256"] = hashlib.file_digest(f, "sha256").hexdigest()
            except OSError as err:  # removed or made unreadable while we walked
                record["problems"].append(err.strerror)
            records.append(record)
    return records

"""Fixtures: a lab CA, the mock API on a free port, a token file, a small tree and a config."""
import subprocess
import threading
from pathlib import Path
 
import mock_api  # lab/mock_api.py, on sys.path through pytest.toml
import pytest
 
LAB = Path(__file__).parent.parent / "lab"
TOKEN = "lab-token-not-secret"
SAMPLE_A = b"ioc-sweep lab sample A: stands in for a malicious file\n"  # in lab/known-bad.txt
KNOWN_BAD = sorted(line.split()[0] for line in (LAB / "known-bad.txt").read_text().splitlines())
TEMPLATE = Path(__file__).parent / "sweep.template.toml"  # sweep.toml with {placeholders}
 
 
@pytest.fixture(scope="session")
def certs(tmp_path_factory):
    """A throwaway CA and server certificate, made once per test run with the lab's own script."""
    directory = tmp_path_factory.mktemp("ca")
    subprocess.run(["bash", LAB / "make_lab_ca.sh"], cwd=directory, check=True,
                   capture_output=True, timeout=30)
    return directory
 
 
@pytest.fixture
def api(certs):
    """The mock API on 127.0.0.1, on a port the OS picks. A test sets api.mode to make it misbehave."""
    server = mock_api.make_server(certs, TOKEN)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    server.url = f"https://localhost:{server.server_address[1]}/v1"
    yield server
    server.shutdown()
    server.server_close()
 
 
@pytest.fixture
def tree(tmp_path):
    """A known-bad sample, two ordinary files, and a symbolic link that points out of the tree."""
    root = tmp_path / "tree"
    (root / "docs").mkdir(parents=True)
    (root / "invoice.pdf").write_bytes(SAMPLE_A)  # SHA-256 1f72...
    (root / "docs" / "notes.txt").write_text("meeting notes\n")  # 2f96...
    (root / "docs" / "backup.log").write_text("backup done\n")  # 83b7..., the 8-f shard
    (root / "passwd").symlink_to("/etc/passwd")
    return root
 
 
@pytest.fixture
def make_config(tmp_path, certs, api):
    """Return a function that writes sweep.toml for this test's API and files; keywords change values."""
    token_file = tmp_path / "api-token"
    token_file.write_text(TOKEN + "\n")
    token_file.chmod(0o600)
 
    def make(**changes) -> Path:
        values = {"url": api.url, "certs": certs, "tmp": tmp_path, "deadline": 5,
                  "max_retry_after": 2, "scanner": LAB / "fakescan"}
        values.update(changes)
        path = tmp_path / "sweep.toml"
        path.write_text(TEMPLATE.read_text().format(**values))
        return path
 
    return make

[api]
url = "{url}"
ca_file = "{certs}/ca.pem"
token_file = "{tmp}/api-token"
timeout = 1
batch_size = 2
max_pages = 3
attempts = 3
deadline = {deadline}
max_retry_after = {max_retry_after}
 
[scan]
max_size = 1000
scanner = "{scanner}"
scanner_timeout = 5
 
[output]
summary_csv = "{tmp}/summary.csv"
quarantine_dir = "{tmp}/quarantine"

[pytest]
testpaths = ["tests"]
pythonpath = ["lab"]  # the mock API is a lab script, not part of the package

"""load_config() and read_token(): what they accept, and the one line they say about the rest."""
import pytest
 
from ioc_sweep.config import ConfigError, load_config
from ioc_sweep.secret import SecretError, read_token
 
 
def test_values_are_typed(make_config):
    settings = load_config(make_config())
    assert settings["timeout"] == 1.0 and type(settings["timeout"]) is float
    assert settings["batch_size"] == 2
    assert settings["scanner"].is_absolute()
 
 
@pytest.mark.parametrize(
    "old, new, message",
    [
        ("batch_size = 2", "batch_sise = 2", "unknown key api.batch_sise"),
        ("max_pages = 3", 'max_pages = "3"', "api.max_pages must be int, not str"),
        ("attempts = 3", "attempts = true", "api.attempts must be int, not bool"),
        ("attempts = 3", "attempts = 0", "api.attempts must be greater than 0"),
        ('url = "https', 'url = "http', "api.url must start with https://"),
        ('summary_csv = "/', 'summary_csv = "', "output.summary_csv must be an absolute path"),
        ("[output]", "[extra]\nx = 1\n[output]", "unknown table [extra]"),
        ("timeout = 1", "timeout = 1s", "(at line 5, column 12)"),
    ],
    ids=["typo", "string", "bool", "zero", "http", "relative", "table", "not-toml"],
)
def test_bad_setting_is_one_clear_error(make_config, old, new, message):
    path = make_config()
    path.write_text(path.read_text().replace(old, new, 1))
    with pytest.raises(ConfigError) as err:
        load_config(path)
    assert message in str(err.value)
 
 
def test_token_file_must_be_private(tmp_path):
    token_file = tmp_path / "api-token"
    token_file.write_text("lab-token-not-secret\n")
    token_file.chmod(0o644)
    with pytest.raises(SecretError, match="-rw-r--r-- lets other users read it"):
        read_token(token_file)
    token_file.chmod(0o600)
    assert read_token(token_file) == "lab-token-not-secret"
 
 
def test_systemd_credential_comes_first(tmp_path, monkeypatch):
    creds = tmp_path / "credentials"
    creds.mkdir()
    (creds / "api-token").write_text("lab-token-from-systemd\n")
    (creds / "api-token").chmod(0o440)  # group-readable, as systemd's copy is through its ACL
    monkeypatch.setenv("CREDENTIALS_DIRECTORY", str(creds))
    assert read_token(tmp_path / "no-such-file") == "lab-token-from-systemd"

"""inventory(): regular files only, links skipped, a size cap, and unreadable places recorded."""
import hashlib
 
from conftest import SAMPLE_A
from ioc_sweep.walk import inventory
 
 
def test_regular_files_hashed_and_link_skipped(tree):
    records = inventory(tree, max_size=1000)
    assert [r["path"].relative_to(tree).as_posix() for r in records] == [
        "invoice.pdf", "docs/backup.log", "docs/notes.txt"]
    assert records[0]["sha256"] == hashlib.sha256(SAMPLE_A).hexdigest()
    assert all(not r["problems"] for r in records)
 
 
def test_problems_are_recorded_not_skipped(tree):
    (tree / "big.iso").write_bytes(b"x" * 2000)
    (tree / "locked").mkdir()
    (tree / "locked").chmod(0o000)  # a directory walk() cannot list
    try:
        problems = {r["path"].name: r["problems"] for r in inventory(tree, max_size=1000) if r["problems"]}
    finally:
        (tree / "locked").chmod(0o755)  # so pytest can delete it afterwards
    assert problems == {"big.iso": ["2000 bytes, over scan.max_size"], "locked": ["Permission denied"]}

"""One GET request to the threat-intel API: verified TLS, the token, time limits, checked JSON."""
import http.client
import json
import ssl
import time
import urllib.error
import urllib.request
 
MAX_BODY = 1_000_000  # bytes; a larger answer is a bug or an attack, not data
 
 
class ApiError(Exception):
    """This request will not work however often it is sent: a wrong status or a wrong answer."""
 
 
class SetupError(ApiError):
    """Every request would fail the same way: a wrong api.url, a certificate that does not verify."""
 
 
class AuthError(SetupError):
    """401 or 403: the token is wrong or not allowed. Retrying cannot help and may lock it."""
 
 
class Unavailable(ApiError):
    """A timeout, a refused or dropped connection, 5xx or 429: the same request may work later."""
 
    def __init__(self, message: str, retry_after: float | None = None):
        super().__init__(message)
        self.retry_after = retry_after  # seconds the server asked us to wait, if it said
 
 
def get_json(url: str, token: str, context: ssl.SSLContext, timeout: float, deadline: float) -> dict:
    """timeout limits each step; deadline (a time.monotonic() value) also limits reading the body."""
    request = urllib.request.Request(url, headers={"Accept": "application/json"})
    request.add_unredirected_header("Authorization", f"Bearer {token}")  # not resent on a redirect
    try:
        with urllib.request.urlopen(request, timeout=timeout, context=context) as response:
            content_type = response.headers.get_content_type()
            if content_type != "application/json":
                raise ApiError(f"expected JSON, got {content_type}")
            body = read_body(response, deadline)
    except urllib.error.HTTPError as err:  # the server answered with 4xx or 5xx
        message = f"HTTP {err.code} {err.reason}"
        if err.code in (401, 403):
            raise AuthError(message) from err
        if err.code == 404:  # the path in api.url is wrong, for every request alike
            raise SetupError(message) from err
        if err.code == 429 or err.code >= 500:
            raise Unavailable(message, retry_after_seconds(err.headers.get("Retry-After"))) from err
        raise ApiError(message) from err
    except (OSError, http.client.HTTPException) as err:  # no usable answer (see the HTTP lesson)
        reason = getattr(err, "reason", err)
        if isinstance(reason, ssl.SSLCertVerificationError):  # wrong name, expired, unknown CA
            raise SetupError(f"TLS: {reason}") from err
        raise Unavailable(str(reason)) from err
    try:
        data = json.loads(body)
    except ValueError as err:  # not JSON, or not UTF-8
        raise ApiError(f"invalid JSON: {err}") from err
    if not isinstance(data, dict):
        raise ApiError(f"expected a JSON object, got {type(data).__name__}")
    return data
 
 
def read_body(response, deadline: float) -> bytes:
    """Chunk by chunk, so that a server sending one byte at a time cannot outlast the deadline."""
    body = b""
    while chunk := response.read1(65536):
        body += chunk
        if len(body) > MAX_BODY:
            raise ApiError(f"answer larger than {MAX_BODY} bytes")
        if time.monotonic() > deadline:
            raise Unavailable("answer still incomplete at api.deadline")
    return body
 
 
def retry_after_seconds(value: str | None) -> float | None:
    """Retry-After in seconds; an HTTP date or anything else counts as 'too long'."""
    if value is None:
        return None
    return float(value) if value.isascii() and value.isdigit() else float("inf")

"""One request, retried while the API is unavailable: bounded, jittered, inside a deadline."""
import logging
import random
import time
 
from .api import Unavailable, get_json
 
log = logging.getLogger(__name__)
 
 
def get_with_retries(url: str, token: str, context, settings: dict, deadline: float) -> dict:
    for attempt in range(1, settings["attempts"] + 1):
        left = deadline - time.monotonic()
        if left <= 0:
            raise Unavailable(f"api.deadline of {settings['deadline']:g}s reached")
        try:
            return get_json(url, token, context, timeout=min(settings["timeout"], left), deadline=deadline)
        except Unavailable as err:
            if attempt == settings["attempts"]:  # the last attempt: give up now, no sleep
                raise Unavailable(f"{err} ({attempt} attempts)") from err
            delay = random.uniform(0, min(4.0, 0.5 * 2 ** (attempt - 1)))  # full jitter
            if err.retry_after is not None:  # a 429 that says how long to wait
                if err.retry_after > settings["max_retry_after"]:
                    raise Unavailable(f"{err}, Retry-After {err.retry_after:g}s is too long") from err
                delay = err.retry_after
            if delay >= deadline - time.monotonic():
                raise Unavailable(f"{err}, no time left before api.deadline") from err
            log.warning("%s; retry %d in %.1fs", err, attempt, delay)
            time.sleep(delay)

"""Look digests up in batches: every page of a batch, inside limits the API cannot break."""
import logging
import time
from urllib.parse import urlencode
 
from .api import ApiError, SetupError, Unavailable
from .retry import get_with_retries
 
log = logging.getLogger(__name__)
VERDICTS = {"malicious", "suspicious"}
 
 
def lookup_all(digests: list[str], settings: dict, token: str, context) -> tuple[dict, set]:
    """Return ({sha256: match}, {digests whose batch failed}). SetupError stops everything."""
    size = settings["batch_size"]
    batches = [digests[i:i + size] for i in range(0, len(digests), size)]
    matches, failed, temporary = {}, set(), False
    for number, batch in enumerate(batches, 1):
        try:
            matches.update(lookup_batch(batch, settings, token, context))
        except SetupError:
            raise  # the same URL, certificate or token would fail every batch: stop sending
        except ApiError as err:  # Unavailable included: this batch gave up, try the next
            log.warning("batch %d of %d failed: %s", number, len(batches), err)
            failed.update(batch)
            temporary = temporary or isinstance(err, Unavailable)
    if digests and len(failed) == len(digests) and not temporary:
        # every batch got a wrong answer, none of them temporary: waiting will not fix that
        raise SetupError(f"all {len(batches)} batches failed, none of them temporary")
    return matches, failed
 
 
def lookup_batch(batch: list[str], settings: dict, token: str, context) -> dict:
    """Every page of one batch; one deadline covers all its pages and retries."""
    deadline = time.monotonic() + settings["deadline"]
    matches, seen, cursor = {}, set(), None
    for page in range(1, settings["max_pages"] + 1):
        query = {"sha256": ",".join(batch)}
        if cursor is not None:
            query["cursor"] = cursor
        url = f"{settings['url']}/hashes?{urlencode(query)}"
        data = get_with_retries(url, token, context, settings, deadline)
        items, cursor = data.get("matches"), data.get("next_cursor")
        if not isinstance(items, list) or not (cursor is None or isinstance(cursor, str)):
            raise ApiError(f"page {page}: not a {{matches, next_cursor}} answer")
        for item in items:
            if not isinstance(item, dict) or item.get("sha256") not in batch \
                    or item.get("verdict") not in VERDICTS or not isinstance(item.get("name"), str):
                raise ApiError(f"page {page}: unexpected match {item!r}")
            matches[item["sha256"]] = item
        if cursor is None:
            return matches
        if cursor in seen:
            raise ApiError(f"page {page} repeats cursor {cursor!r}: the API is looping")
        seen.add(cursor)
    raise ApiError(f"more than api.max_pages ({settings['max_pages']}) pages")

"""lookup_all() against the mock API over real HTTPS, once for each way the API behaves."""
import ssl
import time
 
import pytest
 
from conftest import KNOWN_BAD, TOKEN
from ioc_sweep.api import AuthError, SetupError
from ioc_sweep.config import load_config
from ioc_sweep.intel import lookup_all
 
A, B = KNOWN_BAD  # both start with 0-7
LOW, HIGH = "0" * 64, "e" * 64  # unknown hashes, one on each shard
 
 
@pytest.fixture
def lookup(make_config):
    """Call lookup_all() with this test's config; keyword arguments change config values."""
    def run(digests, token=TOKEN, **changes):
        settings = load_config(make_config(**changes))
        context = ssl.create_default_context(cafile=settings["ca_file"])
        return lookup_all(digests, settings, token, context)
    return run
 
 
def test_two_matches_two_pages(api, lookup):
    assert lookup([A, B]) == ({A: {"sha256": A, "verdict": "malicious", "name": "lab-dropper-a"},
                               B: {"sha256": B, "verdict": "suspicious", "name": "lab-loader-b"}}, set())
    assert api.requests == 2  # one batch of two, one match per page
 
 
def test_wrong_token_is_not_retried(api, lookup):
    with pytest.raises(AuthError, match="401"):
        lookup([A], token="lab-token-revoked")
    assert api.requests == 1
 
 
def test_429_waits_for_retry_after(api, lookup):
    api.mode = "throttle"
    start = time.monotonic()
    assert list(lookup([A])[0]) == [A]
    assert api.requests == 2 and time.monotonic() - start >= 1
 
 
@pytest.mark.parametrize("mode, failed, requests", [
    ("down", {LOW, A, HIGH}, 6),  # two batches, three attempts each
    ("partial", {HIGH}, 4),       # batch 2 holds a hash from the broken shard
], ids=["down", "partial"])
def test_failed_batches_are_reported(api, lookup, mode, failed, requests):
    api.mode = mode
    assert lookup([LOW, A, HIGH])[1] == failed
    assert api.requests == requests
 
 
def test_every_batch_wrong_is_a_setup_error(api, lookup):
    api.mode = "loop"  # in each batch, page 2 repeats the cursor of page 1: nothing temporary
    with pytest.raises(SetupError, match="none of them temporary"):
        lookup([LOW, A, HIGH])
    assert api.requests == 4
 
 
def test_dripping_api_is_bounded_by_the_deadline(api, lookup):
    api.mode, api.delay = "drip", 0.2  # every read gets a byte in time; the whole answer never does
    start = time.monotonic()
    assert lookup([A], deadline=2)[1] == {A}
    assert time.monotonic() - start < 3
 
 
def test_stalled_api_is_bounded_by_the_deadline(api, lookup):
    api.mode, api.delay = "slow", 3
    start = time.monotonic()
    assert lookup([A], deadline=2)[1] == {A}
    assert time.monotonic() - start < 2.5

#!/bin/sh
# fakescan: a stand-in for a malware scanner, for the ioc-sweep lab.
# Usage: fakescan --json -- FILE
# Prints a JSON report; a finding when FILE holds the lab's test signature.
# Exit status: 0 report printed, 2 usage error or unreadable file.
if [ "$#" -ne 3 ] || [ "$1" != --json ] || [ "$2" != -- ]; then
  echo "usage: fakescan --json -- FILE" >&2
  exit 2
fi
grep -q -F -e LAB-TEST-SIGNATURE -- "$3"
case $? in
  0) echo '{"findings": [{"rule": "lab-test-signature", "severity": "high"}]}' ;;
  1) echo '{"findings": []}' ;;
  *) exit 2 ;;
esac

"""Run the external scanner on one file: a list, --, a timeout and a scrubbed environment."""
import json
import subprocess
from pathlib import Path
 
CHILD_ENV = {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8"}  # no token, nothing else inherited
 
 
class ScanError(Exception):
    """No usable answer from the scanner: missing, failed, hung, or not its JSON report."""
 
 
def scan_file(scanner: Path, path: Path, timeout: float) -> list[dict]:
    try:
        result = subprocess.run(
            [scanner, "--json", "--", path],  # "--": a file named -x is not an option
            capture_output=True, timeout=timeout, env=CHILD_ENV,
            encoding="utf-8", errors="replace",  # a file name in its output may not be UTF-8
        )
    except OSError as err:  # FileNotFoundError, PermissionError: it never started
        raise ScanError(f"cannot run {scanner}: {err.strerror}") from err
    except subprocess.TimeoutExpired as err:
        raise ScanError(f"no answer within {timeout:g}s") from err
    if result.returncode != 0:
        raise ScanError(f"exited {result.returncode}: {result.stderr.strip()}")
    try:
        findings = json.loads(result.stdout)["findings"]
        return [{"rule": str(f["rule"]), "severity": str(f["severity"])} for f in findings]
    except (json.JSONDecodeError, KeyError, TypeError) as err:
        raise ScanError("its output is not the JSON report") from err

"""scan_file() with the lab's fakescan and with stubs that fail, hang or print their environment."""
from pathlib import Path
 
import pytest
 
from conftest import LAB
from ioc_sweep.scanner import ScanError, scan_file
 
 
@pytest.fixture
def stub(tmp_path):
    """Return a function that writes an executable stub scanner running the given sh code."""
    def write(sh_code: str) -> Path:
        path = tmp_path / "stubscan"
        path.write_text(f"#!/bin/sh\n{sh_code}\n")
        path.chmod(0o755)
        return path
    return write
 
 
def test_signature_found_and_option_like_name_is_a_file(tmp_path, monkeypatch):
    monkeypatch.chdir(tmp_path)
    Path("--help").write_text("text LAB-TEST-SIGNATURE text\n")  # a name that looks like an option
    assert scan_file(LAB / "fakescan", Path("--help"), 5) == [{"rule": "lab-test-signature", "severity": "high"}]
 
 
@pytest.mark.parametrize("sh_code, timeout, message", [
    ("echo 'signature database missing' >&2; exit 2", 5, "exited 2: signature database missing"),
    ("echo 'Scanning... done'", 5, "not the JSON report"),
    ("exec sleep 30", 0.5, "no answer within 0.5s"),  # exec: the kill on timeout reaches sleep
], ids=["fails", "garbage", "hangs"])
def test_scanner_failures(stub, sh_code, timeout, message):
    with pytest.raises(ScanError, match=message):
        scan_file(stub(sh_code), Path("x"), timeout)
 
 
def test_missing_scanner(tmp_path):
    with pytest.raises(ScanError, match="No such file or directory"):
        scan_file(tmp_path / "nowhere", Path("x"), 5)
 
 
def test_scanner_gets_no_token(stub, tmp_path, monkeypatch):
    monkeypatch.setenv("IOC_SWEEP_TOKEN", "lab-token-not-secret")  # as if a wrapper exported it
    scan_file(stub(f"""env > {tmp_path}/env; echo '{{"findings": []}}'"""), Path("x"), 5)
    env = (tmp_path / "env").read_text()
    assert "lab-token-not-secret" not in env
    assert "PATH=/usr/bin:/bin\n" in env

"""Findings on stdout as NDJSON or CSV, and the per-file summary CSV, replaced atomically."""
import csv
import json
import os
import sys
import tempfile
from pathlib import Path
 
FINDING_FIELDS = ["path", "sha256", "source", "detail"]
SUMMARY_FIELDS = ["path", "sha256", "status", "detail"]
 
 
def printable(path) -> str:
    """A path as text. Linux file names are bytes; bytes that are not UTF-8 are written as \\xNN."""
    return os.fsencode(path).decode("utf-8", "backslashreplace")
 
 
def neutralise(value) -> str:
    """Stop a spreadsheet from running a cell as a formula: file names come from outside."""
    text = str(value)
    return "'" + text if text.startswith(("=", "+", "-", "@", "\t", "\r", "\n")) else text
 
 
def write_findings(findings: list[dict], fmt: str) -> None:
    if fmt == "ndjson":
        for finding in findings:
            print(json.dumps(finding))  # one object per line; json.dumps escapes any file name
        return
    writer = csv.DictWriter(sys.stdout, FINDING_FIELDS)
    writer.writeheader()
    for finding in findings:
        writer.writerow({key: neutralise(finding[key]) for key in FINDING_FIELDS})
 
 
def write_summary(rows: list[dict], target: Path) -> None:
    """Readers see the old summary or the new one, never half of one."""
    fd, tmp = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.")
    try:
        with open(fd, "w", encoding="utf-8", newline="") as f:
            writer = csv.DictWriter(f, SUMMARY_FIELDS)
            writer.writeheader()
            for row in rows:
                writer.writerow({key: neutralise(row[key]) for key in SUMMARY_FIELDS})
            f.flush()
            os.fsync(f.fileno())
        os.replace(tmp, target)
    except BaseException:
        os.unlink(tmp)
        raise

"""Everything that changes files: the summary and the quarantine. --dry-run stops here, only here."""
import logging
 
from .output import printable, write_summary
from .quarantine import QuarantineError, destination, prepare, quarantine
 
log = logging.getLogger(__name__)
 
 
def apply(records: list[dict], findings: list[dict], settings: dict, move: bool, dry_run: bool) -> bool:
    """Write the summary and, with move, quarantine each file with a finding. False: not all done."""
    found = {}  # path -> the details of its findings
    for f in findings:
        found.setdefault(f["path"], []).append(f"{f['source']} {f['detail']}")
    rows = [summary_row(r, found.get(printable(r["path"]))) for r in records]
    done = True
    if dry_run:
        log.info("dry run: would write %d rows to %s", len(rows), settings["summary_csv"])
    else:
        try:
            write_summary(rows, settings["summary_csv"])
        except OSError as err:
            log.error("summary not written: %s: %s", settings["summary_csv"], err.strerror)
            done = False
    if not move:
        return done
    into = settings["quarantine_dir"]
    try:
        prepare(into, create=not dry_run)
    except (OSError, QuarantineError) as err:
        log.error("nothing quarantined: %s", err)
        return False
    for r in records:
        if printable(r["path"]) not in found:
            continue
        if dry_run:  # the plan comes from the same code path as the real moves
            log.info("dry run: would move %s -> %s", r["path"], destination(into, r["path"], r["sha256"]))
            continue
        try:
            log.info("quarantined %s -> %s", r["path"], quarantine(r["path"], r["sha256"], into))
        except (OSError, QuarantineError) as err:  # FileExistsError included: never overwrite
            log.warning("%s: not moved: %s", r["path"], err)
            done = False
    return done
 
 
def summary_row(record: dict, details: list[str] | None) -> dict:
    if details:
        status = "finding"
    else:
        status, details = ("incomplete" if record["problems"] else "clean"), record["problems"]
    return {"path": printable(record["path"]), "sha256": record["sha256"], "status": status,
            "detail": "; ".join(details)}

"""The checks: hash the tree, look the digests up in threat intel, run the scanner on each file."""
import logging
 
from .intel import lookup_all
from .output import printable
from .scanner import ScanError, scan_file
from .walk import inventory
 
log = logging.getLogger(__name__)
 
 
class ApiUnavailable(Exception):
    """Every lookup failed. Without threat intel, "no findings" would be a false all-clear."""
 
 
def check_tree(root, settings: dict, token: str, context) -> tuple[list[dict], list[dict]]:
    """Return (records, findings). A record's problems say why its file was not fully checked."""
    records = inventory(root, settings["max_size"])
    if not records:  # an unmounted share looks exactly like this
        log.warning("no files at all under %s: is it the right directory, and mounted?", root)
    digests = sorted({r["sha256"] for r in records if r["sha256"]})  # each content once
    log.info("%d files under %s, %d distinct digests to look up", len(records), root, len(digests))
    matches, failed = lookup_all(digests, settings, token, context)
    if digests and len(failed) == len(digests):
        raise ApiUnavailable(f"all {len(digests)} lookups failed")
    findings = []
    for record in records:
        if not record["sha256"]:
            continue  # not hashed: the reason is already in its problems
        if record["sha256"] in failed:
            record["problems"].append("lookup failed")
        match = matches.get(record["sha256"])
        if match:
            findings.append(finding(record, "intel", f"{match['verdict']} {match['name']}"))
        try:
            hits = scan_file(settings["scanner"], record["path"], settings["scanner_timeout"])
        except ScanError as err:
            log.warning("%s: scanner: %s", record["path"], err)
            record["problems"].append(f"scanner: {err}")
            continue
        for hit in hits:
            findings.append(finding(record, "scanner", f"{hit['severity']} {hit['rule']}"))
    return records, findings
 
 
def finding(record: dict, source: str, detail: str) -> dict:
    return {"path": printable(record["path"]), "sha256": record["sha256"], "source": source, "detail": detail}

"""The command line of ioc-sweep."""
import argparse
from pathlib import Path
 
 
def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="ioc-sweep", description="Sweep a directory for known-bad files.", allow_abbrev=False,
        epilog="exit status: 0 clean, 1 findings, 2 fix the setup, 3 partial, 4 API unavailable, "
               "70 internal error")
    commands = parser.add_subparsers(dest="command", required=True, metavar="COMMAND")
    scan = commands.add_parser("scan", allow_abbrev=False, help="sweep one directory tree")
    scan.add_argument("dir", type=Path, metavar="DIR", help="the directory to sweep")
    scan.add_argument("--config", required=True, type=Path, metavar="FILE", help="settings (TOML)")
    scan.add_argument("--dry-run", action="store_true", help="look everything up, change nothing")
    scan.add_argument("--quarantine", action="store_true", help="move files with findings away")
    scan.add_argument("--format", choices=["ndjson", "csv"], default="ndjson",
                      help="how findings are written to stdout (default: %(default)s)")
    scan.add_argument("-v", "--verbose", action="store_true", help="log debug messages too")
    return parser

"""ioc-sweep: sweep a directory for known-bad files.
 
Exit status: 0 clean; 1 findings; 2 usage, config, token or API setup problem (fix it, then run
again); 3 partial: some files were not fully checked (wins over 1); 4 threat-intel API
unavailable (try later); 70 internal error: a bug or an input the tool did not expect.
"""
import logging
import ssl
 
from .actions import apply
from .api import AuthError, SetupError
from .cli_args import build_parser
from .config import ConfigError, load_config
from .output import write_findings
from .secret import RedactSecrets, SecretError, read_token
from .sweep import ApiUnavailable, check_tree
 
CLEAN, FINDINGS, USAGE, PARTIAL, UNAVAILABLE, INTERNAL = 0, 1, 2, 3, 4, 70
log = logging.getLogger("ioc_sweep")  # the parent of every module's logger in the package
 
 
def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)  # a usage error exits 2 here
    handler = logging.StreamHandler()  # stderr
    handler.setFormatter(logging.Formatter("ioc-sweep: %(levelname)s %(message)s"))
    log.handlers = [handler]  # replace, not add: tests call main() many times in one process
    log.setLevel(logging.DEBUG if args.verbose else logging.INFO)
    try:
        return sweep(args, handler)
    except Exception:  # never let a crash exit 1, which would read as "findings, all checked"
        log.exception("internal error; the sweep is incomplete")
        return INTERNAL
 
 
def sweep(args, handler: logging.Handler) -> int:
    try:
        settings = load_config(args.config)
        token = read_token(settings["token_file"])
        context = ssl.create_default_context(cafile=settings["ca_file"])
    except (ConfigError, SecretError) as err:
        log.error("%s", err)
        return USAGE
    except OSError as err:  # the CA file: missing, or not a certificate (ssl.SSLError)
        log.error("api.ca_file %s: %s", settings["ca_file"], err)
        return USAGE
    handler.addFilter(RedactSecrets([token]))  # from here on, no log line can show the token
    if not args.dir.is_dir() or settings["quarantine_dir"].resolve().is_relative_to(args.dir.resolve()):
        log.error("%s: not a directory, or output.quarantine_dir is inside it", args.dir)
        return USAGE
    try:
        records, findings = check_tree(args.dir, settings, token, context)
    except AuthError as err:  # a subclass of SetupError, so it comes first
        log.error("the API refused the token: %s (not retried)", err)
        return USAGE
    except SetupError as err:
        log.error("the API cannot work with these settings: %s (not retried)", err)
        return USAGE
    except ApiUnavailable as err:
        log.error("threat-intel API unavailable: %s; nothing reported", err)
        return UNAVAILABLE
    write_findings(findings, args.format)
    if not apply(records, findings, settings, move=args.quarantine, dry_run=args.dry_run):
        return PARTIAL
    if any(r["problems"] for r in records):
        return PARTIAL
    return FINDINGS if findings else CLEAN

"""ioc-sweep through main(argv): exit statuses, stdout and stderr, and the files it changes."""
import json
import os
 
import pytest
 
from ioc_sweep import cli
from ioc_sweep.cli import main
 
 
def files_under(root):
    return sorted(str(p.relative_to(root)) for p in root.rglob("*"))
 
 
def sweep(tree, config, *options):
    return main(["scan", str(tree), "--config", str(config), *options])
 
 
@pytest.mark.parametrize("mode, status", [("ok", 1), ("partial", 3), ("down", 4)])
def test_exit_status_follows_the_api(api, tree, make_config, capsys, mode, status):
    api.mode = mode
    assert sweep(tree, make_config()) == status
    out = capsys.readouterr().out
    if status == 4:
        assert out == ""  # without threat intel nothing is reported, not even "clean"
    else:
        assert json.loads(out)["path"] == str(tree / "invoice.pdf")
 
 
def test_clean_tree(api, tree, make_config):
    (tree / "invoice.pdf").unlink()
    assert sweep(tree, make_config()) == 0
 
 
def test_config_error_and_usage_error(api, tree, make_config, capsys):
    config = make_config()
    config.write_text(config.read_text().replace("batch_size", "batch_sise"))
    assert sweep(tree, config) == 2
    assert "unknown key api.batch_sise" in capsys.readouterr().err
    with pytest.raises(SystemExit) as exc:
        sweep(tree, config, "--format", "yaml")
    assert exc.value.code == 2
 
 
@pytest.mark.parametrize("old, new, requests", [
    ("/v1", "/v2", 1),              # a wrong path: 404, the same for every request
    ("localhost", "127.0.0.1", 0),  # a name the certificate does not list: no request at all
], ids=["404", "tls-name"])
def test_setup_error_is_2_not_4(api, tree, make_config, old, new, requests):
    assert sweep(tree, make_config(url=api.url.replace(old, new))) == 2
    assert api.requests == requests  # stopped at once, never retried
 
 
def test_refused_token_is_redacted(api, tree, make_config, capsys):
    config = make_config()
    (config.parent / "api-token").write_text("lab-token-revoked\n")
    assert sweep(tree, config) == 2
    err = capsys.readouterr().err
    assert "Unknown token [REDACTED]" in err and "lab-token-revoked" not in err
 
 
def test_dry_run_changes_nothing(api, tree, make_config, tmp_path):
    config = make_config()
    before = files_under(tmp_path)
    assert sweep(tree, config, "--dry-run", "--quarantine") == 1
    assert files_under(tmp_path) == before
 
 
def test_quarantine_then_rerun_is_a_no_op(api, tree, make_config, tmp_path):
    config = make_config()
    assert sweep(tree, config, "--quarantine") == 1
    manifest = (tmp_path / "quarantine" / "manifest.ndjson").read_text()
    assert not (tree / "invoice.pdf").exists() and manifest.count("\n") == 1
    assert (tmp_path / "quarantine").stat().st_mode & 0o777 == 0o700
    assert sweep(tree, config, "--quarantine") == 0
    assert (tmp_path / "quarantine" / "manifest.ndjson").read_text() == manifest
 
 
def test_name_that_is_not_utf8(api, tree, make_config, tmp_path, capsys):
    odd = tree / os.fsdecode(b"caf\xe9.pdf")  # a Latin-1 name, as an upload can have
    odd.write_bytes((tree / "invoice.pdf").read_bytes())
    assert sweep(tree, make_config()) == 1
    paths = [json.loads(line)["path"] for line in capsys.readouterr().out.splitlines()]
    assert f"{tree}/caf\\xe9.pdf" in paths
    assert "caf\\xe9.pdf,1f72" in (tmp_path / "summary.csv").read_text()
 
 
def test_internal_error_is_70_not_findings(api, tree, make_config, monkeypatch, capsys):
    def broken(*args):
        raise RuntimeError("a bug")
 
    monkeypatch.setattr(cli, "check_tree", broken)  # the name cli.sweep() looks up
    assert sweep(tree, make_config()) == 70
    assert "internal error" in capsys.readouterr().err

#!/usr/bin/env bash
# Builds the trees the sweep runs on, from scratch each time:
#   srv/clean    three ordinary files
#   srv/uploads  two known-bad samples that share a name, a file with the scanner's test
#                signature, ordinary files, a file whose name is not valid UTF-8, and a
#                symbolic link that points out of the tree
set -euo pipefail
rm -rf srv
mkdir -p srv/clean/docs srv/uploads/alice srv/uploads/bob srv/uploads/shared
printf 'quarterly report, nothing unusual\n' > srv/clean/docs/report.txt
printf 'meeting notes\n' > srv/clean/docs/notes.txt
printf 'backup done\n' > srv/clean/backup.log
printf 'ioc-sweep lab sample A: stands in for a malicious file\n' > srv/uploads/alice/invoice.pdf
printf 'ioc-sweep lab sample B: a different file with the same name\n' > srv/uploads/bob/invoice.pdf
printf 'macro loader text LAB-TEST-SIGNATURE end\n' > srv/uploads/bob/macro.docm
printf 'quarterly report, nothing unusual\n' > srv/uploads/shared/report.txt
printf 'meeting notes\n' > srv/uploads/shared/notes.txt
printf 'staff rota for October\n' > srv/uploads/shared/rota.txt
# A name with byte 0xE9 (a Latin-1 "é"), which is not valid UTF-8: whoever uploads chooses the name.
printf 'meeting notes\n' > "srv/uploads/shared/caf$(printf '\351').txt"
ln -s /etc/passwd srv/uploads/shared/passwd
# %q shows each name as Bash would quote it, so the byte 0xE9 appears as \351
find srv -type f -print0 | sort -z | while IFS= read -r -d '' name; do printf '%q\n' "$name"; done
find srv -type l -printf '%p -> %l\n'

#!/usr/bin/env bash
# One scheduled sweep, run the way cron should run it: one run at a time, a time limit, and the
# findings and the log kept in a directory that only this user can read.
# Exit status: ioc-sweep's own (0, 1, 2, 3, 4, 70); 75 another sweep still holds the lock;
# 124 the time limit was reached (137 if the sweep then had to be killed).
set -euo pipefail
umask 077
dir=${1:?usage: nightly-sweep.sh DIR}
root=$(cd -- "$(dirname -- "$0")" && pwd -P)
results=$root/results
mkdir -p -- "$results"
# The script opens its own log only after mkdir: a crontab redirection into results/ would be
# opened by the shell before this script runs, and fail while results/ does not exist yet.
exec 2>> "$results/sweep.log"
printf 'nightly-sweep: %s sweep of %s\n' "$(date -u +%FT%TZ)" "$dir" >&2
 
exec 9> "$results/.lock"
if ! flock -n 9; then
  echo "nightly-sweep: another sweep still holds the lock" >&2
  exit 75
fi
tmp=$(mktemp -- "$results/.findings.XXXXXX")
trap 'rm -f -- "$tmp"' EXIT
 
status=0
timeout -k 1m 6h "$root/prod/bin/ioc-sweep" scan "$dir" --config "$root/sweep.toml" > "$tmp" || status=$?
case $status in
  0 | 1 | 3) mv -- "$tmp" "$results/findings.ndjson" ;; # a result, complete or partial: publish it
  *) echo "nightly-sweep: findings.ndjson not replaced" >&2 ;;
esac
echo "nightly-sweep: ioc-sweep exit status $status" >&2
exit "$status"