#!/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())