zone-engine/tests/test_zone_exceptions.py

from copy import deepcopy
from datetime import datetime, timezone
import unittest

from tools.check_zone_exceptions import evaluate_exceptions


def policy():
    return {
        "standard": "security-zone-exception-policy_v0.1",
        "policy_id": "ops-warden/security-zone-exceptions",
        "version": "1",
        "controls": {
            "flex-auth/pre-sign": {
                "grant_authorities": ["ops-warden/security-owner"],
                "maximum_duration_seconds": 7200,
            }
        },
    }


def exception_record():
    return {
        "exception_id": "zone-exc-001",
        "security_zone": "z2-protected",
        "control": "flex-auth/pre-sign",
        "workloads": ["issue-core"],
        "base": {"stance": "enforced", "failure_mode": "fail_closed"},
        "relaxation": {"stance": "advisory"},
        "justification": "bounded migration",
        "requested_by": "issue-core",
        "granted_by": "ops-warden/security-owner",
        "issued_at": "2026-08-23T09:00:00Z",
        "not_before": "2026-08-23T10:00:00Z",
        "not_after": "2026-08-23T12:00:00Z",
        "maximum_duration_policy": "ops-warden/security-zone-exceptions@1",
        "change_ref": "ops-warden@abc123",
        "durable_authorities": [
            {"id": "ssh-cert:123", "not_after": "2026-08-23T11:00:00Z"}
        ],
    }


def evaluate(record, at="2026-08-23T10:00:00+00:00"):
    return evaluate_exceptions(
        {
            "standard": "security-zone-exceptions_v0.1",
            "exceptions": [record],
        },
        policy(),
        at=datetime.fromisoformat(at),
    )


class ExceptionConformanceTest(unittest.TestCase):
    def test_not_before_is_inclusive(self):
        result = evaluate(exception_record())
        self.assertTrue(result["ok"])
        self.assertEqual(result["active_exception_ids"], ["zone-exc-001"])
        self.assertEqual(result["results"][0]["state"], "active")

    def test_not_after_is_exclusive(self):
        result = evaluate(exception_record(), "2026-08-23T12:00:00+00:00")
        self.assertTrue(result["ok"])
        self.assertFalse(result["results"][0]["active"])
        self.assertEqual(result["results"][0]["state"], "expired")

    def test_future_record_is_valid_but_inactive(self):
        result = evaluate(exception_record(), "2026-08-23T09:30:00+00:00")
        self.assertTrue(result["ok"])
        self.assertEqual(result["results"][0]["state"], "future")

    def test_wrong_grant_authority_is_invalid_and_inactive(self):
        record = exception_record()
        record["granted_by"] = "workload/self"
        result = evaluate(record)
        self.assertFalse(result["ok"])
        self.assertFalse(result["results"][0]["active"])
        self.assertIn("designated", result["results"][0]["errors"][0])

    def test_duration_beyond_owner_maximum_is_invalid(self):
        record = exception_record()
        record["not_after"] = "2026-08-23T12:00:01Z"
        result = evaluate(record)
        self.assertFalse(result["ok"])
        self.assertTrue(
            any(
                "exceeds owner maximum" in error
                for error in result["results"][0]["errors"]
            )
        )

    def test_wildcard_and_unknown_workloads_are_forbidden(self):
        for workload in ("*", "unknown"):
            with self.subTest(workload=workload):
                record = exception_record()
                record["workloads"] = [workload]
                result = evaluate(record)
                self.assertFalse(result["ok"])
                self.assertIn("exact resolved", result["results"][0]["errors"][0])

    def test_durable_authority_must_not_outlive_exception(self):
        record = exception_record()
        record["durable_authorities"][0]["not_after"] = "2026-08-23T12:00:01Z"
        result = evaluate(record)
        self.assertFalse(result["ok"])
        self.assertIn("outlives", result["results"][0]["errors"][0])

    def test_failure_mode_can_only_relax_closed_to_open(self):
        record = exception_record()
        record["relaxation"] = {"failure_mode": "fail_open"}
        result = evaluate(record)
        self.assertTrue(result["ok"])
        record["base"]["failure_mode"] = "fail_open"
        result = evaluate(record)
        self.assertFalse(result["ok"])
        self.assertIn("fail_closed", result["results"][0]["errors"][0])

    def test_overlapping_grants_for_same_control_and_workload_are_rejected(self):
        first = exception_record()
        second = deepcopy(first)
        second["exception_id"] = "zone-exc-002"
        document = {
            "standard": "security-zone-exceptions_v0.1",
            "exceptions": [first, second],
        }
        result = evaluate_exceptions(
            document,
            policy(),
            at=datetime(2026, 8, 23, 10, tzinfo=timezone.utc),
        )
        self.assertFalse(result["ok"])
        self.assertTrue(all("overlaps" in item["errors"][0] for item in result["results"]))

    def test_renewal_requires_a_new_existing_id(self):
        record = exception_record()
        record["renews"] = record["exception_id"]
        result = evaluate(record)
        self.assertFalse(result["ok"])
        self.assertIn("different existing", result["results"][0]["errors"][0])

    def test_timezone_is_required(self):
        record = exception_record()
        record["not_after"] = "2026-08-23T12:00:00"
        result = evaluate(record)
        self.assertFalse(result["ok"])
        self.assertIn("timezone", result["results"][0]["errors"][0])


if __name__ == "__main__":
    unittest.main()