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
f4259e0194c491eb53fd60708daca7264563b89b
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-07-29T17:10:35Z
README.org | 2 +- applications/aws/README.md | 13 +- applications/aws/audit.py | 10 +- applications/aws/collectors/api.py | 6 + applications/aws/collectors/iam.py | 26 +++- applications/aws/collectors/monitoring.py | 81 ++++++++++++ applications/aws/collectors/security_groups.py | 63 ++++++++++ tui/aws_runner.py | 30 ++++- tui/tests/test_aws_collectors.py | 166 +++++++++++++++++++++++++ 9 files changed, 391 insertions(+), 6 deletions(-) @@ -17,7 +17,7 @@ connection details, choose which checks to run, and watch live progress. | Directory | Description | |----------------------+------------------------------------------------------------------------------| -| =applications/aws/= | AWS IAM users, password policy, and S3 bucket analysis | +| =applications/aws/= | AWS IAM users, account/root security, password policy, S3 public access, open security groups, CloudTrail, Config, SSO | | =applications/github/= | GitHub admin enumeration, org security settings, webhooks, deploy keys, secret-scanning/Dependabot alerts, audit log, branch protections, commits | | =applications/gitlab/= | GitLab group/project members, branch protections, approvals, pipelines, audit events | | =databases/mongo/= | MongoDB admin enumeration | @@ -1,9 +1,11 @@ > **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 +> `Get*`/`List*`, S3 `s3:GetBucket*` + `s3:ListAllMyBuckets`, EC2 +> `ec2:DescribeRegions`/`DescribeSecurityGroups`, `cloudtrail:DescribeTrails` + +> `GetTrailStatus`, `config:DescribeConfigurationRecorders*`, and for the SSO > check `sso:List*`/`sso:Describe*`, `identitystore:Describe*`, and -> `organizations:ListAccounts`. +> `organizations:ListAccounts`. The SecurityAudit managed policy covers these. --- @@ -42,10 +44,17 @@ Creates a directory: `<out>/aws_audit_<profile>_<YYYY-MM-DD>/` |---|---| | `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) | +| `account_security.csv` | Account summary — root MFA, root access keys, and resource counts | | `s3_public_access.csv` | Per-bucket Public Access Block, policy public status, and ACL public exposure | +| `open_security_groups.csv` | Security-group ingress rules open to `0.0.0.0/0` or `::/0`, across all regions | +| `cloudtrail.csv` | CloudTrail trails — logging status, multi-region, log-file validation | +| `config_recorders.csv` | AWS Config recording status per region (gaps are flagged) | | `sso_assignments.csv` | IAM Identity Center permission-set assignments per account (Identity Center + Organizations) | | `summary.txt` | Row counts per section | +`open_security_groups.csv` and `config_recorders.csv` scan every enabled region, +so they take longer on accounts with many regions. + Checks that aren't available (no password policy, no Identity Center instance, missing permissions) are skipped with a warning; the rest still run. @@ -31,7 +31,7 @@ import sys from datetime import date import config -from collectors import iam, s3, sso +from collectors import iam, monitoring, s3, security_groups, sso from reporters import csv_reporter @@ -86,7 +86,15 @@ def run(): collect("IAM users", iam.iam_users, "iam_users.csv") collect("Password policy", iam.password_policy, "password_policy.csv") + collect("Account security", iam.account_security, "account_security.csv") collect("S3 public access", s3.s3_public_access, "s3_public_access.csv") + collect( + "Open security groups", + security_groups.security_groups, + "open_security_groups.csv", + ) + collect("CloudTrail", monitoring.cloudtrail, "cloudtrail.csv") + collect("AWS Config recorders", monitoring.config_recorders, "config_recorders.csv") collect("SSO assignments", sso.sso_assignments, "sso_assignments.csv") print() @@ -35,3 +35,9 @@ def account_id(cfg): return cfg["session"].client("sts").get_caller_identity()["Account"] except Exception: return "" + + +def enabled_regions(cfg): + """Return the region names enabled for the account (for region-scoped checks).""" + ec2 = cfg["session"].client("ec2", region_name=cfg.get("region") or "us-east-1") + return [r["RegionName"] for r in ec2.describe_regions().get("Regions", [])] @@ -1,5 +1,6 @@ """ -Collect IAM user hygiene and the account password policy. +Collect IAM user hygiene, the account password policy, and account-level +security summary (root MFA, root access keys). """ import sys @@ -8,6 +9,29 @@ from datetime import datetime, timezone from botocore.exceptions import ClientError +def account_security(cfg): + """ + One row of account-level security signals from the IAM account summary: + whether the root user has MFA and access keys, plus resource counts. + """ + iam = cfg["session"].client("iam") + s = iam.get_account_summary()["SummaryMap"] + return [ + { + "root_mfa_enabled": bool(s.get("AccountMFAEnabled", 0)), + "root_access_keys_present": bool(s.get("AccountAccessKeysPresent", 0)), + "root_signing_certs_present": bool( + s.get("AccountSigningCertificatesPresent", 0) + ), + "mfa_devices": s.get("MFADevices", 0), + "users": s.get("Users", 0), + "groups": s.get("Groups", 0), + "roles": s.get("Roles", 0), + "policies": s.get("Policies", 0), + } + ] + + def iam_users(cfg): """ One row per IAM user: MFA status, access-key count and oldest key age, new file mode 100644 @@ -0,0 +1,81 @@ +""" +Collect audit-logging posture: CloudTrail trails and AWS Config recorders. +""" + +import sys + +from botocore.exceptions import ClientError + +from .api import enabled_regions + + +def cloudtrail(cfg): + """ + One row per CloudTrail trail: whether it is logging, multi-region, and has + log-file validation. An empty result means no trails are configured. + """ + region = cfg.get("region") or "us-east-1" + ct = cfg["session"].client("cloudtrail", region_name=region) + rows = [] + for trail in ct.describe_trails(includeShadowTrails=False).get("trailList", []): + try: + status = ct.get_trail_status(Name=trail["TrailARN"]) + except ClientError: + status = {} + rows.append( + { + "name": trail.get("Name", ""), + "home_region": trail.get("HomeRegion", ""), + "multi_region": trail.get("IsMultiRegionTrail"), + "log_file_validation": trail.get("LogFileValidationEnabled"), + "is_logging": status.get("IsLogging"), + "s3_bucket": trail.get("S3BucketName", ""), + } + ) + return rows + + +def config_recorders(cfg): + """ + One row per region: whether AWS Config is recording. Regions with no + recorder are reported so gaps are visible. + """ + session = cfg["session"] + rows = [] + for region in enabled_regions(cfg): + try: + cc = session.client("config", region_name=region) + recorders = cc.describe_configuration_recorders().get( + "ConfigurationRecorders", [] + ) + statuses = { + s["name"]: s + for s in cc.describe_configuration_recorder_status().get( + "ConfigurationRecordersStatus", [] + ) + } + except ClientError as e: + print(f" Skipping {region}: config returned {e}", file=sys.stderr) + continue + + if not recorders: + rows.append( + { + "region": region, + "recorder": "(none)", + "recording": False, + "last_status": "", + } + ) + continue + for r in recorders: + st = statuses.get(r["name"], {}) + rows.append( + { + "region": region, + "recorder": r["name"], + "recording": st.get("recording"), + "last_status": st.get("lastStatus", ""), + } + ) + return rows new file mode 100644 @@ -0,0 +1,63 @@ +""" +Collect security-group ingress rules open to the internet, across all regions. + +Only rules allowing 0.0.0.0/0 or ::/0 are reported — one row per open rule. +""" + +import sys + +from botocore.exceptions import ClientError + +from .api import enabled_regions + +OPEN_V4 = "0.0.0.0/0" +OPEN_V6 = "::/0" + + +def security_groups(cfg): + session = cfg["session"] + rows = [] + for region in enabled_regions(cfg): + try: + ec2 = session.client("ec2", region_name=region) + groups = _all_groups(ec2) + except ClientError as e: + print(f" Skipping {region}: ec2 returned {e}", file=sys.stderr) + continue + for sg in groups: + rows.extend(_open_rules(region, sg)) + return rows + + +def _all_groups(ec2): + groups = [] + for page in ec2.get_paginator("describe_security_groups").paginate(): + groups.extend(page.get("SecurityGroups", [])) + return groups + + +def _open_rules(region, sg): + rows = [] + for perm in sg.get("IpPermissions", []): + open_to = [ + r["CidrIp"] for r in perm.get("IpRanges", []) if r.get("CidrIp") == OPEN_V4 + ] + open_to += [ + r["CidrIpv6"] + for r in perm.get("Ipv6Ranges", []) + if r.get("CidrIpv6") == OPEN_V6 + ] + if not open_to: + continue + rows.append( + { + "region": region, + "group_id": sg.get("GroupId", ""), + "group_name": sg.get("GroupName", ""), + "protocol": perm.get("IpProtocol", ""), + "from_port": perm.get("FromPort", "all"), + "to_port": perm.get("ToPort", "all"), + "open_to": ", ".join(open_to), + } + ) + return rows @@ -21,7 +21,14 @@ _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.collectors import ( + api, + iam, + monitoring, + s3, + security_groups, + sso, +) from applications.aws.reporters import csv_reporter # --- Check registry --------------------------------------------------------- @@ -34,12 +41,33 @@ CHECKS: list[Check] = [ iam.password_policy, "password_policy.csv", ), + Check( + "account_security", + "Account security (root MFA)", + iam.account_security, + "account_security.csv", + ), Check( "s3_public_access", "S3 public access", s3.s3_public_access, "s3_public_access.csv", ), + Check( + "security_groups", + "Open security groups", + security_groups.security_groups, + "open_security_groups.csv", + note="scans all regions", + ), + Check("cloudtrail", "CloudTrail", monitoring.cloudtrail, "cloudtrail.csv"), + Check( + "config_recorders", + "AWS Config recorders", + monitoring.config_recorders, + "config_recorders.csv", + note="scans all regions", + ), Check( "sso_assignments", "SSO assignments", new file mode 100644 @@ -0,0 +1,166 @@ +"""Unit tests for the new AWS security collectors, mocking boto3 clients.""" + +from unittest.mock import MagicMock + +from applications.aws.collectors import iam as aws_iam +from applications.aws.collectors import monitoring +from applications.aws.collectors import security_groups as sg + + +class FakeSession: + """Dispatch .client(service, region_name=...) to preconfigured mocks.""" + + def __init__(self, clients): + self._clients = clients + + def client(self, service, region_name=None): + return self._clients[service] + + +def _cfg(clients): + return {"session": FakeSession(clients), "region": "us-east-1"} + + +# --- account_security ------------------------------------------------------- + + +def test_account_security(): + iam = MagicMock() + iam.get_account_summary.return_value = { + "SummaryMap": { + "AccountMFAEnabled": 1, + "AccountAccessKeysPresent": 0, + "Users": 5, + "Roles": 12, + } + } + rows = aws_iam.account_security(_cfg({"iam": iam})) + assert rows[0]["root_mfa_enabled"] is True + assert rows[0]["root_access_keys_present"] is False + assert rows[0]["users"] == 5 + assert rows[0]["roles"] == 12 + + +# --- cloudtrail ------------------------------------------------------------- + + +def test_cloudtrail(): + ct = MagicMock() + ct.describe_trails.return_value = { + "trailList": [ + { + "Name": "org-trail", + "TrailARN": "arn:aws:cloudtrail:...:trail/org-trail", + "HomeRegion": "us-east-1", + "IsMultiRegionTrail": True, + "LogFileValidationEnabled": True, + "S3BucketName": "logs", + } + ] + } + ct.get_trail_status.return_value = {"IsLogging": True} + rows = monitoring.cloudtrail(_cfg({"cloudtrail": ct})) + assert rows[0]["is_logging"] is True + assert rows[0]["multi_region"] is True + assert rows[0]["s3_bucket"] == "logs" + + +# --- config_recorders ------------------------------------------------------- + + +def _ec2_one_region(): + ec2 = MagicMock() + ec2.describe_regions.return_value = {"Regions": [{"RegionName": "us-east-1"}]} + return ec2 + + +def test_config_recorders_recording(): + config = MagicMock() + config.describe_configuration_recorders.return_value = { + "ConfigurationRecorders": [{"name": "default"}] + } + config.describe_configuration_recorder_status.return_value = { + "ConfigurationRecordersStatus": [ + {"name": "default", "recording": True, "lastStatus": "SUCCESS"} + ] + } + rows = monitoring.config_recorders( + _cfg({"ec2": _ec2_one_region(), "config": config}) + ) + assert rows == [ + { + "region": "us-east-1", + "recorder": "default", + "recording": True, + "last_status": "SUCCESS", + } + ] + + +def test_config_recorders_reports_gap(): + config = MagicMock() + config.describe_configuration_recorders.return_value = { + "ConfigurationRecorders": [] + } + config.describe_configuration_recorder_status.return_value = { + "ConfigurationRecordersStatus": [] + } + rows = monitoring.config_recorders( + _cfg({"ec2": _ec2_one_region(), "config": config}) + ) + assert rows[0]["recorder"] == "(none)" + assert rows[0]["recording"] is False + + +# --- security_groups -------------------------------------------------------- + + +def _ec2_with_groups(groups): + ec2 = _ec2_one_region() + paginator = MagicMock() + paginator.paginate.return_value = [{"SecurityGroups": groups}] + ec2.get_paginator.return_value = paginator + return ec2 + + +def test_security_groups_flags_open_ingress(): + groups = [ + { + "GroupId": "sg-1", + "GroupName": "web", + "IpPermissions": [ + { + "IpProtocol": "tcp", + "FromPort": 22, + "ToPort": 22, + "IpRanges": [{"CidrIp": "0.0.0.0/0"}], + "Ipv6Ranges": [], + } + ], + } + ] + rows = sg.security_groups(_cfg({"ec2": _ec2_with_groups(groups)})) + assert len(rows) == 1 + assert rows[0]["group_id"] == "sg-1" + assert rows[0]["from_port"] == 22 + assert rows[0]["open_to"] == "0.0.0.0/0" + + +def test_security_groups_ignores_scoped_ingress(): + groups = [ + { + "GroupId": "sg-2", + "GroupName": "internal", + "IpPermissions": [ + { + "IpProtocol": "tcp", + "FromPort": 5432, + "ToPort": 5432, + "IpRanges": [{"CidrIp": "10.0.0.0/8"}], + "Ipv6Ranges": [], + } + ], + } + ] + rows = sg.security_groups(_cfg({"ec2": _ec2_with_groups(groups)})) + assert rows == []