Terraform Patterns

# main.tf — Terraform config with deliberate security issues
#
# This Terraform configuration contains common security anti-patterns
# for the terraform-patterns skill scanner to detect:
#   - Overly permissive CIDR blocks (0.0.0.0/0)
#   - Public S3 bucket
#   - Missing encryption (EBS, RDS, S3)
#   - Hardcoded credentials
#   - No logging or monitoring
#   - Overly broad IAM policies
#   - Missing tags
#   - Default VPC usage

terraform {
  required_version = ">= 1.5.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = "us-east-1"
  # SECURITY ISSUE: hardcoded credentials
  access_key = "AKIAIOSFODNN7EXAMPLE"
  secret_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
}

# SECURITY ISSUE: security group allows all inbound traffic
resource "aws_security_group" "web" {
  name        = "web-sg"
  description = "Web server security group"

  ingress {
    description = "SSH from anywhere"
    from_port   = 22
    to_port     = 22
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]  # ISSUE: SSH open to the world
  }

  ingress {
    description = "HTTP"
    from_port   = 80
    to_port     = 80
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]
  }

  ingress {
    description = "HTTPS"
    from_port   = 443
    to_port     = 443
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]
  }

  ingress {
    description = "Database"
    from_port   = 5432
    to_port     = 5432
    protocol    = "tcp"
    cidr_blocks = ["0.0.0.0/0"]  # ISSUE: DB port open to the world
  }

  egress {
    from_port   = 0
    to_port     = 0
    protocol    = "-1"
    cidr_blocks = ["0.0.0.0/0"]
  }
}

# SECURITY ISSUE: public S3 bucket with no encryption
resource "aws_s3_bucket" "data" {
  bucket = "acme-corp-customer-data"
  # ISSUE: no tags
}

resource "aws_s3_bucket_public_access_block" "data" {
  bucket = aws_s3_bucket.data.id

  # ISSUE: public access not blocked
  block_public_acls       = false
  block_public_policy     = false
  ignore_public_acls      = false
  restrict_public_buckets = false
}

# ISSUE: no server-side encryption configured
# (missing: aws_s3_bucket_server_side_encryption_configuration)

# ISSUE: no versioning enabled
# (missing: aws_s3_bucket_versioning)

# ISSUE: no access logging
# (missing: aws_s3_bucket_logging)

# SECURITY ISSUE: EC2 instance with no encryption, public IP, no IMDSv2
resource "aws_instance" "web" {
  ami           = "ami-0c55b159cbfafe1f0"
  instance_type = "t3.large"

  vpc_security_group_ids = [aws_security_group.web.id]

  associate_public_ip_address = true  # ISSUE: public IP assigned

  # ISSUE: no encryption on root volume
  root_block_device {
    volume_size = 50
    volume_type = "gp3"
    encrypted   = false  # ISSUE: unencrypted EBS
  }

  # ISSUE: no IMDSv2 enforcement
  # (missing: metadata_options { http_tokens = "required" })

  # ISSUE: user data with inline secrets
  user_data = <<-EOF
    #!/bin/bash
    echo "DB_PASSWORD=pr0duct10n_s3cret" >> /etc/environment
    echo "API_KEY=sk-live-abc123def456" >> /etc/environment
  EOF

  # ISSUE: no tags
}

# SECURITY ISSUE: RDS with no encryption, public access, weak password
resource "aws_db_instance" "main" {
  identifier     = "acme-production-db"
  engine         = "postgres"
  engine_version = "15.4"
  instance_class = "db.t3.medium"

  allocated_storage = 100
  storage_type      = "gp3"

  db_name  = "acme_production"
  username = "admin"
  password = "admin123"  # ISSUE: weak hardcoded password

  publicly_accessible = true   # ISSUE: database publicly accessible
  storage_encrypted   = false  # ISSUE: storage not encrypted

  skip_final_snapshot       = true  # ISSUE: no final snapshot on deletion
  backup_retention_period   = 0     # ISSUE: no automated backups

  # ISSUE: no multi-AZ for production
  multi_az = false

  # ISSUE: no tags
}

