tegwick
ee4e901611
Assistant: codex Assistant-Model: gpt-5.6-luna Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
254 lines
10 KiB
Python
254 lines
10 KiB
Python
"""PIP claim + validate join (SECRETS-WP-0007-T04 / SECRETS-WP-0008-T02).
|
|
|
|
These cover the seam that was previously a `return None` stub: the engine now
|
|
reproduces the exact CheckRequest, fetches the durable ActionAuthorization, and
|
|
validates it before offering a consume binding. A half-configured PEP must
|
|
raise rather than look like an unconfigured one.
|
|
"""
|
|
import copy
|
|
import io
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from secrets_engine.approval_claim import (
|
|
binding_from_check_request,
|
|
claim_binding_digest,
|
|
)
|
|
from secrets_engine.approval_consume import resolve_approval_observation, authorize_action
|
|
from tests.authorization_stub import AuthorizationStub
|
|
from secrets_engine.authorization import build_action_request, request_digest
|
|
from secrets_engine.catalog import validate_entry
|
|
from secrets_engine.errors import DecisionError
|
|
from tests.test_catalog import VALID
|
|
|
|
AUTH_ID = "8bfc20be-47a4-4fb0-97a2-bf0a920afad8"
|
|
|
|
|
|
class _Cfg:
|
|
def __init__(self, token_file, **over):
|
|
self.approval_url = "https://approval.example"
|
|
self.approval_token_file = token_file
|
|
self.authorization_subject_id = "user:alice"
|
|
self.authorization_subject_type = "Human"
|
|
self.authorization_policy_package = "secrets-engine.lifecycle"
|
|
self.authorization_policy_version = "v1"
|
|
self.pdp_url = "http://127.0.0.1:1234"
|
|
self.pdp_token_file = token_file
|
|
self.authorization_min_approvals = 2
|
|
for k, v in over.items():
|
|
setattr(self, k, v)
|
|
|
|
|
|
def _entry():
|
|
raw = copy.deepcopy(VALID)
|
|
raw["approval"] = dict(raw.get("approval") or {})
|
|
raw["approval"]["authorization_id"] = AUTH_ID
|
|
raw["approval"]["purpose"] = "contract-test"
|
|
return validate_entry(raw)
|
|
|
|
|
|
def _token(tmp_path):
|
|
f = tmp_path / "approval.token"
|
|
f.write_text("token-value\n")
|
|
f.chmod(0o600)
|
|
return f
|
|
|
|
|
|
def _expected_request(entry, action="deactivate", fields=("api_token",)):
|
|
return build_action_request(
|
|
entry, action,
|
|
subject_id="user:alice", subject_type="Human", purpose="contract-test",
|
|
fields=list(fields),
|
|
policy_targets=[entry.policy_name], auth_targets=[entry.role_name],
|
|
)
|
|
|
|
|
|
def _served(entry=None, action="deactivate", fields=("api_token",), **over):
|
|
"""An approval-engine approval-claim bound to the proposed action."""
|
|
entry = entry or _entry()
|
|
request = _expected_request(entry, action, fields)
|
|
binding = binding_from_check_request(request)
|
|
now = datetime.now(timezone.utc)
|
|
claim = {
|
|
"schema_version": "0.1",
|
|
"kind": "approval-claim",
|
|
"issuer": "approval-engine",
|
|
"approval_id": AUTH_ID,
|
|
"state": "valid",
|
|
"valid_now": True,
|
|
"consumed": False,
|
|
"binding": {
|
|
**binding,
|
|
"digest": claim_binding_digest(**binding),
|
|
# The issuer recorded the PDP digest at issue time. That is the only
|
|
# comparison usable today: the native digest speaks
|
|
# approval-engine's vocabulary and no mapping to ours is published.
|
|
"pdp_digest": request_digest(request),
|
|
"pdp_path": True,
|
|
},
|
|
"freshness": {
|
|
"observed_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"ttl_seconds": 30,
|
|
"not_after": (now + timedelta(seconds=30)).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
},
|
|
"validity": {
|
|
"not_before": (now - timedelta(minutes=5)).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"expires_at": (now + timedelta(minutes=10)).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
},
|
|
"reason_code": "ok",
|
|
}
|
|
claim.update(over)
|
|
return claim
|
|
|
|
|
|
def _opener(envelope, status=200):
|
|
def _open(request, timeout=None):
|
|
body = json.dumps(envelope).encode()
|
|
resp = io.BytesIO(body)
|
|
resp.status = status
|
|
resp.__enter__ = lambda s=resp: s
|
|
resp.__exit__ = lambda s, *a: False
|
|
return resp
|
|
return _open
|
|
|
|
|
|
def _resolve(cfg, entry, envelope, action="deactivate"):
|
|
pdp = AuthorizationStub(approval_id=AUTH_ID, package=cfg.authorization_policy_package, version=cfg.authorization_policy_version)
|
|
def check(request, timeout=None):
|
|
submitted = json.loads(request.data)
|
|
return _opener(pdp.decision(submitted))(request)
|
|
return authorize_action(cfg, entry, action, None, fields=("api_token",),
|
|
policy_targets=(entry.policy_name,), auth_targets=(entry.role_name,),
|
|
opener=_opener(envelope), pdp_opener=check)
|
|
|
|
|
|
def test_unconfigured_serving_path_stays_fail_closed(tmp_path):
|
|
"""No URL/token/authorization id: None, exactly as before the join existed."""
|
|
cfg = _Cfg(None, approval_url="", approval_token_file=None)
|
|
assert resolve_approval_observation(cfg, _entry(), "deactivate", None) is None
|
|
|
|
|
|
def test_valid_authorization_yields_binding_with_canonical_digest(tmp_path):
|
|
entry = _entry()
|
|
cfg = _Cfg(_token(tmp_path))
|
|
binding = _resolve(cfg, entry, _served())
|
|
assert binding is not None
|
|
assert binding.binding.approval_id == AUTH_ID
|
|
assert binding.binding.request_digest != request_digest(_expected_request(entry))
|
|
|
|
|
|
def test_missing_subject_raises_instead_of_returning_none(tmp_path):
|
|
"""Half-configured must not be mistaken for unconfigured."""
|
|
cfg = _Cfg(_token(tmp_path), authorization_subject_id="")
|
|
with pytest.raises(DecisionError, match="SUBJECT_ID"):
|
|
_resolve(cfg, _entry(), _served())
|
|
|
|
|
|
def test_policy_pin_is_not_enforced_on_the_claim_path(tmp_path):
|
|
"""flex-auth: the published example vocabulary is not a live pin."""
|
|
# The pin is a step-2 (DecisionEnvelope) concern after GH-DEC-2026-005 and
|
|
# is asserted in tests/test_action_authorization.py, not on the claim path.
|
|
cfg = _Cfg(_token(tmp_path), authorization_policy_package="")
|
|
assert resolve_approval_observation(cfg, _entry(), "deactivate", None, opener=_opener(_served())) is not None
|
|
|
|
|
|
def test_wrong_field_set_fails_closed(tmp_path):
|
|
claim = _served(fields=("other_field",))
|
|
with pytest.raises(DecisionError, match="pdp digest does not match"):
|
|
_resolve(_Cfg(_token(tmp_path)), _entry(), claim)
|
|
|
|
|
|
def test_action_mismatch_fails_closed(tmp_path):
|
|
"""A destroy must never ride a deactivate authorization."""
|
|
entry = _entry()
|
|
cfg = _Cfg(_token(tmp_path))
|
|
with pytest.raises(DecisionError):
|
|
_resolve(cfg, entry, _served(), action="destroy")
|
|
|
|
|
|
def test_unreachable_approval_engine_fails_closed(tmp_path):
|
|
from urllib.error import URLError
|
|
|
|
def _boom(request, timeout=None):
|
|
raise URLError("no route")
|
|
|
|
with pytest.raises(DecisionError, match="unreachable"):
|
|
resolve_approval_observation(
|
|
_Cfg(_token(tmp_path)), _entry(), "deactivate", None,
|
|
fields=("api_token",), opener=_boom,
|
|
)
|
|
|
|
|
|
def test_superseded_claim_fails_closed(tmp_path):
|
|
with pytest.raises(DecisionError):
|
|
_resolve(_Cfg(_token(tmp_path)), _entry(), _served(valid_now=False, reason_code="superseded"))
|
|
|
|
|
|
def test_claim_without_pdp_path_fails_closed(tmp_path):
|
|
"""pdp_path is a declaration, and it is never inferred from a digest.
|
|
|
|
approval-engine schema v3 refuses to issue pdp_path true without a
|
|
pdp_digest, so a true declaration guarantees the digest. The converse does
|
|
not hold: a digest recorded for some other reason is not a statement that
|
|
this approval was requested against a bound CheckRequest, and approvals
|
|
issued before v3 carry pdp_path false regardless of any digest they hold
|
|
(GH-DEC-2026-008).
|
|
"""
|
|
claim = _served()
|
|
claim["binding"]["pdp_path"] = False
|
|
with pytest.raises(DecisionError, match="does not declare binding.pdp_path"):
|
|
_resolve(_Cfg(_token(tmp_path)), _entry(), claim)
|
|
|
|
|
|
def test_claim_omitting_pdp_path_fails_closed(tmp_path):
|
|
"""An absent declaration is not a true one."""
|
|
claim = _served()
|
|
claim["binding"].pop("pdp_path")
|
|
with pytest.raises(DecisionError, match="does not declare binding.pdp_path"):
|
|
_resolve(_Cfg(_token(tmp_path)), _entry(), claim)
|
|
|
|
|
|
def test_claim_without_pdp_digest_fails_closed_naming_the_missing_mapping(tmp_path):
|
|
"""No published vocabulary mapping means the claim cannot be tied to this action.
|
|
|
|
approval-engine's binding.action/target use their vocabulary
|
|
("secrets.kv.destroy", {"id":..., "stage":...}); ours uses the catalog's.
|
|
flex-auth makes no cross-check and states the correspondence is ours via
|
|
pdp_digest. Without one there is nothing sound to compare, so this must
|
|
refuse rather than fall back to comparing two different languages.
|
|
"""
|
|
claim = _served()
|
|
claim["binding"].pop("pdp_digest")
|
|
with pytest.raises(DecisionError, match="no published mapping"):
|
|
_resolve(_Cfg(_token(tmp_path)), _entry(), claim)
|
|
|
|
|
|
def test_wrong_pdp_digest_fails_closed(tmp_path):
|
|
claim = _served()
|
|
claim["binding"]["pdp_digest"] = "sha256:" + "c" * 64
|
|
with pytest.raises(DecisionError, match="pdp digest does not match"):
|
|
_resolve(_Cfg(_token(tmp_path)), _entry(), claim)
|
|
|
|
|
|
def test_observation_is_not_a_consume_binding(tmp_path):
|
|
claim = _served()
|
|
claim["binding"]["pdp_digest"] = "sha256:" + "e" * 64
|
|
observation = resolve_approval_observation(_Cfg(_token(tmp_path)), _entry(), "deactivate", None,
|
|
opener=_opener(claim))
|
|
assert observation.claim == claim
|
|
assert observation.request["context"]["approval"] == claim
|
|
assert not hasattr(observation, "request_digest")
|
|
assert not hasattr(observation, "decision_id")
|
|
|
|
|
|
def test_claim_that_expires_during_check_cannot_yield_consume_binding(tmp_path, monkeypatch):
|
|
import secrets_engine.approval_consume as consumer
|
|
from secrets_engine.approval_claim import validate_approval_claim
|
|
def after_check(*args, **kwargs):
|
|
return validate_approval_claim(*args, **kwargs, now=datetime.now(timezone.utc) + timedelta(seconds=31))
|
|
monkeypatch.setattr(consumer, "validate_approval_claim", after_check)
|
|
with pytest.raises(DecisionError, match="observation is stale"):
|
|
_resolve(_Cfg(_token(tmp_path)), _entry(), _served())
|