merge(phase/03): v1.28 P3 idp-auth complete (REQ-333..335, CAP-036)

.ciagent/CHECKPOINT.json

@@ -1,18 +1,19 @@
 {
-  "phase": 2,
+  "phase": 3,
   "stage": "complete",
   "milestone": "v1.28",
   "phase_role": "execution",
   "attempts": 0,
-  "updated_at": "2026-08-19T22:00:00Z",
+  "updated_at": "2026-08-19T22:30:00Z",
   "project": "acdl",
   "projects": ["acdl", "nova-blockchain-exchange"],
   "active_milestone": "v1.28",
   "milestone_branch": "milestone/v1.28-cli-identity",
-  "phase_branch": "phase/02-lambda-packaging",
+  "phase_branch": "phase/03-idp-auth",
   "tag_line": "v1.27.x",
-  "phase_name": "lambda-packaging",
-  "reqs_covered": ["REQ-329", "REQ-330", "REQ-331", "REQ-332"],
-  "tests": {"p2_specific": 43, "total_passing": 922, "failures": 0},
-  "notes": "v1.28 P2 SHIP. lambda-packaging complete. Tag v1.27.2. Merged phase/02 -> milestone/v1.28-cli-identity. 4 REQs covered (REQ-329..332). Dual-use refactor (>=80% share), local env synthesizer, JWS-from-PAT HKDF (C-5.2 grill fix), attestations dir. Next: P3 idp-auth."
+  "phase_name": "idp-auth",
+  "reqs_covered": ["REQ-333", "REQ-334", "REQ-335"],
+  "caps_verified": ["CAP-036"],
+  "tests": {"p3_specific": 22, "total_passing": 944, "failures": 0},
+  "notes": "v1.28 P3 SHIP. idp-auth complete. Tag v1.27.3. Merged phase/03 -> milestone/v1.28-cli-identity. 3 REQs covered (REQ-333..335), CAP-036 verified. nova-idp-auth Lambda (sign-up/sign-in/session), Argon2id t=3 m=65536 p=1 fail-closed, 4 DDB tables. Next: P4 token-vend-pat (highest-risk, double-length)."
 }

core/lambda/nova_idp_auth.py