# SECURITY ISSUE: overly permissive IAM policy
resource "aws_iam_role" "app" {
  name = "acme-app-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action    = "sts:AssumeRole"
      Effect    = "Allow"
      Principal = { Service = "ec2.amazonaws.com" }
    }]
  })
}

resource "aws_iam_role_policy" "app" {
  name = "acme-app-policy"
  role = aws_iam_role.app.id

  # ISSUE: wildcard permissions — grants full access to everything
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect   = "Allow"
      Action   = "*"
      Resource = "*"
    }]
  })
}

#!/usr/bin/env python3
"""
Terraform Module Analyzer - Analyze Terraform modules for complexity and quality.

Examines module structure, variable documentation, output coverage,
resource complexity, and dependency patterns.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import json
import os
import re
import sys
from dataclasses import dataclass, asdict, field
from pathlib import Path
from typing import List, Dict, Any, Optional, Set


@dataclass
class Variable:
    """A Terraform variable."""
    name: str
    type_defined: bool
    has_default: bool
    has_description: bool
    description: str = ""


@dataclass
class Output:
    """A Terraform output."""
    name: str
    has_description: bool
    has_value: bool


@dataclass
class Resource:
    """A Terraform resource."""
    type: str
    name: str
    provider: str


@dataclass
class ModuleCall:
    """A module call."""
    name: str
    source: str


@dataclass
class ModuleAnalysis:
    """Complete module analysis result."""
    path: str
    files: List[str]
    resources: List[Resource]
    variables: List[Variable]
    outputs: List[Output]
    module_calls: List[ModuleCall]
    data_sources: List[str]
    complexity_score: int
    findings: List[Dict[str, str]]


class TerraformParser:
    """Parse Terraform HCL files for structure analysis."""

    RESOURCE_PATTERN = re.compile(r'resource\s+"(\w+)"\s+"(\w+)"')
    DATA_PATTERN = re.compile(r'data\s+"(\w+)"\s+"(\w+)"')
    VARIABLE_PATTERN = re.compile(r'variable\s+"(\w+)"')
    OUTPUT_PATTERN = re.compile(r'output\s+"(\w+)"')
    MODULE_PATTERN = re.compile(r'module\s+"(\w+)"')
    TYPE_PATTERN = re.compile(r'^\s+type\s*=')
    DEFAULT_PATTERN = re.compile(r'^\s+default\s*=')
    DESCRIPTION_PATTERN = re.compile(r'^\s+description\s*=\s*"([^"]*)"')
    VALUE_PATTERN = re.compile(r'^\s+value\s*=')
    SOURCE_PATTERN = re.compile(r'^\s+source\s*=\s*"([^"]*)"')

    def parse_file(self, filepath: Path) -> Dict[str, Any]:
        """Parse a single .tf file."""
        content = filepath.read_text()
        lines = content.split("\n")

        resources = []
        data_sources = []
        variables = []
        outputs = []
        module_calls = []

        i = 0
        while i < len(lines):
            line = lines[i]

            # Resources
            m = self.RESOURCE_PATTERN.search(line)
            if m:
                provider = m.group(1).split("_")[0]
                resources.append(Resource(type=m.group(1), name=m.group(2), provider=provider))

            # Data sources
            m = self.DATA_PATTERN.search(line)
            if m:
                data_sources.append(f"{m.group(1)}.{m.group(2)}")

            # Variables
            m = self.VARIABLE_PATTERN.search(line)
            if m:
                var = self._parse_variable_block(lines, i, m.group(1))
                variables.append(var)

            # Outputs
            m = self.OUTPUT_PATTERN.search(line)
            if m:
                out = self._parse_output_block(lines, i, m.group(1))
                outputs.append(out)

            # Module calls
            m = self.MODULE_PATTERN.search(line)
            if m:
                mod = self._parse_module_block(lines, i, m.group(1))
                module_calls.append(mod)

            i += 1

        return {
            "resources": resources,
            "data_sources": data_sources,
            "variables": variables,
            "outputs": outputs,
            "module_calls": module_calls,
        }

    def _parse_variable_block(self, lines: List[str], start: int, name: str) -> Variable:
        """Parse a variable block for type, default, description."""
        has_type = False
        has_default = False
        has_description = False
        description = ""
        brace_depth = 0
        started = False

        for i in range(start, min(start + 30, len(lines))):
            line = lines[i]
            if "{" in line:
                brace_depth += line.count("{")
                started = True
            if "}" in line:
                brace_depth -= line.count("}")
            if started and brace_depth <= 0:
                break

            if self.TYPE_PATTERN.search(line):
                has_type = True
            if self.DEFAULT_PATTERN.search(line):
                has_default = True
            m = self.DESCRIPTION_PATTERN.search(line)
            if m:
                has_description = True
                description = m.group(1)

        return Variable(name=name, type_defined=has_type, has_default=has_default,
                       has_description=has_description, description=description)

    def _parse_output_block(self, lines: List[str], start: int, name: str) -> Output:
        """Parse an output block."""
        has_description = False
        has_value = False
        brace_depth = 0
        started = False

        for i in range(start, min(start + 15, len(lines))):
            line = lines[i]
            if "{" in line:
                brace_depth += line.count("{")
                started = True
            if "}" in line:
                brace_depth -= line.count("}")
            if started and brace_depth <= 0:
                break
            if self.DESCRIPTION_PATTERN.search(line):
                has_description = True
            if self.VALUE_PATTERN.search(line):
                has_value = True

        return Output(name=name, has_description=has_description, has_value=has_value)

    def _parse_module_block(self, lines: List[str], start: int, name: str) -> ModuleCall:
        """Parse a module call block."""
        source = ""
        brace_depth = 0
        started = False

        for i in range(start, min(start + 30, len(lines))):
            line = lines[i]
            if "{" in line:
                brace_depth += line.count("{")
                started = True
            if "}" in line:
                brace_depth -= line.count("}")
            if started and brace_depth <= 0:
                break
            m = self.SOURCE_PATTERN.search(line)
            if m:
                source = m.group(1)

        return ModuleCall(name=name, source=source)


def calculate_complexity(analysis: Dict[str, Any]) -> int:
    """Calculate complexity score 0-100."""
    score = 0
    num_resources = len(analysis["resources"])
    num_variables = len(analysis["variables"])
    num_outputs = len(analysis["outputs"])
    num_modules = len(analysis["module_calls"])
    num_data = len(analysis["data_sources"])

    # Resource count contribution (0-30)
    if num_resources > 25:
        score += 30
    elif num_resources > 15:
        score += 20
    elif num_resources > 8:
        score += 10
    else:
        score += min(num_resources, 5)

    # Variable sprawl (0-25)
    if num_variables > 30:
        score += 25
    elif num_variables > 20:
        score += 15
    elif num_variables > 10:
        score += 8
    else:
        score += min(num_variables, 4)

    # Module dependencies (0-20)
    score += min(num_modules * 4, 20)

    # Data source complexity (0-15)
    score += min(num_data * 3, 15)

    # Output count (0-10)
    score += min(num_outputs, 10)

    return min(score, 100)


def generate_findings(analysis: Dict[str, Any], complexity: int) -> List[Dict[str, str]]:
    """Generate findings from analysis."""
    findings = []

    # Undocumented variables
    undoc_vars = [v for v in analysis["variables"] if not v.has_description]
    if undoc_vars:
        findings.append({
            "severity": "warning",
            "category": "documentation",
            "message": f"{len(undoc_vars)} variable(s) missing description: {', '.join(v.name for v in undoc_vars[:5])}",
            "recommendation": "Add description to all variables for maintainability.",
        })

    # Untyped variables
    untyped = [v for v in analysis["variables"] if not v.type_defined]
    if untyped:
        findings.append({
            "severity": "warning",
            "category": "quality",
            "message": f"{len(untyped)} variable(s) missing type constraint: {', '.join(v.name for v in untyped[:5])}",
            "recommendation": "Add type constraints to prevent misconfiguration.",
        })

    # Undocumented outputs
    undoc_out = [o for o in analysis["outputs"] if not o.has_description]
    if undoc_out:
        findings.append({
            "severity": "info",
            "category": "documentation",
            "message": f"{len(undoc_out)} output(s) missing description.",
            "recommendation": "Add descriptions to outputs for downstream consumers.",
        })

    # High complexity
    if complexity > 80:
        findings.append({
            "severity": "critical",
            "category": "complexity",
            "message": f"Module complexity score is {complexity}/100 (critical).",
            "recommendation": "Break this module into smaller, focused sub-modules.",
        })
    elif complexity > 60:
        findings.append({
            "severity": "warning",
            "category": "complexity",
            "message": f"Module complexity score is {complexity}/100 (high).",
            "recommendation": "Consider splitting this module to reduce complexity.",
        })

    # Multiple providers
    providers = set(r.provider for r in analysis["resources"])
    if len(providers) > 2:
        findings.append({
            "severity": "info",
            "category": "structure",
            "message": f"Module uses {len(providers)} different providers: {', '.join(providers)}.",
            "recommendation": "Consider separating resources by provider into distinct modules.",
        })

    return findings


def analyze_module(path: Path) -> Optional[ModuleAnalysis]:
    """Analyze a single Terraform module directory."""
    tf_files = list(path.glob("*.tf"))
    if not tf_files:
        return None

    parser = TerraformParser()
    all_resources = []
    all_variables = []
    all_outputs = []
    all_modules = []
    all_data = []

    for tf_file in tf_files:
        try:
            parsed = parser.parse_file(tf_file)
            all_resources.extend(parsed["resources"])
            all_variables.extend(parsed["variables"])
            all_outputs.extend(parsed["outputs"])
            all_modules.extend(parsed["module_calls"])
            all_data.extend(parsed["data_sources"])
        except Exception as e:
            pass

    combined = {
        "resources": all_resources,
        "variables": all_variables,
        "outputs": all_outputs,
        "module_calls": all_modules,
        "data_sources": all_data,
    }

    complexity = calculate_complexity(combined)
    findings = generate_findings(combined, complexity)

    return ModuleAnalysis(
        path=str(path),
        files=[f.name for f in tf_files],
        resources=all_resources,
        variables=all_variables,
        outputs=all_outputs,
        module_calls=all_modules,
        data_sources=all_data,
        complexity_score=complexity,
        findings=findings,
    )


def format_text(results: List[ModuleAnalysis]) -> str:
    """Format as human-readable text."""
    lines = []
    lines.append("=" * 60)
    lines.append("TERRAFORM MODULE ANALYSIS REPORT")
    lines.append("=" * 60)

    for mod in results:
        lines.append(f"\nModule: {mod.path}")
        lines.append(f"  Files: {', '.join(mod.files)}")
        lines.append(f"  Resources: {len(mod.resources)}")
        lines.append(f"  Variables: {len(mod.variables)}")
        lines.append(f"  Outputs: {len(mod.outputs)}")
        lines.append(f"  Module calls: {len(mod.module_calls)}")
        lines.append(f"  Data sources: {len(mod.data_sources)}")
        lines.append(f"  Complexity: {mod.complexity_score}/100")
        lines.append("-" * 40)

        if mod.findings:
            for f in mod.findings:
                lines.append(f"  [{f['severity'].upper()}] {f['message']}")
                lines.append(f"    Fix: {f['recommendation']}")
        else:
            lines.append("  No issues found.")
        lines.append("")

    lines.append("=" * 60)
    return "\n".join(lines)


def format_json(results: List[ModuleAnalysis]) -> str:
    """Format as JSON."""
    data = []
    for mod in results:
        data.append({
            "path": mod.path,
            "files": mod.files,
            "resources": [asdict(r) for r in mod.resources],
            "variables": [asdict(v) for v in mod.variables],
            "outputs": [asdict(o) for o in mod.outputs],
            "module_calls": [asdict(m) for m in mod.module_calls],
            "data_sources": mod.data_sources,
            "complexity_score": mod.complexity_score,
            "findings": mod.findings,
        })
    return json.dumps({"modules": data}, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Analyze Terraform modules for complexity and quality."
    )
    parser.add_argument("--path", "-p", required=True, help="Path to Terraform module or directory")
    parser.add_argument("--recursive", "-r", action="store_true", help="Recursively scan subdirectories")
    parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
    args = parser.parse_args()

    root = Path(args.path)
    if not root.exists():
        print(f"Error: Path not found: {args.path}", file=sys.stderr)
        sys.exit(2)

    results = []
    if args.recursive:
        for dirpath, dirnames, filenames in os.walk(root):
            if any(f.endswith(".tf") for f in filenames):
                result = analyze_module(Path(dirpath))
                if result:
                    results.append(result)
    else:
        result = analyze_module(root)
        if result:
            results.append(result)

    if not results:
        print("No Terraform files found.", file=sys.stderr)
        sys.exit(2)

    if args.format == "json":
        print(format_json(results))
    else:
        print(format_text(results))

    if any(f["severity"] == "critical" for r in results for f in r.findings):
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Terraform Security Scanner - Scan Terraform configs for security misconfigurations.

Detects open ports, public buckets, missing encryption, overly broad IAM,
and other common security anti-patterns in Terraform code.

Author: Claude Skills Engineering Team
License: MIT
"""

