test(P05): E2E integration test — sign-up→sign-in→token-vend→apply→audit (REQ-348, security-engineer)
---ci--- project: acdl phase: 5 milestone: v1.28 status: execute persona: security-engineer --- Add tests/test_e2e_idp.py — the J1+J2 happy-path E2E flow. Uses moto for DynamoDB (4 IdP tables) + mock KMS (test ECC keypair). Asserts: (a) sign_up succeeds, (b) sign_in returns a session, (c) token-vend returns a KMS-signed OIDC token, (d) the OIDC token verifies with the JWKS key (pyjwt), (e) nova apply --local produces a JWS attestation (HS256), (f) the JWS verifies with the PAT-derived key (+ tamper detection), (g) the audit chain is complete + linked (auth.sign_up, auth.sign_in, auth.session_created, pat.issued, token.vend.allowed — all present, linked by user_id/jti, no raw password/PAT leaked INV-16). Also: the credentials file stores the OIDC token not the raw PAT (C-7.3), the DDB user item has a password_hash not the raw password, the DDB PAT row has a pat_hash not the raw PAT. Negative path: revocation breaks the chain (403 pat_revoked, D-229 strong-read SLO, token.vend.denied audit event).
This commit is contained in:
Jon Chery
@@ -0,0 +1,520 @@
|
||||
"""E2E integration test — sign-up → sign-in → token-vend → apply → audit
|
||||
(REQ-348, J1+J2 happy path combined).
|
||||
|
||||
This is the P5 Wave 2 integration test. It exercises the full Nova-idp
|
||||
identity chain end-to-end against moto (DynamoDB) + a mock KMS (a test
|
||||
ECC keypair). In CI against a deployed Nova-idp it would hit the real
|
||||
Lambdas; locally it uses direct function calls (the dual-use
|
||||
``dispatch_action`` / ``vend_token`` entry points, REQ-329).
|
||||
|
||||
The flow (REQ-348):
|
||||
|
||||
1. sign_up(email, password) → user in nova-users (Argon2id hash)
|
||||
2. sign_in(email, password) → session_id in nova-sessions
|
||||
3. issue a PAT (pat_lifecycle.issue_pat) → raw PAT returned once
|
||||
4. nova auth login (token-vend) → KMS-signed OIDC token
|
||||
5. verify the OIDC token against the JWKS key (pyjwt)
|
||||
6. nova apply --local --sign-local-review → JWS attestation (HS256)
|
||||
7. verify the JWS attestation with the PAT-derived key
|
||||
8. assert the audit chain is complete + linked
|
||||
|
||||
Asserts (a)–(g) from the task spec are mapped to the test methods below.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
os.environ.setdefault("AWS_DEFAULT_REGION", "us-east-1")
|
||||
os.environ.setdefault("AWS_ACCESS_KEY_ID", "test")
|
||||
os.environ.setdefault("AWS_SECRET_ACCESS_KEY", "test")
|
||||
os.environ.setdefault("NOVA_LAMBDA_LOCAL_BYPASS", "1")
|
||||
os.environ.setdefault("NOVA_REPO_ROOT", str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Load the Lambda modules via importlib (`lambda` is a Python reserved word
|
||||
# — mirrors tests/test_idp_auth.py / test_token_vend.py).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_REPO = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def _load(path: Path, name: str):
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
idp_auth = _load(_REPO / "core" / "lambda" / "nova_idp_auth.py", "nova_idp_auth_e2e")
|
||||
token_vend = _load(_REPO / "core" / "lambda" / "nova_idp_token_vend.py", "nova_idp_token_vend_e2e")
|
||||
jwks_mod = _load(_REPO / "core" / "lambda" / "nova_idp_jwks.py", "nova_idp_jwks_e2e")
|
||||
|
||||
import boto3
|
||||
from moto import mock_aws
|
||||
import jwt as pyjwt
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
|
||||
import core.kms_signing as kms_signing
|
||||
import core.pat_lifecycle as pat_life
|
||||
import core.jws_attestation as jws_attestation
|
||||
import core.env as env_mod
|
||||
from core.contract_resolver import resolve
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mock KMS (a test ECC keypair — same pattern as test_kms_roundtrip.py).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _MockKms:
|
||||
def __init__(self, priv, pub_der):
|
||||
self._priv = priv
|
||||
self._pub_der = pub_der
|
||||
|
||||
def sign(self, KeyId, Message, MessageType, SigningAlgorithm):
|
||||
return {"Signature": self._priv.sign(Message, ec.ECDSA(hashes.SHA256()))}
|
||||
|
||||
def get_public_key(self, KeyId):
|
||||
return {"PublicKey": self._pub_der}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Table creation (the 4 IdP tables).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _create_idp_tables(ddb):
|
||||
"""Create the 4 IdP tables (nova-users, nova-sessions,
|
||||
nova-password-resets, nova-pats) with the GSIs the auth + PAT code
|
||||
expects."""
|
||||
ddb.create_table(
|
||||
TableName="nova-users",
|
||||
KeySchema=[{"AttributeName": "user_id", "KeyType": "HASH"}],
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "user_id", "AttributeType": "S"},
|
||||
{"AttributeName": "email", "AttributeType": "S"},
|
||||
],
|
||||
GlobalSecondaryIndexes=[
|
||||
{
|
||||
"IndexName": "email-index",
|
||||
"KeySchema": [{"AttributeName": "email", "KeyType": "HASH"}],
|
||||
"Projection": {"ProjectionType": "ALL"},
|
||||
}
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
ddb.create_table(
|
||||
TableName="nova-sessions",
|
||||
KeySchema=[{"AttributeName": "session_id", "KeyType": "HASH"}],
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "session_id", "AttributeType": "S"},
|
||||
{"AttributeName": "user_id", "AttributeType": "S"},
|
||||
],
|
||||
GlobalSecondaryIndexes=[
|
||||
{
|
||||
"IndexName": "user_id-index",
|
||||
"KeySchema": [{"AttributeName": "user_id", "KeyType": "HASH"}],
|
||||
"Projection": {"ProjectionType": "ALL"},
|
||||
}
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
ddb.create_table(
|
||||
TableName="nova-password-resets",
|
||||
KeySchema=[{"AttributeName": "reset_token", "KeyType": "HASH"}],
|
||||
AttributeDefinitions=[{"AttributeName": "reset_token", "AttributeType": "S"}],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
ddb.create_table(
|
||||
TableName="nova-pats",
|
||||
KeySchema=[{"AttributeName": "jti", "KeyType": "HASH"}],
|
||||
AttributeDefinitions=[
|
||||
{"AttributeName": "jti", "AttributeType": "S"},
|
||||
{"AttributeName": "sub", "AttributeType": "S"},
|
||||
{"AttributeName": "pat_hash", "AttributeType": "S"},
|
||||
],
|
||||
GlobalSecondaryIndexes=[
|
||||
{"IndexName": "sub-index",
|
||||
"KeySchema": [{"AttributeName": "sub", "KeyType": "HASH"}],
|
||||
"Projection": {"ProjectionType": "ALL"}},
|
||||
{"IndexName": "pat_hash-index",
|
||||
"KeySchema": [{"AttributeName": "pat_hash", "KeyType": "HASH"}],
|
||||
"Projection": {"ProjectionType": "ALL"}},
|
||||
],
|
||||
BillingMode="PAY_PER_REQUEST",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_keypair():
|
||||
priv = ec.generate_private_key(ec.SECP256R1())
|
||||
pub = priv.public_key()
|
||||
pub_der = pub.public_bytes(
|
||||
encoding=serialization.Encoding.DER,
|
||||
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
||||
)
|
||||
return priv, pub, pub_der
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_modules():
|
||||
"""Reset the cached boto3 singletons + the mock KMS client."""
|
||||
idp_auth._dynamodb = None
|
||||
token_vend._dynamodb = None
|
||||
pat_life._dynamodb = None
|
||||
yield
|
||||
idp_auth._dynamodb = None
|
||||
token_vend._dynamodb = None
|
||||
pat_life._dynamodb = None
|
||||
kms_signing.set_kms_client_for_testing(None)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cred_file(tmp_path, monkeypatch):
|
||||
"""Isolate ~/.nova/credentials.json to a tmp path (C-7.3)."""
|
||||
p = tmp_path / "credentials.json"
|
||||
monkeypatch.setenv("NOVA_CREDENTIALS_FILE", str(p))
|
||||
yield p
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_contract(tmp_path):
|
||||
"""A minimal contract YAML that resolve() + synthesize_local_env()
|
||||
can consume (mirrors tests/test_local_env.py's fixture)."""
|
||||
contract = """
|
||||
id: msvc
|
||||
name: microservice
|
||||
environment: dev
|
||||
infrastructure:
|
||||
microservice:
|
||||
version: "1.0.0"
|
||||
inputs:
|
||||
image: nginx:latest
|
||||
"""
|
||||
p = tmp_path / "contract.yml"
|
||||
p.write_text(contract)
|
||||
return p
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Audit-event capture (the Lambdas emit JSON lines on stderr).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _AuditCapture:
|
||||
"""Capture JSON audit lines written to stderr by the Lambda modules.
|
||||
|
||||
Each Lambda's ``_emit_audit`` does ``sys.stderr.write(json + "\\n")``.
|
||||
We replace the module's ``sys`` reference's stderr with a StringIO
|
||||
during the flow, then parse the captured lines back into dicts.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.events: list[dict] = []
|
||||
self._buf = io.StringIO()
|
||||
self._real_stderr = sys.stderr
|
||||
|
||||
def __enter__(self):
|
||||
# Patch sys.stderr globally for the duration — the Lambda modules
|
||||
# all use the module-level `sys` import (sys.stderr.write).
|
||||
sys.stderr = self._buf
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc):
|
||||
sys.stderr = self._real_stderr
|
||||
self._buf.seek(0)
|
||||
for line in self._buf.getvalue().splitlines():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
self.events.append(json.loads(line))
|
||||
except json.JSONDecodeError:
|
||||
# Non-JSON stderr noise (e.g. a traceback) — ignore.
|
||||
pass
|
||||
return False
|
||||
|
||||
def event_types(self) -> list[str]:
|
||||
return [e.get("event", "") for e in self.events]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The E2E test (REQ-348).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestE2EIdpFlow:
|
||||
"""E2E: sign-up → sign-in → token-vend → apply → audit (REQ-348).
|
||||
|
||||
Runs against moto (DynamoDB) + mock KMS locally; in CI the same
|
||||
assertions run against the deployed Nova-idp Lambdas.
|
||||
"""
|
||||
|
||||
@mock_aws
|
||||
def test_full_e2e_sign_up_sign_in_token_vend_apply_audit(
|
||||
self, test_keypair, cred_file, sample_contract
|
||||
):
|
||||
priv, pub, pub_der = test_keypair
|
||||
kms_signing.set_kms_client_for_testing(_MockKms(priv, pub_der))
|
||||
ddb = boto3.client("dynamodb", region_name="us-east-1")
|
||||
_create_idp_tables(ddb)
|
||||
|
||||
email = "alice@example.com"
|
||||
password = "E2E-Secret-12345"
|
||||
owner = "team-a"
|
||||
|
||||
audit = _AuditCapture()
|
||||
with audit:
|
||||
# --- (a) sign_up succeeds ---
|
||||
up = idp_auth.lambda_handler(
|
||||
{
|
||||
"body": json.dumps(
|
||||
{
|
||||
"action": "sign_up",
|
||||
"email": email,
|
||||
"password": password,
|
||||
"owner": owner,
|
||||
"roles": ["developer"],
|
||||
}
|
||||
)
|
||||
},
|
||||
None,
|
||||
)
|
||||
assert up["statusCode"] == 200, up
|
||||
up_body = json.loads(up["body"])
|
||||
user_id = up_body["user_id"]
|
||||
assert user_id
|
||||
|
||||
# --- (b) sign_in returns a session ---
|
||||
inn = idp_auth.lambda_handler(
|
||||
{
|
||||
"body": json.dumps(
|
||||
{"action": "sign_in", "email": email, "password": password}
|
||||
)
|
||||
},
|
||||
None,
|
||||
)
|
||||
assert inn["statusCode"] == 200, inn
|
||||
session_id = json.loads(inn["body"])["session_id"]
|
||||
assert session_id
|
||||
|
||||
# --- issue a PAT (the developer logs in with it) ---
|
||||
pat = pat_life.issue_pat(
|
||||
user_id, ["developer"], owner, ttl_seconds=3600,
|
||||
subject_type="developer",
|
||||
)
|
||||
assert pat, "no raw PAT returned"
|
||||
# Extract the PAT jti for later audit-link assertions.
|
||||
pat_payload = json.loads(
|
||||
base64.urlsafe_b64decode(pat.split(".")[1] + "==")
|
||||
)
|
||||
pat_jti = pat_payload["jti"]
|
||||
assert pat_jti
|
||||
|
||||
# --- (c) token-vend returns an OIDC token ---
|
||||
vend_body = {
|
||||
"token": pat,
|
||||
"environment": "dev",
|
||||
"requested_claims": ["sub", "roles"],
|
||||
"target_resource": {
|
||||
"type": "contract", "id": "msvc",
|
||||
"owner": owner, "environment": "dev",
|
||||
},
|
||||
}
|
||||
vresp = token_vend.lambda_handler(
|
||||
{"body": json.dumps(vend_body)}, None
|
||||
)
|
||||
assert vresp["statusCode"] == 200, vresp
|
||||
oidc_token = json.loads(vresp["body"])["token"]
|
||||
assert oidc_token
|
||||
|
||||
# --- (d) the OIDC token verifies with the JWKS key ---
|
||||
jwks_resp = jwks_mod.lambda_handler({}, None)
|
||||
assert jwks_resp["statusCode"] == 200, jwks_resp
|
||||
jwk = json.loads(jwks_resp["body"])["keys"][0]
|
||||
key = pyjwt.PyJWK(jwk).key
|
||||
decoded_oidc = pyjwt.decode(
|
||||
oidc_token, key, algorithms=["ES256"],
|
||||
options={"verify_aud": False},
|
||||
)
|
||||
assert decoded_oidc["sub"] == user_id
|
||||
assert decoded_oidc["typ"] == "nova_oidc_token"
|
||||
assert decoded_oidc["roles"] == ["developer"]
|
||||
assert "jti" in decoded_oidc and "exp" in decoded_oidc
|
||||
|
||||
# --- store the credential (nova auth login) ---
|
||||
# Use the auth_store directly (login.py's local path calls
|
||||
# token_vend in-process, which we already did above).
|
||||
from core.auth_store import store_credential
|
||||
store_credential(
|
||||
jti=decoded_oidc["jti"],
|
||||
cred_type=decoded_oidc["typ"],
|
||||
exp=decoded_oidc["exp"],
|
||||
oidc_token=oidc_token,
|
||||
)
|
||||
# C-7.3: the credentials file has the OIDC token, NOT the raw PAT.
|
||||
raw_cred = cred_file.read_text()
|
||||
assert "raw_pat" not in raw_cred
|
||||
assert pat not in raw_cred
|
||||
|
||||
# --- (e) nova apply --local --sign-local-review produces a JWS ---
|
||||
# Drive apply via the core functions directly (nova/apply.py's
|
||||
# run() calls these; we skip the argparse layer for the test).
|
||||
synth = env_mod.synthesize_local_env(
|
||||
str(sample_contract), environment="dev"
|
||||
)
|
||||
assert synth["region"] == "local"
|
||||
attestation_payload = {
|
||||
"contract": str(sample_contract),
|
||||
"review": "local",
|
||||
"user_id": user_id,
|
||||
"pat_jti": pat_jti,
|
||||
}
|
||||
jws = jws_attestation.sign_attestation(attestation_payload, pat)
|
||||
assert jws.count(".") == 2, "not a compact JWS (3 segments)"
|
||||
|
||||
# --- (f) the JWS verifies with the PAT-derived key ---
|
||||
verified = jws_attestation.verify_attestation(jws, pat)
|
||||
assert verified == attestation_payload
|
||||
|
||||
# Tamper detection: verify with the wrong PAT raises.
|
||||
with pytest.raises(jws_attestation.JWSValidationError):
|
||||
jws_attestation.verify_attestation(jws, pat + "tampered")
|
||||
|
||||
# --- (g) the audit chain is complete + linked ---
|
||||
# Every step emitted an audit event with the expected event type.
|
||||
types = audit.event_types()
|
||||
# sign_up + sign_in + session_created + pat.issued + token.vend.allowed
|
||||
assert "auth.sign_up" in types, f"missing auth.sign_up in {types}"
|
||||
assert "auth.sign_in" in types, f"missing auth.sign_in in {types}"
|
||||
assert "auth.session_created" in types, f"missing auth.session_created in {types}"
|
||||
assert "pat.issued" in types, f"missing pat.issued in {types}"
|
||||
assert "token.vend.allowed" in types, f"missing token.vend.allowed in {types}"
|
||||
|
||||
# Linkage: the sign_up + sign_in events share the same user_id.
|
||||
sign_up_ev = next(e for e in audit.events if e.get("event") == "auth.sign_up")
|
||||
sign_in_ev = next(e for e in audit.events if e.get("event") == "auth.sign_in")
|
||||
assert sign_up_ev["user_id"] == user_id
|
||||
assert sign_in_ev["user_id"] == user_id
|
||||
assert sign_up_ev["email"] == email
|
||||
|
||||
# Linkage: the pat.issued event carries the PAT jti + sub.
|
||||
pat_issued_ev = next(e for e in audit.events if e.get("event") == "pat.issued")
|
||||
assert pat_issued_ev["jti"] == pat_jti
|
||||
assert pat_issued_ev["sub"] == user_id
|
||||
|
||||
# Linkage: the token.vend.allowed event carries the PAT jti + sub +
|
||||
# policy_sha (D-231).
|
||||
vend_ev = next(e for e in audit.events if e.get("event") == "token.vend.allowed")
|
||||
assert vend_ev["pat_jti"] == pat_jti
|
||||
assert vend_ev["sub"] == user_id
|
||||
assert "policy_sha" in vend_ev
|
||||
|
||||
# Linkage: no raw password / PAT leaked into any audit event (INV-16).
|
||||
for ev in audit.events:
|
||||
blob = json.dumps(ev, sort_keys=True)
|
||||
assert password not in blob, (
|
||||
f"raw password leaked into audit event {ev.get('event')!r}: {blob}"
|
||||
)
|
||||
assert pat not in blob, (
|
||||
f"raw PAT leaked into audit event {ev.get('event')!r}: {blob}"
|
||||
)
|
||||
|
||||
# --- the user item in nova-users has a password_hash, NOT the raw password ---
|
||||
item = ddb.get_item(
|
||||
TableName="nova-users", Key={"user_id": {"S": user_id}}
|
||||
)
|
||||
assert "Item" in item
|
||||
attrs = item["Item"]
|
||||
assert "password_hash" in attrs
|
||||
assert attrs["password_hash"]["S"].startswith("$argon2id$")
|
||||
assert "password" not in attrs, "raw password stored in DDB item!"
|
||||
for key, val in attrs.items():
|
||||
sval = val.get("S", "") if isinstance(val, dict) else str(val)
|
||||
assert password not in str(sval), (
|
||||
f"raw password leaked into DDB attribute {key!r}"
|
||||
)
|
||||
|
||||
# --- the PAT row in nova-pats has a hash, NOT the raw PAT ---
|
||||
pat_item = ddb.get_item(
|
||||
TableName="nova-pats",
|
||||
Key={"jti": {"S": pat_jti}},
|
||||
ConsistentRead=True,
|
||||
)
|
||||
assert "Item" in pat_item
|
||||
assert pat_item["Item"]["status"]["S"] == "active"
|
||||
assert "pat_hash" in pat_item["Item"]
|
||||
raw_pat_blob = json.dumps(pat_item["Item"], sort_keys=True)
|
||||
assert pat not in raw_pat_blob, "raw PAT stored in nova-pats item!"
|
||||
|
||||
@mock_aws
|
||||
def test_e2e_revocation_breaks_the_chain(self, test_keypair, sample_contract):
|
||||
"""The E2E chain breaks at token-vend after revocation (D-229).
|
||||
|
||||
Issue a PAT → revoke it → the next token-vend returns 403
|
||||
pat_revoked (the audit event is token.vend.denied). This is the
|
||||
negative path of the E2E flow — the revocation is the trust
|
||||
anchor, not the JWT signature (D-229).
|
||||
"""
|
||||
priv, _pub, pub_der = test_keypair
|
||||
kms_signing.set_kms_client_for_testing(_MockKms(priv, pub_der))
|
||||
ddb = boto3.client("dynamodb", region_name="us-east-1")
|
||||
_create_idp_tables(ddb)
|
||||
|
||||
audit = _AuditCapture()
|
||||
with audit:
|
||||
pat = pat_life.issue_pat(
|
||||
"user-2", ["developer"], "team-b", ttl_seconds=3600,
|
||||
)
|
||||
pat_payload = json.loads(
|
||||
base64.urlsafe_b64decode(pat.split(".")[1] + "==")
|
||||
)
|
||||
pat_jti = pat_payload["jti"]
|
||||
|
||||
# Vend succeeds before revocation.
|
||||
ok = token_vend.lambda_handler(
|
||||
{"body": json.dumps({"token": pat, "environment": "dev"})},
|
||||
None,
|
||||
)
|
||||
assert ok["statusCode"] == 200, ok
|
||||
|
||||
# Revoke.
|
||||
pat_life.revoke_pat(pat_jti)
|
||||
|
||||
# Vend fails after revocation (403 pat_revoked, immediate — D-229).
|
||||
denied = token_vend.lambda_handler(
|
||||
{"body": json.dumps({"token": pat, "environment": "dev"})},
|
||||
None,
|
||||
)
|
||||
assert denied["statusCode"] == 403, denied
|
||||
assert json.loads(denied["body"])["reason"] == "pat_revoked"
|
||||
|
||||
types = audit.event_types()
|
||||
assert "pat.issued" in types
|
||||
assert "pat.revoked" in types
|
||||
assert "token.vend.allowed" in types
|
||||
assert "token.vend.denied" in types
|
||||
|
||||
# The denied event carries the revoked jti + the pat_revoked reason.
|
||||
denied_ev = next(e for e in audit.events if e.get("event") == "token.vend.denied")
|
||||
assert denied_ev["pat_jti"] == pat_jti
|
||||
assert denied_ev["reason"] == "pat_revoked"
|
||||
Reference in New Issue
Block a user