From Managing Upstream Infra Dependencies to Autonomous Management of Upstream Infrastructure Dependencies

# project-specific/sftp-export/main.tf — bespoke DevOps pattern
module "sftp_transfer" {
  source      = "terraform-aws-modules/transfer/aws"
  server_name = "client-a-sftp-export"

  users = {
    client_export_user = {
      home_directory      = "/exports/client-a"
      home_directory_type = "LOGICAL"
      policy              = data.aws_iam_policy_document.client_export.json
    }
  }

  tags = merge(local.common_tags, {
    Project = "client-a"
    Custom  = "true"
  })
}

resource "aws_s3_bucket" "client_export" {
  bucket = "client-a-sftp-exports-${var.environment}"
  tags   = merge(local.common_tags, { Project = "client-a" })
}

# drift_detection_agent.py
import boto3
import anthropic
import json
from dataclasses import dataclass
from typing import Any

PLATFORM_STANDARDS = {
    "s3": {
        "block_public_access": True,
        "server_side_encryption": True,
        "versioning": True,
        "cloudtrail_data_events": True,
        "lifecycle_rules": True,
    },
    "iam": {
        "no_wildcard_actions": True,
        "requires_mfa_for_console": True,
        "max_session_duration_seconds": 3600,
    },
    "kms": {
        "key_rotation_enabled": True,
        "key_policy_no_star_principal": True,
    },
}

@dataclass
class DriftFinding:
    resource_id: str
    resource_type: str
    attribute: str
    expected: Any
    actual: Any
    severity: str
    project_tag: str


def assess_s3_bucket(bucket_name: str, project_tag: str) -> list[DriftFinding]:
    s3 = boto3.client("s3")
    findings = []

    # Check block public access
    try:
        pab = s3.get_public_access_block(Bucket=bucket_name)
        config = pab["PublicAccessBlockConfiguration"]
        if not all([
            config.get("BlockPublicAcls"),
            config.get("IgnorePublicAcls"),
            config.get("BlockPublicPolicy"),
            config.get("RestrictPublicBuckets"),
        ]):
            findings.append(DriftFinding(
                resource_id=bucket_name,
                resource_type="S3Bucket",
                attribute="block_public_access",
                expected=True,
                actual=False,
                severity="HIGH",
                project_tag=project_tag,
            ))
    except s3.exceptions.NoSuchPublicAccessBlockConfiguration:
        findings.append(DriftFinding(
            resource_id=bucket_name,
            resource_type="S3Bucket",
            attribute="block_public_access",
            expected=True,
            actual="not_configured",
            severity="HIGH",
            project_tag=project_tag,
        ))

    # Check server-side encryption
    try:
        enc = s3.get_bucket_encryption(Bucket=bucket_name)
        rules = enc.get("ServerSideEncryptionConfiguration", {}).get("Rules", [])
        if not rules:
            findings.append(DriftFinding(
                resource_id=bucket_name,
                resource_type="S3Bucket",
                attribute="server_side_encryption",
                expected=True,
                actual=False,
                severity="HIGH",
                project_tag=project_tag,
            ))
    except s3.exceptions.from_code("ServerSideEncryptionConfigurationNotFoundError"):
        findings.append(DriftFinding(
            resource_id=bucket_name,
            resource_type="S3Bucket",
            attribute="server_side_encryption",
            expected=True,
            actual="not_configured",
            severity="HIGH",
            project_tag=project_tag,
        ))

    return findings


def generate_remediation_plan(findings: list[DriftFinding]) -> str:
    client = anthropic.Anthropic()
    findings_json = json.dumps([f.__dict__ for f in findings], indent=2)

    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": f"""You are an infrastructure remediation specialist.
                
The following drift findings have been detected in project-specific custom infrastructure
that should conform to the Platform's baseline security standards.

Drift Findings:
{findings_json}

For each finding, provide:
1. A precise Terraform patch (HCL) that remediates the drift
2. An estimated risk level if left unremediated
3. Any dependencies or ordering requirements for remediation

