tegwick
be29c28100
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0291a-1e87-7151-9934-fcbfe3f65eb1
358 lines
13 KiB
Python
358 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate security-zone exception records at an explicit instant."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
import yaml
|
|
|
|
try:
|
|
from tools.resolve_zones import ZONE_FLOOR
|
|
except ModuleNotFoundError: # direct ``python tools/...`` execution
|
|
from resolve_zones import ZONE_FLOOR
|
|
|
|
|
|
POLICY_STANDARD = "security-zone-exception-policy_v0.1"
|
|
RECORD_STANDARD = "security-zone-exceptions_v0.1"
|
|
|
|
|
|
class ExceptionConformanceError(ValueError):
|
|
"""The exception input or policy is structurally unusable."""
|
|
|
|
|
|
def _required(mapping: Mapping[str, Any], key: str, where: str) -> Any:
|
|
value = mapping.get(key)
|
|
if value is None or value == "" or value == []:
|
|
raise ExceptionConformanceError(f"{where}.{key} is required")
|
|
return value
|
|
|
|
|
|
def _instant(value: Any, where: str) -> datetime:
|
|
text = str(value)
|
|
if text.endswith("Z"):
|
|
text = text[:-1] + "+00:00"
|
|
try:
|
|
parsed = datetime.fromisoformat(text)
|
|
except ValueError as exc:
|
|
raise ExceptionConformanceError(f"{where} must be an ISO timestamp") from exc
|
|
if parsed.tzinfo is None:
|
|
raise ExceptionConformanceError(f"{where} must include a timezone")
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def _policy(policy: Any) -> dict[str, Any]:
|
|
if not isinstance(policy, Mapping):
|
|
raise ExceptionConformanceError("exception policy must be a mapping")
|
|
if policy.get("standard") != POLICY_STANDARD:
|
|
raise ExceptionConformanceError(
|
|
f"exception policy standard must be {POLICY_STANDARD}"
|
|
)
|
|
policy_id = str(_required(policy, "policy_id", "policy"))
|
|
version = str(_required(policy, "version", "policy"))
|
|
controls = _required(policy, "controls", "policy")
|
|
if not isinstance(controls, Mapping) or not controls:
|
|
raise ExceptionConformanceError("policy.controls must be a non-empty mapping")
|
|
normalized: dict[str, Any] = {}
|
|
for control_id, control in controls.items():
|
|
where = f"policy.controls.{control_id}"
|
|
if not isinstance(control, Mapping):
|
|
raise ExceptionConformanceError(f"{where} must be a mapping")
|
|
if "/" not in str(control_id):
|
|
raise ExceptionConformanceError(f"{where} id must be owner-qualified")
|
|
authorities = _required(control, "grant_authorities", where)
|
|
if not isinstance(authorities, list) or not all(
|
|
isinstance(authority, str) and authority for authority in authorities
|
|
):
|
|
raise ExceptionConformanceError(
|
|
f"{where}.grant_authorities must be a non-empty string list"
|
|
)
|
|
maximum = _required(control, "maximum_duration_seconds", where)
|
|
if not isinstance(maximum, int) or maximum <= 0:
|
|
raise ExceptionConformanceError(
|
|
f"{where}.maximum_duration_seconds must be a positive integer"
|
|
)
|
|
normalized[str(control_id)] = {
|
|
"grant_authorities": set(authorities),
|
|
"maximum_duration_seconds": maximum,
|
|
}
|
|
return {
|
|
"policy_id": policy_id,
|
|
"version": version,
|
|
"policy_ref": f"{policy_id}@{version}",
|
|
"controls": normalized,
|
|
}
|
|
|
|
|
|
def _validate_relaxation(record: Mapping[str, Any], errors: list[str]) -> None:
|
|
base = record.get("base")
|
|
relaxation = record.get("relaxation")
|
|
if not isinstance(base, Mapping) or not isinstance(relaxation, Mapping):
|
|
errors.append("base and relaxation must be mappings")
|
|
return
|
|
changed = False
|
|
if "stance" in relaxation:
|
|
base_stance = base.get("stance")
|
|
relaxed_stance = relaxation.get("stance")
|
|
allowed = {
|
|
"enforced": {"advisory", "exempt"},
|
|
"advisory": {"exempt"},
|
|
}
|
|
if relaxed_stance not in allowed.get(base_stance, set()):
|
|
errors.append("relaxation.stance must strictly relax the base stance")
|
|
else:
|
|
changed = True
|
|
if "failure_mode" in relaxation:
|
|
if base.get("failure_mode") != "fail_closed" or relaxation.get(
|
|
"failure_mode"
|
|
) != "fail_open":
|
|
errors.append(
|
|
"failure-mode relaxation must change fail_closed to fail_open"
|
|
)
|
|
else:
|
|
changed = True
|
|
if not changed and not errors:
|
|
errors.append("relaxation must change stance or failure_mode")
|
|
|
|
|
|
def _record_result(
|
|
record: Any,
|
|
policy: Mapping[str, Any],
|
|
at: datetime,
|
|
) -> dict[str, Any]:
|
|
errors: list[str] = []
|
|
if not isinstance(record, Mapping):
|
|
return {
|
|
"exception_id": None,
|
|
"valid": False,
|
|
"active": False,
|
|
"state": "invalid",
|
|
"errors": ["exception record must be a mapping"],
|
|
}
|
|
exception_id = record.get("exception_id")
|
|
for key in (
|
|
"exception_id",
|
|
"security_zone",
|
|
"control",
|
|
"workloads",
|
|
"base",
|
|
"relaxation",
|
|
"justification",
|
|
"requested_by",
|
|
"granted_by",
|
|
"issued_at",
|
|
"not_before",
|
|
"not_after",
|
|
"maximum_duration_policy",
|
|
"change_ref",
|
|
):
|
|
value = record.get(key)
|
|
if value is None or value == "" or value == () or value == []:
|
|
errors.append(f"{key} is required")
|
|
zone = record.get("security_zone")
|
|
if zone not in ZONE_FLOOR:
|
|
errors.append("security_zone must be a named zone")
|
|
control_id = record.get("control")
|
|
control_policy = policy["controls"].get(control_id)
|
|
if control_policy is None:
|
|
errors.append("control is absent from the owner exception policy")
|
|
workloads = record.get("workloads")
|
|
if not isinstance(workloads, list) or not workloads:
|
|
errors.append("workloads must be a non-empty list")
|
|
workloads = []
|
|
elif any(
|
|
not isinstance(workload, str)
|
|
or not workload
|
|
or workload in {"*", "unknown"}
|
|
for workload in workloads
|
|
):
|
|
errors.append("workloads must contain exact resolved workload ids")
|
|
elif len(set(workloads)) != len(workloads):
|
|
errors.append("workloads must not contain duplicates")
|
|
_validate_relaxation(record, errors)
|
|
|
|
issued = before = after = None
|
|
for key in ("issued_at", "not_before", "not_after"):
|
|
try:
|
|
parsed = _instant(record.get(key), key)
|
|
if key == "issued_at":
|
|
issued = parsed
|
|
elif key == "not_before":
|
|
before = parsed
|
|
else:
|
|
after = parsed
|
|
except ExceptionConformanceError as exc:
|
|
errors.append(str(exc))
|
|
if issued and before and after:
|
|
if issued > before:
|
|
errors.append("issued_at must be at or before not_before")
|
|
if before >= after:
|
|
errors.append("not_before must be before not_after")
|
|
if control_policy and (after - before).total_seconds() > control_policy[
|
|
"maximum_duration_seconds"
|
|
]:
|
|
errors.append("exception duration exceeds owner maximum")
|
|
if control_policy and record.get("granted_by") not in control_policy[
|
|
"grant_authorities"
|
|
]:
|
|
errors.append("granted_by is not a designated control authority")
|
|
if record.get("maximum_duration_policy") != policy["policy_ref"]:
|
|
errors.append("maximum_duration_policy does not match evaluated owner policy")
|
|
|
|
durable = record.get("durable_authorities") or []
|
|
if not isinstance(durable, list):
|
|
errors.append("durable_authorities must be a list")
|
|
else:
|
|
for index, authority in enumerate(durable):
|
|
if not isinstance(authority, Mapping) or not authority.get("id"):
|
|
errors.append(f"durable_authorities[{index}] requires id and not_after")
|
|
continue
|
|
try:
|
|
authority_after = _instant(
|
|
authority.get("not_after"),
|
|
f"durable_authorities[{index}].not_after",
|
|
)
|
|
except ExceptionConformanceError as exc:
|
|
errors.append(str(exc))
|
|
continue
|
|
if after and authority_after > after:
|
|
errors.append(
|
|
f"durable_authorities[{index}] outlives the exception"
|
|
)
|
|
valid = not errors
|
|
active = bool(valid and before and after and before <= at < after)
|
|
if not valid:
|
|
state = "invalid"
|
|
elif at < before:
|
|
state = "future"
|
|
elif at >= after:
|
|
state = "expired"
|
|
else:
|
|
state = "active"
|
|
return {
|
|
"exception_id": exception_id,
|
|
"control": control_id,
|
|
"workloads": sorted(workloads),
|
|
"not_before": before.isoformat() if before else None,
|
|
"not_after": after.isoformat() if after else None,
|
|
"valid": valid,
|
|
"active": active,
|
|
"state": state,
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
def _overlap(left: Mapping[str, Any], right: Mapping[str, Any]) -> bool:
|
|
if left.get("control") != right.get("control"):
|
|
return False
|
|
if not set(left.get("workloads", [])).intersection(right.get("workloads", [])):
|
|
return False
|
|
if not all((left.get("not_before"), left.get("not_after"), right.get("not_before"), right.get("not_after"))):
|
|
return False
|
|
left_before = _instant(left["not_before"], "left.not_before")
|
|
left_after = _instant(left["not_after"], "left.not_after")
|
|
right_before = _instant(right["not_before"], "right.not_before")
|
|
right_after = _instant(right["not_after"], "right.not_after")
|
|
return max(left_before, right_before) < min(left_after, right_after)
|
|
|
|
|
|
def evaluate_exceptions(
|
|
document: Any,
|
|
policy_document: Any,
|
|
*,
|
|
at: datetime,
|
|
) -> dict[str, Any]:
|
|
policy = _policy(policy_document)
|
|
if not isinstance(document, Mapping) or document.get("standard") != RECORD_STANDARD:
|
|
raise ExceptionConformanceError(
|
|
f"exception document standard must be {RECORD_STANDARD}"
|
|
)
|
|
records = document.get("exceptions")
|
|
if not isinstance(records, list):
|
|
raise ExceptionConformanceError("exceptions must be a list")
|
|
results = [_record_result(record, policy, at) for record in records]
|
|
ids: dict[str, list[int]] = {}
|
|
for index, result in enumerate(results):
|
|
if result["exception_id"]:
|
|
ids.setdefault(str(result["exception_id"]), []).append(index)
|
|
for exception_id, indexes in ids.items():
|
|
if len(indexes) > 1:
|
|
for index in indexes:
|
|
results[index]["errors"].append(
|
|
f"duplicate exception_id {exception_id}"
|
|
)
|
|
results[index].update(valid=False, active=False, state="invalid")
|
|
source_by_id = {
|
|
str(record.get("exception_id")): record
|
|
for record in records
|
|
if isinstance(record, Mapping) and record.get("exception_id")
|
|
}
|
|
for index, record in enumerate(records):
|
|
if not isinstance(record, Mapping) or not record.get("renews"):
|
|
continue
|
|
renewed = str(record["renews"])
|
|
if renewed == str(record.get("exception_id")) or renewed not in source_by_id:
|
|
results[index]["errors"].append(
|
|
"renews must name a different existing exception id"
|
|
)
|
|
results[index].update(valid=False, active=False, state="invalid")
|
|
for left in range(len(records)):
|
|
for right in range(left + 1, len(records)):
|
|
if not results[left]["valid"] or not results[right]["valid"]:
|
|
continue
|
|
try:
|
|
overlapping = _overlap(records[left], records[right])
|
|
except ExceptionConformanceError:
|
|
overlapping = False
|
|
if overlapping:
|
|
for index in (left, right):
|
|
results[index]["errors"].append(
|
|
f"overlaps exception {results[right if index == left else left]['exception_id']}"
|
|
)
|
|
results[index].update(valid=False, active=False, state="invalid")
|
|
return {
|
|
"ok": all(result["valid"] for result in results),
|
|
"standard": RECORD_STANDARD,
|
|
"evaluated_at": at.astimezone(timezone.utc).isoformat(),
|
|
"policy": {"id": policy["policy_id"], "version": policy["version"]},
|
|
"results": sorted(results, key=lambda result: str(result["exception_id"])),
|
|
"active_exception_ids": sorted(
|
|
str(result["exception_id"])
|
|
for result in results
|
|
if result["active"]
|
|
),
|
|
}
|
|
|
|
|
|
def _load(path: Path) -> dict[str, Any]:
|
|
value = yaml.safe_load(path.read_text()) or {}
|
|
if not isinstance(value, dict):
|
|
raise ExceptionConformanceError(f"{path} must contain a mapping")
|
|
return value
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("exceptions", type=Path)
|
|
parser.add_argument("--policy", required=True, type=Path)
|
|
parser.add_argument("--at", required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
result = evaluate_exceptions(
|
|
_load(args.exceptions),
|
|
_load(args.policy),
|
|
at=_instant(args.at, "--at"),
|
|
)
|
|
except (OSError, yaml.YAMLError, ExceptionConformanceError) as exc:
|
|
result = {"ok": False, "errors": [str(exc)], "results": []}
|
|
print(json.dumps(result, indent=2, sort_keys=True))
|
|
return 0 if result["ok"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|