audit-labs/audit-tools
A collection of scripts, queries, and other goodies you can use in an audit.
clone: git clone https://gitbay.org/audit-labs/audit-tools.git
a8c4c49d6810b4c1037ba2636dae13fabfe92cad
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-07-29T04:45:16Z
README.org | 2 +- applications/aws/README.md | 56 ++++++++++++ applications/aws/__init__.py | 0 applications/aws/audit.py | 99 ++++++++++++++++++++ applications/aws/collectors/__init__.py | 0 applications/aws/collectors/api.py | 37 ++++++++ applications/aws/collectors/iam.py | 82 +++++++++++++++++ applications/aws/collectors/s3.py | 70 ++++++++++++++ applications/aws/collectors/sso.py | 141 +++++++++++++++++++++++++++++ applications/aws/config.py | 25 +++++ applications/aws/reporters/__init__.py | 0 applications/aws/reporters/csv_reporter.py | 47 ++++++++++ requirements.txt | 1 + tui/README.md | 21 +++-- tui/app.py | 2 +- tui/aws_runner.py | 113 +++++++++++++++++++++++ tui/platforms.py | 53 ++++++++++- tui/tests/test_app.py | 40 ++++++++ tui/tests/test_aws_runner.py | 117 ++++++++++++++++++++++++ 19 files changed, 893 insertions(+), 13 deletions(-) @@ -66,7 +66,7 @@ To pick a platform and be walked through an audit interactively: python audit_tui.py #+end_src -See =tui/README.md= for details. GitHub and GitLab are supported. +See =tui/README.md= for details. GitHub, GitLab, and AWS are supported. ** Contributing @@ -1,3 +1,59 @@ +> **NOTE**: Authentication uses the standard AWS credential chain (environment +> variables, shared config/credentials, SSO profiles, instance roles). This tool +> never handles access keys directly. Read-only permissions are enough — IAM +> `Get*`/`List*`, S3 `s3:GetBucket*` + `s3:ListAllMyBuckets`, and for the SSO +> check `sso:List*`/`sso:Describe*`, `identitystore:Describe*`, and +> `organizations:ListAccounts`. + +--- + +# `audit.py` — Unified AWS Audit Tool + +Runs all collectors against the account reachable with your active AWS +credentials and writes a timestamped audit package to disk. This is the tool the +interactive TUI (`audit_tui.py`) drives. + +## Setup + +```bash +export AWS_PROFILE=my-profile # optional; else the default chain +export AWS_DEFAULT_REGION=us-east-1 # optional +export AWS_AUDIT_ACCOUNT=my-account # optional; only for the SSO check +``` + +## Usage + +```bash +# Basic run — uses the active credentials / default profile +python audit.py + +# Named profile and region, custom output directory +python audit.py --profile my-profile --region us-east-1 --out ./output + +# SSO assignments for a specific account in the organization +python audit.py --account my-account +``` + +## Output + +Creates a directory: `<out>/aws_audit_<profile>_<YYYY-MM-DD>/` + +| File | Contents | +|---|---| +| `iam_users.csv` | IAM users with MFA status, access-key count/age, console password, last use | +| `password_policy.csv` | Account IAM password policy (length, complexity, rotation, reuse) | +| `s3_public_access.csv` | Per-bucket Public Access Block, policy public status, and ACL public exposure | +| `sso_assignments.csv` | IAM Identity Center permission-set assignments per account (Identity Center + Organizations) | +| `summary.txt` | Row counts per section | + +Checks that aren't available (no password policy, no Identity Center instance, +missing permissions) are skipped with a warning; the rest still run. + +The shell scripts below remain for CloudShell or CLI-only environments where +Python and boto3 aren't set up. + +--- + # `aws_iam_users.sh` *Note*: This example uses an account titled `cmc`, which has access provisioned to it through IAM. new file mode 100644 new file mode 100644 @@ -0,0 +1,99 @@ +""" +AWS audit CLI. + +Runs all collectors against the account reachable with the active AWS +credentials and writes a timestamped audit package to an output directory. + +Credentials come from the standard AWS chain (environment variables, shared +config/credentials, SSO profiles, instance roles) — no access keys are passed +to this tool. + +Usage: + export AWS_PROFILE=my-profile # optional + export AWS_DEFAULT_REGION=us-east-1 # optional + + python audit.py + python audit.py --profile my-profile --region us-east-1 + python audit.py --account my-account --out ./output + +Output: + <out>/aws_audit_<profile>_<date>/ + iam_users.csv + password_policy.csv + s3_public_access.csv + sso_assignments.csv + summary.txt +""" + +import argparse +import os +import sys +from datetime import date + +import config +from collectors import iam, s3, sso +from reporters import csv_reporter + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Generate an AWS audit package for the current account." + ) + parser.add_argument( + "--profile", help="AWS named profile. Overrides AWS_PROFILE env var." + ) + parser.add_argument( + "--region", help="AWS region. Overrides AWS_DEFAULT_REGION env var." + ) + parser.add_argument( + "--account", + help="Account name for the SSO assignments check. " + "Overrides AWS_AUDIT_ACCOUNT. Defaults to the current account.", + ) + parser.add_argument( + "--out", + default="./output", + help="Directory to write the audit package into. Default: ./output", + ) + return parser.parse_args() + + +def run(): + args = parse_args() + cfg = config.load(args.profile, args.region, args.account) + subject = cfg.get("profile") or "default" + + output_dir = os.path.join( + args.out, f"aws_audit_{subject}_{date.today().isoformat()}" + ) + + print(f"AWS Audit — profile: {subject}") + print(f"Output directory: {output_dir}") + print() + + sections = [] + + def collect(label, fn, filename): + print(f"Collecting: {label}...") + try: + rows = fn(cfg) + except Exception as e: + print(f" Error: {e}", file=sys.stderr) + rows = [] + csv_reporter.write(output_dir, filename, rows) + sections.append((label, len(rows))) + return rows + + collect("IAM users", iam.iam_users, "iam_users.csv") + collect("Password policy", iam.password_policy, "password_policy.csv") + collect("S3 public access", s3.s3_public_access, "s3_public_access.csv") + collect("SSO assignments", sso.sso_assignments, "sso_assignments.csv") + + print() + csv_reporter.write_summary(output_dir, subject, sections) + print() + print("Done.") + + +if __name__ == "__main__": + run() new file mode 100644 new file mode 100644 @@ -0,0 +1,37 @@ +"""Shared AWS session helpers. + +Authentication uses the standard boto3 credential chain (environment variables, +shared config/credentials files, SSO profiles, instance roles). No access keys +are ever passed in or stored by this tool. +""" + +import boto3 + + +def build_cfg(profile="", region="", account=""): + """Build the config dict the collectors expect. + + ``profile`` and ``region`` are optional; when empty, boto3's default + resolution applies. ``account`` is an optional account name used only by the + SSO assignments collector. + """ + kwargs = {} + if profile: + kwargs["profile_name"] = profile + if region: + kwargs["region_name"] = region + session = boto3.Session(**kwargs) + return { + "session": session, + "profile": profile, + "region": region, + "account": account, + } + + +def account_id(cfg): + """Return the AWS account ID for the active credentials, or '' on failure.""" + try: + return cfg["session"].client("sts").get_caller_identity()["Account"] + except Exception: + return "" new file mode 100644 @@ -0,0 +1,82 @@ +""" +Collect IAM user hygiene and the account password policy. +""" + +import sys +from datetime import datetime, timezone + +from botocore.exceptions import ClientError + + +def iam_users(cfg): + """ + One row per IAM user: MFA status, access-key count and oldest key age, + whether a console password is set, and last password use. + """ + iam = cfg["session"].client("iam") + now = datetime.now(timezone.utc) + rows = [] + + for page in iam.get_paginator("list_users").paginate(): + for u in page["Users"]: + name = u["UserName"] + mfa = iam.list_mfa_devices(UserName=name).get("MFADevices", []) + keys = iam.list_access_keys(UserName=name).get("AccessKeyMetadata", []) + key_ages = [(now - k["CreateDate"]).days for k in keys] + + try: + iam.get_login_profile(UserName=name) + console = True + except ClientError as e: + if e.response["Error"]["Code"] == "NoSuchEntity": + console = False + else: + raise + + last_used = u.get("PasswordLastUsed") + rows.append( + { + "user": name, + "mfa_enabled": bool(mfa), + "access_keys": len(keys), + "oldest_key_age_days": max(key_ages) if key_ages else "", + "console_password": console, + "password_last_used": last_used.isoformat() if last_used else "", + "created": u["CreateDate"].isoformat() + if u.get("CreateDate") + else "", + } + ) + return rows + + +def password_policy(cfg): + """ + One row describing the account IAM password policy. Returns an empty list + with a warning if no policy is set. + """ + iam = cfg["session"].client("iam") + try: + p = iam.get_account_password_policy()["PasswordPolicy"] + except ClientError as e: + if e.response["Error"]["Code"] == "NoSuchEntity": + print( + "Warning: no IAM password policy is set for this account -- skipping.", + file=sys.stderr, + ) + return [] + raise + + return [ + { + "minimum_length": p.get("MinimumPasswordLength"), + "require_symbols": p.get("RequireSymbols"), + "require_numbers": p.get("RequireNumbers"), + "require_uppercase": p.get("RequireUppercaseCharacters"), + "require_lowercase": p.get("RequireLowercaseCharacters"), + "allow_users_to_change": p.get("AllowUsersToChangePassword"), + "max_age_days": p.get("MaxPasswordAge", "N/A"), + "reuse_prevention": p.get("PasswordReusePrevention", "N/A"), + "hard_expiry": p.get("HardExpiry", False), + } + ] new file mode 100644 @@ -0,0 +1,70 @@ +""" +Collect S3 bucket public-access exposure. + +For each bucket, reports the Public Access Block state, whether S3 considers the +bucket policy public, and whether the ACL grants access to AllUsers. +""" + +from botocore.exceptions import ClientError + +ALL_USERS = "http://acs.amazonaws.com/groups/global/AllUsers" + + +def s3_public_access(cfg): + s3 = cfg["session"].client("s3") + rows = [] + for b in s3.list_buckets().get("Buckets", []): + name = b["Name"] + rows.append( + { + "bucket": name, + "region": _bucket_region(s3, name), + "public_access_block": _pab_status(s3, name), + "policy_public": _policy_public(s3, name), + "acl_public": _acl_public(s3, name), + } + ) + return rows + + +def _bucket_region(s3, name): + try: + loc = s3.get_bucket_location(Bucket=name).get("LocationConstraint") + return loc or "us-east-1" + except ClientError: + return "unknown" + + +def _pab_status(s3, name): + try: + pab = s3.get_public_access_block(Bucket=name)["PublicAccessBlockConfiguration"] + except ClientError as e: + if e.response["Error"]["Code"] == "NoSuchPublicAccessBlockConfiguration": + return "MISSING" + return "error" + all_on = all( + [ + pab.get("BlockPublicAcls"), + pab.get("IgnorePublicAcls"), + pab.get("BlockPublicPolicy"), + pab.get("RestrictPublicBuckets"), + ] + ) + return "fully-restricted" if all_on else "partial" + + +def _policy_public(s3, name): + try: + return s3.get_bucket_policy_status(Bucket=name)["PolicyStatus"]["IsPublic"] + except ClientError as e: + if e.response["Error"]["Code"] == "NoSuchBucketPolicy": + return False + return "error" + + +def _acl_public(s3, name): + try: + grants = s3.get_bucket_acl(Bucket=name).get("Grants", []) + except ClientError: + return "error" + return any(g.get("Grantee", {}).get("URI") == ALL_USERS for g in grants) new file mode 100644 @@ -0,0 +1,141 @@ +""" +Collect IAM Identity Center (SSO) permission-set assignments for an account. + +Requires AWS Organizations + IAM Identity Center. Returns an empty list with a +warning if no Identity Center instance is available or the target account can't +be resolved. When no account name is configured, the current account is used. +""" + +import sys + +from botocore.exceptions import ClientError + +from . import api + + +def sso_assignments(cfg): + session = cfg["session"] + sso = session.client("sso-admin") + + instances = sso.list_instances().get("Instances", []) + if not instances: + print( + "Warning: no IAM Identity Center instance found -- skipping.", + file=sys.stderr, + ) + return [] + instance_arn = instances[0]["InstanceArn"] + identity_store_id = instances[0]["IdentityStoreId"] + + account = _resolve_account(cfg) + if not account: + return [] + + ids = session.client("identitystore") + ps_cache = {} + name_cache = {} + rows = [] + + for ps_arn in _provisioned_permission_sets(sso, instance_arn, account): + assignments = _account_assignments(sso, instance_arn, account, ps_arn) + if not assignments: + continue + ps = _permission_set(sso, instance_arn, ps_arn, ps_cache) + for a in assignments: + rows.append( + { + "principal": _principal_name(ids, identity_store_id, a, name_cache), + "type": a["PrincipalType"], + "permission_set": ps["name"], + "managed_policies": ps["managed"], + "inline_policy": ps["inline"], + } + ) + return rows + + +def _resolve_account(cfg): + """Resolve the target account ID from the configured account name, or fall + back to the current account when no name is given.""" + name = cfg.get("account", "") + if not name: + return api.account_id(cfg) + + orgs = cfg["session"].client("organizations") + try: + for page in orgs.get_paginator("list_accounts").paginate(): + for acct in page["Accounts"]: + if acct["Name"] == name and acct["Status"] == "ACTIVE": + return acct["Id"] + except ClientError: + print( + "Warning: could not list organization accounts (needs the management " + "account) -- skipping SSO assignments.", + file=sys.stderr, + ) + return "" + + print(f"Warning: no active account named '{name}' -- skipping.", file=sys.stderr) + return "" + + +def _provisioned_permission_sets(sso, instance_arn, account): + arns = [] + paginator = sso.get_paginator("list_permission_sets_provisioned_to_account") + for page in paginator.paginate(InstanceArn=instance_arn, AccountId=account): + arns.extend(page.get("PermissionSets", [])) + return arns + + +def _account_assignments(sso, instance_arn, account, ps_arn): + out = [] + paginator = sso.get_paginator("list_account_assignments") + for page in paginator.paginate( + InstanceArn=instance_arn, AccountId=account, PermissionSetArn=ps_arn + ): + out.extend(page.get("AccountAssignments", [])) + return out + + +def _permission_set(sso, instance_arn, ps_arn, cache): + if ps_arn in cache: + return cache[ps_arn] + name = sso.describe_permission_set( + InstanceArn=instance_arn, PermissionSetArn=ps_arn + )["PermissionSet"]["Name"] + managed = [ + m["Arn"] + for m in sso.list_managed_policies_in_permission_set( + InstanceArn=instance_arn, PermissionSetArn=ps_arn + ).get("AttachedManagedPolicies", []) + ] + inline = sso.get_inline_policy_for_permission_set( + InstanceArn=instance_arn, PermissionSetArn=ps_arn + ).get("InlinePolicy", "") + cache[ps_arn] = { + "name": name, + "managed": ", ".join(managed) or "(none)", + "inline": "yes" if inline else "no", + } + return cache[ps_arn] + + +def _principal_name(ids, store_id, assignment, cache): + pid = assignment["PrincipalId"] + if pid in cache: + return cache[pid] + ptype = assignment["PrincipalType"] + name = pid + try: + if ptype == "USER": + name = ids.describe_user(IdentityStoreId=store_id, UserId=pid).get( + "UserName", pid + ) + elif ptype == "GROUP": + name = ids.describe_group(IdentityStoreId=store_id, GroupId=pid).get( + "DisplayName", pid + ) + except ClientError: + pass + cache[pid] = name + return name new file mode 100644 @@ -0,0 +1,25 @@ +""" +Configuration loader for the AWS audit tool. + +Reads AWS_PROFILE, AWS_DEFAULT_REGION, and AWS_AUDIT_ACCOUNT from the +environment. Credentials themselves come from the standard boto3 credential +chain — this tool never handles access keys directly. + +Usage: + export AWS_PROFILE=my-profile # optional; else default chain + export AWS_DEFAULT_REGION=us-east-1 # optional + export AWS_AUDIT_ACCOUNT=my-account # optional; only for SSO assignments +""" + +import os + +from collectors.api import build_cfg + + +def load(profile_override=None, region_override=None, account_override=None): + """Return a config dict. AWS needs no required token to validate here; + missing or invalid credentials surface at call time.""" + profile = profile_override or os.environ.get("AWS_PROFILE", "").strip() + region = region_override or os.environ.get("AWS_DEFAULT_REGION", "").strip() + account = account_override or os.environ.get("AWS_AUDIT_ACCOUNT", "").strip() + return build_cfg(profile, region, account) new file mode 100644 new file mode 100644 @@ -0,0 +1,47 @@ +"""CSV reporter: writes one CSV file per data section into an output directory.""" + +import csv +import os + + +def write(output_dir, filename, rows): + """ + Write a list of dicts to a CSV file in output_dir. + Skips writing if rows is empty, but logs the skip. + """ + if not rows: + print(f" {filename}: no data, skipping") + return + + os.makedirs(output_dir, exist_ok=True) + path = os.path.join(output_dir, filename) + + with open(path, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=rows[0].keys()) + writer.writeheader() + writer.writerows(rows) + + print(f" {filename}: {len(rows)} rows -> {path}") + + +def write_summary(output_dir, subject, sections): + """ + Write a plain-text summary file listing section names and row counts. + sections: list of (label, row_count) tuples + """ + os.makedirs(output_dir, exist_ok=True) + path = os.path.join(output_dir, "summary.txt") + lines = [ + "AWS Audit Package", + f"Profile: {subject}", + "", + "Section Rows", + f"{'─' * 40}", + ] + for label, count in sections: + lines.append(f"{label:<35}{count}") + + with open(path, "w", encoding="utf-8") as f: + f.write("\n".join(lines) + "\n") + + print(f" summary.txt -> {path}") @@ -4,6 +4,7 @@ xlrd PyYAML pytest requests +boto3 textual dash plotly @@ -4,8 +4,8 @@ A terminal UI that walks you through running an audit. It presents a platform menu, collects connection details and check selection, then runs the existing collectors with live progress. -GitHub and GitLab are supported. Adding a platform is a matter of writing a -runner and a `Platform` descriptor in `tui/platforms.py` — the screens are +GitHub, GitLab, and AWS are supported. Adding a platform is a matter of writing +a runner and a `Platform` descriptor in `tui/platforms.py` — the screens are platform-agnostic. ## Run it @@ -26,8 +26,16 @@ export GITHUB_TOKEN=ghp_... # needs read:org and repo scopes export GITLAB_GROUP=my-group export GITLAB_TOKEN=glpat-... # needs read_api scope export GITLAB_URL=https://gitlab.example.com/api/v4 # self-hosted only + +# AWS (credentials come from the standard AWS chain, not a form field) +export AWS_PROFILE=my-profile +export AWS_DEFAULT_REGION=us-east-1 +export AWS_AUDIT_ACCOUNT=my-account # optional; only for the SSO check ``` +AWS never asks for an access key in the UI — it uses your configured profile / +credential chain (env vars, `~/.aws`, SSO). Read-only permissions are enough. + ## Walkthrough 1. **Platform** — choose GitHub or GitLab. @@ -41,10 +49,11 @@ export GITLAB_URL=https://gitlab.example.com/api/v4 # self-hosted only ## Output -The TUI writes the same package the platform's `audit.py` produces: -`<output>/github_audit_<org>_<date>/` or `<output>/gitlab_audit_<group>_<date>/`, -one CSV per check plus a `summary.txt`. It reuses each platform's collectors and -CSV reporter unchanged — the TUI is only an interactive driver around them. +The TUI writes the same package the platform's `audit.py` produces — +`github_audit_<org>_<date>/`, `gitlab_audit_<group>_<date>/`, or +`aws_audit_<profile>_<date>/` under the output directory — one CSV per check +plus a `summary.txt`. It reuses each platform's collectors and CSV reporter +unchanged; the TUI is only an interactive driver around them. ## Keys @@ -207,7 +207,7 @@ class RunScreen(Screen): keys = self.app.selected_keys self.sub_title = f"{platform.label} · running" self.output_dir = platform.output_dir(settings) - target = settings[platform.id_key] + target = platform.subject(settings) self.query_one("#run-target", Static).update( f"Auditing [b]{target}[/] · {len(keys)} checks · → {self.output_dir}" ) new file mode 100644 @@ -0,0 +1,113 @@ +""" +Drive the AWS audit collectors from the TUI. + +Reuses the collectors and CSV reporter under ``applications/aws`` unchanged. +Mirrors the other runners: a ``CHECKS`` registry plus ``run_audit`` that writes +the same package ``applications/aws/audit.py`` produces and reports progress +through a callback. + +AWS collectors take a single config dict (a boto3 session plus region/account); +there is no per-item cache, so every check is called as ``fn(cfg)``. +""" + +import os +import sys +from collections.abc import Iterable +from datetime import date + +from tui.common import Check, ProgressCallback, ProgressEvent + +_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if _REPO_ROOT not in sys.path: + sys.path.insert(0, _REPO_ROOT) + +from applications.aws.collectors import api, iam, s3, sso +from applications.aws.reporters import csv_reporter + +# --- Check registry --------------------------------------------------------- + +CHECKS: list[Check] = [ + Check("iam_users", "IAM users", iam.iam_users, "iam_users.csv"), + Check( + "password_policy", + "Password policy", + iam.password_policy, + "password_policy.csv", + ), + Check( + "s3_public_access", + "S3 public access", + s3.s3_public_access, + "s3_public_access.csv", + ), + Check( + "sso_assignments", + "SSO assignments", + sso.sso_assignments, + "sso_assignments.csv", + note="requires Identity Center + Organizations", + ), +] + +DEFAULT_SELECTION = [c.key for c in CHECKS if c.key != "sso_assignments"] + + +# --- Output helper ---------------------------------------------------------- + + +def default_output_dir(out: str, profile: str) -> str: + """Match the folder naming used by applications/aws/audit.py.""" + subject = profile or "default" + return os.path.join(out, f"aws_audit_{subject}_{date.today().isoformat()}") + + +# --- Runner ----------------------------------------------------------------- + + +def run_audit( + *, + profile: str, + region: str, + account: str, + output_dir: str, + selected_keys: Iterable[str], + on_event: ProgressCallback, +) -> list[tuple[str, int]]: + """ + Run the selected checks and write the audit package to ``output_dir``. + + A collector that raises is reported as an error and recorded with a count + of 0, so one bad check never aborts the whole run. If the AWS session itself + can't be built (e.g. an unknown profile), that is reported and the run ends + cleanly. + """ + subject = profile or "default" + selected = set(selected_keys) + checks = [c for c in CHECKS if c.key in selected] + + try: + cfg = api.build_cfg(profile, region, account) + except Exception as e: + on_event(ProgressEvent("error", "AWS session", message=str(e))) + csv_reporter.write_summary(output_dir, subject, []) + on_event(ProgressEvent("summary", output_dir, count=0)) + return [] + + sections: list[tuple[str, int]] = [] + for c in checks: + on_event(ProgressEvent("start", c.label)) + try: + rows = c.fn(cfg) + except Exception as e: + on_event(ProgressEvent("error", c.label, message=str(e))) + sections.append((c.label, 0)) + continue + + csv_reporter.write(output_dir, c.filename, rows) + sections.append((c.label, len(rows))) + on_event(ProgressEvent("done", c.label, count=len(rows))) + + csv_reporter.write_summary(output_dir, subject, sections) + total = sum(n for _, n in sections) + on_event(ProgressEvent("summary", output_dir, count=total)) + return sections @@ -10,7 +10,7 @@ import os from collections.abc import Callable from dataclasses import dataclass, field -from tui import github_runner, gitlab_runner +from tui import aws_runner, github_runner, gitlab_runner from tui.common import Check @@ -31,7 +31,9 @@ class Field: class Platform: key: str label: str - id_key: str # which field is the audit subject (org / group) + subject: Callable[ + [dict], str + ] # (settings) -> audit subject shown on the run screen fields: list[Field] checks: list[Check] default_selection: list[str] @@ -79,10 +81,25 @@ def _gitlab_run(s: dict, output_dir, selected_keys, on_event): ) +def _aws_output_dir(s: dict) -> str: + return aws_runner.default_output_dir(s["out"], s["profile"]) + + +def _aws_run(s: dict, output_dir, selected_keys, on_event): + return aws_runner.run_audit( + profile=s["profile"], + region=s["region"], + account=s["account"], + output_dir=output_dir, + selected_keys=selected_keys, + on_event=on_event, + ) + + GITHUB = Platform( key="github", label="GitHub", - id_key="org", + subject=lambda s: s["org"], fields=[ Field("org", "Organization", "my-org", required=True, env="GITHUB_ORG"), Field( @@ -105,7 +122,7 @@ GITHUB = Platform( GITLAB = Platform( key="gitlab", label="GitLab", - id_key="group", + subject=lambda s: s["group"], fields=[ Field( "group", @@ -136,7 +153,33 @@ GITLAB = Platform( run=_gitlab_run, ) -PLATFORMS = [GITHUB, GITLAB] +AWS = Platform( + key="aws", + label="AWS", + subject=lambda s: s["profile"] or "default", + fields=[ + Field( + "profile", + "AWS profile", + "default chain, or a named / SSO profile", + env="AWS_PROFILE", + ), + Field("region", "Region", "e.g. us-east-1", env="AWS_DEFAULT_REGION"), + Field( + "account", + "Account name (SSO check only)", + "optional; defaults to current account", + env="AWS_AUDIT_ACCOUNT", + ), + Field("out", "Output directory", default="./output"), + ], + checks=aws_runner.CHECKS, + default_selection=aws_runner.DEFAULT_SELECTION, + output_dir=_aws_output_dir, + run=_aws_run, +) + +PLATFORMS = [GITHUB, GITLAB, AWS] def prefill(f: Field) -> str: @@ -125,3 +125,43 @@ def test_gitlab_navigation(monkeypatch): assert app.screen.query_one("#menu", Button).disabled is False _run(scenario()) + + +def test_aws_navigation(monkeypatch): + for var in ("AWS_PROFILE", "AWS_DEFAULT_REGION", "AWS_AUDIT_ACCOUNT"): + monkeypatch.delenv(var, raising=False) + + def fake_run_audit( + *, profile, region, account, output_dir, selected_keys, on_event + ): + on_event(gh.ProgressEvent("done", "IAM users", count=5)) + on_event(gh.ProgressEvent("summary", output_dir, count=5)) + return [("IAM users", 5)] + + monkeypatch.setattr("tui.aws_runner.run_audit", fake_run_audit) + + async def scenario(): + app = AuditApp() + async with app.run_test(size=(120, 40)) as pilot: + await pilot.pause() + await pilot.click("#aws") + await pilot.pause() + assert isinstance(app.screen, ConfigScreen) + + # AWS has no required fields — continue with defaults (default chain). + await pilot.click("#continue") + await pilot.pause() + assert isinstance(app.screen, ChecksScreen) + assert app.settings["profile"] == "" + + await pilot.click("#run") + await pilot.pause() + assert isinstance(app.screen, RunScreen) + # Empty profile renders as "default" in the folder name. + assert "aws_audit_default" in app.screen.output_dir + + await app.workers.wait_for_complete() + await pilot.pause() + assert app.screen.query_one("#menu", Button).disabled is False + + _run(scenario()) new file mode 100644 @@ -0,0 +1,117 @@ +"""Tests for the TUI's AWS audit orchestration. + +Stub the boto3 session and the collectors, then verify run_audit's wiring: +each check is called with the config, per-check errors don't abort the run, a +failed session build is reported cleanly, and the CSV package is written. +""" + +import csv +import os + +import pytest + +from tui import aws_runner as r + + +@pytest.fixture +def fake_checks(monkeypatch): + calls = {} + + def users_fn(cfg): + calls["users"] = cfg + return [{"user": "alice"}, {"user": "bob"}] + + def policy_fn(cfg): + calls["policy"] = cfg + return [{"minimum_length": 14}] + + def boom_fn(cfg): + raise RuntimeError("kaboom") + + checks = [ + r.Check("users", "Users", users_fn, "users.csv"), + r.Check("policy", "Policy", policy_fn, "policy.csv"), + r.Check("boom", "Boom", boom_fn, "boom.csv"), + ] + monkeypatch.setattr(r, "CHECKS", checks) + + # Replace build_cfg so no real boto3 session is created. + monkeypatch.setattr( + r.api, + "build_cfg", + lambda profile, region, account: { + "session": "SESSION", + "profile": profile, + "region": region, + "account": account, + }, + ) + return calls + + +def run(tmp_path, keys, profile="", region="us-east-1", account=""): + events = [] + sections = r.run_audit( + profile=profile, + region=region, + account=account, + output_dir=str(tmp_path), + selected_keys=keys, + on_event=events.append, + ) + return events, sections + + +def test_dispatch_and_files(tmp_path, fake_checks): + calls = fake_checks + run(tmp_path, ["users", "policy"], region="eu-west-1") + + # Each collector received the config dict. + assert calls["users"]["region"] == "eu-west-1" + assert calls["policy"]["session"] == "SESSION" + + for name in ("users.csv", "policy.csv", "summary.txt"): + assert os.path.exists(tmp_path / name), name + + with open(tmp_path / "users.csv", newline="") as f: + assert len(list(csv.DictReader(f))) == 2 + + +def test_failing_check_does_not_abort_run(tmp_path, fake_checks): + events, sections = run(tmp_path, ["boom", "users"]) + + kinds = [(e.kind, e.label) for e in events] + assert ("error", "Boom") in kinds + assert ("done", "Users") in kinds + + labels = dict(sections) + assert labels["Boom"] == 0 + assert labels["Users"] == 2 + + +def test_session_build_failure_is_reported(tmp_path, fake_checks, monkeypatch): + def boom_cfg(profile, region, account): + raise RuntimeError("ProfileNotFound") + + monkeypatch.setattr(r.api, "build_cfg", boom_cfg) + + events, sections = run(tmp_path, ["users"], profile="ghost") + + kinds = [(e.kind, e.label) for e in events] + assert ("error", "AWS session") in kinds + # Run still ends with a summary and writes the (empty) package. + summary = [e for e in events if e.kind == "summary"] + assert summary and summary[0].count == 0 + assert sections == [] + assert os.path.exists(tmp_path / "summary.txt") + + +def test_subject_defaults_to_default(tmp_path, fake_checks): + run(tmp_path, ["users"], profile="") + # An empty profile is folder-named "default". + assert r.default_output_dir("./out", "") == r.default_output_dir("./out", "default") + + +def test_sso_off_by_default(): + assert "sso_assignments" not in r.DEFAULT_SELECTION + assert "iam_users" in r.DEFAULT_SELECTION