118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
TEMPLATE_SCRIPT = ROOT / "scripts" / "reboot-recovery" / "post-reboot-owner-response-template.py"
|
|
PREFLIGHT_SCRIPT = ROOT / "scripts" / "reboot-recovery" / "post-reboot-owner-response-preflight.py"
|
|
|
|
|
|
ESCROW_ITEMS = [
|
|
"restic_repository_password",
|
|
"offsite_provider_credentials",
|
|
"break_glass_admin_credentials",
|
|
"dns_registrar_recovery",
|
|
"oauth_ai_provider_recovery",
|
|
]
|
|
|
|
|
|
def write_packet(tmp_path: Path, gate_ids: list[str]) -> Path:
|
|
owner_packets = []
|
|
for gate_id in gate_ids:
|
|
packet = {
|
|
"packet_id": gate_id,
|
|
"title": f"{gate_id} owner evidence",
|
|
"priority": "P0",
|
|
"required_items": ESCROW_ITEMS if gate_id == "credential_escrow_evidence" else [],
|
|
}
|
|
owner_packets.append(packet)
|
|
|
|
packet_path = tmp_path / "owner-packets.json"
|
|
packet_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": "awoooi_post_reboot_next_gate_owner_packets_v1",
|
|
"source": {"next_required_gates": gate_ids},
|
|
"owner_packets": owner_packets,
|
|
},
|
|
indent=2,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return packet_path
|
|
|
|
|
|
def run_template(packet_path: Path, output_path: Path | None = None) -> dict:
|
|
cmd = [
|
|
sys.executable,
|
|
str(TEMPLATE_SCRIPT),
|
|
"--owner-packet-file",
|
|
str(packet_path),
|
|
]
|
|
if output_path:
|
|
cmd.extend(["--output", str(output_path)])
|
|
|
|
result = subprocess.run(cmd, text=True, capture_output=True, check=True)
|
|
if output_path:
|
|
return json.loads(output_path.read_text(encoding="utf-8"))
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def test_generates_only_active_credential_gate(tmp_path: Path) -> None:
|
|
packet_path = write_packet(tmp_path, ["credential_escrow_evidence"])
|
|
template = run_template(packet_path)
|
|
|
|
assert template["schema_version"] == "awoooi_post_reboot_next_gate_owner_response_v1"
|
|
assert template["active_gates"] == ["credential_escrow_evidence"]
|
|
assert [item["gate_id"] for item in template["responses"]] == ["credential_escrow_evidence"]
|
|
assert "wazuh_manager_registry_export" not in json.dumps(template)
|
|
|
|
response = template["responses"][0]
|
|
assert response["runtime_action_requested"] is False
|
|
assert response["runtime_action_authorized"] is False
|
|
assert response["host_write_requested"] is False
|
|
assert response["host_write_authorized"] is False
|
|
assert response["secret_value_included"] is False
|
|
assert response["secret_value_collection_allowed"] is False
|
|
assert response["credential_marker_write_requested"] is False
|
|
assert response["credential_marker_write_authorized"] is False
|
|
assert [item["item_id"] for item in response["escrow_items"]] == ESCROW_ITEMS
|
|
assert all(item["contains_secret_value"] is False for item in response["escrow_items"])
|
|
|
|
|
|
def test_placeholder_template_fails_closed_against_current_preflight(tmp_path: Path) -> None:
|
|
packet_path = write_packet(tmp_path, ["credential_escrow_evidence"])
|
|
response_path = tmp_path / "owner-response-template.json"
|
|
run_template(packet_path, output_path=response_path)
|
|
|
|
result = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(PREFLIGHT_SCRIPT),
|
|
"--owner-packet-file",
|
|
str(packet_path),
|
|
"--response-file",
|
|
str(response_path),
|
|
"--json",
|
|
],
|
|
text=True,
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
preflight = json.loads(result.stdout)
|
|
|
|
assert preflight["status"] == "blocked_waiting_owner_response_content"
|
|
assert preflight["expected_gate_count"] == 1
|
|
assert preflight["expected_gates"] == ["credential_escrow_evidence"]
|
|
assert preflight["owner_response_received_count"] == 0
|
|
assert preflight["owner_response_accepted_count"] == 0
|
|
assert preflight["runtime_gate_count"] == 0
|
|
assert preflight["runtime_action_authorized"] is False
|
|
assert not any("unknown_gate_ids" in blocker for blocker in preflight["blockers"])
|
|
assert any("credential_escrow_evidence.owner_role_missing" in blocker for blocker in preflight["blockers"])
|