acdl/core/lambda/nova_idp_setup.py

"""Nova IdP setup logic — check / apply / verify (REQ-340, REQ-341, C-2.1).

Backing logic for ``nova idp setup``. The CLI (``nova/idp/setup.py``)
is a thin ≤50-line delegate to this module (CAP-034).
"""

from __future__ import annotations

import importlib.util
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any


def _load_cfn():
    """Load core/lambda/nova_idp_cfn.py via importlib (`lambda` is reserved)."""
    p = Path(__file__).parent / "nova_idp_cfn.py"
    spec = importlib.util.spec_from_file_location("nova_idp_cfn", p)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


_cfn = _load_cfn()
generate_template = _cfn.generate_template
resource_summary = _cfn.resource_summary


def check_prerequisites() -> dict[str, Any]:
    """Check IdP setup prerequisites (AWS creds, CFN/IAM/KMS perms).

    Returns a report dict:
      ``{"aws_creds": bool, "region": str|None, "missing": [str], "iam_delta": [str]}``
    """
    report: dict[str, Any] = {"aws_creds": False, "region": None, "missing": [], "iam_delta": []}
    # AWS creds check.
    try:
        who = subprocess.check_output(
            ["aws", "sts", "get-caller-identity"], stderr=subprocess.DEVNULL, text=True, timeout=10
        )
        report["aws_creds"] = bool(json.loads(who).get("Account"))
    except Exception:
        report["missing"].append("aws_credentials (run `aws configure`)")
    # Region.
    region = os.environ.get("AWS_DEFAULT_REGION") or os.environ.get("AWS_REGION")
    report["region"] = region
    if not region:
        report["missing"].append("aws_region (set AWS_DEFAULT_REGION)")
    # IAM policy delta (the grants the deploying principal needs).
    report["iam_delta"] = [
        "cloudformation:*",
        "iam:CreateRole",
        "iam:PassRole",
        "lambda:CreateFunction",
        "lambda:CreateFunctionUrlConfig",
        "dynamodb:CreateTable",
        "kms:CreateKey",
        "kms:CreateAlias",
    ]
    return report


def generate_and_deploy(
    public_jwks_domain: str | None = None,
    dry_run: bool = False,
    approve_fn=None,
) -> dict[str, Any]:
    """Generate the CFN template + deploy (REQ-341, NFR-10 y/N approval).

    Args:
        public_jwks_domain: optional custom JWKS domain.
        dry_run: if True, print the resource summary only (no deploy).
        approve_fn: callable returning True/False for the y/N prompt
            (defaults to stdin readline).

    Returns:
        ``{"template": <dict>, "summary": <dict>, "deployed": bool}``.
    """
    template = generate_template(public_jwks_domain)
    summary = resource_summary(template)
    if dry_run:
        return {"template": template, "summary": summary, "deployed": False}
    # NFR-10: explicit y/N approval before cloudformation deploy.
    print("Resource summary:")
    for rtype, count in sorted(summary.items()):
        print(f"  {rtype}: {count}")
    # Print template to a temp file + open $PAGER.
    tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False, encoding="utf-8")
    json.dump(template, tmp, indent=2); tmp.flush(); tmp.close()
    pager = os.environ.get("PAGER")
    if pager and sys.stdin.isatty():
        try:
            subprocess.run([pager, tmp.name])
        except Exception:
            print(f"(template at {tmp.name})")
    else:
        print(f"(template at {tmp.name})")
    # y/N prompt.
    if approve_fn is None:
        answer = input("Apply? [y/N] ").strip().lower()
    else:
        answer = "y" if approve_fn() else "n"
    if answer != "y":
        print("aborted (no approval)")
        return {"template": template, "summary": summary, "deployed": False}
    # cloudformation deploy.
    stack_name = os.environ.get("NOVA_IDP_STACK_NAME", "nova-idp")
    try:
        subprocess.check_call([
            "aws", "cloudformation", "deploy",
            "--stack-name", stack_name,
            "--template-file", tmp.name,
            "--capabilities", "CAPABILITY_IAM",
        ])
        deployed = True
    except Exception as e:
        print(f"deploy failed: {e}", file=sys.stderr)
        deployed = False
    return {"template": template, "summary": summary, "deployed": deployed}


def verify() -> dict[str, Any]:
    """Run the KMS round-trip verification (REQ-340 --verify).

    Delegates to the CAP-037 test logic: sign a JWT (mock KMS) → JWKS →
    pyjwt verify. Returns ``{"passed": bool, "detail": str}``.
    """
    try:
        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

        priv = ec.generate_private_key(ec.SECP256R1())
        pub_der = priv.public_key().public_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )

        class _MockKms:
            def sign(self, KeyId, Message, MessageType, SigningAlgorithm):
                return {"Signature": priv.sign(Message, ec.ECDSA(hashes.SHA256()))}
            def get_public_key(self, KeyId):
                return {"PublicKey": pub_der}

        kms_signing.set_kms_client_for_testing(_MockKms())
        token = kms_signing.sign_jwt({"sub": "verify", "exp": 9999999999, "iat": 1, "jti": "v"})
        jwk = kms_signing.get_jwk()
        key = pyjwt.PyJWK(jwk).key
        decoded = pyjwt.decode(token, key, algorithms=["ES256"], options={"verify_aud": False})
        ok = decoded["sub"] == "verify"
        return {"passed": ok, "detail": "KMS round-trip OK" if ok else "mismatch"}
    except Exception as e:
        return {"passed": False, "detail": f"verify error: {e}"}
    finally:
        try:
            kms_signing.set_kms_client_for_testing(None)
        except Exception:
            pass


if __name__ == "__main__":  # pragma: no cover - CLI inspection helper
    mode = sys.argv[1] if len(sys.argv) > 1 else "--check"
    if mode == "--check":
        print(json.dumps(check_prerequisites(), indent=2))
    elif mode == "--dry-run":
        print(json.dumps(generate_and_deploy(dry_run=True)["summary"], indent=2))
    elif mode == "--verify":
        print(json.dumps(verify(), indent=2))
    else:
        print("usage: nova_idp_setup.py --check|--dry-run|--verify", file=sys.stderr)