409 lines
14 KiB
Python
Executable File
409 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Preflight owner responses for post-reboot next gates.
|
|
|
|
Read-only by design. This script validates an owner response JSON file against
|
|
the current post-reboot owner packets. It never sends requests, reads secrets,
|
|
writes credential markers, or modifies host/runtime state.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
OWNER_PACKET_GENERATOR = (
|
|
ROOT / "scripts" / "reboot-recovery" / "post-reboot-next-gate-owner-packets.py"
|
|
)
|
|
|
|
EXPECTED_SCHEMA = "awoooi_post_reboot_next_gate_owner_response_v1"
|
|
EXPECTED_OWNER_PACKET_SCHEMA = "awoooi_post_reboot_next_gate_owner_packets_v1"
|
|
PLACEHOLDER_VALUES = {
|
|
"",
|
|
"pending",
|
|
"todo",
|
|
"tbd",
|
|
"n/a",
|
|
"na",
|
|
"owner_role_here",
|
|
"owner_team_here",
|
|
"decision_reason_here",
|
|
"redacted_evidence_ref_here",
|
|
"non_secret_evidence_ref_here",
|
|
"registry_export_ref_here",
|
|
"followup_owner_here",
|
|
"reviewer_here",
|
|
}
|
|
|
|
ESCROW_ITEM_IDS = {
|
|
"restic_repository_password",
|
|
"offsite_provider_credentials",
|
|
"break_glass_admin_credentials",
|
|
"dns_registrar_recovery",
|
|
"oauth_ai_provider_recovery",
|
|
}
|
|
|
|
EXPECTED_HOST_ALIASES = {
|
|
"core-110",
|
|
"gateway-188",
|
|
"k3s-control-120",
|
|
"k3s-control-121",
|
|
"security-observer-112",
|
|
"dev-workstation-111",
|
|
}
|
|
|
|
FORBIDDEN_BOOLEAN_FIELDS = {
|
|
"runtime_action_requested",
|
|
"runtime_action_authorized",
|
|
"host_write_requested",
|
|
"host_write_authorized",
|
|
"secret_value_included",
|
|
"secret_value_collection_allowed",
|
|
"credential_marker_write_requested",
|
|
"credential_marker_write_authorized",
|
|
"wazuh_active_response_requested",
|
|
"wazuh_active_response_authorized",
|
|
"agent_reenroll_requested",
|
|
"wazuh_restart_requested",
|
|
"kali_active_scan_requested",
|
|
}
|
|
|
|
SECRET_VALUE_PATTERNS = [
|
|
re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
|
|
re.compile(r"\bBearer\s+[A-Za-z0-9._~+/=-]{12,}", re.IGNORECASE),
|
|
re.compile(r"\bAuthorization\s*:\s*", re.IGNORECASE),
|
|
re.compile(r"\bgh[pousr]_[A-Za-z0-9_]{20,}"),
|
|
re.compile(r"\bsk-[A-Za-z0-9]{20,}"),
|
|
re.compile(r"\bAIza[0-9A-Za-z_-]{20,}"),
|
|
re.compile(r"\b[0-9]{8,10}:[A-Za-z0-9_-]{20,}\b"),
|
|
re.compile(r"\b(password|token|secret)\s*[:=]\s*[^,\s]+", re.IGNORECASE),
|
|
re.compile(r"\bclient\.keys\b", re.IGNORECASE),
|
|
]
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Validate post-reboot owner response evidence without opening runtime gates.",
|
|
)
|
|
parser.add_argument("--response-file", type=Path, help="Owner response JSON to validate.")
|
|
parser.add_argument(
|
|
"--owner-packet-file",
|
|
type=Path,
|
|
help="Use an existing owner packet JSON instead of generating one.",
|
|
)
|
|
parser.add_argument(
|
|
"--summary-file",
|
|
type=Path,
|
|
help="Generate owner packets from an existing readiness summary file.",
|
|
)
|
|
parser.add_argument("--json", action="store_true", help="Print machine-readable JSON.")
|
|
parser.add_argument(
|
|
"--no-color",
|
|
action="store_true",
|
|
help="Pass --no-color when generating owner packets.",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def load_json(path: Path, label: str = "response_file") -> dict[str, Any]:
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise SystemExit(f"{label}_not_found={path}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise SystemExit(f"{label}_json_invalid={exc}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise SystemExit(f"{label}_json_not_object")
|
|
return payload
|
|
|
|
|
|
def generate_owner_packet(no_color: bool, summary_file: Path | None) -> dict[str, Any]:
|
|
cmd = [str(OWNER_PACKET_GENERATOR)]
|
|
if summary_file:
|
|
cmd.extend(["--summary-file", str(summary_file)])
|
|
if no_color:
|
|
cmd.append("--no-color")
|
|
completed = subprocess.run(
|
|
cmd,
|
|
cwd=ROOT,
|
|
check=False,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
if completed.returncode != 0:
|
|
raise SystemExit(
|
|
"owner_packet_generation_failed "
|
|
f"rc={completed.returncode}\n{completed.stdout}"
|
|
)
|
|
try:
|
|
packet = json.loads(completed.stdout)
|
|
except json.JSONDecodeError as exc:
|
|
raise SystemExit(f"owner_packet_json_invalid={exc}") from exc
|
|
if not isinstance(packet, dict):
|
|
raise SystemExit("owner_packet_json_not_object")
|
|
return packet
|
|
|
|
|
|
def load_owner_packet(args: argparse.Namespace) -> dict[str, Any]:
|
|
if args.owner_packet_file:
|
|
return load_json(args.owner_packet_file, label="owner_packet_file")
|
|
return generate_owner_packet(no_color=args.no_color, summary_file=args.summary_file)
|
|
|
|
|
|
def as_list(value: Any) -> list[Any]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, list):
|
|
return value
|
|
return [value]
|
|
|
|
|
|
def normalized(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value).strip()
|
|
|
|
|
|
def is_placeholder(value: Any) -> bool:
|
|
return normalized(value).lower() in PLACEHOLDER_VALUES
|
|
|
|
|
|
def collect_strings(value: Any, path: str = "$") -> list[tuple[str, str]]:
|
|
strings: list[tuple[str, str]] = []
|
|
if isinstance(value, str):
|
|
strings.append((path, value))
|
|
elif isinstance(value, dict):
|
|
for key, child in value.items():
|
|
strings.extend(collect_strings(child, f"{path}.{key}"))
|
|
elif isinstance(value, list):
|
|
for index, child in enumerate(value):
|
|
strings.extend(collect_strings(child, f"{path}[{index}]"))
|
|
return strings
|
|
|
|
|
|
def find_forbidden_strings(response: dict[str, Any]) -> list[str]:
|
|
failures: list[str] = []
|
|
for path, value in collect_strings(response):
|
|
if path.endswith(".item_id") and value in ESCROW_ITEM_IDS:
|
|
continue
|
|
if path.endswith(".gate_id") and value in {
|
|
"credential_escrow_evidence",
|
|
"wazuh_manager_registry_export",
|
|
"host_188_hygiene_maintenance_window",
|
|
}:
|
|
continue
|
|
for pattern in SECRET_VALUE_PATTERNS:
|
|
if pattern.search(value):
|
|
failures.append(f"forbidden_payload_at={path}")
|
|
break
|
|
return failures
|
|
|
|
|
|
def find_forbidden_booleans(value: Any, path: str = "$") -> list[str]:
|
|
failures: list[str] = []
|
|
if isinstance(value, dict):
|
|
for key, child in value.items():
|
|
child_path = f"{path}.{key}"
|
|
if key in FORBIDDEN_BOOLEAN_FIELDS and child is not False:
|
|
failures.append(f"{child_path}={child!r}")
|
|
failures.extend(find_forbidden_booleans(child, child_path))
|
|
elif isinstance(value, list):
|
|
for index, child in enumerate(value):
|
|
failures.extend(find_forbidden_booleans(child, f"{path}[{index}]"))
|
|
return failures
|
|
|
|
|
|
def owner_packet_gate_ids(packet: dict[str, Any]) -> set[str]:
|
|
if packet.get("schema_version") != EXPECTED_OWNER_PACKET_SCHEMA:
|
|
raise SystemExit(f"owner_packet_schema={packet.get('schema_version')!r}")
|
|
return {
|
|
str(item.get("packet_id"))
|
|
for item in as_list(packet.get("owner_packets"))
|
|
if isinstance(item, dict) and item.get("packet_id")
|
|
}
|
|
|
|
|
|
def response_by_gate(response: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
responses = as_list(response.get("responses"))
|
|
by_gate: dict[str, dict[str, Any]] = {}
|
|
for item in responses:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
gate_id = normalized(item.get("gate_id"))
|
|
if gate_id:
|
|
by_gate[gate_id] = item
|
|
return by_gate
|
|
|
|
|
|
def validate_common(gate_id: str, item: dict[str, Any]) -> list[str]:
|
|
failures: list[str] = []
|
|
for key in (
|
|
"owner_role",
|
|
"owner_team",
|
|
"decision",
|
|
"decision_reason",
|
|
"affected_scope",
|
|
"followup_owner",
|
|
):
|
|
if is_placeholder(item.get(key)):
|
|
failures.append(f"{gate_id}.{key}_missing")
|
|
decision = normalized(item.get("decision")).lower()
|
|
if decision not in {"accepted", "rejected", "needs_supplement"}:
|
|
failures.append(f"{gate_id}.decision_invalid={decision!r}")
|
|
evidence_refs = [
|
|
ref for ref in as_list(item.get("redacted_evidence_refs")) if not is_placeholder(ref)
|
|
]
|
|
if not evidence_refs:
|
|
failures.append(f"{gate_id}.redacted_evidence_refs_missing")
|
|
return failures
|
|
|
|
|
|
def validate_credential_escrow(item: dict[str, Any]) -> list[str]:
|
|
failures: list[str] = []
|
|
escrow_items = as_list(item.get("escrow_items"))
|
|
seen = {
|
|
normalized(entry.get("item_id"))
|
|
for entry in escrow_items
|
|
if isinstance(entry, dict)
|
|
}
|
|
missing = sorted(ESCROW_ITEM_IDS - seen)
|
|
if missing:
|
|
failures.append(f"credential_escrow_evidence.missing_items={missing}")
|
|
for entry in escrow_items:
|
|
if not isinstance(entry, dict):
|
|
failures.append("credential_escrow_evidence.escrow_item_not_object")
|
|
continue
|
|
item_id = normalized(entry.get("item_id"))
|
|
if item_id not in ESCROW_ITEM_IDS:
|
|
failures.append(f"credential_escrow_evidence.unknown_item={item_id!r}")
|
|
for key in ("non_secret_evidence_ref", "recovery_owner", "reviewer", "last_reviewed_at"):
|
|
if is_placeholder(entry.get(key)):
|
|
failures.append(f"credential_escrow_evidence.{item_id}.{key}_missing")
|
|
if entry.get("contains_secret_value") is not False:
|
|
failures.append(f"credential_escrow_evidence.{item_id}.contains_secret_value_not_false")
|
|
return failures
|
|
|
|
|
|
def validate_wazuh_registry(item: dict[str, Any]) -> list[str]:
|
|
failures: list[str] = []
|
|
for key in (
|
|
"registry_export_ref",
|
|
"registry_time_window",
|
|
"dashboard_api_connection_status",
|
|
"dashboard_api_version_status",
|
|
"reviewer",
|
|
):
|
|
if is_placeholder(item.get(key)):
|
|
failures.append(f"wazuh_manager_registry_export.{key}_missing")
|
|
aliases = {normalized(alias) for alias in as_list(item.get("expected_host_aliases"))}
|
|
missing_aliases = sorted(EXPECTED_HOST_ALIASES - aliases)
|
|
if missing_aliases:
|
|
failures.append(f"wazuh_manager_registry_export.missing_aliases={missing_aliases}")
|
|
if not isinstance(item.get("manager_registry_count"), int):
|
|
failures.append("wazuh_manager_registry_export.manager_registry_count_not_int")
|
|
if normalized(item.get("dashboard_api_connection_status")).lower() != "ok":
|
|
failures.append("wazuh_manager_registry_export.dashboard_api_connection_not_ok")
|
|
if normalized(item.get("dashboard_api_version_status")).lower() != "ok":
|
|
failures.append("wazuh_manager_registry_export.dashboard_api_version_not_ok")
|
|
return failures
|
|
|
|
|
|
def evaluate(packet: dict[str, Any], response: dict[str, Any] | None) -> dict[str, Any]:
|
|
expected_gates = owner_packet_gate_ids(packet)
|
|
result: dict[str, Any] = {
|
|
"schema_version": "awoooi_post_reboot_owner_response_preflight_v1",
|
|
"expected_gate_count": len(expected_gates),
|
|
"expected_gates": sorted(expected_gates),
|
|
"owner_response_received_count": 0,
|
|
"owner_response_accepted_count": 0,
|
|
"runtime_gate_count": 0,
|
|
"runtime_action_authorized": False,
|
|
"host_write_authorized": False,
|
|
"secret_value_collection_allowed": False,
|
|
"status": "blocked_waiting_owner_response_file",
|
|
"blockers": [],
|
|
}
|
|
if response is None:
|
|
result["blockers"] = ["owner_response_file_missing"]
|
|
return result
|
|
|
|
failures: list[str] = []
|
|
if response.get("schema_version") != EXPECTED_SCHEMA:
|
|
failures.append(f"schema_version={response.get('schema_version')!r}")
|
|
failures.extend(find_forbidden_strings(response))
|
|
failures.extend(find_forbidden_booleans(response))
|
|
|
|
by_gate = response_by_gate(response)
|
|
gate_ids = set(by_gate)
|
|
unknown_gates = sorted(gate_ids - expected_gates)
|
|
missing_gates = sorted(expected_gates - gate_ids)
|
|
if unknown_gates:
|
|
failures.append(f"unknown_gate_ids={unknown_gates}")
|
|
if missing_gates:
|
|
failures.append(f"missing_gate_responses={missing_gates}")
|
|
|
|
received = 0
|
|
accepted = 0
|
|
for gate_id in sorted(expected_gates & gate_ids):
|
|
item = by_gate[gate_id]
|
|
gate_failures = validate_common(gate_id, item)
|
|
if gate_id == "credential_escrow_evidence":
|
|
gate_failures.extend(validate_credential_escrow(item))
|
|
elif gate_id == "wazuh_manager_registry_export":
|
|
gate_failures.extend(validate_wazuh_registry(item))
|
|
else:
|
|
gate_failures.append(f"{gate_id}.unsupported_for_response_preflight")
|
|
if gate_failures:
|
|
failures.extend(gate_failures)
|
|
else:
|
|
received += 1
|
|
if normalized(item.get("decision")).lower() == "accepted":
|
|
accepted += 1
|
|
|
|
result["owner_response_received_count"] = received
|
|
result["owner_response_accepted_count"] = accepted
|
|
result["blockers"] = failures
|
|
if failures:
|
|
result["status"] = "blocked_waiting_owner_response_content"
|
|
elif accepted == len(expected_gates):
|
|
result["status"] = "ready_for_independent_reviewer_acceptance"
|
|
else:
|
|
result["status"] = "blocked_waiting_owner_acceptance"
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
packet = load_owner_packet(args)
|
|
response = load_json(args.response_file, label="response_file") if args.response_file else None
|
|
result = evaluate(packet, response)
|
|
|
|
if args.json:
|
|
print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
|
|
else:
|
|
prefix = (
|
|
"POST_REBOOT_OWNER_RESPONSE_PREFLIGHT_OK"
|
|
if result["status"] == "ready_for_independent_reviewer_acceptance"
|
|
else "POST_REBOOT_OWNER_RESPONSE_PREFLIGHT_BLOCKED"
|
|
)
|
|
print(
|
|
f"{prefix} status={result['status']} "
|
|
f"expected_gates={result['expected_gate_count']} "
|
|
f"received={result['owner_response_received_count']} "
|
|
f"accepted={result['owner_response_accepted_count']} "
|
|
f"runtime_gate={result['runtime_gate_count']} "
|
|
f"blockers={len(result['blockers'])}"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|