Commit 4143ddbf32
4143ddbf32ecff882b1fe31faddc89e3aa2bfe35
parent: ecaef9b176
Unsigned
cmc <hello@cleberg.net> · 2026-07-29 04:48 UTC
refactor: address SonarCloud findings on AWS code
- iam.py: extract _user_row / _has_console_password helpers to bring iam_users
cognitive complexity under the threshold (S3776).
- s3.py: annotate the AWS AllUsers ACL grantee URI — a fixed identifier, not a
network endpoint — so it isn't flagged as insecure HTTP (S5332).
- platforms.py: share a single _OUT_FIELD across platforms instead of repeating
the "Output directory" / "./output" literals (S1192).
Layout: unified · split
applications/aws/collectors/iam.py
+28 −29
| @@ -16,38 +16,37 @@ def iam_users(cfg): |
| 16 | 16 | iam = cfg["session"].client("iam") |
| 17 | 17 | now = datetime.now(timezone.utc) |
| 18 | 18 | rows = [] |
| 19 | | |
| 20 | 19 | 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] |
| 20 | rows.extend(_user_row(iam, u, now) for u in page["Users"]) |
| 21 | return rows |
| 26 | 22 | |
| 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 | 23 | |
| 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 |
| 24 | def _user_row(iam, user, now): |
| 25 | name = user["UserName"] |
| 26 | mfa = iam.list_mfa_devices(UserName=name).get("MFADevices", []) |
| 27 | keys = iam.list_access_keys(UserName=name).get("AccessKeyMetadata", []) |
| 28 | key_ages = [(now - k["CreateDate"]).days for k in keys] |
| 29 | last_used = user.get("PasswordLastUsed") |
| 30 | created = user.get("CreateDate") |
| 31 | return { |
| 32 | "user": name, |
| 33 | "mfa_enabled": bool(mfa), |
| 34 | "access_keys": len(keys), |
| 35 | "oldest_key_age_days": max(key_ages) if key_ages else "", |
| 36 | "console_password": _has_console_password(iam, name), |
| 37 | "password_last_used": last_used.isoformat() if last_used else "", |
| 38 | "created": created.isoformat() if created else "", |
| 39 | } |
| 40 | |
| 41 | |
| 42 | def _has_console_password(iam, name): |
| 43 | try: |
| 44 | iam.get_login_profile(UserName=name) |
| 45 | return True |
| 46 | except ClientError as e: |
| 47 | if e.response["Error"]["Code"] == "NoSuchEntity": |
| 48 | return False |
| 49 | raise |
| 51 | 50 | |
| 52 | 51 | |
| 53 | 52 | def password_policy(cfg): |
applications/aws/collectors/s3.py
+4 −1
| @@ -7,7 +7,10 @@ bucket policy public, and whether the ACL grants access to AllUsers. |
| 7 | 7 | |
| 8 | 8 | from botocore.exceptions import ClientError |
| 9 | 9 | |
| 10 | | ALL_USERS = "http://acs.amazonaws.com/groups/global/AllUsers" |
| 10 | # AWS's fixed identifier for the "all users" ACL grantee. It is an opaque URI |
| 11 | # used as a group ID in ACL grants, not a network endpoint this tool connects |
| 12 | # to, so the http scheme is expected. NOSONAR: not an insecure URL. |
| 13 | ALL_USERS = "http://acs.amazonaws.com/groups/global/AllUsers" # NOSONAR |
| 11 | 14 | |
| 12 | 15 | |
| 13 | 16 | def s3_public_access(cfg): |
tui/platforms.py
+7 −3
| @@ -43,6 +43,10 @@ class Platform: |
| 43 | 43 | note: str = field(default="") |
| 44 | 44 | |
| 45 | 45 | |
| 46 | # Shared connection fields reused across platforms. |
| 47 | _OUT_FIELD = Field("out", "Output directory", default="./output") |
| 48 | |
| 49 | |
| 46 | 50 | def _prefill(f: Field) -> str: |
| 47 | 51 | if f.env: |
| 48 | 52 | value = os.environ.get(f.env, "").strip() |
| @@ -110,7 +114,7 @@ GITHUB = Platform( |
| 110 | 114 | required=True, |
| 111 | 115 | env="GITHUB_TOKEN", |
| 112 | 116 | ), |
| 113 | | Field("out", "Output directory", default="./output"), |
| 117 | _OUT_FIELD, |
| 114 | 118 | Field("branch", "Branch (for commit history)", default="main"), |
| 115 | 119 | ], |
| 116 | 120 | checks=github_runner.CHECKS, |
| @@ -145,7 +149,7 @@ GITLAB = Platform( |
| 145 | 149 | default="https://gitlab.com/api/v4", |
| 146 | 150 | env="GITLAB_URL", |
| 147 | 151 | ), |
| 148 | | Field("out", "Output directory", default="./output"), |
| 152 | _OUT_FIELD, |
| 149 | 153 | ], |
| 150 | 154 | checks=gitlab_runner.CHECKS, |
| 151 | 155 | default_selection=gitlab_runner.DEFAULT_SELECTION, |
| @@ -171,7 +175,7 @@ AWS = Platform( |
| 171 | 175 | "optional; defaults to current account", |
| 172 | 176 | env="AWS_AUDIT_ACCOUNT", |
| 173 | 177 | ), |
| 174 | | Field("out", "Output directory", default="./output"), |
| 178 | _OUT_FIELD, |
| 175 | 179 | ], |
| 176 | 180 | checks=aws_runner.CHECKS, |
| 177 | 181 | default_selection=aws_runner.DEFAULT_SELECTION, |