xpk/code-dumps
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

UPD: Added IAM permission check for lambda and ec2 roles

Browse Source
This commit is contained in:
xpk
2025-09-17 20:56:57 +08:00
parent 33a976be9a
commit 2942483744
1 changed files with 193 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aws/AwsEnvReview.py
+193 -3
View File
@@ -1,4 +1,4 @@
#!/usr/bin/python3
#!/usr/bin/env python3
"""
Review AWS environment based on 6 WAR pillars, namely:
1. Operational Excellence
@@ -10,11 +10,14 @@ Review AWS environment based on 6 WAR pillars, namely:
"""
import boto3
import botocore
from botocore.exceptions import ClientError
import jmespath
import re
from pprint import pprint
from datetime import date
from mdutils.mdutils import MdUtils
import os
import json
def printTitle(level: int, title: str):
@@ -58,13 +61,16 @@ aid = sts.get_caller_identity().get("Account")
client = boto3.client('ec2', region_name="us-east-1")
regions = getAllRegions(client)
mdFile.write("-" * 5)
printTitle(3, f"Primary region: {os.environ['AWS_DEFAULT_REGION']}\n")
mdFile.write("-" * 5)
outTable = []
"""Check instances stopped for long time"""
printTitle(1, "Ec2 service review")
printTitle(2, "[Cost Optimization] Instances stopped for over 14 days")
printTitle(3, "Consider backing up and terminate instances "
"or use AutoScalingGroup to spin up and down instances as needed.")
outTable = []
for r in regions:
client = boto3.client('ec2', region_name=r)
@@ -75,6 +81,7 @@ for r in regions:
outTable.append([r, aid, i[0].get("InstanceId"), getAgeFromDate(i[0].get("UsageOperationUpdateTime"))])
printResult(outTable, "Region, AccountID, InstanceId, DaysStopped")
"""Check IMDS version"""
printTitle(2, "[Security] Insecure IDMSv1 allowed")
printTitle(3, "Consider requiring IDMSv2. For more information, "
"see https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html")
@@ -94,6 +101,7 @@ for r in regions:
])
printResult(outTable, "Region, AccountID, InstanceId, IDMSv2")
"""Check EC2 instance generation"""
printTitle(2,"[Sustainability] Use of early generation instance type")
printTitle(3, "Consider using current generation instances")
outTable = []
@@ -112,6 +120,7 @@ for r in regions:
])
printResult(outTable, "Region, AccountID, InstanceId, InstanceName, InstanceType")
"""Check unattached EBS volumes"""
printTitle(2, "[Cost Optimization] Unattached EBS volumes")
printTitle(3, "Consider backing up the volumes and delete them")
outTable = []
@@ -129,6 +138,7 @@ for r in regions:
outTable.append([r, aid, i.get("VolumeId"), i.get("Size"), i.get("VolumeType")])
printResult(outTable, "Region, AccountID, VolumeId, Size, VolumeType")
"""Check stale EBS snapshots"""
printTitle(2, "[Cost Optimization] EBS snapshots more than 365 days old")
printTitle(3,"Consider removing snapshots if no longer needed")
outTable = []
@@ -144,6 +154,7 @@ for r in regions:
[r, aid, i.get("SnapshotId"), i.get("Description")[:70], getAgeFromDate(i.get("StartTime"))])
printResult(outTable, "Region, AccountID, SnapshotId, Description, SnapshotAge")
"""Check EBS encryption"""
printTitle(2, "[Security] Unencrypted EBS volumes")
printTitle(3, "Consider replacing volume with encrypted ones. "
"One can do so by stopping the Ec2 instance, creating snapshot for the unencrypted volume, "
@@ -169,6 +180,7 @@ for r in regions:
outTable.append([r, aid, i.get("VolumeId"), i.get("Size"), i.get("VolumeType")])
printResult(outTable, "Region, AccountID, VolumeId, Size, VolumeType")
"""Check unused EIP"""
printTitle(2, "[Cost Optimization] Unused Elastic IP")
printTitle(3, "Consider deleting unused EIP")
outTable = []
@@ -181,6 +193,7 @@ for r in regions:
outTable.append([r, aid, i.get("PublicIp")])
printResult(outTable, "Region, AccountID, PublicIp")
"""Check unsafe security groups"""
printTitle(1, "Security group review")
printTitle(2, "[Security] Security group rules allowing ingress from 0.0.0.0/0")
printTitle(3, "Consider setting more restrictive rules allowing access from specific sources.")
@@ -200,6 +213,7 @@ for r in regions:
[r, aid, sgr.get("GroupId"), sgr.get("SecurityGroupRuleId"), sgr.get("FromPort"), sgr.get("ToPort")])
printResult(outTable, "Region, AccountID, SecurityGroup, Rule, FromPort, ToPort")
"""Check RDS encryption setting"""
printTitle(1, "Rds service review")
printTitle(2, "[Security] Unencrypted RDS instances")
printTitle(3, "Consider encrypting RDS instances. For more detail, see "
@@ -217,6 +231,7 @@ for r in regions:
outTable.append([r, aid, i.get("DBClusterIdentifier"), i.get("Engine")])
printResult(outTable, "Region, AccountID, DBIdentifier, Engine")
"""Check RDS instance in single AZ"""
printTitle(2, "[Reliability] RDS instance running in single availability zone")
printTitle(3, "Consider enabling multi-az for production use.")
outTable = []
@@ -246,6 +261,7 @@ for r in regions:
outTable.append([r, aid, i.get("FunctionName"), i.get("Runtime")])
printResult(outTable, "Region, AccountID, FunctionName, Runtime")
"""Check IAM access key rotation"""
printTitle(1, "Iam service review")
printTitle(2, "[Security] Iam user access key not rotated for 180 days")
printTitle(3, "Consider rotating access key")
@@ -261,6 +277,7 @@ for u in users:
outTable.append([aid, u, i.get("AccessKeyId"), getAgeFromDate(i.get("CreateDate"))])
printResult(outTable, "AccountID, UserName, AccessKeyId, AccessKeyAge")
"""Check IAM entity with admin access"""
printTitle(2, "[Security] Iam AdministratorAccess policy attached")
printTitle(3, "Consider granting minimum privileges "
"to users/groups/roles. AWS managed policies for job functions are recommended. See "
@@ -279,6 +296,65 @@ for role in jmespath.search("PolicyRoles[*].RoleName", entityResp):
outTable.append([aid, "Role", role])
printResult(outTable, "AccountID, Type, Name")
"""Check Lambda role and IAM instance profile permissions"""
printTitle(2, "[Security] Check permissions of Lamda roles and Ec2 instance roles")
printTitle(3, "Typically these roles should not have admin or iam permissions.")
outTable = []
# Get a list of roles
client = boto3.client('lambda')
roles = set()
paginator = client.get_paginator('list_functions')
for page in paginator.paginate():
for function in page['Functions']:
role_arn = function.get('Role')
if role_arn:
roles.add(role_arn.split('/')[-1])
client = boto3.client('iam')
paginator = client.get_paginator('list_instance_profiles')
for page in paginator.paginate():
for profile in page['InstanceProfiles']:
profile_name = profile['InstanceProfileName']
instance_roles = profile.get('Roles', [])
for role in instance_roles:
roles.add(role['RoleName'])
# Check inline policies for each role
client = boto3.client('iam', region_name="us-east-1")
for role in roles:
inline_policy_names = client.list_role_policies(RoleName=role)['PolicyNames']
for policy_name in inline_policy_names:
response = client.get_role_policy(RoleName=role, PolicyName=policy_name)
policy = response['PolicyDocument']
flat_actions = jmespath.search('Statement[].Action[]', policy)
if "*" in flat_actions or "iam:*" in flat_actions:
outTable.append([aid, role, policy_name, "Inline policy contains * or iam:*, please review it"])
# Check managed policies for each role
for role in roles:
attached_policies = client.list_attached_role_policies(RoleName=role)['AttachedPolicies']
for policy in attached_policies:
policy_arn = policy['PolicyArn']
policy_name = policy['PolicyName']
# Get the policy default version
policy_info = client.get_policy(PolicyArn=policy_arn)['Policy']
default_version_id = policy_info['DefaultVersionId']
# Get the policy document of the default version
version = client.get_policy_version(PolicyArn=policy_arn, VersionId=default_version_id)
policy_document = version['PolicyVersion']['Document']
flat_actions = jmespath.search('Statement[].Action[]', policy_document)
if "*" in flat_actions or "iam:*" in flat_actions:
outTable.append([aid, role, policy_name, "Managed policy contains * or iam:*, please review it"])
printResult(outTable, "AccountID, RoleName, PolicyName, Issue")
"""Check cloudwatch log group retention"""
printTitle(1, "Cloudwatch service review")
printTitle(2, "[Cost Optimization] Cloudwatch LogGroups without retention period")
printTitle(3, "Consider setting retention")
@@ -292,6 +368,7 @@ for r in regions:
outTable.append([r, aid, i.get("logGroupName"), int(round(i.get("storedBytes") / 1024 / 1024, 0))])
printResult(outTable, "Region, AccountID, LogGroup, SizeMiB")
"""Check unencrypted cloudwatch log groups"""
printTitle(2, "[Security] Cloudwatch LogGroups unencrypted")
printTitle(3, "Consider encrypting LogGroups")
outTable = []
@@ -304,6 +381,7 @@ for r in regions:
outTable.append([r, aid, i.get("logGroupName")])
printResult(outTable, "Region, AccountID, LogGroup")
"""Check AWS Backup plan"""
printTitle(1, "Backup service review")
printTitle(2, "[Reliability] Ec2/Rds instances found but AWSBackup plan missing")
printTitle(3, "Consider setting up AWSBackup plans to backup AWS resources.")
@@ -323,6 +401,7 @@ for r in regions:
outTable.append([r, aid, "AWSBackup plan missing", instanceCount])
printResult(outTable, "Region, AccountID, BackupPlan, Ec2RdsInstances")
"""Check S3 bucket policy"""
printTitle(1, "S3 service review")
printTitle(2, "[Security] S3 bucket policy missing")
printTitle(3, "Consider creating bucket policy and restrict access to bucket")
@@ -337,6 +416,25 @@ for i in jmespath.search("Buckets[*].Name", response):
outTable.append([aid, i])
printResult(outTable, "AccountID, BucketName")
"""Check s3 public access block"""
printTitle(2, "[Security] S3 public access block")
printTitle(3, "Blocking public access prevents accidental data leak due to misconfigurations in bucket policy or acl")
# get account id
sts = boto3.client("sts")
account_id = sts.get_caller_identity()["Account"]
s3control = boto3.client("s3control")
try:
response = s3control.get_public_access_block(AccountId=account_id)
config = response["PublicAccessBlockConfiguration"]
is_blocked = all(config.values())
printTitle(3, "Account-level Public Access blocked.")
except ClientError as e:
if e.response["Error"]["Code"] == "NoSuchPublicAccessBlockConfiguration":
printTitle(3, "Account-level Public Access not blocked.")
"""Check elasticache platform"""
printTitle(1, "ElastiCache review")
printTitle(2, "[Sustainability] ElastiCache instances on x64 platform")
printTitle(3, "Consider Graviton instances such as t4g/r7g to optimize your infrastructure investment.")
@@ -350,6 +448,7 @@ for r in regions:
outTable.append([r, aid, i.get("CacheClusterId"), i.get("CacheNodeType")])
printResult(outTable, "Region, AccountID, CacheClusterId, CacheNodeType")
"""Check target group with no target"""
printTitle(1, "LoadBalancer service review")
printTitle(2, "[Cost Optimization] LB Target group without targets")
printTitle(3, "Consider removing empty target groups")
@@ -364,6 +463,8 @@ for r in regions:
outTable.append([r, aid, i.get("TargetGroupName")])
printResult(outTable, "Region, AccountID, TargetGroup")
"""Check KMS key rotation"""
printTitle(1, "KMS service review")
printTitle(2, "[Security] Customer Managed Keys do not have auto rotation enabled")
printTitle(3, "Consider enabling auto key rotation. When a key is rotated, previous ones "
@@ -385,6 +486,8 @@ for r in regions:
pass
printResult(outTable, "Region, AccountID, KeyId")
"""Check API gateway resource policy"""
printTitle(1, "ApiGateway service review")
printTitle(2, "[Security] ApiGateway resource policy missing")
printTitle(3, "Consider restricting access to private API with a "
@@ -433,6 +536,60 @@ for r in regions:
printResult(outTable, "Region, AccountID, Status")
printTitle(1, "Vpc service review")
"""Check VPC flow log"""
printTitle(2, "[Security] VPC flow log not enabled")
printTitle(3, "Consider enabling VPC flowlog for audit purpose or delete the default VPCs if not in use")
for r in regions:
client = boto3.client('ec2', region_name=r)
response = client.describe_vpcs()
vpc_ids = [vpc['VpcId'] for vpc in response['Vpcs']]
for vpc_id in vpc_ids:
flow_logs_response = client.describe_flow_logs(
Filters=[
{
'Name': 'resource-id',
'Values': [vpc_id]
}
]
)
flow_logs = flow_logs_response.get('FlowLogs', [])
if not flow_logs:
outTable.append([r, aid, vpc_id])
printResult(outTable, "Region, AccountID, VpcId")
"""Check default VPCs"""
printTitle(2, "[Security] Default VPCs should be removed")
printTitle(3, "Consider deleting the default VPCs if not in use")
outTable = []
for r in regions:
client = boto3.client('ec2', region_name=r)
response = client.describe_vpcs(
Filters=[{'Name': 'isDefault', 'Values': ['true']}]
)
vpc_ids = [vpc['VpcId'] for vpc in response['Vpcs']]
for vpc_id in vpc_ids:
outTable.append([r, aid, vpc_id])
printResult(outTable, "Region, AccountID, DefaultVpcId")
"""Check VPN endpoints"""
printTitle(2, "[Security] Use of VPC endpoints")
printTitle(3, "Consider deploying VPC endpoints and connect to AWS api endpoints privately")
outTable = []
for r in regions:
client = boto3.client('ec2', region_name=r)
response = client.describe_vpc_endpoints()
if len(response['VpcEndpoints']) <= 0:
outTable.append([r, aid, "Vpc endpoint not found"])
printResult(outTable, "Region, AccountID, VpcEndpointCount")
"""Check VPN tunnels"""
printTitle(2, "[Reliability] Insufficient VPN tunnels")
printTitle(3, "Consider having 2 tunnels for each site VPN connection. "
"AWS performs VPN tunnel endpoint maintenance rather frequently. Having 2 tunnel reduces the risk "
@@ -448,6 +605,38 @@ for r in regions:
len(jmespath.search("Options.TunnelOptions[*].OutsideIpAddress", i))])
printResult(outTable, "Region, AccountID, VpnConnection, TunnelCount")
"""Check CF-ALB which allows public access"""
printTitle(2, "[Security] Cloudfront origins allow public access")
printTitle(3, "Your ALB is exposed to public, which bypass edge and WAF protection. Consider restricting access from Cloudfront only")
outTable = []
client = boto3.client('elbv2')
response = client.describe_load_balancers()
alb_sgs = {}
for alb in response['LoadBalancers']:
dns_name = alb.get('DNSName')
security_groups = alb.get('SecurityGroups', [])
alb_sgs[dns_name] = security_groups
client = boto3.client('cloudfront')
distributions = client.list_distributions()
for dist in distributions.get('DistributionList', {}).get('Items', []):
origins = dist['Origins']['Items']
for origin in origins:
if re.match(".*elb.*", origin['DomainName']):
for sg in alb_sgs[origin['DomainName']]:
ec2client = boto3.client('ec2')
s = ec2client.describe_security_groups(GroupIds=[sg])['SecurityGroups'][0]
for source_ip in jmespath.search('IpPermissions[*].IpRanges[*].CidrIp', s):
if '0.0.0.0/0' in source_ip:
outTable.append([aid, dist['Id'], origin['DomainName'], sg])
break
printResult(outTable, "AccountID, CFDistribution, Origin, SecurityGroup")
"""EKS node OS version"""
printTitle(1, "Eks service review")
printTitle(2, "[Sustainability] Eks node running on AmazonLinux2 (AL2)")
printTitle(3, "Consider using AmazonLinux2023. "
@@ -469,6 +658,7 @@ for r in regions:
outTable.append([r, aid, cluster, ng, ngResp.get("nodegroup").get("amiType")])
printResult(outTable, "Region, AccountID, Cluster, NodeGroup, AmiType")
"""Check outdated EKS control plane"""
printTitle(2, "[Sustainability] Eks control plane version outdated")
printTitle(3, "Consider using upgrading Eks cluster. "
"Reference https://docs.aws.amazon.com/eks/latest/userguide/kubernetes-versions.html for a list "