"""Argon2 fail-closed test (C-1.2, REQ-334, D-228). Verifies the three pillars of D-228 (amended): 1. **ImportError → Argon2UnavailableError** — when the ``argon2`` C extension fails to load, ``hash_password`` / ``verify_password`` raise ``Argon2UnavailableError`` (not a crash, not a weak hash, not a return of a plaintext). 2. **Lambda handler → 503** — the handler returns HTTP 503 ``{"error": "argon2_unavailable"}`` when ``_ARGON2_AVAILABLE`` is False (no pure-Python fallback, no weak hash). 3. **No raw passwords in logs** — the password string never appears in any log record (caplog). The module is loaded via importlib (``lambda`` is a Python reserved word — mirrors tests/test_contract_ingestor.py). """ import importlib.util import json import logging 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) # --------------------------------------------------------------------------- # Pillar 1: ImportError → Argon2UnavailableError (not a weak hash) # --------------------------------------------------------------------------- class TestArgon2ImportFailure: """C-1.2: the auth Lambda fails closed when the C extension is missing.""" def test_hash_password_raises_argon2unavailable_when_unavailable(self): """When _ARGON2_AVAILABLE is False, hash_password raises Argon2UnavailableError — NOT a crash, NOT a weak hash, NOT a plaintext return.""" with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): with pytest.raises(idp.Argon2UnavailableError): idp.hash_password("super-secret-123") # And no hash string was produced (no weak fallback). def test_verify_password_raises_argon2unavailable_when_unavailable(self): with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): with pytest.raises(idp.Argon2UnavailableError): idp.verify_password("any", "$argon2id$fake$hash") def test_hash_password_does_not_return_plaintext_on_failure(self): """C-1.2 explicit: the function must not return the raw password or any non-argon2 string when argon2 is unavailable.""" with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): try: result = idp.hash_password("plaintext-to-check") # If we get here, the function FAILED to fail closed. pytest.fail( f"hash_password returned {result!r} instead of raising " f"Argon2UnavailableError (fail-closed violated)" ) except idp.Argon2UnavailableError: pass # correct except Exception as e: pytest.fail( f"hash_password raised {type(e).__name__} instead of " f"Argon2UnavailableError" ) def test_simulated_importerror_at_module_load_raises_unavailable(self): """Simulate the actual cold-start ImportError: reload the module with argon2 import poisoned → _ARGON2_AVAILABLE is False and the hashing functions raise Argon2UnavailableError.""" # Poison sys.modules so `from argon2 import PasswordHasher` fails. with mock.patch.dict(sys.modules, {"argon2": None, "argon2.exceptions": None}): # Reload in the poisoned environment. mod = importlib.util.module_from_spec(_spec) try: _spec.loader.exec_module(mod) except Exception: # If exec_module itself raises (importlib treats None as # "not imported"), that's also acceptable fail-closed # behaviour — but we expect a clean load with the flag False. mod = idp # fall back to the already-loaded module assert mod._ARGON2_AVAILABLE is False, ( "module should mark argon2 unavailable on ImportError" ) with pytest.raises(mod.Argon2UnavailableError): mod.hash_password("x") def test_argon2unavailable_is_a_clean_exception_not_a_crash(self): """The fail-closed signal is a catchable Exception, not a segfault / SystemExit / KeyboardInterrupt.""" with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): try: idp.hash_password("x") except idp.Argon2UnavailableError as e: assert isinstance(e, Exception) # Must NOT be a SystemExit or KeyboardInterrupt. assert not isinstance(e, (SystemExit, KeyboardInterrupt)) # The message should mention argon2 / fail-closed. assert "argon2" in str(e).lower() # --------------------------------------------------------------------------- # Pillar 2: Lambda handler → 503 (not a crash, not a weak hash) # --------------------------------------------------------------------------- class TestHandler503OnArgon2Unavailable: """C-1.2: the handler returns 503 when argon2 is unavailable.""" def test_sign_up_returns_503_when_argon2_unavailable(self): """When _ARGON2_AVAILABLE is False, sign_up → 503 argon2_unavailable (NOT a weak-hash write, NOT a 500 crash).""" event = { "body": json.dumps( { "action": "sign_up", "email": "user@example.com", "password": "SuperSecret-1", "owner": "owner-1", "roles": ["user"], } ) } with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): resp = idp.lambda_handler(event, None) assert resp["statusCode"] == 503, resp body = json.loads(resp["body"]) assert body["error"] == "argon2_unavailable" def test_sign_in_returns_503_when_argon2_unavailable(self): event = { "body": json.dumps( { "action": "sign_in", "email": "user@example.com", "password": "SuperSecret-1", } ) } with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): resp = idp.lambda_handler(event, None) assert resp["statusCode"] == 503, resp assert json.loads(resp["body"])["error"] == "argon2_unavailable" def test_reset_password_returns_503_when_argon2_unavailable(self): event = { "body": json.dumps( { "action": "reset_password", "reset_token": "some-token", "new_password": "NewSecret-2", } ) } with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): resp = idp.lambda_handler(event, None) assert resp["statusCode"] == 503, resp def test_503_is_not_a_500_crash(self): """The fail-closed response is exactly 503, never 500.""" event = { "body": json.dumps( { "action": "sign_up", "email": "u@e.com", "password": "p", "owner": "o", "roles": ["user"], } ) } with mock.patch.object(idp, "_ARGON2_AVAILABLE", False): resp = idp.lambda_handler(event, None) assert resp["statusCode"] != 500, "fail-closed must be 503, not 500" assert resp["statusCode"] != 200, "fail-closed must not succeed" # --------------------------------------------------------------------------- # Pillar 3: no raw passwords in logs (INV-16) # --------------------------------------------------------------------------- class TestNoRawPasswordsInLogs: """INV-16: raw passwords never appear in logs / traces.""" def test_hash_password_does_not_log_password(self, caplog): secret = "NeverLogMe-12345" with caplog.at_level(logging.DEBUG, logger="nova_idp_auth"): idp.hash_password(secret) for record in caplog.records: assert secret not in record.getMessage(), ( f"raw password leaked in log: {record.getMessage()!r}" ) def test_audit_emit_does_not_include_password(self, caplog): """The _emit_audit helper must never include a password field.""" with caplog.at_level(logging.DEBUG): idp._emit_audit( "auth.test", user_id="u1", email="e@e.com", password="leak-me" ) full = "\n".join(r.getMessage() for r in caplog.records) assert "leak-me" not in full, "password leaked via audit emit" # Even though we passed password=, it must be scrubbed. for record in caplog.records: assert "leak-me" not in record.getMessage() def test_sign_up_audit_does_not_log_password(self, caplog, monkeypatch): """End-to-end: a sign_up writes an audit event to stderr that does NOT contain the raw password.""" # Stub DynamoDB so we don't need moto here (just test the audit). from tests.test_idp_auth import _stub_dynamodb_for_audit _stub_dynamodb_for_audit(idp, monkeypatch) secret = "AuditSecret-99887" with caplog.at_level(logging.DEBUG): idp.sign_up( { "email": "audit@example.com", "password": secret, "owner": "owner-1", "roles": ["user"], } ) for record in caplog.records: msg = record.getMessage() assert secret not in msg, ( f"raw password leaked in audit log: {msg!r}" )