image-tools/image_tools/cis.py

"""Is the image what CIS asks of it? (CIS Docker Benchmark 1.6.0 §4, CIS Debian Linux 12)

Two measurements, one verdict:

- Trivy's ``docker-cis-1.6.0`` spec, for the controls it automates (4.1, 4.4,
  4.6, 4.7, 4.9, 4.10); the rest it calls "Manual";
- ``image_tools/cis_os.sh``, run inside the image as root, for what of that can be
  seen from inside (4.3, 4.8, 4.11) and the CIS Debian controls that make sense
  in a container.

The image's own repository decides the rest in ``cis.toml``: ``[[manual]]`` is
met by how the image is made, ``[[exception]]`` fails on purpose, with a reason,
until it expires. ``packages.allow`` there names the packages it may hold. The gate is red when a control fails without an exception, an
exception has expired, a control is neither measured nor decided, a decision
names a control nothing knows, or a measurement could not be taken.

    python -m image_tools.cis <ref> [--decisions cis.toml] [--packages packages.allow] \
        --json cis.json --report cis.md

Both files default to the working directory: the image's repository.

Standard library only; Trivy runs as the container image_tools/verify.py pins.
"""

from __future__ import annotations

import argparse
import datetime
import json
import subprocess
import tomllib
from collections.abc import Callable
from dataclasses import asdict, dataclass, field
from pathlib import Path

from image_tools.verify import TIMEOUT, TRIVY_IMAGE

OS_CHECKS = Path(__file__).with_name("cis_os.sh")
SPEC = "docker-cis-1.6.0"

PASS, FAIL, UNMEASURED = "pass", "fail", "unmeasured"
#: What the gate makes of a control; the first three keep it green.
GREEN = frozenset({"pass", "manual", "excepted"})


@dataclass
class Control:
    id: str
    title: str
    result: str
    detail: str = ""
    state: str = ""
    reason: str = ""


@dataclass
class Decision:
    kind: str
    reason: str
    expires: datetime.date | None = None


@dataclass
class Report:
    ok: bool
    controls: list[Control] = field(default_factory=list)
    stale: list[str] = field(default_factory=list)
    unknown: list[str] = field(default_factory=list)
    error: str | None = None

    def to_json(self) -> str:
        return json.dumps(asdict(self), indent=2, default=str)

    def failing(self) -> set[str]:
        return {c.id for c in self.controls if c.result == FAIL}


def from_trivy(summary: dict) -> list[Control]:
    """Trivy's compliance summary: no ``TotalFail`` means it did not measure."""
    controls = []
    for c in summary.get("SummaryControls") or []:
        fails = c.get("TotalFail")
        result = UNMEASURED if fails is None else (FAIL if fails else PASS)
        detail = f"{fails} failed (Trivy)" if fails else ""
        controls.append(Control(f"docker:{c['ID']}", c.get("Name", ""), result, detail))
    return controls


def from_os(text: str) -> list[Control]:
    """cis_os.sh's lines: id, PASS|FAIL, title, what fails."""
    controls = []
    for line in text.splitlines():
        if not line.strip():
            continue
        cid, result, title, detail = (line.split("\t") + ["", "", ""])[:4]
        if result not in ("PASS", "FAIL"):
            raise ValueError(f"cis_os.sh said: {line!r}")
        controls.append(Control(cid, title, result.lower(), detail))
    return controls


def merge(*sources: list[Control]) -> list[Control]:
    """One control per id, in order of first appearance; a failure wins, then a pass."""
    rank = {FAIL: 2, PASS: 1, UNMEASURED: 0}
    merged: dict[str, Control] = {}
    for controls in sources:
        for c in controls:
            known = merged.get(c.id)
            if known is None or rank[c.result] > rank[known.result]:
                if known is not None and not c.title:
                    c.title = known.title
                merged[c.id] = c
    return list(merged.values())


def load_decisions(text: str) -> dict[str, Decision]:
    data = tomllib.loads(text)
    decisions = {}
    for kind in ("manual", "exception"):
        for entry in data.get(kind) or []:
            if entry["id"] in decisions:
                raise ValueError(f"{entry['id']} is decided twice")
            if not entry.get("reason", "").strip():
                raise ValueError(f"{entry['id']} has no reason")
            if not isinstance(entry.get("expires", datetime.date.min), datetime.date):
                raise ValueError(f"{entry['id']}: expires is a date, not {entry['expires']!r}")
            decisions[entry["id"]] = Decision(kind, entry["reason"], entry.get("expires"))
    return decisions