Format your response as structured remediation steps.""",
            }
        ],
    )

    return message.content[0].text


def run_drift_detection(project_buckets: dict[str, str]) -> None:
    """
    project_buckets: dict mapping bucket_name -> project_tag
    """
    all_findings = []
    for bucket_name, project_tag in project_buckets.items():
        findings = assess_s3_bucket(bucket_name, project_tag)
        all_findings.extend(findings)

    if all_findings:
        print(f"[DRIFT AGENT] Detected {len(all_findings)} drift findings")
        remediation_plan = generate_remediation_plan(all_findings)
        print("[DRIFT AGENT] AI-generated remediation plan:")
        print(remediation_plan)
    else:
        print("[DRIFT AGENT] No drift detected. Custom infrastructure is aligned with Platform standards.")

# impact_assessment_agent.py
import boto3
import anthropic
import json
import subprocess


def get_current_custom_infra_state(project_prefix: str) -> dict:
    """Retrieve all custom infrastructure resources tagged with the project prefix."""
    session = boto3.Session()
    tagging = session.client("resourcegroupstaggingapi")

    resources = []
    paginator = tagging.get_paginator("get_resources")

    for page in paginator.paginate(
        TagFilters=[{"Key": "Project", "Values": [project_prefix]}]
    ):
        resources.extend(page["ResourceTagMappingList"])

    return {"project": project_prefix, "resources": resources}


def get_platform_change_plan(terraform_plan_output: str) -> dict:
    """Parse terraform plan output to extract planned changes."""
    # In practice this would parse the structured JSON output of `terraform show -json`
    return {"raw_plan": terraform_plan_output}


def assess_change_impact(
    platform_change_plan: dict,
    custom_infra_state: dict,
) -> str:
    client = anthropic.Anthropic()

    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": f"""You are an infrastructure change impact analyst.

A Platform infrastructure change is planned. Assess whether this change will affect
any of the project-specific custom infrastructure listed below.

Platform Change Plan:
{json.dumps(platform_change_plan, indent=2)}

Custom Infrastructure State:
{json.dumps(custom_infra_state, indent=2)}

For each affected custom infrastructure resource, provide:
1. The nature of the impact (breaking, degraded, cosmetic, none)
2. Specific attributes that will be affected
3. Required changes to the custom infrastructure to maintain parity after the Platform change
4. Recommended sequencing (apply Platform change first, or custom infra change first)

If no custom infrastructure is affected, state that explicitly.""",
            }
        ],
    )

    return message.content[0].text


def run_impact_assessment(
    terraform_plan_json_path: str,
    project_tags: list[str],
) -> None:
    with open(terraform_plan_json_path) as f:
        platform_plan = json.load(f)

    for project_tag in project_tags:
        custom_state = get_current_custom_infra_state(project_tag)

        print(f"\n[IMPACT AGENT] Assessing impact for project: {project_tag}")
        impact_report = assess_change_impact(platform_plan, custom_state)
        print(impact_report)

        # In production: post impact report to PR comment, Slack, or JIRA ticket

# security_assurance_agent.py
import boto3
import anthropic
import json
from typing import Any


MANDATORY_SECURITY_CONTROLS = {
    "S3": [
        "aws:SecureTransport enforcement in bucket policy",
        "Server-side encryption enabled",
        "Block public access enabled",
        "Versioning enabled for data buckets",
        "Access logging enabled",
    ],
    "IAM": [
        "No wildcard (*) in Action with wildcard (*) in Resource",
        "No inline policies on users",
        "MFA required for console access",
    ],
    "KMS": [
        "Key rotation enabled",
        "Key policy does not allow principal *",
    ],
    "EC2": [
        "IMDSv2 required (HttpTokens: required)",
        "EBS encryption by default",
        "No public IP on instances unless explicitly required",
    ],
    "RDS": [
        "Storage encrypted",
        "Deletion protection enabled in production",
        "Multi-AZ for production workloads",
        "Automated backups enabled",
    ],
}


def audit_iam_policy(policy_document: dict) -> list[str]:
    """Check IAM policy for security control violations."""
    violations = []
    for statement in policy_document.get("Statement", []):
        effect = statement.get("Effect", "")
        actions = statement.get("Action", [])
        resources = statement.get("Resource", [])

        if isinstance(actions, str):
            actions = [actions]
        if isinstance(resources, str):
            resources = [resources]

        if effect == "Allow" and "*" in actions and "*" in resources:
            violations.append(
                "Policy allows Action:* on Resource:* — violates Platform IAM standard"
            )

    return violations


def generate_security_exception_assessment(
    resource_id: str,
    control: str,
    justification: str,
) -> str:
    """Use AI to assess whether a security exception is warranted."""
    client = anthropic.Anthropic()

    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": f"""You are a security control review agent for cloud infrastructure.

A request has been made to exempt a custom infrastructure resource from a mandatory Platform security control.

Resource: {resource_id}
Security Control: {control}
Provided Justification: {justification}