@@ -0,0 +1,613 @@
+"""Nova IdP auth Lambda — sign-up / sign-in / session (REQ-333, REQ-334).
+
+Invoked via a Function URL (IAM auth) by the Nova CLI and consumer
+pipelines. Mirrors the ``contract_ingestor.py`` pattern: lazy
+``boto3.resource`` DynamoDB singleton, env-var table names,
+``NOVA_LAMBDA_LOCAL_BYPASS`` for local testing, ``__main__`` CLI block
+for dual-use (REQ-329).
+
+## Argon2id password hashing (REQ-334, D-228, C-7.2)
+
+Passwords are hashed with Argon2id via ``argon2-cffi``:
+
+    PasswordHasher(time_cost=3, memory_cost=65536, parallelism=1)
+
+These are the OWASP minimum parameters (t=3, m=65536 KiB, p=1).
+Lambda memory **MUST be ≥ 512 MB** (Argon2id memory_cost ~64 MiB +
+runtime overhead).
+
+**D-228 (amended) — fail-closed:** there is no maintained pure-Python
+Argon2 implementation; a pure-Python crypto fallback is a liability
+(weaker hashing, violates INV-16's spirit). If the ``argon2`` C
+extension fails to import, the Lambda **fails closed** —
+``_ARGON2_AVAILABLE`` is set ``False`` at cold-start, and
+:func:`hash_password` / :func:`verify_password` raise
+``Argon2UnavailableError``. The handler catches this and returns
+**HTTP 503** (``{"error": "argon2_unavailable"}``) — **no pure-Python
+fallback, no weak hash, no crash.** This is verified by the explicit
+``test_argon2_fail_closed`` test (C-1.2).
+
+## No raw passwords anywhere (INV-16)
+
+Raw passwords are NEVER:
+  * written to DynamoDB (only ``password_hash`` is stored),
+  * logged (the handler never logs the password argument),
+  * put in traces / env vars / X-Ray segments.
+
+Audit events (``auth.sign_up``, ``auth.sign_in``,
+``auth.session_created``) are emitted to stderr as JSON; they carry the
+``user_id`` / ``email`` but **never** the password.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import os
+import sys
+import uuid
+
+import boto3
+
+# ---------------------------------------------------------------------------
+# Argon2id — fail-closed import (REQ-334, D-228, C-7.2)
+# ---------------------------------------------------------------------------
+#
+# try-import the C extension. If it fails (missing abi3 wheel, wrong
+# glibc, etc.), _ARGON2_AVAILABLE becomes False and hash/verify raise
+# Argon2UnavailableError. The handler returns 503. NO pure-Python fallback.
+_ARGON2_AVAILABLE = False
+_PasswordHasher = None
+
+try:  # pragma: no cover - import success path covered by round-trip test
+    from argon2 import PasswordHasher
+    from argon2.exceptions import VerifyMismatchError
+
+    _PasswordHasher = PasswordHasher
+    _ARGON2_AVAILABLE = True
+except ImportError:  # pragma: no cover - exercised via mock in tests
+    _ARGON2_AVAILABLE = False
+
+    # Define a stand-in so `verify_password` can raise the right type
+    # even when argon2 isn't importable. VerifyMismatchError is only
+    # raised by verify() which itself raises Argon2UnavailableError first.
+    class VerifyMismatchError(Exception):
+        """Raised by verify_password when the password does not match."""
+
+
+class Argon2UnavailableError(Exception):
+    """Raised when the Argon2 C extension is unavailable (D-228 fail-closed).
+
+    The handler catches this and returns HTTP 503 — no pure-Python
+    fallback, no weak hash.
+    """
+
+
+# OWASP-minimum Argon2id parameters (C-7.2):
+#   time_cost=3, memory_cost=65536 KiB (64 MiB), parallelism=1
+_ARGON2_TIME_COST = 3
+_ARGON2_MEMORY_COST = 65536  # KiB
+_ARGON2_PARALLELISM = 1
+
+
+def _get_hasher():
+    """Return a PasswordHasher configured with the OWASP-min params.
+
+    Raises Argon2UnavailableError if the C extension is not loaded.
+    """
+    if not _ARGON2_AVAILABLE or _PasswordHasher is None:
+        raise Argon2UnavailableError(
+            "argon2 C extension unavailable — refusing to hash with a "
+            "weak fallback (D-228 fail-closed)"
+        )
+    return _PasswordHasher(
+        time_cost=_ARGON2_TIME_COST,
+        memory_cost=_ARGON2_MEMORY_COST,
+        parallelism=_ARGON2_PARALLELISM,
+    )
+
+
+def hash_password(password: str) -> str:
+    """Hash a password with Argon2id (OWASP-min params).
+
+    Returns the Argon2id hash string (includes the salt + params).
+
+    Raises:
+        Argon2UnavailableError: if the ``argon2`` C extension is not
+            importable (D-228 fail-closed — NO pure-Python fallback).
+    """
+    if not _ARGON2_AVAILABLE:
+        raise Argon2UnavailableError(
+            "argon2 C extension unavailable — refusing to hash (D-228)"
+        )
+    # NOTE: the password argument is NEVER logged. Do not add debug
+    # prints here that include `password`.
+    return _get_hasher().hash(password)
+
+
+def verify_password(password: str, hash_str: str) -> bool:
+    """Verify a password against an Argon2id hash.
+
+    Returns ``True`` if the password matches.
+
+    Raises:
+        Argon2UnavailableError: if the ``argon2`` C extension is not
+            importable.
+        VerifyMismatchError: if the password does not match the hash.
+    """
+    if not _ARGON2_AVAILABLE:
+        raise Argon2UnavailableError(
+            "argon2 C extension unavailable — refusing to verify (D-228)"
+        )
+    # argon2.PasswordHasher().verify raises VerifyMismatchError on
+    # mismatch (and InvalidHash on a malformed hash). We let those
+    # propagate; the handler maps them to 401 / 500.
+    _get_hasher().verify(hash_str, password)
+    return True
+
+
+# ---------------------------------------------------------------------------
+# Config (env-var table names, mirroring contract_ingestor.py)
+# ---------------------------------------------------------------------------
+
+USERS_TABLE = os.environ.get("NOVA_USERS_TABLE", "nova-users")
+SESSIONS_TABLE = os.environ.get("NOVA_SESSIONS_TABLE", "nova-sessions")
+PASSWORD_RESETS_TABLE = os.environ.get(
+    "NOVA_PASSWORD_RESETS_TABLE", "nova-password-resets"
+)
+# Session lifetime (seconds). Default 24h.
+SESSION_TTL_SECONDS = int(os.environ.get("NOVA_SESSION_TTL_SECONDS", "86400"))
+# Password-reset token lifetime (seconds). Default 15 min.
+RESET_TTL_SECONDS = int(os.environ.get("NOVA_RESET_TTL_SECONDS", "900"))
+
+_dynamodb = None
+
+
+def _get_dynamodb():
+    """Lazy boto3 DynamoDB resource singleton (mirrors contract_ingestor)."""
+    global _dynamodb
+    if _dynamodb is None:
+        _dynamodb = boto3.resource("dynamodb")
+    return _dynamodb
+
+
+def _iso8601_now() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).strftime(
+        "%Y-%m-%dT%H:%M:%SZ"
+    )
+
+
+def _epoch_now() -> int:
+    return int(datetime.datetime.now(datetime.timezone.utc).timestamp())
+
+
+def _emit_audit(event_type: str, **fields) -> None:
+    """Emit an audit event to stderr as JSON (never includes passwords)."""
+    payload = {"event": event_type, "ts": _iso8601_now(), **fields}
+    # Defense-in-depth: scrub any field literally named 'password' or
+    # 'password_hash' value from the audit payload (they should never be
+    # passed here, but a stray kwarg would leak — INV-16).
+    for _k in ("password", "new_password", "old_password"):
+        payload.pop(_k, None)
+    sys.stderr.write(json.dumps(payload, sort_keys=True) + "\n")
+    sys.stderr.flush()
+
+
+# ---------------------------------------------------------------------------
+# Business logic (sign_up / sign_in / create_session / reset flows)
+# ---------------------------------------------------------------------------
+
+
+def _require(fields, payload):
+    """Validate required fields; raise ValueError (→ 400) if missing."""
+    for f in fields:
+        if f not in payload or payload[f] in (None, ""):
+            raise ValueError(f"missing field: {f}")
+
+
+def _lookup_user_by_email(email: str):
+    """Query nova-users GSI1 (email-index) → return the user item or None."""
+    table = _get_dynamodb().Table(USERS_TABLE)
+    resp = table.query(
+        IndexName="email-index",
+        KeyConditionExpression="email = :e",
+        ExpressionAttributeValues={":e": email},
+        Limit=1,
+    )
+    items = resp.get("Items", [])
+    return items[0] if items else None
+
+
+def sign_up(payload):
+    """Create a new user. Fails closed (503) if argon2 is unavailable.
+
+    Payload: { email, password, owner, roles }
+    Writes to nova-users: PK user_id (uuid4), email, password_hash,
+    owner, roles, created_at. The raw password is NEVER stored.
+    """
+    _require(("email", "password", "owner", "roles"), payload)
+    if not _ARGON2_AVAILABLE:
+        raise Argon2UnavailableError("argon2 unavailable")
+    email = payload["email"]
+    password = payload["password"]
+    owner = payload["owner"]
+    roles = payload["roles"]
+    if not isinstance(roles, list):
+        raise ValueError("roles must be a list")
+
+    # Duplicate-email check → 409.
+    if _lookup_user_by_email(email) is not None:
+        raise _DuplicateEmailError(email)
+
+    user_id = str(uuid.uuid4())
+    password_hash = hash_password(password)  # fail-closed here
+    created_at = _iso8601_now()
+    item = {
+        "user_id": user_id,
+        "email": email,
+        "password_hash": password_hash,
+        "owner": owner,
+        "roles": roles,
+        "created_at": created_at,
+    }
+    table = _get_dynamodb().Table(USERS_TABLE)
+    table.put_item(TableName=USERS_TABLE, Item=item)
+    _emit_audit("auth.sign_up", user_id=user_id, email=email)
+    return {
+        "status": "ok",
+        "action": "sign_up",
+        "user_id": user_id,
+        "email": email,
+        "created_at": created_at,
+    }
+
+
+class _DuplicateEmailError(Exception):
+    """Raised when sign_up is called with an already-registered email → 409."""
+
+    def __init__(self, email: str):
+        self.email = email
+        super().__init__(f"email already registered: {email}")
+
+
+def create_session(user_id: str) -> str:
+    """Create a session row in nova-sessions; return the session_id.
+
+    TTL: expires_at = now + SESSION_TTL_SECONDS (epoch seconds).
+    """
+    session_id = str(uuid.uuid4())
+    now = _epoch_now()
+    expires_at = now + SESSION_TTL_SECONDS
+    created_at = _iso8601_now()
+    table = _get_dynamodb().Table(SESSIONS_TABLE)
+    table.put_item(
+        TableName=SESSIONS_TABLE,
+        Item={
+            "session_id": session_id,
+            "user_id": user_id,
+            "expires_at": expires_at,
+            "created_at": created_at,
+        },
+    )
+    _emit_audit("auth.session_created", user_id=user_id, session_id=session_id)
+    return session_id
+
+
+def sign_in(payload):
+    """Sign in by email + password → return a session_id.
+
+    On wrong password → raises VerifyMismatchError (→ 401).
+    On unknown email → raises _UnknownUserError (→ 401, same code to
+    avoid user-enumeration via timing — the message is generic).
+    On argon2 unavailable → Argon2UnavailableError (→ 503).
+    """
+    _require(("email", "password"), payload)
+    if not _ARGON2_AVAILABLE:
+        raise Argon2UnavailableError("argon2 unavailable")
+    email = payload["email"]
+    password = payload["password"]
+    user = _lookup_user_by_email(email)
+    if user is None:
+        # Generic 401 — do not reveal whether the email is registered
+        # (user-enumeration defense).
+        raise _UnknownUserError("invalid credentials")
+    try:
+        verify_password(password, user["password_hash"])
+    except VerifyMismatchError:
+        raise _UnknownUserError("invalid credentials")
+    session_id = create_session(user["user_id"])
+    _emit_audit("auth.sign_in", user_id=user["user_id"], email=email)
+    return {
+        "status": "ok",
+        "action": "sign_in",
+        "user_id": user["user_id"],
+        "session_id": session_id,
+    }
+
+
+class _UnknownUserError(Exception):
+    """Generic 'invalid credentials' — 401 (no user enumeration)."""
+
+
+def request_password_reset(payload):
+    """Generate a reset token (uuid4) → write to nova-password-resets (15 min TTL).
+
+    Returns the token directly (in a real system this would be emailed;
+    for v1.28 it is returned so tests / the CLI can drive reset_password).
+    """
+    _require(("email",), payload)
+    email = payload["email"]
+    user = _lookup_user_by_email(email)
+    if user is None:
+        # Return ok regardless (no user enumeration via reset endpoint).
+        # We still return a (fake) token shape so the response is uniform;
+        # the token is single-use and reset_password validates against DDB.
+        _emit_audit("auth.password_reset_requested", email=email, found=False)
+        return {
+            "status": "ok",
+            "action": "request_password_reset",
+            "reset_token": None,
+            "message": "if the email is registered, a reset token was issued",
+        }
+    reset_token = str(uuid.uuid4())
+    now = _epoch_now()
+    expires_at = now + RESET_TTL_SECONDS
+    table = _get_dynamodb().Table(PASSWORD_RESETS_TABLE)
+    table.put_item(
+        TableName=PASSWORD_RESETS_TABLE,
+        Item={
+            "reset_token": reset_token,
+            "user_id": user["user_id"],
+            "expires_at": expires_at,
+            "created_at": _iso8601_now(),
+        },
+    )
+    _emit_audit(
+        "auth.password_reset_requested",
+        user_id=user["user_id"],
+        email=email,
+        found=True,
+    )
+    return {
+        "status": "ok",
+        "action": "request_password_reset",
+        "reset_token": reset_token,
+        "expires_at": expires_at,
+    }
+
+
+def reset_password(payload):
+    """Validate a reset token → set a new password → delete the token.
+
+    Payload: { reset_token, new_password }
+    On invalid/expired token → ValueError (→ 400).
+    On argon2 unavailable → Argon2UnavailableError (→ 503).
+    """
+    _require(("reset_token", "new_password"), payload)
+    if not _ARGON2_AVAILABLE:
+        raise Argon2UnavailableError("argon2 unavailable")
+    reset_token = payload["reset_token"]
+    new_password = payload["new_password"]
+    resets = _get_dynamodb().Table(PASSWORD_RESETS_TABLE)
+    resp = resets.get_item(
+        TableName=PASSWORD_RESETS_TABLE,
+        Key={"reset_token": reset_token},
+    )
+    item = resp.get("Item")
+    if not item:
+        raise ValueError("invalid or expired reset token")
+    if item.get("expires_at", 0) < _epoch_now():
+        # Token expired (TTL may not have reaped it yet).
+        raise ValueError("reset token expired")
+    user_id = item["user_id"]
+    new_hash = hash_password(new_password)  # fail-closed
+    users = _get_dynamodb().Table(USERS_TABLE)
+    users.update_item(
+        TableName=USERS_TABLE,
+        Key={"user_id": user_id},
+        UpdateExpression="SET password_hash = :h",
+        ExpressionAttributeValues={":h": new_hash},
+    )
+    resets.delete_item(
+        TableName=PASSWORD_RESETS_TABLE,
+        Key={"reset_token": reset_token},
+    )
+    _emit_audit("auth.password_reset", user_id=user_id)
+    return {
+        "status": "ok",
+        "action": "reset_password",
+        "user_id": user_id,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Dispatch (shared by Lambda handler + CLI — REQ-329 dual-use)
+# ---------------------------------------------------------------------------
+
+
+def dispatch_action(payload, event=None):
+    """Shared business-logic dispatch for the IdP auth Lambda (REQ-329).
+
+    Both the AWS Lambda handler (``lambda_handler``) and the CLI path
+    (``cli_main`` / ``__main__``) call this so the two paths share a
+    single source of truth for action routing.
+
+    Args:
+        payload: the decoded action envelope dict, e.g.
+            ``{ action: "sign_up", email, password, owner, roles }``.
+        event: the raw Lambda Function-URL event (unused for identity —
+            the IAM auth is enforced at the Function URL layer; kept for
+            signature symmetry with contract_ingestor).
+
+    Returns:
+        The action result dict on success. Raises on error — the caller
+        maps exceptions to status codes via :func:`_to_http_response`.
+    """
+    action = payload.get("action")
+    if action == "sign_up":
+        return sign_up(payload)
+    if action == "sign_in":
+        return sign_in(payload)
+    if action == "create_session":
+        _require(("user_id",), payload)
+        sid = create_session(payload["user_id"])
+        return {"status": "ok", "action": "create_session", "session_id": sid}
+    if action == "request_password_reset":
+        return request_password_reset(payload)
+    if action == "reset_password":
+        return reset_password(payload)
+    raise ValueError(f"unknown action: {action!r}")
+
+
+def _to_http_response(result_or_error):
+    """Map a dispatch result / exception to a Lambda HTTP response."""
+    if isinstance(result_or_error, Exception):
+        # Fail-closed: argon2 unavailable → 503 (NO weak hash, NO crash).
+        if isinstance(result_or_error, Argon2UnavailableError):
+            return {
+                "statusCode": 503,
+                "body": json.dumps({"error": "argon2_unavailable"}),
+            }
+        if isinstance(result_or_error, _DuplicateEmailError):
+            return {
+                "statusCode": 409,
+                "body": json.dumps({"error": "email_already_registered"}),
+            }
+        if isinstance(result_or_error, _UnknownUserError):
+            return {
+                "statusCode": 401,
+                "body": json.dumps({"error": "invalid_credentials"}),
+            }
+        if isinstance(result_or_error, ValueError):
+            return {
+                "statusCode": 400,
+                "body": json.dumps({"error": str(result_or_error)}),
+            }
+        return {
+            "statusCode": 500,
+            "body": json.dumps({"error": str(result_or_error)}),
+        }
+    return {"statusCode": 200, "body": json.dumps(result_or_error)}
+
+
+def lambda_handler(event, context):
+    """AWS Lambda handler entry point (thin wrapper, REQ-329 dual-use).
+
+    Accepts a Function-URL-style event whose ``body`` is a JSON string
+    containing ``{ action, email, password, ... }``. Parses the envelope
+    then delegates to :func:`dispatch_action`.
+    """
+    # Fail-closed fast-path: if argon2 is unavailable, sign_up / sign_in /
+    # reset_password all raise Argon2UnavailableError which maps to 503.
+    # We do NOT short-circuit here so non-password actions (create_session)
+    # still work when argon2 is down — only the hashing paths fail closed.
+    try:
+        body = event.get("body", "{}")
+        payload = json.loads(body) if isinstance(body, str) else body
+        result = dispatch_action(payload, event=event)
+        return _to_http_response(result)
+    except Exception as e:
+        return _to_http_response(e)
+
+
+# ---------------------------------------------------------------------------
+# CLI (dual-use, REQ-329 pattern)
+# ---------------------------------------------------------------------------
+
+
+def cli_main(argv=None):
+    """CLI entry point for the IdP auth Lambda (REQ-329 dual-use).
+
+    Usage:
+        python3 -m core.lambda.nova_idp_auth --sign-up <email> <password> <owner>
+        python3 -m core.lambda.nova_idp_auth --sign-in <email> <password>
+        python3 -m core.lambda.nova_idp_auth --create-session <user_id>
+        python3 -m core.lambda.nova_idp_auth --request-reset <email>
+        python3 -m core.lambda.nova_idp_auth --reset-password <token> <new_password>
+        python3 -m core.lambda.nova_idp_auth --dispatch <payload.json>
+        python3 -m core.lambda.nova_idp_auth --dispatch-stdin < <payload.json>
+    """
+    import sys
+
+    raw = argv if argv is not None else sys.argv[1:]
+    local_bypass = os.environ.get("NOVA_LAMBDA_LOCAL_BYPASS")
+    if not local_bypass:
+        os.environ["NOVA_LAMBDA_LOCAL_BYPASS"] = "1"
+    try:
+        if "--dispatch-stdin" in raw:
+            payload = json.loads(sys.stdin.read())
+        elif "--dispatch" in raw:
+            idx = raw.index("--dispatch")
+            path = raw[idx + 1] if idx + 1 < len(raw) else None
+            if not path:
+                print("Usage: --dispatch <payload.json>", file=sys.stderr)
+                return 2
+            with open(path) as fh:
+                payload = json.loads(fh.read())
+        elif "--sign-up" in raw:
+            idx = raw.index("--sign-up")
+            email, password, owner = raw[idx + 1 : idx + 4]
+            roles = ["user"]
+            payload = {
+                "action": "sign_up",
+                "email": email,
+                "password": password,
+                "owner": owner,
+                "roles": roles,
+            }
+        elif "--sign-in" in raw:
+            idx = raw.index("--sign-in")
+            email, password = raw[idx + 1 : idx + 3]
+            payload = {"action": "sign_in", "email": email, "password": password}
+        elif "--create-session" in raw:
+            idx = raw.index("--create-session")
+            user_id = raw[idx + 1]
+            payload = {"action": "create_session", "user_id": user_id}
+        elif "--request-reset" in raw:
+            idx = raw.index("--request-reset")
+            email = raw[idx + 1]
+            payload = {"action": "request_password_reset", "email": email}
+        elif "--reset-password" in raw:
+            idx = raw.index("--reset-password")
+            token, new_password = raw[idx + 1 : idx + 3]
+            payload = {
+                "action": "reset_password",
+                "reset_token": token,
+                "new_password": new_password,
+            }
+        else:
+            print(
+                "Usage: python3 -m core.lambda.nova_idp_auth "
+                "--sign-up <email> <password> <owner> | "
+                "--sign-in <email> <password> | "
+                "--dispatch <payload.json>",
+                file=sys.stderr,
+            )
+            return 2
+        result = dispatch_action(payload, event=None)
+        sys.stdout.write(json.dumps(result, indent=2) + "\n")
+        return 0
+    except Argon2UnavailableError as e:
+        sys.stderr.write(f"error: {e}\n")
+        return 3  # 503-class
+    except ValueError as e:
+        sys.stderr.write(f"error: {e}\n")
+        return 1
+    except _DuplicateEmailError as e:
+        sys.stderr.write(f"error: {e}\n")
+        return 9  # 409-class
+    except _UnknownUserError as e:
+        sys.stderr.write(f"error: {e}\n")
+        return 1  # 401-class
+    except Exception as e:  # pragma: no cover - defensive top-level guard
+        sys.stderr.write(f"internal error: {e}\n")
+        return 2
+    finally:
+        if not local_bypass:
+            os.environ.pop("NOVA_LAMBDA_LOCAL_BYPASS", None)
+
+
+if __name__ == "__main__":  # pragma: no cover - CLI entry
+    import sys
+
+    sys.exit(cli_main())

core/lambda/nova_idp_auth_cfn.py

@@ -0,0 +1,244 @@
+"""CloudFormation snippet for the Nova IdP DynamoDB identity schema (REQ-335).
+
+This module exports :func:`dynamodb_tables_snippet`, which returns a
+CloudFormation fragment (a plain ``dict``) defining the four DynamoDB
+tables that back the Nova identity provider:
+
+  * ``nova-users``           — user records (PK ``user_id``, GSI1 ``email``)
+  * ``nova-sessions``        — session tokens (PK ``session_id``, GSI1
+    ``user_id``, TTL ``expires_at``)
+  * ``nova-password-resets`` — reset tokens (PK ``reset_token``, TTL
+    ``expires_at`` — 15 min)
+  * ``nova-pats``            — personal access tokens (PK ``jti``, GSI1
+    ``sub``, GSI2 ``pat_hash``). This table is consumed in P4 (OIDC/PAT
+    issuance) but is defined here so a single ``nova idp setup``
+    CloudFormation template provisions the complete identity backend.
+
+Design notes (REQ-335):
+  * All tables use ``BillingMode: PAY_PER_REQUEST`` (on-demand) — the
+    IdP traffic is bursty and unpredictable; provisioned capacity would
+    either throttle or waste money.
+  * PITR (``PointInTimeRecoverySpecification``) is enabled on
+    ``nova-users`` — user records are irreplaceable; continuous backup
+    protects against accidental deletes / corrupt writes. The session /
+    reset / PAT tables are ephemeral (TTL-managed) so PITR is not
+    required there, but enabling it is cheap insurance; we enable it on
+    ``nova-users`` per REQ-335 and leave the others as on-demand only
+    (TTL is the recovery mechanism for those).
+  * TTL attributes (``expires_at``) are epoch seconds — DynamoDB TTL
+    silently deletes expired items in the background (best-effort, do
+    not rely on for access control; the handler also checks ``expires_at``
+    on read).
+
+The fragment is composed into the full ``nova idp setup`` template in
+P4 Wave 8 (``nova idp setup --apply``). The keys in the returned dict
+are CloudFormation logical resource IDs (``NovaUsersTable``, etc.) so
+the composer can merge it directly into a template's ``Resources``
+section.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+
+def _attribute(name: str, attr_type: str = "S") -> Dict[str, str]:
+    return {"AttributeName": name, "AttributeType": attr_type}
+
+
+def _key_schema(name: str, key_type: str = "HASH") -> Dict[str, str]:
+    return {"AttributeName": name, "KeyType": key_type}
+
+
+def dynamodb_tables_snippet() -> Dict[str, Dict[str, Any]]:
+    """Return a CloudFormation fragment defining the four IdP DynamoDB tables.
+
+    The returned dict maps logical resource IDs to CloudFormation
+    resource dicts (``Type: AWS::DynamoDB::Table``). It is intended to be
+    merged into the ``Resources`` block of the full
+    ``nova idp setup`` template (P4 Wave 8).
+
+    Tables:
+      * ``NovaUsersTable`` (``nova-users``)
+      * ``NovaSessionsTable`` (``nova-sessions``)
+      * ``NovaPasswordResetsTable`` (``nova-password-resets``)
+      * ``NovaPatsTable`` (``nova-pats``)
+
+    All tables are ``PAY_PER_REQUEST`` (on-demand). PITR is enabled on
+    ``nova-users`` (REQ-335). TTL is enabled on the three ephemeral
+    tables (``expires_at`` epoch-seconds attribute).
+    """
+    return {
+        # -----------------------------------------------------------------
+        # nova-users — the user directory (PK user_id, GSI1 email).
+        # PITR enabled: user records are irreplaceable.
+        # -----------------------------------------------------------------
+        "NovaUsersTable": {
+            "Type": "AWS::DynamoDB::Table",
+            "Properties": {
+                "TableName": "nova-users",
+                "BillingMode": "PAY_PER_REQUEST",
+                "KeySchema": [
+                    _key_schema("user_id", "HASH"),
+                ],
+                "AttributeDefinitions": [
+                    _attribute("user_id", "S"),
+                    _attribute("email", "S"),
+                ],
+                "GlobalSecondaryIndexes": [
+                    {
+                        "IndexName": "email-index",
+                        "KeySchema": [_key_schema("email", "HASH")],
+                        "Projection": {"ProjectionType": "ALL"},
+                    },
+                ],
+                "PointInTimeRecoverySpecification": {
+                    "PointInTimeRecoveryEnabled": True,
+                },
+                # Attribute shape (for documentation / the setup --dry-run
+                # summary; DynamoDB is schemaless so this is not enforced):
+                #   user_id       String  (PK)
+                #   email         String  (GSI1 hash, unique)
+                #   password_hash String  (Argon2id, never the raw password)
+                #   owner         String
+                #   roles         List
+                #   created_at    String  (ISO-8601)
+                "AttributeShape": {
+                    "user_id": "String",
+                    "email": "String",
+                    "password_hash": "String",
+                    "owner": "String",
+                    "roles": "List",
+                    "created_at": "String",
+                },
+            },
+        },
+        # -----------------------------------------------------------------
+        # nova-sessions — session tokens (PK session_id, GSI1 user_id).
+        # TTL: expires_at (epoch seconds). Sessions live 24h.
+        # -----------------------------------------------------------------
+        "NovaSessionsTable": {
+            "Type": "AWS::DynamoDB::Table",
+            "Properties": {
+                "TableName": "nova-sessions",
+                "BillingMode": "PAY_PER_REQUEST",
+                "KeySchema": [
+                    _key_schema("session_id", "HASH"),
+                ],
+                "AttributeDefinitions": [
+                    _attribute("session_id", "S"),
+                    _attribute("user_id", "S"),
+                ],
+                "GlobalSecondaryIndexes": [
+                    {
+                        "IndexName": "user_id-index",
+                        "KeySchema": [_key_schema("user_id", "HASH")],
+                        "Projection": {"ProjectionType": "ALL"},
+                    },
+                ],
+                "TimeToLiveSpecification": {
+                    "AttributeName": "expires_at",
+                    "Enabled": True,
+                },
+                "AttributeShape": {
+                    "session_id": "String",
+                    "user_id": "String",
+                    "expires_at": "String (epoch seconds, TTL)",
+                    "created_at": "String (ISO-8601)",
+                },
+            },
+        },
+        # -----------------------------------------------------------------
+        # nova-password-resets — reset tokens (PK reset_token).
+        # TTL: expires_at (epoch seconds). Tokens live 15 min.
+        # -----------------------------------------------------------------
+        "NovaPasswordResetsTable": {
+            "Type": "AWS::DynamoDB::Table",
+            "Properties": {
+                "TableName": "nova-password-resets",
+                "BillingMode": "PAY_PER_REQUEST",
+                "KeySchema": [
+                    _key_schema("reset_token", "HASH"),
+                ],
+                "AttributeDefinitions": [
+                    _attribute("reset_token", "S"),
+                ],
+                "TimeToLiveSpecification": {
+                    "AttributeName": "expires_at",
+                    "Enabled": True,
+                },
+                "AttributeShape": {
+                    "reset_token": "String",
+                    "user_id": "String",
+                    "expires_at": "String (epoch seconds, TTL; 15 min)",
+                },
+            },
+        },
+        # -----------------------------------------------------------------
+        # nova-pats — personal access tokens (PK jti, GSI1 sub, GSI2 pat_hash).
+        # Consumed in P4 (OIDC/PAT issuance) but defined here so the single
+        # CloudFormation template provisions the complete identity backend.
+        # TTL: expires_at (epoch seconds).
+        # -----------------------------------------------------------------
+        "NovaPatsTable": {
+            "Type": "AWS::DynamoDB::Table",
+            "Properties": {
+                "TableName": "nova-pats",
+                "BillingMode": "PAY_PER_REQUEST",
+                "KeySchema": [
+                    _key_schema("jti", "HASH"),
+                ],
+                "AttributeDefinitions": [
+                    _attribute("jti", "S"),
+                    _attribute("sub", "S"),
+                    _attribute("pat_hash", "S"),
+                ],
+                "GlobalSecondaryIndexes": [
+                    {
+                        "IndexName": "sub-index",
+                        "KeySchema": [_key_schema("sub", "HASH")],
+                        "Projection": {"ProjectionType": "ALL"},
+                    },
+                    {
+                        "IndexName": "pat_hash-index",
+                        "KeySchema": [_key_schema("pat_hash", "HASH")],
+                        "Projection": {"ProjectionType": "ALL"},
+                    },
+                ],
+                "TimeToLiveSpecification": {
+                    "AttributeName": "expires_at",
+                    "Enabled": True,
+                },
+                "AttributeShape": {
+                    "jti": "String (PK)",
+                    "sub": "String (GSI1; subject / user_id)",
+                    "pat_hash": "String (GSI2; SHA-256 of the PAT for lookup)",
+                    "status": "String (active|revoked)",
+                    "issued_at": "String (ISO-8601)",
+                    "expires_at": "String (epoch seconds, TTL)",
+                    "revoked_at": "String (ISO-8601, present iff status=revoked)",
+                    "claims": "Map (JWT claims payload)",
+                },
+            },
+        },
+    }
+
+
+def table_names() -> Dict[str, str]:
+    """Return the logical→physical table-name mapping (for env-var defaults)."""
+    return {
+        "users": "nova-users",
+        "sessions": "nova-sessions",
+        "password_resets": "nova-password-resets",
+        "pats": "nova-pats",
+    }
+
+
+if __name__ == "__main__":  # pragma: no cover - CLI inspection helper
+    import json
+    import sys
+
+    if "--names" in sys.argv:
+        sys.stdout.write(json.dumps(table_names(), indent=2) + "\n")
+    else:
+        sys.stdout.write(json.dumps(dynamodb_tables_snippet(), indent=2) + "\n")

tests/test_argon2_fail_closed.py

@@ -0,0 +1,240 @@
+"""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}"
+            )

tests/test_idp_auth.py

@@ -0,0 +1,444 @@
+"""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}"
+            )