Control: Redshift security groups should allow ingress on the cluster port only from restricted origins

powerpipe control run aws_compliance.control.redshift_cluster_security_group_restrict_ingress_redshift_port

powerpipe login
powerpipe control run aws_compliance.control.redshift_cluster_security_group_restrict_ingress_redshift_port --share

with redshift_security_groups as (
  select
    arn,
    sg ->> 'VpcSecurityGroupId' as security_group_id,
    region,
    account_id,
    coalesce((endpoint ->> 'Port')::int, 5439) as redshift_port
  from
    aws_redshift_cluster,
    jsonb_array_elements(vpc_security_groups) as sg
), ingress_redshift_port as (
  select
    distinct rsg.arn as redshift_arn,
    group_id as sg_id,
    redshift_port
  from
    redshift_security_groups as rsg
    left join aws_vpc_security_group_rule as sgr on sgr.group_id = rsg.security_group_id
  where
    type = 'ingress'
    and
      (cidr_ipv4 = '0.0.0.0/0'
      or cidr_ipv6 = '::/0')
    and
    (
      ( ip_protocol = '-1'
      and from_port is null
      )
      or (
        from_port <= redshift_port
        and to_port >= redshift_port
      )
    )
  group by
    group_id,
    rsg.arn,
    rsg.redshift_port
), redshift_sg_agg as (
  select
    redshift_arn,
    redshift_port,
    array_agg(distinct sg_id) as offending_sgs,
    count(distinct sg_id)  as offending_sg_count
  from ingress_redshift_port
  group by redshift_arn, redshift_port
)
select
  arn as resource,
  case
    when offending_sg_count > 0  then 'alarm'
    else 'ok'
  end as status,
  case
    when offending_sg_count > 0  then title || ' contains ' || offending_sg_count || ' security groups ' || array_to_string(coalesce(offending_sgs, array[]::text[]), ', ') || ' with ingress rule(s) from 0.0.0.0/0 or ::/0 to Redshift port ' ||  p.redshift_port || '.'
    else title || ' has no security groups with ingress from 0.0.0.0/0 or ::/0 to Redshift port.'
  end as reason
  
  , region, account_id
from
  aws_redshift_cluster as r
  left join redshift_sg_agg as p on p.redshift_arn = r.arn;