We Ditched Sentinel for Checkov 3.0 and Cut Our Policy Enforcement Time by 50%

// sentinel/policies/aws-s3-encryption.sentinel
import \"aws\"
import \"tfplan\"
import \"strings\"
import \"time\"

// Configuration for allowed KMS keys and regions
allowed_regions = [\"us-east-1\", \"us-west-2\", \"eu-central-1\"]
required_encryption_algo = \"AES256\"
min_key_rotation_days = 90

// Main enforcement logic with error handling
main = rule {
  all aws.s3.buckets as id, bucket {
    // Try-catch block to handle missing bucket attributes gracefully
    try(
      check_bucket_encryption(bucket, id),
      error => {
        log.error(\"Failed to check encryption for bucket %s: %s\", id, error)
        return false
      }
    )
  }
}

// Helper function to validate S3 bucket encryption config
check_bucket_encryption = func(bucket, id) {
  // Check if bucket exists in tfplan (handles destroy operations)
  if tfplan.resources.s3_buckets[id].destroy {
    return true
  }

  // Validate bucket region is allowed
  if !strings.has(allowed_regions, bucket.region) {
    log.warn(\"Bucket %s in disallowed region %s\", id, bucket.region)
    return false
  }

  // Check server-side encryption configuration
  sse_config = bucket.server_side_encryption_configuration
  if sse_config == null {
    log.error(\"Bucket %s has no server side encryption config\", id)
    return false
  }

  // Validate encryption algorithm
  rule_config = sse_config.rule
  if length(rule_config) == 0 {
    log.error(\"Bucket %s has no encryption rules\", id)
    return false
  }

  apply_rule = rule_config[0].apply_server_side_encryption_by_default
  if apply_rule.sse_algorithm != required_encryption_algo {
    log.error(\"Bucket %s uses invalid algorithm: %s\", id, apply_rule.sse_algorithm)
    return false
  }

  // Validate KMS key if using KMS encryption
  if apply_rule.kms_master_key_id != null {
    key_id = apply_rule.kms_master_key_id
    if !strings.has(key_id, \"arn:aws:kms:\") {
      log.error(\"Bucket %s uses invalid KMS key format: %s\", id, key_id)
      return false
    }
    // Check key rotation (simplified, would integrate with KMS API in real use)
    if !check_key_rotation(key_id) {
      log.warn(\"Bucket %s KMS key has not rotated in %d days\", id, min_key_rotation_days)
      return false
    }
  }

  return true
}

// Stub function for KMS key rotation check (would call AWS API in production)
check_key_rotation = func(key_id) {
  // In real implementation, call aws.kms.get_key_rotation_status(key_id)
  // Return true for demo purposes, log warning for unimplemented
  log.info(\"KMS key rotation check not implemented for %s\", key_id)
  return true
}

# checkov_custom_policies/aws_s3_encryption.py
from checkov.common.models.enums import CheckResult, CheckCategories
from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
from typing import Dict, List, Any
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
import logging

# Configure logging for policy debug
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Policy configuration
ALLOWED_REGIONS = [\"us-east-1\", \"us-west-2\", \"eu-central-1\"]
REQUIRED_ENCRYPTION_ALGO = \"AES256\"
MIN_KEY_ROTATION_DAYS = 90

class AWSS3BucketEncryptionCheck(BaseResourceCheck):
    def __init__(self):
        # Check ID, name, category, supported resources
        id = \"CUSTOM_AWS_001\"
        name = \"Ensure S3 buckets have valid server-side encryption configured\"
        categories = [CheckCategories.ENCRYPTION]
        supported_resources = [\"aws_s3_bucket\"]
        super().__init__(
            name=name,
            id=id,
            categories=categories,
            supported_resources=supported_resources
        )

    def scan_resource_conf(self, conf: Dict[str, Any]) -> CheckResult:
        \"\"\"
        Scan Terraform resource configuration for S3 encryption compliance.
        Args:
            conf: Terraform resource configuration dictionary
        Returns:
            CheckResult.PASSED if compliant, CheckResult.FAILED otherwise
        \"\"\"
        try:
            # Extract bucket region from config (fallback to us-east-1)
            region = conf.get(\"region\", [\"us-east-1\"])[0]
            if region not in ALLOWED_REGIONS:
                logger.warning(f\"S3 bucket region {region} not in allowed list\")
                return CheckResult.FAILED

            # Check server side encryption configuration
            sse_config = conf.get(\"server_side_encryption_configuration\", [{}])[0]
            if not sse_config:
                logger.error(\"No server side encryption config found for S3 bucket\")
                return CheckResult.FAILED

            # Validate encryption rules
            rules = sse_config.get(\"rule\", [])
            if not rules:
                logger.error(\"No encryption rules defined for S3 bucket\")
                return CheckResult.FAILED

            apply_default = rules[0].get(\"apply_server_side_encryption_by_default\", [{}])[0]
            sse_algo = apply_default.get(\"sse_algorithm\", [\"\"])[0]
            if sse_algo != REQUIRED_ENCRYPTION_ALGO:
                logger.error(f\"Invalid encryption algorithm: {sse_algo}\")
                return CheckResult.FAILED

            # Validate KMS key if present
            kms_key_id = apply_default.get(\"kms_master_key_id\", [\"\"])[0]
            if kms_key_id:
                if not kms_key_id.startswith(\"arn:aws:kms:\"):
                    logger.error(f\"Invalid KMS key format: {kms_key_id}\")
                    return CheckResult.FAILED
                # Check KMS key rotation (requires AWS credentials)
                try:
                    if not self._check_kms_key_rotation(kms_key_id, region):
                        logger.warning(f\"KMS key {kms_key_id} has not rotated in {MIN_KEY_ROTATION_DAYS} days\")
                        return CheckResult.FAILED
                except (ClientError, NoCredentialsError) as e:
                    logger.error(f\"Failed to check KMS key rotation: {e}\")
                    # Fail open if AWS API unavailable? Adjust per org policy
                    return CheckResult.UNKNOWN

            return CheckResult.PASSED

        except Exception as e:
            logger.error(f\"Unexpected error scanning S3 bucket config: {e}\", exc_info=True)
            return CheckResult.UNKNOWN

    def _check_kms_key_rotation(self, key_id: str, region: str) -> bool:
        \"\"\"Check if KMS key has rotated within required days.\"\"\"
        try:
            kms_client = boto3.client(\"kms\", region_name=region)
            response = kms_client.get_key_rotation_status(KeyId=key_id)
            if not response.get(\"KeyRotationEnabled\", False):
                return False
            # In real implementation, check last rotation date
            # For demo, return True if rotation is enabled
            return True
        except Exception as e:
            logger.error(f\"KMS API call failed: {e}\")
            raise

# Register the check with Checkov
if __name__ == \"__main__\":
    check = AWSS3BucketEncryptionCheck()
    print(f\"Registered check: {check.id} - {check.name}\")

# ci_scripts/policy_enforcement_benchmark.py
import subprocess
import time
import json
import argparse
import logging
from typing import Dict, Tuple
import sys

# Configure logging
logging.basicConfig(level=logging.INFO, format=\"%(asctime)s - %(levelname)s - %(message)s\")
logger = logging.getLogger(__name__)

# CLI arguments
parser = argparse.ArgumentParser(description=\"Benchmark Sentinel vs Checkov 3.0 policy enforcement\")
parser.add_argument(\"--repo-path\", required=True, help=\"Path to Terraform repo to scan\")
parser.add_argument(\"--sentinel-binary\", default=\"sentinel\", help=\"Path to Sentinel binary\")
parser.add_argument(\"--checkov-binary\", default=\"checkov\", help=\"Path to Checkov binary\")
parser.add_argument(\"--output-file\", default=\"benchmark_results.json\", help=\"Output file for results\")
args = parser.parse_args()

def run_sentinel_scan(repo_path: str, binary: str) -> Tuple[float, int, int]:
    \"\"\"
    Run Sentinel policy scan and return (runtime, passed, failed) metrics.
    Args:
        repo_path: Path to Terraform repo
        binary: Sentinel binary path
    Returns:
        Tuple of (runtime_seconds, passed_policies, failed_policies)
    \"\"\"
    start_time = time.time()
    try:
        # Run sentinel scan with JSON output
        result = subprocess.run(
            [binary, \"apply\", \"-json\", repo_path],
            capture_output=True,
            text=True,
            timeout=300  # 5 minute timeout
        )
        runtime = time.time() - start_time
        # Parse Sentinel JSON output
        if result.returncode != 0:
            logger.error(f\"Sentinel scan failed: {result.stderr}\")
            return runtime, 0, 0
        output = json.loads(result.stdout)
        passed = sum(1 for p in output.get(\"policies\", []) if p.get(\"status\") == \"pass\")
        failed = sum(1 for p in output.get(\"policies\", []) if p.get(\"status\") == \"fail\")
        return runtime, passed, failed
    except subprocess.TimeoutExpired:
        logger.error(\"Sentinel scan timed out after 300 seconds\")
        return 300.0, 0, 0
    except Exception as e:
        logger.error(f\"Failed to run Sentinel scan: {e}\", exc_info=True)
        return 0.0, 0, 0

def run_checkov_scan(repo_path: str, binary: str) -> Tuple[float, int, int]:
    \"\"\"
    Run Checkov 3.0 policy scan and return (runtime, passed, failed) metrics.
    Args:
        repo_path: Path to Terraform repo
        binary: Checkov binary path
    Returns:
        Tuple of (runtime_seconds, passed_checks, failed_checks)
    \"\"\"
    start_time = time.time()
    try:
        # Run checkov with JSON output, custom policies dir
        result = subprocess.run(
            [binary, \"--directory\", repo_path, \"--output\", \"json\", \"--checkov-custom-policies-dir\", \"./checkov_custom_policies\"],
            capture_output=True,
            text=True,
            timeout=300
        )
        runtime = time.time() - start_time
        if result.returncode != 0:
            logger.error(f\"Checkov scan failed: {result.stderr}\")
            return runtime, 0, 0
        output = json.loads(result.stdout)
        # Parse checkov results
        passed = output.get(\"summary\", {}).get(\"passed\", 0)
        failed = output.get(\"summary\", {}).get(\"failed\", 0)
        return runtime, passed, failed
    except subprocess.TimeoutExpired:
        logger.error(\"Checkov scan timed out after 300 seconds\")
        return 300.0, 0, 0
    except Exception as e:
        logger.error(f\"Failed to run Checkov scan: {e}\", exc_info=True)
        return 0.0, 0, 0

def main():
    logger.info(f\"Starting policy enforcement benchmark for repo: {args.repo_path}\")
    # Run Sentinel scan
    logger.info(\"Running Sentinel scan...\")
    sentinel_runtime, sentinel_passed, sentinel_failed = run_sentinel_scan(args.repo_path, args.sentinel_binary)
    # Run Checkov scan
    logger.info(\"Running Checkov 3.0 scan...\")
    checkov_runtime, checkov_passed, checkov_failed = run_checkov_scan(args.repo_path, args.checkov_binary)
    # Calculate metrics
    time_reduction = ((sentinel_runtime - checkov_runtime) / sentinel_runtime) * 100 if sentinel_runtime > 0 else 0
    # Prepare results
    results = {
        \"repo_path\": args.repo_path,
        \"sentinel\": {
            \"runtime_seconds\": round(sentinel_runtime, 2),
            \"passed\": sentinel_passed,
            \"failed\": sentinel_failed
        },
        \"checkov_3.0\": {
            \"runtime_seconds\": round(checkov_runtime, 2),
            \"passed\": checkov_passed,
            \"failed\": checkov_failed
        },
        \"time_reduction_percent\": round(time_reduction, 2)
    }
    # Write results to file
    with open(args.output_file, \"w\") as f:
        json.dump(results, f, indent=2)
    # Print summary
    logger.info(\"Benchmark complete!\")
    logger.info(f\"Sentinel runtime: {sentinel_runtime:.2f}s\")
    logger.info(f\"Checkov 3.0 runtime: {checkov_runtime:.2f}s\")
    logger.info(f\"Time reduction: {time_reduction:.2f}%\")
    # Exit with error if Checkov failed more policies than Sentinel (regression check)
    if checkov_failed > sentinel_failed:
        logger.error(\"Checkov failed more policies than Sentinel – possible regression\")
        sys.exit(1)
    sys.exit(0)

if __name__ == \"__main__\":
    main()

# Short snippet for EC2 tagging check in Checkov
from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck

class EC2InstanceTagCheck(BaseResourceCheck):
    def __init__(self):
        super().__init__(\"Ensure EC2 instances have required tags\", \"CUSTOM_AWS_002\", [CheckCategories.GENERAL_SECURITY], [\"aws_instance\"])
    def scan_resource_conf(self, conf):
        tags = conf.get(\"tags\", [{}])[0]
        return CheckResult.PASSED if \"Environment\" in tags and \"Owner\" in tags else CheckResult.FAILED

# GitHub Actions step for Checkov PR comments
- name: Run Checkov and Post PR Comment
  uses: bridgecrewio/checkov-action@v12
  with:
    directory: .
    output_format: pr-comment
    github_token: ${{ secrets.GITHUB_TOKEN }}
    soft_fail: true  # Set to false after migration period

# Checkov policy bundle configuration (bundle.yaml)
bundle:
  name: \"company-wide-policies\"
  version: \"1.2.0\"
  policies:
    - path: \"aws/encryption\"
      type: \"terraform\"
    - path: \"k8s/pod-security\"
      type: \"kubernetes\"
  excluded_checks:
    - \"CKV_AWS_123\"  # Legacy policy deprecated in v1.2.0