Cloud Resource Tagging Validation: Automated Compliance Testing

# Terraform: AWS Organizations Tag Policy
resource "aws_organizations_policy" "tagging" {
  name = "mandatory-tagging-policy"
  type = "TAG_POLICY"

  content = jsonencode({
    tags = {
      Environment = {
        tag_key = {
          @@assign = "Environment"
        }
        tag_value = {
          @@assign = ["production", "staging", "development", "sandbox"]
        }
        enforced_for = {
          @@assign = ["ec2:instance", "rds:db", "s3:bucket"]
        }
      }
      CostCenter = {
        tag_key = {
          @@assign = "CostCenter"
        }
        tag_value = {
          @@assign = ["CC-\\d{4}"]  # Regex: CC-0000 format
        }
      }
      Owner = {
        tag_key = {
          @@assign = "Owner"
        }
      }
    }
  })
}

# Lambda function for AWS Config custom rule
import boto3
import json

def lambda_handler(event, context):
    """Check if EC2 instances have required tags."""

    config = boto3.client('config')
    required_tags = ['Environment', 'CostCenter', 'Owner', 'Application']

    invoking_event = json.loads(event['invokingEvent'])
    configuration_item = invoking_event['configurationItem']

    if configuration_item['resourceType'] != 'AWS::EC2::Instance':
        return build_evaluation(event, 'NOT_APPLICABLE')

    tags = configuration_item.get('tags', {})
    missing_tags = [tag for tag in required_tags if tag not in tags]

    if missing_tags:
        annotation = f"Missing required tags: {', '.join(missing_tags)}"
        return build_evaluation(event, 'NON_COMPLIANT', annotation)

    # Validate tag value formats
    if tags.get('CostCenter') and not tags['CostCenter'].startswith('CC-'):
        return build_evaluation(event, 'NON_COMPLIANT',
                               'CostCenter must start with CC-')

    return build_evaluation(event, 'COMPLIANT')

def build_evaluation(event, compliance_type, annotation=''):
    return {
        'ComplianceResourceType': event['configurationItem']['resourceType'],
        'ComplianceResourceId': event['configurationItem']['resourceId'],
        'ComplianceType': compliance_type,
        'Annotation': annotation,
        'OrderingTimestamp': event['notificationCreationTime']
    }

{
  "mode": "Indexed",
  "policyRule": {
    "if": {
      "anyOf": [
        {
          "field": "tags['Environment']",
          "exists": "false"
        },
        {
          "field": "tags['CostCenter']",
          "exists": "false"
        },
        {
          "field": "tags['Owner']",
          "exists": "false"
        }
      ]
    },
    "then": {
      "effect": "deny"
    }
  },
  "parameters": {}
}

# Azure Tag Compliance Report
$requiredTags = @('Environment', 'CostCenter', 'Owner', 'Application')
$validEnvironments = @('production', 'staging', 'development', 'sandbox')

$resources = Get-AzResource

$complianceReport = foreach ($resource in $resources) {
    $tags = $resource.Tags
    $issues = @()

    # Check for missing tags
    foreach ($requiredTag in $requiredTags) {
        if (-not $tags.ContainsKey($requiredTag)) {
            $issues += "Missing: $requiredTag"
        }
    }

    # Validate Environment values
    if ($tags.ContainsKey('Environment')) {
        if ($tags['Environment'] -notin $validEnvironments) {
            $issues += "Invalid Environment: $($tags['Environment'])"
        }
    }

    # Validate CostCenter format (CC-XXXX)
    if ($tags.ContainsKey('CostCenter')) {
        if ($tags['CostCenter'] -notmatch '^CC-\d{4}$') {
            $issues += "Invalid CostCenter format: $($tags['CostCenter'])"
        }
    }

    [PSCustomObject]@{
        ResourceName = $resource.Name
        ResourceType = $resource.ResourceType
        ResourceGroup = $resource.ResourceGroupName
        Compliant = ($issues.Count -eq 0)
        Issues = $issues -join '; '
    }
}

# Export report
$complianceReport | Export-Csv -Path "tag-compliance-report.csv" -NoTypeInformation

# Calculate compliance percentage
$totalResources = $complianceReport.Count
$compliantResources = ($complianceReport | Where-Object { $_.Compliant }).Count
$compliancePercent = [math]::Round(($compliantResources / $totalResources) * 100, 2)

Write-Host "Tag Compliance: $compliancePercent% ($compliantResources/$totalResources resources)"

# organization-policy.yaml
constraint: constraints/compute.requireLabels
listPolicy:
  allowedValues:

    - environment
    - cost_center
    - owner
    - application
  inheritFromParent: true

from google.cloud import asset_v1
from google.cloud import resourcemanager_v3
import re

