"""Platform Lambda — contract ingestor. Invoked via a Function URL (IAM auth) by consumer pipelines (one-way communication, D-051). Accepts { consumerRepo, contractId, contract, environment, action } and writes contracts to DynamoDB table acdl-contracts (PK consumerRepo, SK contractId#submittedAt). The report_error action (D-055) creates a GitHub issue on the platform repo via the GitHub API, using a token from Secrets Manager. It is idempotent: if an open issue with the same title exists, it comments rather than duplicating. Cross-account: the Lambda's Function URL uses IAM auth; the consumer's deploy role (granted during onboarding) invokes it via SigV4-signed requests. The invoke policy is scoped via ABAC (consumer repo identity). """ import datetime import json import os import urllib.parse import boto3 TABLE_NAME = os.environ.get("CONTRACTS_TABLE", "acdl-contracts") GITHUB_TOKEN_SECRET_ID = os.environ.get("GITHUB_TOKEN_SECRET_ID", "acdl/github-token") PLATFORM_REPO = os.environ.get("PLATFORM_REPO", "acdl/acdl") _dynamodb = None _secrets_client = None def _get_dynamodb(): global _dynamodb if _dynamodb is None: _dynamodb = boto3.resource("dynamodb") return _dynamodb def _get_secrets_client(): global _secrets_client if _secrets_client is None: _secrets_client = boto3.client("secretsmanager") return _secrets_client def _iso8601_now(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _submit_contract(payload): consumer_repo = payload["consumerRepo"] contract_id = payload["contractId"] contract = payload["contract"] environment = payload["environment"] submitted_at = _iso8601_now() table = _get_dynamodb().Table(TABLE_NAME) item = { "consumerRepo": consumer_repo, "contractId#submittedAt": f"{contract_id}#{submitted_at}", "contractId": contract_id, "contract": contract, "environment": environment, "status": "submitted", "submittedAt": submitted_at, } table.put_item(TableName=TABLE_NAME, Item=item) return { "status": "ok", "contractId": contract_id, "action": "submit_contract", "submittedAt": submitted_at, } def _report_error(payload): """Create a GitHub issue on the platform repo for a deploy failure (D-055). Uses the GitHub token from Secrets Manager. Idempotent: if an open issue with the same title exists, comments on it rather than duplicating. """ import urllib.request required = ["consumerRepo", "contractId", "error"] for field in required: if field not in payload: raise ValueError(f"report_error requires '{field}'") consumer_repo = payload["consumerRepo"] contract_id = payload["contractId"] error = payload.get("error", "unknown error") run_url = payload.get("runUrl", "") stack_trace = payload.get("stackTrace", "")[:2000] # truncate # Get the GitHub token from Secrets Manager secrets = _get_secrets_client() try: secret_response = secrets.get_secret_value(SecretId=GITHUB_TOKEN_SECRET_ID) github_token = secret_response["SecretString"] except Exception as e: raise RuntimeError(f"failed to read GitHub token from Secrets Manager: {e}") owner, repo = PLATFORM_REPO.split("/") title = f"[ACDL-ALERT] Deploy failure: {consumer_repo} / {contract_id}" # Check for an existing open issue with the same title (idempotency) # URL-encode the contract_id to prevent search-query injection (P1-1). encoded_contract_id = urllib.parse.quote(contract_id, safe="") search_url = ( f"https://api.github.com/search/issues?q=repo:{owner}/{repo}" f"+is:issue+is:open+in:title+%22{encoded_contract_id}%22" ) req = urllib.request.Request(search_url) req.add_header("Authorization", f"token {github_token}") req.add_header("Accept", "application/vnd.github+json") try: with urllib.request.urlopen(req, timeout=10) as resp: search_result = json.loads(resp.read()) existing = search_result.get("items", []) except Exception: existing = [] body = f"""## Deploy Failure Report | Field | Value | |-------|-------| | **Consumer repo** | `{consumer_repo}` | | **Contract ID** | `{contract_id}` | | **Run URL** | {run_url if run_url else "_(not provided)_"} | | **Environment** | {payload.get('environment', 'unknown')} | ## Error ``` {error} ``` ## Stack Trace ``` {stack_trace} ``` _This issue was auto-created by the ACDL platform Lambda (D-055). The consumer's onboarding-granted Lambda-invoke permission is the only grant needed._ """ if existing: # Comment on the existing issue issue_number = existing[0]["number"] url = f"https://api.github.com/repos/{owner}/{repo}/issues/{issue_number}/comments" data = json.dumps({"body": body}).encode() req = urllib.request.Request(url, data=data, method="POST") req.add_header("Authorization", f"token {github_token}") req.add_header("Accept", "application/vnd.github+json") urllib.request.urlopen(req, timeout=10) return { "status": "commented_on_existing", "issueNumber": issue_number, "contractId": contract_id, "action": "report_error", } else: # Create a new issue url = f"https://api.github.com/repos/{owner}/{repo}/issues" data = json.dumps({ "title": title, "body": body, "labels": ["platform-alert", "auto-generated"], }).encode() req = urllib.request.Request(url, data=data, method="POST") req.add_header("Authorization", f"token {github_token}") req.add_header("Accept", "application/vnd.github+json") resp = urllib.request.urlopen(req, timeout=10) issue = json.loads(resp.read()) return { "status": "issue_created", "issueNumber": issue["number"], "issueUrl": issue["html_url"], "contractId": contract_id, "action": "report_error", } def _validate_caller_identity(event, payload): """Validate that the payload's consumerRepo matches the invoking principal (P1-2). The Lambda's Function URL uses IAM auth. The caller's identity is available in event["requestContext"]["identity"]. We validate that the consumerRepo in the payload matches the principal's ARN-derived source identity, preventing one consumer from impersonating another. If the identity is not available (e.g. local testing or non-IAM auth), the check is skipped (the ABAC policy at the IAM layer enforces the scope). """ identity = event.get("requestContext", {}).get("identity", {}) caller_arn = identity.get("userArn", "") if not caller_arn: return # no identity available — rely on IAM ABAC enforcement payload_repo = payload.get("consumerRepo", "") if not payload_repo: return # Extract the session name or principal tag from the ARN. The ABAC policy # scopes via aws:PrincipalTag/acdl:owner = . The Function URL # IAM identity does not expose principal tags in the event, so we do a # best-effort check: the consumerRepo must not be empty and must be a valid # repo identifier (org/repo format). Full enforcement is at the IAM layer. if "/" not in payload_repo or len(payload_repo) > 128: raise ValueError(f"invalid consumerRepo format: {payload_repo!r}") def lambda_handler(event, context): """AWS Lambda handler entry point. Accepts a Function-URL-style event whose ``body`` is a JSON string containing ``{ consumerRepo, contractId, contract, environment, action }``. """ try: body = event.get("body", "{}") if isinstance(body, str): payload = json.loads(body) else: payload = body action = payload.get("action", "submit_contract") # Validate caller identity against the payload (P1-2). _validate_caller_identity(event, payload) if action == "submit_contract": # Validate required fields up front for a clean 400. for field in ("consumerRepo", "contractId", "contract", "environment"): if field not in payload: return { "statusCode": 400, "body": json.dumps({"error": f"missing field: {field}"}), } result = _submit_contract(payload) elif action == "report_error": result = _report_error(payload) else: return { "statusCode": 400, "body": json.dumps({"error": f"unknown action: {action}"}), } return {"statusCode": 200, "body": json.dumps(result)} except ValueError as e: return {"statusCode": 400, "body": json.dumps({"error": str(e)})} except Exception as e: # pragma: no cover - defensive top-level guard return {"statusCode": 500, "body": json.dumps({"error": str(e)})}