import argparse
import json
import os
import re
import sys
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import List, Dict, Optional


@dataclass
class SecurityFinding:
    """A security finding."""
    severity: str  # critical, high, medium, low
    category: str
    file: str
    line: int
    resource: str
    message: str
    recommendation: str


SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3}


class TerraformSecurityScanner:
    """Scans Terraform files for security misconfigurations."""

    # Patterns for insecure configurations
    OPEN_CIDR_PATTERN = re.compile(r'cidr_blocks\s*=\s*\[\s*"0\.0\.0\.0/0"\s*\]')
    OPEN_IPV6_PATTERN = re.compile(r'ipv6_cidr_blocks\s*=\s*\[\s*"::/0"\s*\]')
    PUBLIC_ACL_PATTERN = re.compile(r'acl\s*=\s*"(public-read|public-read-write|authenticated-read)"')
    PUBLIC_ACCESS_PATTERN = re.compile(r'publicly_accessible\s*=\s*true')
    WILDCARD_ACTION_PATTERN = re.compile(r'"Action"\s*:\s*"\*"')
    WILDCARD_RESOURCE_PATTERN = re.compile(r'"Resource"\s*:\s*"\*"')
    NO_ENCRYPTION_S3 = re.compile(r'resource\s+"aws_s3_bucket"\s+"(\w+)"')
    RESOURCE_PATTERN = re.compile(r'resource\s+"(\w+)"\s+"(\w+)"')
    INGRESS_PATTERN = re.compile(r'ingress\s*\{')

    def __init__(self, min_severity: str = "low"):
        self.findings: List[SecurityFinding] = []
        self.min_severity = min_severity

    def scan_directory(self, path: Path) -> List[SecurityFinding]:
        """Scan all .tf files in a directory."""
        for dirpath, _, filenames in os.walk(path):
            for fname in filenames:
                if fname.endswith(".tf"):
                    filepath = Path(dirpath) / fname
                    self._scan_file(filepath)
        return self._filter_by_severity()

    def _filter_by_severity(self) -> List[SecurityFinding]:
        """Filter findings by minimum severity."""
        min_order = SEVERITY_ORDER.get(self.min_severity, 3)
        return [f for f in self.findings if SEVERITY_ORDER.get(f.severity, 3) <= min_order]

    def _scan_file(self, filepath: Path):
        """Scan a single Terraform file."""
        try:
            content = filepath.read_text()
        except Exception:
            return

        lines = content.split("\n")
        rel_path = str(filepath)

        self._check_open_cidrs(lines, rel_path)
        self._check_public_access(lines, rel_path)
        self._check_iam_policies(lines, rel_path)
        self._check_encryption(lines, rel_path, content)
        self._check_logging(lines, rel_path, content)
        self._check_security_groups(lines, rel_path)
        self._check_sensitive_outputs(lines, rel_path)

    def _check_open_cidrs(self, lines: List[str], filepath: str):
        """Check for open CIDR blocks in security groups."""
        current_resource = ""
        in_ingress = False

        for i, line in enumerate(lines, 1):
            rm = self.RESOURCE_PATTERN.search(line)
            if rm:
                current_resource = f"{rm.group(1)}.{rm.group(2)}"

            if self.INGRESS_PATTERN.search(line):
                in_ingress = True

            if self.OPEN_CIDR_PATTERN.search(line):
                sev = "critical" if in_ingress else "high"
                self.findings.append(SecurityFinding(
                    severity=sev,
                    category="network",
                    file=filepath,
                    line=i,
                    resource=current_resource,
                    message="Open CIDR block 0.0.0.0/0 allows access from any IP.",
                    recommendation="Restrict to specific CIDR ranges for the required source IPs.",
                ))

            if self.OPEN_IPV6_PATTERN.search(line):
                self.findings.append(SecurityFinding(
                    severity="critical",
                    category="network",
                    file=filepath,
                    line=i,
                    resource=current_resource,
                    message="Open IPv6 CIDR ::/0 allows access from any IPv6 address.",
                    recommendation="Restrict to specific IPv6 CIDR ranges.",
                ))

            if "}" in line and in_ingress:
                in_ingress = False

    def _check_public_access(self, lines: List[str], filepath: str):
        """Check for publicly accessible resources."""
        current_resource = ""

        for i, line in enumerate(lines, 1):
            rm = self.RESOURCE_PATTERN.search(line)
            if rm:
                current_resource = f"{rm.group(1)}.{rm.group(2)}"

            if self.PUBLIC_ACL_PATTERN.search(line):
                self.findings.append(SecurityFinding(
                    severity="critical",
                    category="access",
                    file=filepath,
                    line=i,
                    resource=current_resource,
                    message=f"Public ACL detected on S3 bucket.",
                    recommendation="Remove public ACL. Use bucket policies with specific principal access.",
                ))

            if self.PUBLIC_ACCESS_PATTERN.search(line):
                self.findings.append(SecurityFinding(
                    severity="high",
                    category="access",
                    file=filepath,
                    line=i,
                    resource=current_resource,
                    message="Resource is publicly accessible.",
                    recommendation="Set publicly_accessible = false unless explicitly required.",
                ))

    def _check_iam_policies(self, lines: List[str], filepath: str):
        """Check for overly broad IAM policies."""
        current_resource = ""

        for i, line in enumerate(lines, 1):
            rm = self.RESOURCE_PATTERN.search(line)
            if rm:
                current_resource = f"{rm.group(1)}.{rm.group(2)}"

            if self.WILDCARD_ACTION_PATTERN.search(line):
                self.findings.append(SecurityFinding(
                    severity="critical",
                    category="iam",
                    file=filepath,
                    line=i,
                    resource=current_resource,
                    message="IAM policy uses wildcard Action (*), granting all permissions.",
                    recommendation="Apply least-privilege: specify only the required actions.",
                ))

            if self.WILDCARD_RESOURCE_PATTERN.search(line):
                self.findings.append(SecurityFinding(
                    severity="high",
                    category="iam",
                    file=filepath,
                    line=i,
                    resource=current_resource,
                    message="IAM policy uses wildcard Resource (*), applying to all resources.",
                    recommendation="Scope to specific resource ARNs.",
                ))

            # Check for AssumeRole with broad principal
            if re.search(r'"Principal"\s*:\s*"\*"', line):
                self.findings.append(SecurityFinding(
                    severity="critical",
                    category="iam",
                    file=filepath,
                    line=i,
                    resource=current_resource,
                    message="IAM trust policy allows any principal to assume role.",
                    recommendation="Restrict Principal to specific AWS accounts or services.",
                ))

    def _check_encryption(self, lines: List[str], filepath: str, content: str):
        """Check for missing encryption configurations."""
        current_resource = ""
        current_type = ""
        block_start = 0

        for i, line in enumerate(lines, 1):
            rm = self.RESOURCE_PATTERN.search(line)
            if rm:
                # Check previous resource for missing encryption
                if current_type and block_start:
                    self._check_resource_encryption(current_type, current_resource, filepath, block_start, lines)
                current_type = rm.group(1)
                current_resource = f"{rm.group(1)}.{rm.group(2)}"
                block_start = i

        # Check last resource
        if current_type and block_start:
            self._check_resource_encryption(current_type, current_resource, filepath, block_start, lines)

    def _check_resource_encryption(self, res_type: str, resource: str, filepath: str,
                                    start: int, lines: List[str]):
        """Check a specific resource for encryption."""
        encryption_resources = {
            "aws_s3_bucket": "server_side_encryption_configuration",
            "aws_ebs_volume": "encrypted",
            "aws_rds_instance": "storage_encrypted",
            "aws_rds_cluster": "storage_encrypted",
            "aws_redshift_cluster": "encrypted",
            "aws_efs_file_system": "encrypted",
            "aws_kinesis_firehose_delivery_stream": "server_side_encryption",
        }

        expected = encryption_resources.get(res_type)
        if not expected:
            return

        # Look within the resource block (up to 50 lines)
        block_content = "\n".join(lines[start - 1:min(start + 50, len(lines))])
        if expected not in block_content:
            self.findings.append(SecurityFinding(
                severity="high",
                category="encryption",
                file=filepath,
                line=start,
                resource=resource,
                message=f"Resource {res_type} may be missing encryption ({expected}).",
                recommendation=f"Add {expected} = true or configure encryption block.",
            ))

    def _check_logging(self, lines: List[str], filepath: str, content: str):
        """Check for missing logging configurations."""
        current_resource = ""

        # Check for S3 bucket without logging
        s3_buckets = re.finditer(r'resource\s+"aws_s3_bucket"\s+"(\w+)"', content)
        for match in s3_buckets:
            bucket_name = match.group(1)
            # Simple check: look for logging block after this resource
            start_pos = match.end()
            next_resource = re.search(r'\nresource\s+', content[start_pos:])
            block = content[start_pos:start_pos + (next_resource.start() if next_resource else 500)]
            if "logging" not in block and "aws_s3_bucket_logging" not in content:
                line_num = content[:match.start()].count("\n") + 1
                self.findings.append(SecurityFinding(
                    severity="medium",
                    category="logging",
                    file=filepath,
                    line=line_num,
                    resource=f"aws_s3_bucket.{bucket_name}",
                    message="S3 bucket may not have access logging enabled.",
                    recommendation="Enable S3 access logging with aws_s3_bucket_logging resource.",
                ))

    def _check_security_groups(self, lines: List[str], filepath: str):
        """Check security group configurations."""
        current_resource = ""

        for i, line in enumerate(lines, 1):
            rm = self.RESOURCE_PATTERN.search(line)
            if rm:
                current_resource = f"{rm.group(1)}.{rm.group(2)}"

            # Check for unrestricted egress
            if re.search(r'from_port\s*=\s*0', line):
                # Look ahead for to_port = 0 (all ports)
                if i < len(lines) and re.search(r'to_port\s*=\s*0', lines[i]):
                    if i + 1 < len(lines) and re.search(r'protocol\s*=\s*"-1"', lines[i + 1]):
                        pass  # Egress all is often intentional, skip

            # Check for SSH from anywhere
            if re.search(r'from_port\s*=\s*22', line):
                nearby = "\n".join(lines[max(0, i - 3):min(len(lines), i + 5)])
                if "0.0.0.0/0" in nearby:
                    self.findings.append(SecurityFinding(
                        severity="critical",
                        category="network",
                        file=filepath,
                        line=i,
                        resource=current_resource,
                        message="SSH (port 22) open to the internet (0.0.0.0/0).",
                        recommendation="Restrict SSH access to specific IPs or use a bastion host.",
                    ))

            # Check for RDP from anywhere
            if re.search(r'from_port\s*=\s*3389', line):
                nearby = "\n".join(lines[max(0, i - 3):min(len(lines), i + 5)])
                if "0.0.0.0/0" in nearby:
                    self.findings.append(SecurityFinding(
                        severity="critical",
                        category="network",
                        file=filepath,
                        line=i,
                        resource=current_resource,
                        message="RDP (port 3389) open to the internet (0.0.0.0/0).",
                        recommendation="Restrict RDP access to specific IPs or use a VPN.",
                    ))

    def _check_sensitive_outputs(self, lines: List[str], filepath: str):
        """Check for sensitive values in outputs without sensitive flag."""
        in_output = False
        output_name = ""
        has_sensitive = False
        output_start = 0

        sensitive_keywords = ["password", "secret", "key", "token", "credential"]

        for i, line in enumerate(lines, 1):
            m = re.search(r'output\s+"(\w+)"', line)
            if m:
                if in_output and not has_sensitive:
                    for kw in sensitive_keywords:
                        if kw in output_name.lower():
                            self.findings.append(SecurityFinding(
                                severity="medium",
                                category="secrets",
                                file=filepath,
                                line=output_start,
                                resource=f"output.{output_name}",
                                message=f"Output '{output_name}' may contain sensitive data but is not marked sensitive.",
                                recommendation="Add 'sensitive = true' to this output.",
                            ))
                            break
                in_output = True
                output_name = m.group(1)
                has_sensitive = False
                output_start = i

            if in_output and "sensitive" in line and "true" in line:
                has_sensitive = True