class GCPLabelValidator:
    """Validate GCP resource labels against organizational policy."""

    REQUIRED_LABELS = ['environment', 'cost_center', 'owner', 'application']
    VALID_ENVIRONMENTS = ['production', 'staging', 'development', 'sandbox']
    COST_CENTER_PATTERN = re.compile(r'^cc-\d{4}$')

    def __init__(self, project_id: str):
        self.project_id = project_id
        self.asset_client = asset_v1.AssetServiceClient()

    def list_all_resources(self) -> list:
        """List all resources in the project."""
        parent = f"projects/{self.project_id}"

        request = asset_v1.ListAssetsRequest(
            parent=parent,
            content_type=asset_v1.ContentType.RESOURCE,
            asset_types=[
                "compute.googleapis.com/Instance",
                "storage.googleapis.com/Bucket",
                "sqladmin.googleapis.com/Instance",
                "container.googleapis.com/Cluster"
            ]
        )

        resources = []
        for asset in self.asset_client.list_assets(request=request):
            resources.append(asset)

        return resources

    def validate_resource(self, resource) -> dict:
        """Validate a single resource's labels."""
        labels = resource.resource.data.get('labels', {})
        issues = []

        # Check required labels
        for required in self.REQUIRED_LABELS:
            if required not in labels:
                issues.append(f"Missing label: {required}")

        # Validate environment value
        if 'environment' in labels:
            if labels['environment'] not in self.VALID_ENVIRONMENTS:
                issues.append(f"Invalid environment: {labels['environment']}")

        # Validate cost_center format
        if 'cost_center' in labels:
            if not self.COST_CENTER_PATTERN.match(labels['cost_center']):
                issues.append(f"Invalid cost_center format: {labels['cost_center']}")

        return {
            'resource_name': resource.name,
            'resource_type': resource.asset_type,
            'compliant': len(issues) == 0,
            'issues': issues
        }

    def generate_compliance_report(self) -> dict:
        """Generate full compliance report for the project."""
        resources = self.list_all_resources()
        results = [self.validate_resource(r) for r in resources]

        compliant = sum(1 for r in results if r['compliant'])
        total = len(results)

        return {
            'project': self.project_id,
            'total_resources': total,
            'compliant_resources': compliant,
            'compliance_percentage': round((compliant / total) * 100, 2) if total > 0 else 100,
            'resources': results
        }

# Usage
validator = GCPLabelValidator('my-project-id')
report = validator.generate_compliance_report()
print(f"Compliance: {report['compliance_percentage']}%")

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Dict, Optional
import boto3
from azure.identity import DefaultAzureCredential
from azure.mgmt.resource import ResourceManagementClient
from google.cloud import asset_v1

@dataclass
class TagViolation:
    resource_id: str
    resource_type: str
    cloud_provider: str
    violation_type: str
    details: str

@dataclass
class TagPolicy:
    required_tags: List[str]
    valid_environments: List[str]
    cost_center_pattern: str
    owner_email_pattern: str

class CloudTagValidator(ABC):
    """Abstract base class for cloud-specific tag validators."""

    def __init__(self, policy: TagPolicy):
        self.policy = policy

    @abstractmethod
    def get_resources(self) -> List[Dict]:
        pass

    @abstractmethod
    def get_tags(self, resource: Dict) -> Dict[str, str]:
        pass

    @abstractmethod
    def get_cloud_name(self) -> str:
        pass

    def validate_tags(self, tags: Dict[str, str], resource_id: str,
                      resource_type: str) -> List[TagViolation]:
        """Validate tags against policy."""
        violations = []

        # Check required tags
        for required in self.policy.required_tags:
            if required.lower() not in {k.lower() for k in tags.keys()}:
                violations.append(TagViolation(
                    resource_id=resource_id,
                    resource_type=resource_type,
                    cloud_provider=self.get_cloud_name(),
                    violation_type='MISSING_TAG',
                    details=f"Missing required tag: {required}"
                ))

        # Validate environment value
        env_tag = next((v for k, v in tags.items()
                       if k.lower() == 'environment'), None)
        if env_tag and env_tag.lower() not in self.policy.valid_environments:
            violations.append(TagViolation(
                resource_id=resource_id,
                resource_type=resource_type,
                cloud_provider=self.get_cloud_name(),
                violation_type='INVALID_VALUE',
                details=f"Invalid environment value: {env_tag}"
            ))

        return violations

    def run_validation(self) -> List[TagViolation]:
        """Run validation across all resources."""
        all_violations = []

        for resource in self.get_resources():
            tags = self.get_tags(resource)
            resource_id = resource.get('id', resource.get('name', 'unknown'))
            resource_type = resource.get('type', 'unknown')

            violations = self.validate_tags(tags, resource_id, resource_type)
            all_violations.extend(violations)

        return all_violations

