#!/usr/bin/env python3 """Render the Dedicated account list and legacy ProviderConfigs for one session.""" import argparse import json import re from pathlib import Path def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--account-list", type=Path, required=True) parser.add_argument("--context", type=Path, required=True) parser.add_argument("--output-account-list", type=Path, required=True) parser.add_argument("--output-provider-configs", type=Path, required=True) args = parser.parse_args() context = json.loads(args.context.read_text()) cloud = str(((context.get("target") or {}).get("cloud")) or "").strip() account = str(((context.get("target") or {}).get("account")) or "").strip() host = context.get("host") or {} if cloud == "gcp": gcp = host.get("gcp") or {} session = str(context.get("session_name") or "").strip() project = str(gcp.get("dataplane_project_id") or "").strip() role_arn = str(gcp.get("aws_role_arn") or "").strip() if not re.fullmatch(r"[a-z0-9][a-z0-9-]{1,61}[a-z0-9]", session): raise SystemExit("GCP Dedicated account contract requires session_name") if not re.fullmatch(r"[a-z][a-z0-9-]{4,28}[a-z0-9]", project): raise SystemExit("GCP Dedicated account contract requires host.gcp.dataplane_project_id") if not re.fullmatch(r"arn:aws:iam::\d{12}:role/[^/]+", role_arn): raise SystemExit("GCP Dedicated account contract requires host.gcp.aws_role_arn") account_list = args.account_list.read_text() if account_list.count("__CP9_SESSION_NAME__") != 1: raise SystemExit( "unable to materialize GCP Dedicated account list: missing __CP9_SESSION_NAME__" ) account_list = account_list.replace("__CP9_SESSION_NAME__", session) for pattern, replacement in [ (r'(?m)^(\s*default:)\s*.*$', rf'\1 "{project}"'), (r"(?m)^(\s*projectID:)\s*.*$", rf"\1 {project}"), ]: account_list, count = re.subn(pattern, replacement, account_list, count=1) if count != 1: raise SystemExit( f"unable to materialize GCP Dedicated account list: missing {pattern}" ) args.output_account_list.write_text(account_list) args.output_provider_configs.write_text( f"""apiVersion: aws.crossplane.io/v1beta1 kind: ProviderConfig metadata: name: provider-aws-gr spec: credentials: source: InjectedIdentity --- apiVersion: gcp.crossplane.io/v1beta1 kind: ProviderConfig metadata: name: {project} spec: projectID: {project} credentials: source: InjectedIdentity """ ) return role_arn = str( ((((context.get("terraform") or {}).get("iam") or {}).get("role_arn")) or "") ).strip() match = re.fullmatch(r"arn:aws:iam::(\d{12}):role/([^/]+)", role_arn) if not re.fullmatch(r"\d{12}", account): raise SystemExit("CP9 session context target.account must be a 12-digit AWS account ID") if not match or match.group(1) != account: raise SystemExit( "CP9 session context terraform.iam.role_arn must be a role in target.account" ) account_list = args.account_list.read_text() replacements = [ (r"(?m)^(\s*default:)\s*.*$", rf'\1 "{account}"'), (r"(?m)^(\s*awsAccount:)\s*.*$", rf"\1 {account}"), (r"(?m)^(\s*awsRole:)\s*.*$", rf"\1 {match.group(2)}"), ] for pattern, replacement in replacements: account_list, count = re.subn(pattern, replacement, account_list, count=1) if count != 1: raise SystemExit( f"unable to materialize session account list: missing {pattern}" ) args.output_account_list.write_text(account_list) args.output_provider_configs.write_text( f"""apiVersion: aws.crossplane.io/v1beta1 kind: ProviderConfig metadata: name: default spec: credentials: source: InjectedIdentity --- apiVersion: aws.crossplane.io/v1beta1 kind: ProviderConfig metadata: name: provider-aws-gr spec: credentials: source: InjectedIdentity --- apiVersion: aws.crossplane.io/v1beta1 kind: ProviderConfig metadata: name: "{account}" spec: credentials: source: InjectedIdentity assumeRoleARN: {role_arn} assumeRole: roleARN: {role_arn} """ ) if __name__ == "__main__": main()