def format_text(findings: List[SecurityFinding]) -> str:
    """Format as human-readable text."""
    lines = []
    lines.append("=" * 60)
    lines.append("TERRAFORM SECURITY SCAN REPORT")
    lines.append("=" * 60)

    by_severity = {}
    for f in findings:
        by_severity.setdefault(f.severity, []).append(f)

    total = len(findings)
    lines.append(f"\nTotal findings: {total}")
    for sev in ["critical", "high", "medium", "low"]:
        count = len(by_severity.get(sev, []))
        if count:
            lines.append(f"  {sev.upper()}: {count}")

    lines.append("-" * 60)

    for sev in ["critical", "high", "medium", "low"]:
        group = by_severity.get(sev, [])
        if not group:
            continue
        lines.append(f"\n[{sev.upper()}]")
        for f in group:
            lines.append(f"  [{f.category}] {f.file}:{f.line}")
            lines.append(f"    Resource: {f.resource}")
            lines.append(f"    Issue: {f.message}")
            lines.append(f"    Fix: {f.recommendation}")
            lines.append("")

    if not findings:
        lines.append("\nNo security issues found.")

    lines.append("=" * 60)
    return "\n".join(lines)


def format_json(findings: List[SecurityFinding]) -> str:
    """Format as JSON."""
    return json.dumps({
        "findings": [asdict(f) for f in findings],
        "summary": {
            "total": len(findings),
            "critical": sum(1 for f in findings if f.severity == "critical"),
            "high": sum(1 for f in findings if f.severity == "high"),
            "medium": sum(1 for f in findings if f.severity == "medium"),
            "low": sum(1 for f in findings if f.severity == "low"),
        }
    }, indent=2)


def main():
    parser = argparse.ArgumentParser(
        description="Scan Terraform configurations for security misconfigurations."
    )
    parser.add_argument("--path", "-p", required=True, help="Path to scan")
    parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
    parser.add_argument("--min-severity", choices=["critical", "high", "medium", "low"],
                       default="low", help="Minimum severity to report")
    args = parser.parse_args()

    path = Path(args.path)
    if not path.exists():
        print(f"Error: Path not found: {args.path}", file=sys.stderr)
        sys.exit(2)

    scanner = TerraformSecurityScanner(min_severity=args.min_severity)
    findings = scanner.scan_directory(path)

    if args.format == "json":
        print(format_json(findings))
    else:
        print(format_text(findings))

    if any(f.severity == "critical" for f in findings):
        sys.exit(1)


if __name__ == "__main__":
    main()