class AWSTagValidator(CloudTagValidator):
    def __init__(self, policy: TagPolicy, region: str = 'us-east-1'):
        super().__init__(policy)
        self.ec2 = boto3.client('ec2', region_name=region)
        self.rds = boto3.client('rds', region_name=region)

    def get_cloud_name(self) -> str:
        return 'AWS'

    def get_resources(self) -> List[Dict]:
        resources = []

        # EC2 instances
        instances = self.ec2.describe_instances()
        for reservation in instances['Reservations']:
            for instance in reservation['Instances']:
                resources.append({
                    'id': instance['InstanceId'],
                    'type': 'EC2::Instance',
                    'tags': {t['Key']: t['Value'] for t in instance.get('Tags', [])}
                })

        # RDS instances
        dbs = self.rds.describe_db_instances()
        for db in dbs['DBInstances']:
            tags = self.rds.list_tags_for_resource(
                ResourceName=db['DBInstanceArn']
            )['TagList']
            resources.append({
                'id': db['DBInstanceIdentifier'],
                'type': 'RDS::DBInstance',
                'tags': {t['Key']: t['Value'] for t in tags}
            })

        return resources

    def get_tags(self, resource: Dict) -> Dict[str, str]:
        return resource.get('tags', {})

# Multi-cloud orchestration
def validate_all_clouds(policy: TagPolicy,
                        aws_regions: List[str],
                        azure_subscriptions: List[str],
                        gcp_projects: List[str]) -> Dict:
    """Run tag validation across all cloud providers."""

    all_violations = []

    # AWS validation
    for region in aws_regions:
        validator = AWSTagValidator(policy, region)
        all_violations.extend(validator.run_validation())

    # Azure and GCP validators would follow similar pattern...

    # Generate summary
    by_cloud = {}
    for v in all_violations:
        by_cloud.setdefault(v.cloud_provider, []).append(v)

    return {
        'total_violations': len(all_violations),
        'by_cloud': {cloud: len(violations)
                    for cloud, violations in by_cloud.items()},
        'violations': all_violations
    }

# .github/workflows/tag-validation.yml
name: Infrastructure Tag Validation

on:
  pull_request:
    paths:

      - 'terraform/**'
      - 'cloudformation/**'

jobs:
  validate-tags:
    runs-on: ubuntu-latest
    steps:

      - uses: actions/checkout@v4

      - name: Setup Terraform
        uses: hashicorp/setup-terraform@v3

      - name: Terraform Init
        run: terraform init
        working-directory: terraform/

      - name: Generate Plan
        run: terraform plan -out=tfplan
        working-directory: terraform/

      - name: Validate Tags in Plan
        run: |
          terraform show -json tfplan > plan.json
          python scripts/validate-tags.py plan.json
        working-directory: terraform/

      - name: Run Checkov Tag Validation
        uses: bridgecrewio/checkov-action@v12
        with:
          directory: terraform/
          check: CKV_AWS_153,CKV_AWS_154  # Tag-related checks

      - name: Post Results
        if: failure()
        uses: actions/github-script@v7
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: '❌ Tag validation failed. Please ensure all resources have required tags: Environment, CostCenter, Owner, Application'
            })

#!/usr/bin/env python3
"""Validate tags in Terraform plan output."""

import json
import sys
from pathlib import Path

REQUIRED_TAGS = ['Environment', 'CostCenter', 'Owner', 'Application']
TAGGABLE_RESOURCE_TYPES = [
    'aws_instance', 'aws_s3_bucket', 'aws_rds_cluster',
    'aws_lambda_function', 'aws_ecs_cluster', 'aws_eks_cluster',
    'azurerm_virtual_machine', 'azurerm_storage_account',
    'google_compute_instance', 'google_storage_bucket'
]

def validate_terraform_plan(plan_file: str) -> list:
    """Validate tags in Terraform plan JSON."""

    with open(plan_file) as f:
        plan = json.load(f)

    violations = []

    for change in plan.get('resource_changes', []):
        resource_type = change['type']
        resource_name = change['name']

        # Skip non-taggable resources
        if resource_type not in TAGGABLE_RESOURCE_TYPES:
            continue

        # Skip resources being destroyed
        if change['change']['actions'] == ['delete']:
            continue

        after = change['change'].get('after', {})
        tags = after.get('tags', {}) or {}

        # Check for missing required tags
        missing = [tag for tag in REQUIRED_TAGS if tag not in tags]

        if missing:
            violations.append({
                'resource': f"{resource_type}.{resource_name}",
                'missing_tags': missing
            })

    return violations

if __name__ == '__main__':
    if len(sys.argv) != 2:
        print("Usage: validate-tags.py <plan.json>")
        sys.exit(1)

    violations = validate_terraform_plan(sys.argv[1])

    if violations:
        print("❌ Tag validation failed!\n")
        for v in violations:
            print(f"  Resource: {v['resource']}")
            print(f"  Missing:  {', '.join(v['missing_tags'])}\n")
        sys.exit(1)

    print("✅ All resources have required tags")
    sys.exit(0)