tegwick
31da1af5e4
An entry stored subject_id, assurance and evidence_ref but nothing about what kind of principal bound the approval, and subject_id is a naming convention rather than a verified claim. /entries is not restricted by principal type — only /consume is — and the approval-engine-operator client holds approval:approve, so a service can supply approver evidence today. Whether it may is gate-house doctrine; that it is legible is ours. Add entries.principal_type, populate it from the verified token, surface it on the object and the audit evidence path (not the claim, which stays least-disclosure), and migrate v3 stores leaving legacy rows null rather than back-filling a claim nobody made. Also corrects two statements in the requirements issued to informed-decision: agent tokens are not barred from approval:approve, and an empty assurance object is accepted rather than refused. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HyybaE7DUXrWYrhbnESCTe Assistant: claude-code Assistant-Model: opus Assistant-Process: 1275879@bnt-lap001 Assistant-Session: eb464208-f821-41b2-bc5a-a6c33d92a8ad
425 lines
13 KiB
Python
425 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import jwt
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
|
|
from approval_engine.api import call
|
|
from approval_engine.auth import Identity, JWTAuthenticator, StaticTokenAuthenticator
|
|
from approval_engine.errors import Unauthenticated
|
|
from tests.conftest import binding, validity
|
|
|
|
|
|
class _Key:
|
|
def __init__(self, key):
|
|
self.key = key
|
|
|
|
|
|
class _JWKS:
|
|
def __init__(self, key):
|
|
self.key = key
|
|
|
|
def get_signing_key_from_jwt(self, _token):
|
|
return _Key(self.key)
|
|
|
|
|
|
def _jwt(private_key, **overrides):
|
|
now = datetime.now(timezone.utc)
|
|
claims = {
|
|
"iss": "https://keycape.example",
|
|
"sub": "service:secrets-engine",
|
|
"aud": "approval-engine",
|
|
"iat": int(now.timestamp()),
|
|
"exp": int((now + timedelta(minutes=5)).timestamp()),
|
|
"tenant": "tenant:coulomb",
|
|
"principal_type": "service",
|
|
"roles": ["secrets-engine"],
|
|
"scope": "approval:read approval:consume",
|
|
"assurance": {"level": "aal1", "methods": ["client_secret"], "source": "key-cape"},
|
|
}
|
|
claims.update(overrides)
|
|
return jwt.encode(claims, private_key, algorithm="RS256", headers={"kid": "test"})
|
|
|
|
|
|
def test_jwt_authenticator_verifies_signature_issuer_audience_and_claims():
|
|
private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
auth = JWTAuthenticator(
|
|
issuer="https://keycape.example",
|
|
audience="approval-engine",
|
|
jwks_url="https://keycape.example/jwks",
|
|
jwks_client=_JWKS(private.public_key()),
|
|
)
|
|
identity = auth.authenticate("Bearer " + _jwt(private))
|
|
assert identity.subject == "service:secrets-engine"
|
|
assert identity.principal_type == "service"
|
|
assert identity.scopes == {"approval:read", "approval:consume"}
|
|
assert identity.evidence_ref.startswith("jwt-sha256:")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"claims",
|
|
[
|
|
{"iss": "https://wrong.example"},
|
|
{"aud": "somewhere-else"},
|
|
{"exp": 1},
|
|
{"scope": ""},
|
|
{"principal_type": "unknown"},
|
|
{"assurance": "not-an-object"},
|
|
],
|
|
)
|
|
def test_jwt_authenticator_fails_closed(claims):
|
|
private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
auth = JWTAuthenticator(
|
|
issuer="https://keycape.example",
|
|
audience="approval-engine",
|
|
jwks_url="https://keycape.example/jwks",
|
|
jwks_client=_JWKS(private.public_key()),
|
|
)
|
|
with pytest.raises(Unauthenticated):
|
|
auth.authenticate("Bearer " + _jwt(private, **claims))
|
|
|
|
|
|
def test_jwt_authenticator_rejects_wrong_signature_and_hs256():
|
|
private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
other = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
auth = JWTAuthenticator(
|
|
issuer="https://keycape.example",
|
|
audience="approval-engine",
|
|
jwks_url="https://keycape.example/jwks",
|
|
jwks_client=_JWKS(private.public_key()),
|
|
)
|
|
with pytest.raises(Unauthenticated):
|
|
auth.authenticate("Bearer " + _jwt(other))
|
|
hs = jwt.encode(
|
|
{
|
|
"iss": "https://keycape.example",
|
|
"sub": "service:secrets-engine",
|
|
"aud": "approval-engine",
|
|
"exp": 2**31 - 1,
|
|
"iat": 1,
|
|
"tenant": "tenant:coulomb",
|
|
"principal_type": "service",
|
|
"roles": ["secrets-engine"],
|
|
"scope": "approval:read approval:consume",
|
|
"assurance": {"level": "aal1"},
|
|
},
|
|
"not-an-rsa-key",
|
|
algorithm="HS256",
|
|
)
|
|
with pytest.raises(Unauthenticated):
|
|
auth.authenticate("Bearer " + hs)
|
|
|
|
|
|
def test_api_requires_scope_and_binds_create_actor(engine):
|
|
identity = Identity(
|
|
subject="service:creator",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="service",
|
|
tenant="tenant:platform",
|
|
roles=frozenset(),
|
|
scopes=frozenset({"approval:create"}),
|
|
assurance={"level": "aal1"},
|
|
evidence_ref="test",
|
|
)
|
|
from approval_engine.api import App
|
|
|
|
app = App(engine, StaticTokenAuthenticator({"creator": identity}))
|
|
status, body = call(
|
|
app,
|
|
"POST",
|
|
"/v1/approvals",
|
|
{"binding": binding(), "validity": validity()},
|
|
authorization="Bearer creator",
|
|
)
|
|
assert status == 403
|
|
assert body["error"] == "forbidden"
|
|
|
|
exact = binding(actor="service:creator")
|
|
status, created = call(
|
|
app,
|
|
"POST",
|
|
"/v1/approvals",
|
|
{"binding": exact, "validity": validity()},
|
|
authorization="Bearer creator",
|
|
)
|
|
assert status == 201
|
|
status, body = call(
|
|
app, "GET", f"/v1/approvals/{created['id']}/claim", authorization="Bearer creator"
|
|
)
|
|
assert status == 403
|
|
assert body["error"] == "forbidden"
|
|
|
|
|
|
def test_approval_entry_uses_verified_identity_not_body(app):
|
|
_, created = call(
|
|
app,
|
|
"POST",
|
|
"/v1/approvals",
|
|
{"binding": binding(), "validity": validity()},
|
|
)
|
|
_, approved = call(
|
|
app,
|
|
"POST",
|
|
f"/v1/approvals/{created['id']}/entries",
|
|
{
|
|
"subject_id": "user:spoofed",
|
|
"assurance": "spoofed",
|
|
"evidence_ref": "spoofed",
|
|
},
|
|
)
|
|
assert approved["entries"][0]["subject_id"] == "agt-secrets-engine"
|
|
assert approved["entries"][0]["evidence_ref"] == "test-identity:test-token"
|
|
assert "spoofed" not in approved["entries"][0]["assurance"]
|
|
|
|
|
|
def test_missing_token_is_unauthenticated(app):
|
|
status, body = call(app, "GET", "/v1/cadence", authorization=None)
|
|
assert status == 401
|
|
assert body["error"] == "unauthenticated"
|
|
|
|
|
|
def test_wrong_tenant_is_forbidden(engine):
|
|
identity = Identity(
|
|
subject="agt-secrets-engine",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="service",
|
|
tenant="another-tenant",
|
|
roles=frozenset(),
|
|
scopes=frozenset({"approval:observe"}),
|
|
assurance={},
|
|
evidence_ref="test",
|
|
)
|
|
from approval_engine.api import App
|
|
|
|
app = App(engine, StaticTokenAuthenticator({"wrong": identity}))
|
|
status, body = call(app, "GET", "/v1/cadence", authorization="Bearer wrong")
|
|
assert status == 403
|
|
assert body["error"] == "forbidden"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"tenant",
|
|
[
|
|
"platform",
|
|
"tenant:coulomb",
|
|
"TENANT:PLATFORM",
|
|
"Tenant:Platform",
|
|
"tenant:platform ",
|
|
" tenant:platform",
|
|
"tenant:platform:",
|
|
"",
|
|
],
|
|
)
|
|
def test_near_miss_tenant_spellings_are_forbidden(engine, tenant):
|
|
"""Decision 5ed3fb35 accepted exactly `tenant:platform` with no alias.
|
|
|
|
The store comparison is exact string equality, so every near miss below must
|
|
be refused: the bare `platform` this repo used to serve, the `tenant:coulomb`
|
|
the registrations used to request, case variants, and whitespace. Varying the
|
|
field is the point — a suite whose fixtures all carry the sanctioned value
|
|
proves nothing about the tenant check.
|
|
"""
|
|
identity = Identity(
|
|
subject="agt-secrets-engine",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="service",
|
|
tenant=tenant,
|
|
roles=frozenset(),
|
|
scopes=frozenset({"approval:observe"}),
|
|
assurance={},
|
|
evidence_ref="test",
|
|
)
|
|
from approval_engine.api import App
|
|
|
|
app = App(engine, StaticTokenAuthenticator({"near": identity}))
|
|
status, body = call(app, "GET", "/v1/cadence", authorization="Bearer near")
|
|
assert status == 403, f"{tenant!r} was admitted as an alias"
|
|
assert body["error"] == "forbidden"
|
|
|
|
|
|
def test_exact_sanctioned_tenant_is_admitted(engine):
|
|
"""The other half of the pin: the exact spelling must actually work.
|
|
|
|
Without this, a bug that rejected every tenant would pass the near-miss test
|
|
above while denying the sanctioned caller too.
|
|
"""
|
|
assert engine.tenant == "tenant:platform"
|
|
identity = Identity(
|
|
subject="agt-secrets-engine",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="service",
|
|
tenant="tenant:platform",
|
|
roles=frozenset(),
|
|
scopes=frozenset({"approval:observe"}),
|
|
assurance={},
|
|
evidence_ref="test",
|
|
)
|
|
from approval_engine.api import App
|
|
|
|
app = App(engine, StaticTokenAuthenticator({"exact": identity}))
|
|
status, _ = call(app, "GET", "/v1/cadence", authorization="Bearer exact")
|
|
assert status == 200
|
|
|
|
|
|
def test_deny_all_default_does_not_mutate(engine):
|
|
from approval_engine.api import App
|
|
|
|
app = App(engine)
|
|
status, body = call(
|
|
app,
|
|
"POST",
|
|
"/v1/approvals",
|
|
{"binding": binding(), "validity": validity()},
|
|
authorization="Bearer anything",
|
|
)
|
|
assert status == 401
|
|
assert body["error"] == "unauthenticated"
|
|
assert engine.transition_counts()["issuance"] == 0
|
|
|
|
|
|
def test_human_principal_cannot_consume(engine):
|
|
from approval_engine.api import App
|
|
|
|
service = Identity(
|
|
subject="agt-secrets-engine",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="service",
|
|
tenant="tenant:platform",
|
|
roles=frozenset(),
|
|
scopes=frozenset(
|
|
{"approval:create", "approval:approve", "approval:read", "approval:consume"}
|
|
),
|
|
assurance={"level": "aal1"},
|
|
evidence_ref="service",
|
|
)
|
|
human = Identity(
|
|
subject="user:alice",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="human",
|
|
tenant="tenant:platform",
|
|
roles=frozenset(),
|
|
scopes=frozenset({"approval:consume"}),
|
|
assurance={"level": "aal2"},
|
|
evidence_ref="human",
|
|
)
|
|
app = App(
|
|
engine,
|
|
StaticTokenAuthenticator({"service": service, "human": human}),
|
|
)
|
|
_, created = call(
|
|
app,
|
|
"POST",
|
|
"/v1/approvals",
|
|
{"binding": binding(), "validity": validity()},
|
|
authorization="Bearer service",
|
|
)
|
|
call(
|
|
app,
|
|
"POST",
|
|
f"/v1/approvals/{created['id']}/entries",
|
|
{},
|
|
authorization="Bearer service",
|
|
)
|
|
status, body = call(
|
|
app,
|
|
"POST",
|
|
f"/v1/approvals/{created['id']}/consume",
|
|
{"request_digest": "sha256:" + "ab" * 32},
|
|
authorization="Bearer human",
|
|
)
|
|
assert status == 403
|
|
assert body["error"] == "forbidden"
|
|
status, claim = call(
|
|
app,
|
|
"GET",
|
|
f"/v1/approvals/{created['id']}/claim",
|
|
authorization="Bearer service",
|
|
)
|
|
assert status == 200
|
|
assert claim["consumed"] is False
|
|
assert claim["valid_now"] is True
|
|
|
|
|
|
def test_entry_records_the_verified_principal_type(engine):
|
|
"""Approver evidence must say what kind of principal bound the approval.
|
|
|
|
`subject_id` alone cannot answer it: `user:alice` is a naming convention,
|
|
not a verified claim. The recorded value comes from the token and from
|
|
nowhere else, so an evidence reader can tell a human bind from a service
|
|
one without trusting a string's shape.
|
|
"""
|
|
from approval_engine.api import App
|
|
|
|
creator = Identity(
|
|
subject="agt-secrets-engine",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="service",
|
|
tenant="tenant:platform",
|
|
roles=frozenset(),
|
|
scopes=frozenset({"approval:create", "approval:read"}),
|
|
assurance={"level": "aal1"},
|
|
evidence_ref="service",
|
|
)
|
|
human = Identity(
|
|
subject="user:alice",
|
|
issuer="test",
|
|
audiences=("approval-engine",),
|
|
principal_type="human",
|
|
tenant="tenant:platform",
|
|
roles=frozenset(),
|
|
scopes=frozenset({"approval:approve"}),
|
|
assurance={"level": "aal2", "amr": ["pwd", "otp"]},
|
|
evidence_ref="human",
|
|
)
|
|
app = App(engine, StaticTokenAuthenticator({"creator": creator, "human": human}))
|
|
_, created = call(
|
|
app,
|
|
"POST",
|
|
"/v1/approvals",
|
|
{"binding": binding(), "validity": validity()},
|
|
authorization="Bearer creator",
|
|
)
|
|
_, approved = call(
|
|
app,
|
|
"POST",
|
|
f"/v1/approvals/{created['id']}/entries",
|
|
{"principal_type": "service"},
|
|
authorization="Bearer human",
|
|
)
|
|
entry = approved["entries"][0]
|
|
assert entry["subject_id"] == "user:alice"
|
|
assert entry["principal_type"] == "human"
|
|
|
|
_, claim = call(
|
|
app, "GET", f"/v1/approvals/{created['id']}/claim", authorization="Bearer creator"
|
|
)
|
|
assert "approvers" not in claim
|
|
|
|
|
|
def test_non_human_approver_is_recorded_as_such(app):
|
|
"""A service principal holding `approval:approve` is not refused — but it
|
|
is not silently indistinguishable from a human either.
|
|
|
|
Whether a non-human may supply approver evidence at all is approval
|
|
doctrine and belongs to gate-house; the `approval-engine-operator`
|
|
registration holds `approval:approve` today. This engine's obligation is
|
|
that the evidence chain records which it was.
|
|
"""
|
|
_, created = call(
|
|
app,
|
|
"POST",
|
|
"/v1/approvals",
|
|
{"binding": binding(), "validity": validity()},
|
|
)
|
|
_, approved = call(app, "POST", f"/v1/approvals/{created['id']}/entries", {})
|
|
assert approved["status"] == "approved"
|
|
assert approved["entries"][0]["principal_type"] == "service"
|