Evaluate this exception request against the following criteria:
1. Is the justification technically valid?
2. Are there compensating controls that mitigate the risk of exemption?
3. What is the residual risk if this exception is granted?
4. What is the recommended time-bound duration for this exception?
5. What monitoring should be put in place if the exception is granted?

Provide a structured assessment with a APPROVE / DENY / ESCALATE recommendation.""",
            }
        ],
    )

    return message.content[0].text


def run_security_audit(resources: list[dict[str, Any]]) -> None:
    """Audit a list of custom infrastructure resources for security control compliance."""
    violations_found = False

    for resource in resources:
        resource_type = resource.get("type")
        resource_id = resource.get("id")
        config = resource.get("config", {})

        print(f"\n[SECURITY AGENT] Auditing {resource_type}: {resource_id}")

        if resource_type == "IAM_POLICY":
            violations = audit_iam_policy(config.get("policy_document", {}))
            if violations:
                violations_found = True
                for v in violations:
                    print(f"  [VIOLATION] {v}")

        # Additional resource-type checks would be added here for S3, KMS, EC2, RDS

    if not violations_found:
        print("[SECURITY AGENT] All custom infrastructure resources comply with Platform security controls.")

# lifecycle_alignment_agent.py
import boto3
import anthropic
import json
from datetime import datetime, timezone


def get_custom_infra_manifest() -> list[dict]:
    """
    Returns the manifest of all registered custom infrastructure.
    In practice, this is stored in a DynamoDB table or an S3-backed JSON registry.
    """
    return [
        {
            "project": "client-a",
            "resource_type": "aws_transfer_server",
            "resource_id": "s-abc123",
            "created_date": "2022-04-15",
            "platform_version_at_creation": "v3.2.0",
            "owner_team": "platform-integrations",
            "last_reviewed": "2023-11-01",
            "review_frequency_days": 180,
        },
        {
            "project": "client-a",
            "resource_type": "aws_s3_bucket",
            "resource_id": "client-a-sftp-exports-prod",
            "created_date": "2022-04-15",
            "platform_version_at_creation": "v3.2.0",
            "owner_team": "platform-integrations",
            "last_reviewed": "2023-11-01",
            "review_frequency_days": 180,
        },
    ]


def get_current_platform_version() -> str:
    """Retrieve the current Platform infrastructure version from the registry."""
    # In practice: fetch from SSM Parameter Store or a version API
    return "v5.1.0"


def get_platform_changelog(from_version: str, to_version: str) -> str:
    """Retrieve Platform changelog between two versions."""
    # In practice: fetch from Git tags or a changelog API
    return f"""
Platform Changes from {from_version} to {to_version}:
- v4.0.0: S3 block public access now mandatory for all buckets
- v4.1.0: KMS key rotation enforced via SCP
- v4.2.0: CloudTrail S3 data event logging mandatory
- v5.0.0: AWS Transfer Family SFTP servers must use VPC endpoint
- v5.1.0: IAM session duration reduced to 1 hour for cross-account roles
"""


def assess_lifecycle_alignment(
    manifest: list[dict],
    current_platform_version: str,
    changelog: str,
) -> str:
    client = anthropic.Anthropic()

    now = datetime.now(timezone.utc)
    stale_resources = [
        r for r in manifest
        if (now - datetime.fromisoformat(r["last_reviewed"]).replace(tzinfo=timezone.utc)).days
        > r["review_frequency_days"]
    ]

    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": f"""You are an infrastructure lifecycle alignment agent.

The Platform has evolved from the versions at which the following custom infrastructure 
resources were created. Identify which resources are out of alignment with the current 
Platform version and what changes are required.

Current Platform Version: {current_platform_version}

Custom Infrastructure Manifest:
{json.dumps(manifest, indent=2)}

Platform Changelog (relevant sections):
{changelog}

Resources overdue for review (based on review schedule):
{json.dumps(stale_resources, indent=2)}

For each resource, determine:
1. Which Platform changes since its creation version affect it
2. What configuration updates are required to achieve parity
3. Whether any breaking changes require the resource to be replaced rather than updated
4. Priority (CRITICAL / HIGH / MEDIUM / LOW) and recommended remediation timeline

Produce a structured lifecycle alignment report.""",
            }
        ],
    )

    return message.content[0].text


def run_lifecycle_alignment() -> None:
    manifest = get_custom_infra_manifest()
    current_version = get_current_platform_version()
    changelog = get_platform_changelog("v3.2.0", current_version)

    print("[LIFECYCLE AGENT] Running lifecycle alignment assessment...")
    report = assess_lifecycle_alignment(manifest, current_version, changelog)
    print(report)