tegwick
2370f69927
Close remaining in-repo APPROVAL-WP-0002 gaps: drive GH-DEC-2026-003 against the real HTTP surface, fail closed on JWT/human-consume/static-token paths, treat audit 200 duplicates as drained, and ask KeyCape for the production audience and client grants. Assistant: grok Assistant-Session: 01a06253-e557-7971-93d9-4f4c2cfbf455
282 lines
8.4 KiB
Python
282 lines
8.4 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="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"
|
|
|
|
|
|
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="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="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
|