scrivas/scripts/ec2_code_discovery.py

#!/usr/bin/env python3
"""
Scrivas — EC2 source-code discovery (stage 1: inventory)
========================================================
Read-only. Enumerates EC2 instances in active regions and collects the
signals that indicate where application source code lives:

  - instance identity, state, type, AMI, launch time, tags
  - IAM instance profile (what the box can reach)
  - SSM managed status (can we run a read-only inspection on it?)
  - EC2 user data (frequently contains the bootstrap/deploy script,
    git remote URLs, artifact bucket names, image registries)

Context: the client does not hold the source code for their own platform;
it was built under contract by a third party. This inventories what is
running in the client's own account to locate their code.

Usage:  python3 scripts/ec2_code_discovery.py [--profile PROFILE]
Output: findings/ec2_code_discovery.json
"""
import argparse
import base64
import json
import os
from datetime import datetime, date

import boto3
from botocore.exceptions import ClientError, BotoCoreError

DEFAULT_PROFILE = "dasnuve-scrivas-louis-impersonation"
REGIONS = ["us-east-2", "us-east-1"]
FINDINGS = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "findings")


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 tag_map(tags):
    return {t["Key"]: t["Value"] for t in (tags or [])}


def user_data(ec2, iid):
    resp, err = safe(ec2.describe_instance_attribute, InstanceId=iid, Attribute="userData")
    if err:
        return {"error": err}
    raw = (resp.get("UserData") or {}).get("Value")
    if not raw:
        return None
    try:
        return {"text": base64.b64decode(raw).decode("utf-8", "replace")}
    except Exception as e:  # noqa: BLE001
        return {"error": f"decode failed: {e}"}


def ssm_managed(session, region):
    """Instance IDs SSM can reach — these are inspectable without SSH keys."""
    ssm = session.client("ssm", region_name=region)
    out = {}
    try:
        for page in ssm.get_paginator("describe_instance_information").paginate():
            for i in page.get("InstanceInformationList", []):
                out[i.get("InstanceId")] = {
                    "ping_status": i.get("PingStatus"),
                    "platform": f'{i.get("PlatformName")} {i.get("PlatformVersion")}',
                    "agent": i.get("AgentVersion"),
                    "last_ping": i.get("LastPingDateTime"),
                }
    except (ClientError, BotoCoreError) as e:
        return {"_error": str(e)}
    return out


def scan_region(session, region):
    ec2 = session.client("ec2", region_name=region)
    managed = ssm_managed(session, region)
    instances = []

    pages, err = safe(lambda: list(ec2.get_paginator("describe_instances").paginate()))
    if err:
        return {"error": err}

    for page in pages:
        for res in page.get("Reservations", []):
            for i in res.get("Instances", []):
                iid = i["InstanceId"]
                tags = tag_map(i.get("Tags"))
                instances.append({
                    "instance_id": iid,
                    "name": tags.get("Name"),
                    "state": i["State"]["Name"],
                    "type": i.get("InstanceType"),
                    "image_id": i.get("ImageId"),
                    "launch_time": i.get("LaunchTime"),
                    "key_name": i.get("KeyName"),
                    "private_ip": i.get("PrivateIpAddress"),
                    "public_ip": i.get("PublicIpAddress"),
                    "iam_instance_profile": (i.get("IamInstanceProfile") or {}).get("Arn"),
                    "security_groups": [g["GroupId"] for g in i.get("SecurityGroups", [])],
                    "volumes": [
                        b["Ebs"]["VolumeId"]
                        for b in i.get("BlockDeviceMappings", []) if b.get("Ebs")
                    ],
                    "tags": tags,
                    "ssm": managed.get(iid),
                    "user_data": user_data(ec2, iid),
                })
    return {
        "region": region,
        "ssm_managed_count": len([k for k in managed if not k.startswith("_")]),
        "instance_count": len(instances),
        "instances": instances,
    }


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--profile", default=DEFAULT_PROFILE)
    ap.add_argument("--regions", nargs="*", default=REGIONS)
    args = ap.parse_args()

    session = boto3.Session(profile_name=args.profile)
    ident = session.client("sts").get_caller_identity()

    report = {
        "generated": datetime.now().astimezone().isoformat(),
        "account": ident["Account"],
        "principal": ident["Arn"],
        "regions": [scan_region(session, r) for r in args.regions],
    }

    os.makedirs(FINDINGS, exist_ok=True)
    path = os.path.join(FINDINGS, "ec2_code_discovery.json")
    with open(path, "w") as fh:
        json.dump(report, fh, indent=2, default=_default)
    print(f"wrote {path}")

    for r in report["regions"]:
        if r.get("error"):
            print(f'{r}')
            continue
        print(f'\n== {r["region"]}: {r["instance_count"]} instances, {r["ssm_managed_count"]} SSM-managed')
        for i in r["instances"]:
            ud = i["user_data"]
            ud_flag = "userdata" if (ud and ud.get("text")) else "-"
            ssm_flag = i["ssm"]["ping_status"] if i.get("ssm") else "-"
            print(f'  {i["instance_id"]:22} {i["state"]:10} {i["type"]:14} '
                  f'ssm={ssm_flag:8} {ud_flag:9} {i.get("name")}')


if __name__ == "__main__":
    main()