Alvaro Del Valle
3cce1fa61f
Read-only cloud discovery of the Scrivas AWS Organization (o-qfj0pvhhv7) to inform a proposal. - scripts/: boto3 org assessment, member-account assessment, fast discovery - findings/: self-contained HTML dashboards, written report, summary JSON, and the rendered Prowler benchmark report - docs/full_discovery_plan.md: phased full-discovery plan - index.html: landing page linking all reports - Pipfile/.python-version: reproducible pipenv env (Python 3.12.11) Large raw scans (OCSF JSON, CSV, compliance/) are git-ignored.
295 lines
12 KiB
Python
295 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Scrivas — Fast-Track Discovery (hours, not days)
|
|
================================================
|
|
Read-only. Prioritizes the four highest-leverage phases:
|
|
- Cost baseline (Cost Explorer, payer account)
|
|
- Resource census (active regions only)
|
|
- Security exposure (public S3, open SGs, GuardDuty/SecurityHub/CloudTrail)
|
|
- IAM quick hits (users, MFA, key age, root usage, password policy)
|
|
|
|
Runs across the whole org: management account directly, members via assume-role.
|
|
Profile is passed into the boto3 Session object.
|
|
|
|
Usage: python3 fast_discovery.py [--profile PROFILE]
|
|
Output: findings/fast_discovery.json
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, date, timezone, timedelta
|
|
|
|
import boto3
|
|
from botocore.exceptions import ClientError, BotoCoreError
|
|
|
|
DEFAULT_PROFILE = "dasnuve-scrivas-louis-impersonation"
|
|
FINDINGS = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "findings")
|
|
MEMBER_ROLE = "OrganizationAccountAccessRole"
|
|
|
|
|
|
def _default(o):
|
|
return o.isoformat() if isinstance(o, (datetime, date)) else str(o)
|
|
|
|
|
|
def safe(fn, *a, **k):
|
|
try:
|
|
return fn(*a, **k), None
|
|
except (ClientError, BotoCoreError) as e:
|
|
return None, str(e)
|
|
|
|
|
|
def pg(client, op, key, **k):
|
|
out = []
|
|
try:
|
|
for page in client.get_paginator(op).paginate(**k):
|
|
out.extend(page.get(key, []))
|
|
except (ClientError, BotoCoreError) as e:
|
|
return out, str(e)
|
|
return out, None
|
|
|
|
|
|
# ---------------------------------------------------------------- regions
|
|
def active_regions(session):
|
|
"""Enabled regions that actually contain EC2 instances or EBS volumes,
|
|
plus us-east-1 (global services). Keeps the scan fast."""
|
|
ec2 = session.client("ec2", region_name="us-east-1")
|
|
regs, err = safe(ec2.describe_regions, AllRegions=False)
|
|
enabled = [r["RegionName"] for r in regs["Regions"]] if regs else ["us-east-1", "us-east-2"]
|
|
active = set(["us-east-1"])
|
|
for r in enabled:
|
|
c = session.client("ec2", region_name=r)
|
|
insts, _ = safe(c.describe_instances, MaxResults=5)
|
|
if insts and any(res.get("Instances") for res in insts.get("Reservations", [])):
|
|
active.add(r); continue
|
|
vols, _ = safe(c.describe_volumes, MaxResults=5)
|
|
if vols and vols.get("Volumes"):
|
|
active.add(r)
|
|
return sorted(active), enabled
|
|
|
|
|
|
# ---------------------------------------------------------------- IAM
|
|
def iam_quickhits(session):
|
|
iam = session.client("iam")
|
|
out = {}
|
|
users, err = pg(iam, "list_users", "Users")
|
|
if err:
|
|
return {"error": err}
|
|
detail = []
|
|
no_mfa = 0
|
|
stale_keys = []
|
|
for u in users:
|
|
name = u["UserName"]
|
|
mfa, _ = safe(iam.list_mfa_devices, UserName=name)
|
|
has_mfa = bool(mfa and mfa.get("MFADevices"))
|
|
if not has_mfa:
|
|
no_mfa += 1
|
|
keys, _ = safe(iam.list_access_keys, UserName=name)
|
|
kinfo = []
|
|
for k in (keys or {}).get("AccessKeyMetadata", []):
|
|
age = (datetime.now(timezone.utc) - k["CreateDate"]).days
|
|
lu, _ = safe(iam.get_access_key_last_used, AccessKeyId=k["AccessKeyId"])
|
|
last = (lu or {}).get("AccessKeyLastUsed", {}).get("LastUsedDate")
|
|
kinfo.append({"id": k["AccessKeyId"][-4:], "status": k["Status"],
|
|
"age_days": age, "last_used": last})
|
|
if age > 90:
|
|
stale_keys.append(f"{name}:***{k['AccessKeyId'][-4:]} ({age}d)")
|
|
detail.append({"user": name, "mfa": has_mfa, "keys": kinfo})
|
|
pw, _ = safe(iam.get_account_password_policy)
|
|
summ, _ = safe(iam.get_account_summary)
|
|
out = {
|
|
"user_count": len(users),
|
|
"users_without_mfa": no_mfa,
|
|
"stale_keys_over_90d": stale_keys,
|
|
"password_policy": (pw or {}).get("PasswordPolicy", "none set"),
|
|
"account_summary": (summ or {}).get("SummaryMap", {}),
|
|
"users": detail,
|
|
}
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------- census + exposure
|
|
def census_and_exposure(session, regions):
|
|
census = {}
|
|
exposure = {"public_buckets": [], "open_security_groups": [], "public_rds": []}
|
|
|
|
# S3 (global list, per-bucket public-access)
|
|
s3 = session.client("s3")
|
|
buckets, err = safe(s3.list_buckets)
|
|
blist = (buckets or {}).get("Buckets", [])
|
|
census["s3_buckets"] = len(blist)
|
|
for b in blist:
|
|
name = b["Name"]
|
|
pab, _ = safe(s3.get_public_access_block, Bucket=name)
|
|
cfg = (pab or {}).get("PublicAccessBlockConfiguration", {})
|
|
blocked = all([cfg.get("BlockPublicAcls"), cfg.get("IgnorePublicAcls"),
|
|
cfg.get("BlockPublicPolicy"), cfg.get("RestrictPublicBuckets")])
|
|
if not blocked:
|
|
exposure["public_buckets"].append(
|
|
{"bucket": name, "public_access_block": cfg or "none"})
|
|
|
|
# per-region compute/db/net
|
|
per_region = {}
|
|
for r in regions:
|
|
rr = {}
|
|
ec2 = session.client("ec2", region_name=r)
|
|
insts, _ = pg(ec2, "describe_instances", "Reservations")
|
|
rr["ec2_instances"] = sum(len(x.get("Instances", [])) for x in insts)
|
|
vols, _ = pg(ec2, "describe_volumes", "Volumes")
|
|
rr["ebs_volumes"] = len(vols)
|
|
rr["ebs_unattached"] = sum(1 for v in vols if not v.get("Attachments"))
|
|
eips, _ = safe(ec2.describe_addresses)
|
|
addrs = (eips or {}).get("Addresses", [])
|
|
rr["eips"] = len(addrs)
|
|
rr["eips_unassociated"] = sum(1 for a in addrs if not a.get("AssociationId"))
|
|
sgs, _ = pg(ec2, "describe_security_groups", "SecurityGroups")
|
|
for sg in sgs:
|
|
for perm in sg.get("IpPermissions", []):
|
|
for ipr in perm.get("IpRanges", []):
|
|
if ipr.get("CidrIp") == "0.0.0.0/0":
|
|
exposure["open_security_groups"].append({
|
|
"region": r, "group": sg["GroupId"],
|
|
"from_port": perm.get("FromPort"), "to_port": perm.get("ToPort"),
|
|
"proto": perm.get("IpProtocol")})
|
|
rds = session.client("rds", region_name=r)
|
|
dbs, _ = pg(rds, "describe_db_instances", "DBInstances")
|
|
rr["rds_instances"] = len(dbs)
|
|
for d in dbs:
|
|
if d.get("PubliclyAccessible"):
|
|
exposure["public_rds"].append({"region": r, "db": d["DBInstanceIdentifier"]})
|
|
lam = session.client("lambda", region_name=r)
|
|
fns, _ = pg(lam, "list_functions", "Functions")
|
|
rr["lambda_functions"] = len(fns)
|
|
eks = session.client("eks", region_name=r)
|
|
clusters, _ = pg(eks, "list_clusters", "clusters")
|
|
rr["eks_clusters"] = len(clusters)
|
|
rr["eks_cluster_names"] = clusters
|
|
elbv2 = session.client("elbv2", region_name=r)
|
|
lbs, _ = pg(elbv2, "describe_load_balancers", "LoadBalancers")
|
|
rr["load_balancers"] = len(lbs)
|
|
per_region[r] = {k: v for k, v in rr.items() if v not in (0, [], None)} or {"(empty)": True}
|
|
census["by_region"] = per_region
|
|
return census, exposure
|
|
|
|
|
|
# ---------------------------------------------------------------- security services
|
|
def security_services(session, regions):
|
|
out = {}
|
|
# CloudTrail (any region lists org+shadow trails)
|
|
ct = session.client("cloudtrail", region_name="us-east-1")
|
|
trails, _ = safe(ct.describe_trails, includeShadowTrails=True)
|
|
tl = (trails or {}).get("trailList", [])
|
|
out["cloudtrail"] = [{"name": t["Name"], "multi_region": t.get("IsMultiRegionTrail"),
|
|
"org_trail": t.get("IsOrganizationTrail"),
|
|
"log_validation": t.get("LogFileValidationEnabled")} for t in tl]
|
|
# GuardDuty / SecurityHub per active region
|
|
gd_findings = {}
|
|
shub = {}
|
|
for r in regions:
|
|
g = session.client("guardduty", region_name=r)
|
|
dets, _ = safe(g.list_detectors)
|
|
for d in (dets or {}).get("DetectorIds", []):
|
|
stats, _ = safe(g.get_findings_statistics, DetectorId=d,
|
|
FindingCriteria={}, FindingStatisticTypes=["COUNT_BY_SEVERITY"])
|
|
gd_findings[r] = (stats or {}).get("FindingStatistics", {}).get("CountBySeverity", {})
|
|
s = session.client("securityhub", region_name=r)
|
|
desc, err = safe(s.describe_hub)
|
|
shub[r] = "enabled" if desc else "not enabled"
|
|
out["guardduty_findings_by_severity"] = gd_findings
|
|
out["securityhub"] = shub
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------- cost
|
|
def cost_baseline(session):
|
|
ce = session.client("ce", region_name="us-east-1")
|
|
end = date.today().replace(day=1)
|
|
start = (end - timedelta(days=185)).replace(day=1)
|
|
res, err = safe(ce.get_cost_and_usage,
|
|
TimePeriod={"Start": start.isoformat(), "End": end.isoformat()},
|
|
Granularity="MONTHLY", Metrics=["UnblendedCost"],
|
|
GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}])
|
|
if err:
|
|
return {"error": err}
|
|
months = []
|
|
service_totals = {}
|
|
for period in res["ResultsByTime"]:
|
|
m = period["TimePeriod"]["Start"]
|
|
total = 0.0
|
|
for g in period["Groups"]:
|
|
amt = float(g["Metrics"]["UnblendedCost"]["Amount"])
|
|
svc = g["Keys"][0]
|
|
service_totals[svc] = service_totals.get(svc, 0.0) + amt
|
|
total += amt
|
|
months.append({"month": m, "total": round(total, 2)})
|
|
top = sorted(service_totals.items(), key=lambda x: -x[1])[:12]
|
|
return {"monthly_total": months,
|
|
"top_services_6mo": [{"service": s, "cost": round(v, 2)} for s, v in top]}
|
|
|
|
|
|
# ---------------------------------------------------------------- driver
|
|
def assess_account(session, account_id, name, is_payer):
|
|
print(f"\n=== {name} ({account_id}) ===")
|
|
regions, enabled = active_regions(session)
|
|
print(f" active regions: {regions} (of {len(enabled)} enabled)")
|
|
census, exposure = census_and_exposure(session, regions)
|
|
result = {
|
|
"account_id": account_id, "name": name,
|
|
"active_regions": regions, "enabled_region_count": len(enabled),
|
|
"iam": iam_quickhits(session),
|
|
"census": census,
|
|
"exposure": exposure,
|
|
"security_services": security_services(session, regions),
|
|
}
|
|
if is_payer:
|
|
result["cost"] = cost_baseline(session)
|
|
print(f" ec2/eks/rds/lambda scanned | public buckets: {len(exposure['public_buckets'])}"
|
|
f" | open SGs: {len(exposure['open_security_groups'])}"
|
|
f" | users w/o MFA: {result['iam'].get('users_without_mfa')}")
|
|
return result
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--profile", default=DEFAULT_PROFILE)
|
|
args = ap.parse_args()
|
|
|
|
base = boto3.Session(profile_name=args.profile)
|
|
payer = base.client("sts").get_caller_identity()["Account"]
|
|
org = base.client("organizations")
|
|
accounts, _ = pg(org, "list_accounts", "Accounts")
|
|
|
|
report = {"generated": datetime.now(timezone.utc).isoformat(),
|
|
"payer_account": payer, "accounts": []}
|
|
|
|
for a in accounts:
|
|
if a["Status"] != "ACTIVE":
|
|
continue
|
|
aid, name = a["Id"], a["Name"]
|
|
if aid == payer:
|
|
sess = base
|
|
else:
|
|
sts = base.client("sts")
|
|
creds, err = safe(sts.assume_role,
|
|
RoleArn=f"arn:aws:iam::{aid}:role/{MEMBER_ROLE}",
|
|
RoleSessionName="scrivas-fast-discovery")
|
|
if err:
|
|
report["accounts"].append({"account_id": aid, "name": name,
|
|
"error": f"assume-role failed: {err}"})
|
|
continue
|
|
c = creds["Credentials"]
|
|
sess = boto3.Session(aws_access_key_id=c["AccessKeyId"],
|
|
aws_secret_access_key=c["SecretAccessKey"],
|
|
aws_session_token=c["SessionToken"])
|
|
report["accounts"].append(assess_account(sess, aid, name, aid == payer))
|
|
|
|
out = os.path.join(FINDINGS, "fast_discovery.json")
|
|
os.makedirs(FINDINGS, exist_ok=True)
|
|
with open(out, "w") as f:
|
|
json.dump(report, f, indent=2, default=_default)
|
|
print(f"\nWritten to: {out}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|