audit-labs/audit-tools

A collection of scripts, queries, and other goodies you can use in an audit. audit automation compliance evidence scripts

Commit ecaef9b176

ecaef9b176b07d3253840a90ab7a6625a1d8e3c1

parent: d44f75329c

Unsigned

cmc <hello@cleberg.net> · 2026-07-29 04:45 UTC 

feat: add AWS support to the interactive TUI audit app

Add AWS as a third platform in the TUI, alongside GitHub and GitLab.

- applications/aws/collectors/: boto3-based collectors — IAM users (MFA,
  access-key age, console password), account password policy, S3 public-access
  exposure (PAB/policy/ACL), and IAM Identity Center (SSO) assignments. Plus
  reporters/, config.py, and a unified audit.py CLI mirroring the other apps.
- Auth uses the standard AWS credential chain (profile/region/SSO); the TUI
  never collects access keys. SSO is off by default and unavailable checks skip
  gracefully.
- tui/aws_runner.py drives the collectors; platforms.py gains an AWS Platform.
  Generalized Platform.id_key into a subject callable so a defaulted AWS profile
  renders on the run screen.
- Add boto3 to requirements. Add offline aws_runner tests (stubbed session +
  collectors, including the bad-profile path) and an AWS navigation smoke test.
- Keep the existing AWS shell scripts for CloudShell/CLI-only use; document the
  new tool in the AWS README.

Layout: unified · split

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