S3 Bucket Policy Confused Deputy Protection for Service Principals

AnalysisType: policy
Filename: aws_s3_bucket_policy_confused_deputy.py
PolicyID: "AWS.S3.Bucket.PolicyConfusedDeputyProtection"
DisplayName: "S3 Bucket Policy Confused Deputy Protection for Service Principals"
Enabled: true
ResourceTypes:
  - AWS.S3.Bucket
Tags:
  - AWS
  - Security Control
  - Best Practices
Severity: High
Description: >
  Ensures that S3 bucket policies with service principals include conditions to prevent the confused deputy problem.
Runbook: |
  Update the bucket policy to include conditions such as aws:SourceArn, aws:SourceAccount,
  aws:SourceOrgID, or aws:SourceOrgPaths when a service principal is specified.
Reference: https://docs.aws.amazon.com/IAM/latest/UserGuide/confused-deputy.html
Tests:
  - Name: Compliant Policy with Service Principal and Condition
    ExpectedResult: true
    Resource:
      {
        "Policy": '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"cloudtrail.amazonaws.com"},"Action":"s3:PutObject","Resource":"arn:aws:s3:::my-example-bucket/*","Condition":{"StringEquals":{"aws:SourceAccount":"123456789012"}}}]}'
      }

  - Name: Non-Compliant Policy with Service Principal and No Condition
    ExpectedResult: false
    Resource:
      {
        "Policy": '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"cloudtrail.amazonaws.com"},"Action":"s3:PutObject","Resource":"arn:aws:s3:::my-example-bucket/*"}]}'
      }
  
  - Name: Policy without Service Principal
    ExpectedResult: true
    Resource:
      {
        "Policy": '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":"arn:aws:iam::123456789012:root"},"Action":"s3:GetObject","Resource":"arn:aws:s3:::my-example-bucket/*"}]}'
      }


# ------ paired body: aws_s3_bucket_policy_confused_deputy.py ------

import json

REQUIRED_CONDITIONS = {
    "aws:SourceArn",
    "aws:SourceAccount",
    "aws:SourceOrgID",
    "aws:SourceOrgPaths",
}


def policy(resource):
    bucket_policy = resource.get("Policy")
    if bucket_policy is None:
        return True  # Pass if there is no bucket policy

    policy_statements = json.loads(bucket_policy).get("Statement", [])
    for statement in policy_statements:
        # Check if the statement includes a service principal and allows access
        principal = statement.get("Principal", {})
        if "Service" in principal and statement["Effect"] == "Allow":
            conditions = statement.get("Condition", {})
            # Flatten nested condition keys (e.g., inside "StringEquals")
            flat_condition_keys = set()
            for condition in conditions.values():
                if isinstance(condition, dict):
                    flat_condition_keys.update(condition.keys())
            # Check if any required condition key is present
            if not {str.casefold(x) for x in REQUIRED_CONDITIONS} & {
                str.casefold(x) for x in flat_condition_keys
            }:
                return False
    return True