Jon Chery
1863a85144
---ci--- project: acdl phase: 4 milestone: v1.28 status: execute persona: backend-engineer ---
236 lines
8.9 KiB
Python
236 lines
8.9 KiB
Python
"""CloudFormation template for the Nova IdP (REQ-340, REQ-341, C-2.1).
|
|
|
|
Composes the DynamoDB snippet (from P3 ``nova_idp_auth_cfn.py``) + 3
|
|
Lambdas (``nova-idp-auth``, ``nova-idp-token-vend``, ``nova-idp-jwks``)
|
|
+ KMS key (``alias/nova-oidc-signing``, ``ECC_NIST_P256``,
|
|
``SIGN_VERIFY``) + function URLs + IAM roles + optional
|
|
CloudFront/WAF/ACM (when ``public_jwks_domain`` is provided).
|
|
|
|
:func:`generate_template` returns a CloudFormation template dict (no
|
|
troposphere dependency — raw dict → JSON).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
|
|
def _load_auth_cfn():
|
|
"""Load core/lambda/nova_idp_auth_cfn.py via importlib (`lambda` is reserved)."""
|
|
p = Path(__file__).parent / "nova_idp_auth_cfn.py"
|
|
spec = importlib.util.spec_from_file_location("nova_idp_auth_cfn", p)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
_auth_cfn = _load_auth_cfn()
|
|
dynamodb_tables_snippet = _auth_cfn.dynamodb_tables_snippet
|
|
table_names = _auth_cfn.table_names
|
|
|
|
|
|
def _lambda_role(logical_id: str, table_envs: dict[str, str], kms: bool = False) -> dict:
|
|
"""Build an IAM role for a Nova IdP Lambda."""
|
|
statements = [
|
|
{
|
|
"Effect": "Allow",
|
|
"Action": ["logs:CreateLogStream", "logs:PutLogEvents"],
|
|
"Resource": {"Fn::Sub": "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/*"},
|
|
},
|
|
{
|
|
"Effect": "Allow",
|
|
"Action": ["logs:CreateLogGroup"],
|
|
"Resource": {"Fn::Sub": "arn:aws:logs:${AWS::Region}:${AWS::AccountId}:*"},
|
|
},
|
|
]
|
|
if table_envs:
|
|
statements.append({
|
|
"Effect": "Allow",
|
|
"Action": ["dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:UpdateItem",
|
|
"dynamodb:Query", "dynamodb:DeleteItem"],
|
|
"Resource": [
|
|
{"Fn::Sub": f"arn:aws:dynamodb:${{AWS::Region}}:${{AWS::AccountId}}:table/{name}"}
|
|
for name in table_envs.values()
|
|
],
|
|
})
|
|
if kms:
|
|
statements.append({
|
|
"Effect": "Allow",
|
|
"Action": ["kms:Sign", "kms:GetPublicKey", "kms:DescribeKey"],
|
|
"Resource": {"Fn::GetAtt": "NovaOidcSigningKey.Arn"},
|
|
})
|
|
return {
|
|
"Type": "AWS::IAM::Role",
|
|
"Properties": {
|
|
"AssumeRolePolicyDocument": {
|
|
"Version": "2012-10-17",
|
|
"Statement": [{
|
|
"Effect": "Allow",
|
|
"Principal": {"Service": {"Fn::Sub": "lambda.${AWS::Region}.amazonaws.com"}},
|
|
"Action": "sts:AssumeRole",
|
|
}],
|
|
},
|
|
"Policies": [{"PolicyName": f"{logical_id}Policy", "PolicyDocument": {
|
|
"Version": "2012-10-17", "Statement": statements,
|
|
}}],
|
|
},
|
|
}
|
|
|
|
|
|
def _lambda_function(logical_id: str, handler: str, role_ref: str,
|
|
env_vars: dict[str, str], memory: int = 512) -> dict:
|
|
return {
|
|
"Type": "AWS::Lambda::Function",
|
|
"Properties": {
|
|
"Handler": handler,
|
|
"Runtime": "python3.12",
|
|
"MemorySize": memory,
|
|
"Timeout": 30,
|
|
"Role": {"Fn::GetAtt": [role_ref, "Arn"]},
|
|
"Environment": {"Variables": env_vars},
|
|
"Code": {"ZipFile": "def lambda_handler(event, context):\n return {}"},
|
|
},
|
|
}
|
|
|
|
|
|
def _function_url(logical_id: str, auth_type: str = "AWS_IAM") -> dict:
|
|
return {
|
|
"Type": "AWS::Lambda::Url",
|
|
"Properties": {
|
|
"TargetFunction": {"Ref": logical_id},
|
|
"AuthType": auth_type,
|
|
},
|
|
}
|
|
|
|
|
|
def generate_template(public_jwks_domain: str | None = None) -> Dict[str, Any]:
|
|
"""Generate the full Nova IdP CloudFormation template (REQ-340).
|
|
|
|
Args:
|
|
public_jwks_domain: optional custom domain for the JWKS endpoint.
|
|
When provided, CloudFront + ACM + WAF resources are added.
|
|
|
|
Returns:
|
|
A CloudFormation template dict (``{"Resources": {...}}``).
|
|
"""
|
|
resources: Dict[str, Any] = {}
|
|
# DynamoDB tables (from P3).
|
|
resources.update(dynamodb_tables_snippet())
|
|
names = table_names()
|
|
|
|
# KMS key (ECC_NIST_P256, SIGN_VERIFY) + alias.
|
|
resources["NovaOidcSigningKey"] = {
|
|
"Type": "AWS::KMS::Key",
|
|
"Properties": {
|
|
"Description": "Nova OIDC token signing key (REQ-337, ECC_NIST_P256)",
|
|
"KeySpec": "ECC_NIST_P256",
|
|
"KeyUsage": "SIGN_VERIFY",
|
|
"KeyPolicy": {
|
|
"Version": "2012-10-17",
|
|
"Statement": [{
|
|
"Effect": "Allow",
|
|
"Principal": {"AWS": {"Fn::Sub": "arn:aws:iam::${AWS::AccountId}:root"}},
|
|
"Action": "kms:*",
|
|
"Resource": "*",
|
|
}],
|
|
},
|
|
},
|
|
}
|
|
resources["NovaOidcSigningKeyAlias"] = {
|
|
"Type": "AWS::KMS::Alias",
|
|
"Properties": {
|
|
"AliasName": "alias/nova-oidc-signing",
|
|
"TargetKeyId": {"Fn::GetAtt": "NovaOidcSigningKey.Arn"},
|
|
},
|
|
}
|
|
|
|
# Lambda roles.
|
|
auth_tables = {"users": names["users"], "sessions": names["sessions"],
|
|
"password_resets": names["password_resets"]}
|
|
resources["NovaIdpAuthRole"] = _lambda_role("NovaIdpAuth", auth_tables)
|
|
resources["NovaIdpTokenVendRole"] = _lambda_role(
|
|
"NovaIdpTokenVend", {"pats": names["pats"]}, kms=True)
|
|
resources["NovaIdpJwksRole"] = _lambda_role("NovaIdpJwks", {}, kms=True)
|
|
|
|
# Lambda functions.
|
|
common_env = {
|
|
"NOVA_USERS_TABLE": names["users"],
|
|
"NOVA_SESSIONS_TABLE": names["sessions"],
|
|
"NOVA_PASSWORD_RESETS_TABLE": names["password_resets"],
|
|
"NOVA_PATS_TABLE": names["pats"],
|
|
}
|
|
resources["NovaIdpAuthFunction"] = _lambda_function(
|
|
"NovaIdpAuth", "nova_idp_auth.lambda_handler", "NovaIdpAuthRole", common_env)
|
|
resources["NovaIdpTokenVendFunction"] = _lambda_function(
|
|
"NovaIdpTokenVend", "nova_idp_token_vend.lambda_handler", "NovaIdpTokenVendRole",
|
|
{**common_env, "NOVA_OIDC_KMS_KEY_ID": "alias/nova-oidc-signing"})
|
|
resources["NovaIdpJwksFunction"] = _lambda_function(
|
|
"NovaIdpJwks", "nova_idp_jwks.lambda_handler", "NovaIdpJwksRole",
|
|
{"NOVA_OIDC_KMS_KEY_ID": "alias/nova-oidc-signing"}, memory=256)
|
|
|
|
# Function URLs (auth Lambda: IAM; token-vend: IAM; jwks: NONE — public).
|
|
resources["NovaIdpAuthUrl"] = _function_url("NovaIdpAuthFunction", "AWS_IAM")
|
|
resources["NovaIdpTokenVendUrl"] = _function_url("NovaIdpTokenVendFunction", "AWS_IAM")
|
|
resources["NovaIdpJwksUrl"] = _function_url("NovaIdpJwksFunction", "NONE")
|
|
|
|
# Optional: CloudFront + ACM + WAF for a custom JWKS domain.
|
|
if public_jwks_domain:
|
|
resources["NovaJwksCloudFront"] = {
|
|
"Type": "AWS::CloudFront::Distribution",
|
|
"Properties": {
|
|
"DistributionConfig": {
|
|
"Enabled": True,
|
|
"Aliases": [public_jwks_domain],
|
|
"Origins": [{
|
|
"DomainName": {"Fn::GetAtt": "NovaIdpJwksUrl.Endpoint"},
|
|
"Id": "JwksOrigin",
|
|
"CustomOriginConfig": {"OriginProtocolPolicy": "https-only"},
|
|
}],
|
|
"DefaultCacheBehavior": {
|
|
"TargetOriginId": "JwksOrigin",
|
|
"ViewerProtocolPolicy": "redirect-to-https",
|
|
"ForwardedValues": {"QueryString": False},
|
|
},
|
|
"ViewerCertificate": {
|
|
"AcmCertificateArn": {"Ref": "NovaJwksAcmCert"},
|
|
"SslSupportMethod": "sni-only",
|
|
},
|
|
}
|
|
},
|
|
}
|
|
resources["NovaJwksAcmCert"] = {
|
|
"Type": "AWS::CertificateManager::Certificate",
|
|
"Properties": {"DomainName": public_jwks_domain,
|
|
"ValidationMethod": "DNS"},
|
|
}
|
|
resources["NovaJwksWafRateRule"] = {
|
|
"Type": "AWS::WAFv2::RateBasedRule",
|
|
"Properties": {
|
|
"Name": "nova-jwks-rate-limit",
|
|
"Scope": "CLOUDFRONT",
|
|
"RateLimit": 100,
|
|
"Action": {"Block": {}},
|
|
"ComparisonOperator": "GreaterThan",
|
|
"AggregateKeyType": "IP",
|
|
"DefaultCaptchaConfig": {"ImmunityTimeProperty": {"ImmunityTime": 60}},
|
|
},
|
|
}
|
|
|
|
return {"Resources": resources}
|
|
|
|
|
|
def resource_summary(template: dict) -> dict[str, int]:
|
|
"""Return ``{resource_type: count}`` for a template (for --dry-run)."""
|
|
counts: dict[str, int] = {}
|
|
for res in template.get("Resources", {}).values():
|
|
t = res.get("Type", "Unknown")
|
|
counts[t] = counts.get(t, 0) + 1
|
|
return counts
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover - CLI inspection helper
|
|
import json, sys
|
|
domain = sys.argv[1] if len(sys.argv) > 1 else None
|
|
print(json.dumps(generate_template(domain), indent=2)) |