#!/usr/bin/env python3 """Validate and resolve the executable S1 host-baseline contract.""" from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any import yaml SCHEMA_VERSION = "2.0" PROFILE_MODES = {"ufw", "external"} REQUIRED_PACKAGES = {"curl", "git", "vim", "htop", "ufw", "fail2ban"} REQUIRED_SERVICES = {"fail2ban", "ssh.socket"} REQUIRED_SSH = { "PasswordAuthentication": "no", "PermitRootLogin": "no", "PubkeyAuthentication": "yes", "ChallengeResponseAuthentication": "no", } CONSUMER_MARKERS = { "ansible/roles/base/tasks/main.yml": [ "baseline_required_packages", "baseline_ssh_directives", "baseline_user", "baseline_security", "baseline_firewall", ], "goss/baseline.yaml.j2": [ "baseline_required_packages", "baseline_required_services", "baseline_ssh_directives", "baseline_user", "baseline_security", "baseline_firewall", ], } class BaselineError(ValueError): """The baseline model or its repository consumers are inconsistent.""" def _strings( value: Any, label: str, errors: list[str], *, allow_empty: bool = False ) -> list[str]: if not isinstance(value, list) or (not value and not allow_empty) or not all( isinstance(item, str) and item.strip() for item in value ): errors.append(f"{label} must be a non-empty string list") return [] if len(set(value)) != len(value): errors.append(f"{label} contains duplicates") return value def validate_spec(payload: Any) -> dict[str, Any]: if not isinstance(payload, dict): raise BaselineError("baseline must be a YAML object") errors: list[str] = [] if str(payload.get("version")) != SCHEMA_VERSION: errors.append(f"version must be {SCHEMA_VERSION!r}") defaults = payload.get("defaults") profiles = payload.get("profiles") if not isinstance(defaults, dict): errors.append("defaults must be an object") defaults = {} if not isinstance(profiles, dict) or not profiles: errors.append("profiles must be a non-empty object") profiles = {} packages = _strings(defaults.get("packages"), "defaults.packages", errors) services = _strings(defaults.get("services"), "defaults.services", errors) missing_packages = sorted(REQUIRED_PACKAGES - set(packages)) missing_services = sorted(REQUIRED_SERVICES - set(services)) if missing_packages: errors.append(f"defaults.packages omits governed packages {', '.join(missing_packages)}") if missing_services: errors.append(f"defaults.services omits governed services {', '.join(missing_services)}") ssh = defaults.get("ssh_directives") if not isinstance(ssh, dict) or not ssh or not all( isinstance(key, str) and isinstance(value, str) for key, value in (ssh or {}).items() ): errors.append("defaults.ssh_directives must be a non-empty string map") elif {key: ssh.get(key) for key in REQUIRED_SSH} != REQUIRED_SSH: errors.append("defaults.ssh_directives weakens a governed SSH directive") user = defaults.get("user") if not isinstance(user, dict): errors.append("defaults.user must be an object") else: for key in ("name", "shell", "sudo"): if not isinstance(user.get(key), str) or not user[key]: errors.append(f"defaults.user.{key} must be a non-empty string") if user.get("sudo") != "NOPASSWD": errors.append("defaults.user.sudo must be NOPASSWD") security = defaults.get("security") if not isinstance(security, dict): errors.append("defaults.security must be an object") else: if not isinstance(security.get("histcontrol"), str): errors.append("defaults.security.histcontrol must be a string") _strings( security.get("fail2ban_jails"), "defaults.security.fail2ban_jails", errors, ) for name, profile in profiles.items(): label = f"profiles.{name}" if not isinstance(profile, dict): errors.append(f"{label} must be an object") continue _strings( profile.get("services", []), f"{label}.services", errors, allow_empty=True, ) firewall = profile.get("firewall") if not isinstance(firewall, dict): errors.append(f"{label}.firewall must be an object") continue mode = firewall.get("mode") if mode not in PROFILE_MODES: errors.append(f"{label}.firewall.mode must be one of {sorted(PROFILE_MODES)}") expected_managed = mode == "ufw" if firewall.get("managed") is not expected_managed: errors.append(f"{label}.firewall.managed must be {expected_managed}") verification = firewall.get("verification") if not isinstance(verification, dict): errors.append(f"{label}.firewall.verification must be an object") else: if not isinstance(verification.get("command"), str): errors.append(f"{label}.firewall.verification.command must be a string") _strings( verification.get("stdout"), f"{label}.firewall.verification.stdout", errors, ) if mode == "external": replacement = firewall.get("replacement_control") if not isinstance(replacement, dict): errors.append(f"{label}.firewall.replacement_control must be an object") else: for key in ("description", "owner", "removal_condition"): if not isinstance(replacement.get(key), str) or not replacement[key]: errors.append( f"{label}.firewall.replacement_control.{key} must be a string" ) for required_profile, required_mode in { "ufw-managed": "ufw", "external-firewall": "external", }.items(): if profiles.get(required_profile, {}).get("firewall", {}).get("mode") != required_mode: errors.append( f"profiles.{required_profile} must declare firewall mode {required_mode}" ) if errors: raise BaselineError("baseline contract failed:\n- " + "\n- ".join(errors)) return payload def load_spec(path: Path) -> dict[str, Any]: try: payload = yaml.safe_load(path.read_text(encoding="utf-8")) except (OSError, yaml.YAMLError) as exc: raise BaselineError(f"cannot read {path}: {exc}") from exc return validate_spec(payload) def profile_hostvars(payload: dict[str, Any], profile_name: str) -> dict[str, Any]: validate_spec(payload) try: profile = payload["profiles"][profile_name] except KeyError as exc: raise BaselineError(f"unknown baseline profile {profile_name!r}") from exc defaults = payload["defaults"] services = list(dict.fromkeys(defaults["services"] + profile["services"])) return { "baseline_profile": profile_name, "baseline_required_packages": defaults["packages"], "baseline_required_services": services, "baseline_ssh_directives": defaults["ssh_directives"], "baseline_user": defaults["user"], "baseline_security": defaults["security"], "baseline_firewall": profile["firewall"], "ufw_manage": profile["firewall"]["managed"], } def validate_repo_consumers(root: Path) -> None: errors: list[str] = [] for relative, markers in CONSUMER_MARKERS.items(): path = root / relative try: text = path.read_text(encoding="utf-8") except OSError as exc: errors.append(f"cannot read {relative}: {exc}") continue missing = [marker for marker in markers if marker not in text] if missing: errors.append(f"{relative} does not consume {', '.join(missing)}") if errors: raise BaselineError("baseline consumer parity failed:\n- " + "\n- ".join(errors)) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "spec", nargs="?", type=Path, default=Path("spec/server-baseline.yaml") ) parser.add_argument("--profile", help="resolve one profile as Ansible hostvars") parser.add_argument("--check-repo", action="store_true") args = parser.parse_args() try: payload = load_spec(args.spec) if args.check_repo: validate_repo_consumers(Path(__file__).resolve().parents[1]) result: dict[str, Any] = { "ok": True, "version": payload["version"], "profiles": sorted(payload["profiles"]), } if args.profile: result["hostvars"] = profile_hostvars(payload, args.profile) except BaselineError as exc: print(exc, file=sys.stderr) return 1 print(json.dumps(result, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())