Approach: crawl account IAM (users, roles, groups, policies), parse policy statements, detect overly-permissive patterns (Action:"*", Resource:"*", Principal wide, NotAction/NotResource holes), compute a risk score per statement and per-policy using weighted heuristics (scope, wildcard presence, effect=Allow, condition absence, service criticality). Produce prioritized remediation list with least-privilege suggestions (narrow actions/resources, add conditions, replace with managed policy templates, suggest resource ARNs).Pseudocode and example implementation sketch:python
# python (pseudocode)
import boto3
from collections import defaultdict
# weights for scoring
WEIGHTS = {
'action_wildcard': 40,
'resource_wildcard': 30,
'allow_effect': 20,
'no_condition': 5,
'privileged_service': 25
}
PRIVILEGED_SERVICES = {'iam', 'kms', 'ec2', 's3', 'sts', 'rds'}
def list_policies():
iam = boto3.client('iam')
# combine inline, attached, role/user/group policies
return iam.list_policies(Scope='Local')['Policies']
def analyze_statement(stmt):
score = 0
reasons = []
if stmt.get('Effect','') != 'Allow':
return 0, reasons
if '*' in stmt.get('Action',[] ) or stmt.get('Action') == '*':
score += WEIGHTS['action_wildcard']; reasons.append('action:*')
if '*' in stmt.get('Resource',[]) or stmt.get('Resource') == '*':
score += WEIGHTS['resource_wildcard']; reasons.append('resource:*')
if not stmt.get('Condition'):
score += WEIGHTS['no_condition']
# privileged service check
actions = stmt.get('Action') if isinstance(stmt.get('Action'), list) else [stmt.get('Action')]
if any(a.split(':')[0] in PRIVILEGED_SERVICES for a in actions if isinstance(a,str)):
score += WEIGHTS['privileged_service']; reasons.append('privileged_service')
return score, reasons
def suggest_least_privilege(stmt):
suggestions=[]
if stmt.get('Action') == '*' or '*' in stmt.get('Action',[]):
# recommend enumerate allowed actions by service: use cloudtrail/iam access advisor to identify used actions
suggestions.append('Replace Action:"*" with service-specific actions; use access logs to enumerate used actions')
if stmt.get('Resource') == '*' or '*' in stmt.get('Resource',[]):
suggestions.append('Restrict Resource ARNs to specific buckets, tables, roles or use condition keys for prefixes')
if not stmt.get('Condition'):
suggestions.append('Add Condition (aws:SourceIp, aws:PrincipalTag, aws:RequestTag, aws:ResourceTag) where applicable')
return suggestions
def scan_and_report():
policies = list_policies()
findings=[]
for p in policies:
doc = get_policy_document(p) # fetch full JSON
policy_score=0
details=[]
for stmt in doc['Statement']:
s_score, reasons = analyze_statement(stmt)
if s_score>0:
policy_score += s_score
details.append({'statement':stmt,'score':s_score,'reasons':reasons,'suggestions':suggest_least_privilege(stmt)})
if details:
findings.append({'policy':p['PolicyName'],'score':policy_score,'details':details})
# sort descending by score
return sorted(findings, key=lambda x: x['score'], reverse=True)
