approval-engine/tests/test_auth.py

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},
    ],
)
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_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"