"""CAP-036 E2E auth flow test (REQ-333, CAP-036). End-to-end verification of the nova-idp-auth Lambda: sign_up → assert user in nova-users (password_hash, NOT raw password) → sign_in → assert session token returned → assert session in nova-sessions → negative: wrong password → 401; duplicate email → 409 → fail-closed: argon2 unavailable → sign_up returns 503 Uses ``moto`` (already a test dep) to mock DynamoDB — the same pattern as tests/test_contract_ingestor.py. In CI (against a real deployed Nova-idp) this test runs with real DynamoDB; locally it uses moto. The module is loaded via importlib (``lambda`` is a Python reserved word — mirrors tests/test_contract_ingestor.py). """ import importlib.util import json import os import sys from pathlib import Path from unittest import mock import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) _SOURCE_PATH = ( Path(__file__).resolve().parent.parent / "core" / "lambda" / "nova_idp_auth.py" ) _spec = importlib.util.spec_from_file_location("nova_idp_auth", _SOURCE_PATH) idp = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(idp) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- def _create_idp_tables(dynamodb_client): """Create the 3 IdP tables (nova-users, nova-sessions, nova-password-resets).""" # nova-users with email-index GSI dynamodb_client.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", ) # nova-sessions dynamodb_client.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", ) # nova-password-resets dynamodb_client.create_table( TableName="nova-password-resets", KeySchema=[{"AttributeName": "reset_token", "KeyType": "HASH"}], AttributeDefinitions=[ {"AttributeName": "reset_token", "AttributeType": "S"}, ], BillingMode="PAY_PER_REQUEST", ) @pytest.fixture def moto_idp_tables(monkeypatch): """Spin up moto-backed DynamoDB with the 3 IdP tables.""" from moto import mock_aws import boto3 monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1") monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing") monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing") with mock_aws(): client = boto3.client("dynamodb", region_name="us-east-1") _create_idp_tables(client) # Reset the cached boto3 resource so the idp module picks up moto. saved = idp._dynamodb idp._dynamodb = None monkeypatch.setattr(idp, "USERS_TABLE", "nova-users") monkeypatch.setattr(idp, "SESSIONS_TABLE", "nova-sessions") monkeypatch.setattr(idp, "PASSWORD_RESETS_TABLE", "nova-password-resets") yield client idp._dynamodb = saved # Helper used by tests/test_argon2_fail_closed.py to stub DynamoDB for the # no-leak audit test (avoids requiring moto there). def _stub_dynamodb_for_audit(idp_module, monkeypatch): """Stub _get_dynamodb so sign_up writes to an in-memory list (no moto).""" class _Tbl: def __init__(self, name, store): self.name = name self.store = store def put_item(self, *, TableName=None, Item=None, **kw): self.store.setdefault(self.name, []).append(Item) return {} def query(self, **kw): return {"Items": []} def get_item(self, **kw): return {} def update_item(self, **kw): return {} def delete_item(self, **kw): return {} class _Res: def __init__(self): self.store = {} def Table(self, name): return _Tbl(name, self.store) res = _Res() monkeypatch.setattr(idp_module, "_dynamodb", res) # --------------------------------------------------------------------------- # CAP-036: E2E sign-up → sign-in → session # --------------------------------------------------------------------------- class TestCap036E2E: """CAP-036: the E2E auth flow runs against moto locally (real DDB in CI).""" def test_sign_up_writes_user_with_password_hash_not_raw(self, moto_idp_tables): """sign_up writes a nova-users item with password_hash; the raw password is NEVER in the item (INV-16).""" password = "E2E-Secret-12345" resp = idp.lambda_handler( { "body": json.dumps( { "action": "sign_up", "email": "alice@example.com", "password": password, "owner": "owner-alice", "roles": ["user"], } ) }, None, ) assert resp["statusCode"] == 200, resp body = json.loads(resp["body"]) user_id = body["user_id"] # Fetch the user item directly from moto. item = moto_idp_tables.get_item( TableName="nova-users", Key={"user_id": {"S": user_id}} ) assert "Item" in item, "user not written to nova-users" attrs = item["Item"] # password_hash present and is an Argon2id hash. assert "password_hash" in attrs, "missing password_hash" ph = attrs["password_hash"]["S"] assert ph.startswith("$argon2id$"), f"not an argon2id hash: {ph!r}" # CRITICAL: the raw password must NOT be stored anywhere in the item. 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}: {sval!r}" ) def test_full_e2e_sign_up_sign_in_session(self, moto_idp_tables): """CAP-036 headline: sign_up → sign_in → session in nova-sessions.""" password = "E2E-Secret-67890" # 1. sign_up up = idp.lambda_handler( { "body": json.dumps( { "action": "sign_up", "email": "bob@example.com", "password": password, "owner": "owner-bob", "roles": ["user"], } ) }, None, ) assert up["statusCode"] == 200, up # 2. sign_in inn = idp.lambda_handler( { "body": json.dumps( { "action": "sign_in", "email": "bob@example.com", "password": password, } ) }, None, ) assert inn["statusCode"] == 200, inn session_id = json.loads(inn["body"])["session_id"] assert session_id, "no session_id returned" # 3. session is in nova-sessions sitem = moto_idp_tables.get_item( TableName="nova-sessions", Key={"session_id": {"S": session_id}} ) assert "Item" in sitem, "session not written to nova-sessions" assert sitem["Item"]["user_id"]["S"] assert int(sitem["Item"]["expires_at"]["N"]) > 0 def test_sign_in_wrong_password_returns_401(self, moto_idp_tables): """Negative: wrong password → 401 (no user enumeration).""" idp.lambda_handler( { "body": json.dumps( { "action": "sign_up", "email": "carol@example.com", "password": "Correct-1", "owner": "owner-carol", "roles": ["user"], } ) }, None, ) resp = idp.lambda_handler( { "body": json.dumps( { "action": "sign_in", "email": "carol@example.com", "password": "Wrong-2", } ) }, None, ) assert resp["statusCode"] == 401, resp body = json.loads(resp["body"]) assert body["error"] == "invalid_credentials" def test_sign_up_duplicate_email_returns_409(self, moto_idp_tables): """Negative: duplicate email → 409.""" payload = { "action": "sign_up", "email": "dup@example.com", "password": "First-1", "owner": "owner-dup", "roles": ["user"], } first = idp.lambda_handler({"body": json.dumps(payload)}, None) assert first["statusCode"] == 200, first second = idp.lambda_handler({"body": json.dumps(payload)}, None) assert second["statusCode"] == 409, second assert json.loads(second["body"])["error"] == "email_already_registered" def test_sign_in_unknown_email_returns_401(self, moto_idp_tables): """Unknown email → 401 (same as wrong password, no enumeration).""" resp = idp.lambda_handler( { "body": json.dumps( { "action": "sign_in", "email": "nobody@example.com", "password": "x", } ) }, None, ) assert resp["statusCode"] == 401, resp def test_create_session_standalone(self, moto_idp_tables): """create_session action writes a session row.""" resp = idp.lambda_handler( {"body": json.dumps({"action": "create_session", "user_id": "u-xyz"})}, None, ) assert resp["statusCode"] == 200, resp sid = json.loads(resp["body"])["session_id"] item = moto_idp_tables.get_item( TableName="nova-sessions", Key={"session_id": {"S": sid}} ) assert "Item" in item # --------------------------------------------------------------------------- # Fail-closed (also covered in test_argon2_fail_closed.py, but verify E2E) # --------------------------------------------------------------------------- class TestFailClosedE2E: def test_sign_up_503_when_argon2_unavailable(self, moto_idp_tables): """E2E fail-closed: argon2 unavailable → sign_up returns 503 and does NOT write a user (no weak hash write).""" with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): resp = idp.lambda_handler( { "body": json.dumps( { "action": "sign_up", "email": "fail@example.com", "password": "p", "owner": "o", "roles": ["user"], } ) }, None, ) assert resp["statusCode"] == 503, resp # No user should have been written. items = moto_idp_tables.scan(TableName="nova-users").get("Items", []) assert not items, "user was written despite argon2 unavailable (weak hash!)" # --------------------------------------------------------------------------- # Password reset flow # --------------------------------------------------------------------------- class TestPasswordReset: def test_request_then_reset_password(self, moto_idp_tables): password = "Original-1" idp.lambda_handler( { "body": json.dumps( { "action": "sign_up", "email": "reset@example.com", "password": password, "owner": "owner-reset", "roles": ["user"], } ) }, None, ) # request reset req = idp.lambda_handler( {"body": json.dumps({"action": "request_password_reset", "email": "reset@example.com"})}, None, ) assert req["statusCode"] == 200, req token = json.loads(req["body"])["reset_token"] assert token, "no reset token returned" # reset password new_pw = "NewSecret-2" rst = idp.lambda_handler( {"body": json.dumps({"action": "reset_password", "reset_token": token, "new_password": new_pw})}, None, ) assert rst["statusCode"] == 200, rst # sign in with the new password works inn = idp.lambda_handler( {"body": json.dumps({"action": "sign_in", "email": "reset@example.com", "password": new_pw})}, None, ) assert inn["statusCode"] == 200, inn # old password now fails old = idp.lambda_handler( {"body": json.dumps({"action": "sign_in", "email": "reset@example.com", "password": password})}, None, ) assert old["statusCode"] == 401, old def test_reset_with_invalid_token_returns_400(self, moto_idp_tables): resp = idp.lambda_handler( {"body": json.dumps({"action": "reset_password", "reset_token": "bogus", "new_password": "x"})}, None, ) assert resp["statusCode"] == 400, resp # --------------------------------------------------------------------------- # No raw passwords in logs (verification step 5) # --------------------------------------------------------------------------- class TestNoRawPasswordsInLogs: def test_sign_up_does_not_log_password(self, moto_idp_tables, caplog): """Verification step 5: the password string is NOT in any log record.""" import logging secret = "LogSecret-55512" with caplog.at_level(logging.DEBUG): idp.lambda_handler( { "body": json.dumps( { "action": "sign_up", "email": "log@example.com", "password": secret, "owner": "owner-log", "roles": ["user"], } ) }, None, ) for record in caplog.records: assert secret not in record.getMessage(), ( f"raw password leaked in log: {record.getMessage()!r}" )