Use bounded Railiance time for protected validity checks
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 2s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a09cbb-87c6-7900-a145-4ce53ba9f1a6
This commit is contained in:
tegwick 2026-09-15 22:16:31 +02:00
parent 5841d8e88a
commit f4b4dd6b17
8 changed files with 514 additions and 14 deletions

View file

@ -18,6 +18,7 @@ from datetime import datetime, timezone
from typing import Any
from secrets_engine.errors import DecisionError
from secrets_engine.application_time import validity_bounds
SCHEMA_VERSION = "0.1"
KIND = "approval-claim"
@ -104,6 +105,7 @@ def _validate_observation(
approval_id: str,
require_human_control: bool = False,
now: datetime | None = None,
time_window=None,
) -> dict[str, Any]:
"""Validate issuer, shape, validity and freshness; no action correspondence."""
if not isinstance(claim, dict):
@ -138,34 +140,34 @@ def _validate_observation(
raise DecisionError("approval claim binding must be an object")
if require_human_control and binding.get("human_control") is not True:
raise DecisionError("approval claim does not declare binding.human_control true; request a human-controlled approval at issue")
current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
lower, upper = validity_bounds(now, time_window)
freshness = claim.get("freshness")
if not isinstance(freshness, dict):
raise DecisionError("approval claim freshness must be an object")
if _parse_time(freshness.get("not_after"), "freshness.not_after") <= current:
if _parse_time(freshness.get("not_after"), "freshness.not_after") <= upper:
raise DecisionError("approval claim observation is stale; re-fetch")
validity = claim.get("validity")
if not isinstance(validity, dict):
raise DecisionError("approval claim validity must be an object")
if _parse_time(validity.get("expires_at"), "validity.expires_at") <= current:
if _parse_time(validity.get("expires_at"), "validity.expires_at") <= upper:
raise DecisionError("approval claim validity window has expired")
if validity.get("not_before") is not None:
if _parse_time(validity.get("not_before"), "validity.not_before") > current:
if _parse_time(validity.get("not_before"), "validity.not_before") > lower:
raise DecisionError("approval claim is not yet valid")
return claim
def observe_pdp_approval_claim(claim: object, *, approval_id: str,
require_human_control: bool = False,
now: datetime | None = None) -> dict[str, Any]:
now: datetime | None = None, time_window=None) -> dict[str, Any]:
"""Validate a fresh fact for submission to the PDP, not authority to consume.
Action correspondence cannot be established until the evaluator returns its
enriched approval binding. This function deliberately makes no such claim.
"""
observed = _validate_observation(claim, approval_id=approval_id,
require_human_control=require_human_control, now=now)
require_human_control=require_human_control, now=now, time_window=time_window)
binding = observed["binding"]
if binding.get("pdp_path") is not True:
raise DecisionError("approval claim does not declare binding.pdp_path; request an approval bound at issue")
@ -178,17 +180,17 @@ def observe_pdp_approval_claim(claim: object, *, approval_id: str,
def validate_approval_claim(claim: object, *, approval_id: str,
expected_binding_digest: str = "", expected_pdp_digest: str = "",
require_human_control: bool = False,
now: datetime | None = None) -> dict[str, Any]:
now: datetime | None = None, time_window=None) -> dict[str, Any]:
"""Validate the fact and compare an independently supplied binding.
A PDP digest supplied here must come from the evaluated decision, never a
local reconstruction of the unenriched request.
"""
observed = _validate_observation(claim, approval_id=approval_id,
require_human_control=require_human_control, now=now)
require_human_control=require_human_control, now=now, time_window=time_window)
if expected_pdp_digest:
observe_pdp_approval_claim(observed, approval_id=approval_id,
require_human_control=require_human_control, now=now)
require_human_control=require_human_control, now=now, time_window=time_window)
if observed["binding"]["pdp_digest"] != expected_pdp_digest:
raise DecisionError("approval claim pdp digest does not match the request")
elif expected_binding_digest: