b93af8cc78
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02994-7685-7940-bf34-3555b8256018
240 lines
9 KiB
Python
240 lines
9 KiB
Python
#!/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())
|