def judge(controls: list[Control], decisions: dict[str, Decision], today: datetime.date) -> Report:
    """The gate on the measured *controls* and what ``cis.toml`` decided."""
    stale = []
    for c in controls:
        d = decisions.get(c.id)
        c.reason = d.reason if d else ""
        if d and d.kind == "manual":
            c.state = "manual"
        elif c.result == PASS:
            c.state = "pass"
            if d:
                stale.append(c.id)
        elif d and d.expires and d.expires < today:
            c.state = "expired"
        elif d:
            c.state = "excepted"
        else:
            c.state = "fail" if c.result == FAIL else "undecided"
    unknown = sorted(set(decisions) - {c.id for c in controls})
    ok = bool(controls) and not unknown and all(c.state in GREEN for c in controls)
    return Report(ok, controls, stale, unknown)


def measure(
    ref: str,
    packages: list[str],
    *,
    trivy: Callable[[str], dict] | None = None,
    os_checks: Callable[[str, list[str]], str] | None = None,
) -> list[Control]:
    trivy = trivy or trivy_compliance
    os_checks = os_checks or run_os_checks
    return merge(from_trivy(trivy(ref)), from_os(os_checks(ref, packages)))


def check(
    ref: str,
    decisions: dict[str, Decision],
    packages: list[str],
    today: datetime.date,
    **seams,
) -> Report:
    """Measure *ref* and judge it; never raises."""
    try:
        return judge(measure(ref, packages, **seams), decisions, today)
    except (OSError, ValueError, KeyError, subprocess.SubprocessError) as exc:
        return Report(False, error=f"{type(exc).__name__}: {exc}")


STATE_MARK = {
    "pass": "✅ pass",
    "manual": "📝 manual",
    "excepted": "⚠️ exception",
    "fail": "❌ fail",
    "expired": "❌ exception expired",
    "undecided": "❌ not measured, not decided",
}


def render(report: Report, ref: str) -> str:
    head = "**CIS: green**" if report.ok else "**CIS: red**"
    lines = [f"{head} for `{ref}` (CIS Docker 1.6.0 §4, CIS Debian Linux 12).", ""]
    if report.error:
        lines += [f"The measurement could not be taken: {report.error}", ""]
    if report.unknown:
        lines += [f"Decided in cis.toml but measured by nothing: {', '.join(report.unknown)}", ""]
    if report.stale:
        lines += [f"Exceptions for controls that pass now: {', '.join(report.stale)}", ""]
    if report.controls:
        lines += ["| Control | | Title | Detail |", "|---|---|---|---|"]
        for c in report.controls:
            detail = c.reason if c.state in ("manual", "excepted") else c.detail
            detail = detail.replace("|", "\\|").replace("\n", " ")
            lines.append(f"| {c.id} | {STATE_MARK.get(c.state, c.state)} | {c.title} | {detail} |")
    return "\n".join(lines).rstrip() + "\n"


def load_packages(text: str) -> list[str]:
    return [ln.strip() for ln in text.splitlines() if ln.strip() and not ln.startswith("#")]


def trivy_compliance(ref: str) -> dict:  # pragma: no cover - container seam
    """Trivy's summary of *ref* against the spec, with the checks it was released with."""
    proc = subprocess.run(
        [
            "docker",
            "run",
            "--rm",
            "-v",
            "/var/run/docker.sock:/var/run/docker.sock",
            "-e",
            "TRIVY_USERNAME",
            "-e",
            "TRIVY_PASSWORD",
            TRIVY_IMAGE,
            "image",
            "--quiet",
            "--disable-telemetry",
            "--skip-version-check",
            "--skip-check-update",
            # 4.4 is the matcher's (cis.toml); no vulnerability DB to fetch
            "--scanners",
            "misconfig,secret",
            "--compliance",
            SPEC,
            "--format",
            "json",
            ref,
        ],
        capture_output=True,
        text=True,
        timeout=TIMEOUT,
        check=True,
    )
    return json.loads(proc.stdout)


def run_os_checks(ref: str, packages: list[str]) -> str:  # pragma: no cover - container seam
    """cis_os.sh inside *ref*, as root, fed on stdin: nothing to mount."""
    proc = subprocess.run(
        [
            "docker",
            "run",
            "--rm",
            "-i",
            "--user",
            "0",
            "--network",
            "none",
            "-e",
            "CIS_PACKAGES=" + " ".join(packages),
            "--entrypoint",
            "sh",
            ref,
            "-s",
        ],
        input=OS_CHECKS.read_text(),
        capture_output=True,
        text=True,
        timeout=TIMEOUT,
        check=True,
    )
    return proc.stdout


def main() -> int:  # pragma: no cover - CLI seam
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("ref")
    parser.add_argument("--decisions", default="cis.toml")
    parser.add_argument("--packages", default="packages.allow")
    parser.add_argument("--json")
    parser.add_argument("--report")
    args = parser.parse_args()
    decisions = load_decisions(Path(args.decisions).read_text())
    packages = load_packages(Path(args.packages).read_text())
    report = check(args.ref, decisions, packages, datetime.date.today())
    text = render(report, args.ref)
    if args.json:
        Path(args.json).write_text(report.to_json())
    if args.report:
        Path(args.report).write_text(text)
    print(text)
    return 0 if report.ok else 1


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())