acdl/tests/test_argon2_fail_closed.py

"""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}"
            )