Announcing the AWS Foundational Security Best Practices for Snowflake

with bad_statements as (
SELECT
    p.id
FROM
    aws_iam_policies p
    , lateral flatten(input => p.POLICY_VERSION_LIST) as f
    , lateral flatten(input => parse_json(f.value:Document):Statement) as s
where f.value:IsDefaultVersion = 'true'
    and s.value:Effect = 'Allow'
            and (s.value:Action = '*' or s.value:Action = '*:*')
            and s.value:Resource = '*' 
)
select
    'IAM policies should not allow full * administrative privileges' as title,
    account_id,
    arn as resource_id,
    CASE
        WHEN b.id is not null THEN 'fail'
        ELSE 'pass'
    END as status
from
    aws_iam_policies as p
LEFT JOIN bad_statements as b
    ON p.id = b.id

SELECT 
  'AWS KMS keys should not be deleted unintentionally' AS title,
    account_id,
    arn AS resource_id,
    CASE 
        WHEN key_state IN ('PendingDeletion', 'PendingReplicaDeletion') AND key_manager = 'CUSTOMER' THEN 'fail'
        ELSE 'pass'
    END AS status
FROM aws_kms_keys;

select
    'CloudTrail should have encryption at rest enabled' as title,
    account_id,
    arn as resource_id,
    case
        when kms_key_id is NULL then 'fail'
        else 'pass'
    end as status
FROM aws_cloudtrail_trails

select
  'EC2 instances should use IMDSv2' as title,
  account_id,
  instance_id as resource_id,
  case when
    metadata_options:HttpTokens is distinct from 'required'
    then 'fail'
    else 'pass'
  end as status
from aws_ec2_instances

select
  'Classic Load Balancer listeners should be configured with HTTPS or TLS termination' as title,
  lb.account_id,
  lb.arn as resource_id,
  case when
    li.value:Listener:Protocol not in ('HTTPS', 'SSL')
    then 'fail'
    else 'pass'
  end as status
from aws_elbv1_load_balancers lb, lateral flatten(input => parse_json(lb.listener_descriptions)) as li

select
    'RDS DB instances should prohibit public access, determined by the PubliclyAccessible configuration' as title,
    account_id,
    arn AS resource_id,
    case when publicly_accessible = TRUE then 'fail' else 'pass' end as status
from aws_rds_instances

select
    'S3 Block Public Access setting should be enabled' as title,
    aws_iam_accounts.account_id,
    aws_iam_accounts.account_id AS resource_id,
    case when
        config_exists is distinct from TRUE
        or block_public_acls is distinct from TRUE
        or block_public_policy is distinct from TRUE
        or ignore_public_acls is distinct from TRUE
        or restrict_public_buckets is distinct from TRUE
    then 'fail' else 'pass' end as status
from
    aws_iam_accounts
left join
    aws_s3_accounts on
        aws_iam_accounts.account_id = aws_s3_accounts.account_id

WITH snapshot_access_groups AS (
    SELECT account_id,
           region,
           snapshot_id,
           create_volume_permissions.value:Group AS "group",
           create_volume_permissions.value:UserId AS user_id
    FROM aws_ec2_ebs_snapshot_attributes, lateral flatten(input => parse_json(aws_ec2_ebs_snapshot_attributes.create_volume_permissions)) as create_volume_permissions
)
SELECT DISTINCT
  'Amazon EBS snapshots should not be public, determined by the ability to be restorable by anyone' as title,
  account_id,
  snapshot_id as resource_id,
  case when
    "group" = 'all'
    -- this is under question because
    -- trusted accounts(user_id) do not violate this control
        OR user_id IS DISTINCT FROM ''
    then 'fail'
    else 'pass'
  end as status
FROM snapshot_access_groups

SELECT
  'Auto Scaling group launch configurations should configure EC2 instances to require Instance Metadata Service Version 2' AS "title",
  account_id,
  arn AS resource_id,
  case
  when METADATA_OPTIONS:HttpTokens = 'required' then 'pass'
    else 'fail'
  END
    AS status
FROM
  aws_autoscaling_launch_configurations;

SELECT DISTINCT
    'Lambda function policies should prohibit public access' AS title,
    account_id,
    arn AS resource_id,
    'fail' AS status
FROM (
  SELECT
    account_id,
    arn
  FROM
  aws_lambda_functions,
  table(flatten(policy_document, 'Statement')) as statement
where
  statement.value:Effect = 'Allow'
  and
    statement.value:Principal = '*'
    or
    statement.value:Principal:AWS = '*'
  )