aws2-2 Exercise 3: Roles, External IDs, Session Limits, Offline Linting and a Least-Privilege Policy from a Log ================================================================================================================= Run with Python 3.14.0 on Windows 11 in a scratch venv: boto3/botocore 1.43.112 (used offline, through botocore's Stubber), parliament 1.6.4 and setuptools 80.10.2 (parliament imports pkg_resources, which newer setuptools removed). Setup: py -m venv .venv ; .venv/Scripts/python.exe -m pip install boto3 parliament "setuptools<81". VERIFICATION LEVEL: 1 (run here; no AWS account, no AWS CLI, no network call to AWS). The evaluator is MY MODEL of the IAM User Guide and STS API pages read on 2026-10-12 (through a page-fetch tool that summarises pages): it is not AWS, and AWS may change its rules. It raises NotModelled for anything it does not implement. The event log in the last part is INVENTED; the dollar-free, account-free method is the point. --- t_trust.py --- """t_trust.py - role assumption, external IDs, service principals and session policies, run through iam_eval.py. A MODEL of the documented rules (AssumeRole API page, Cross-account page, Confused deputy page, Condition pages), not AWS. 'docs' says where the expected answer comes from; 'derived' means I derived it by combining two documented rules; its columns say whether the Deny statement APPLIES. """ import sys from iam_eval import decide, decide_trust, decide_cross, cond_ok, NotModelled A, B = "111111111111", "222222222222" def user(acct, name): return {"type": "user", "arn": f"arn:aws:iam::{acct}:user/{name}", "account": acct} def req(who, action, resource="*", **ctx): return {"who": who, "action": action, "resource": resource, "ctx": ctx} def allow(a, r="*", **kw): return dict(Effect="Allow", Action=a, Resource=r, **kw) def deny(a, r="*", **kw): return dict(Effect="Deny", Action=a, Resource=r, **kw) def trust(principal, **kw): return [dict(Effect="Allow", Principal=principal, Action="sts:AssumeRole", **kw)] ROLE_A, ROLE_B = f"arn:aws:iam::{A}:role/Deploy", f"arn:aws:iam::{B}:role/Deploy" dev = user(A, "dev") rows = [] def check(src, label, expected, fn): try: ok, why = fn() except NotModelled as e: ok, why = None, "NotModelled: " + str(e) rows.append((src, label, expected, ok, why)) # ---- same account (AssumeRole page: "Attach a policy ... (as long as the role's trust policy trusts the account)" OR "Add the user as a principal directly") call = req(dev, "sts:AssumeRole", ROLE_A) check("docs", "same account: trust names the USER directly, user has no identity policy", True, lambda: decide_trust(call, {"identity": []}, trust({"AWS": dev["arn"]}))) check("docs", "same account: trust names the ACCOUNT, user has no identity policy", False, lambda: decide_trust(call, {"identity": []}, trust({"AWS": f"arn:aws:iam::{A}:root"}))) check("docs", "same account: trust names the ACCOUNT, user's policy allows sts:AssumeRole on this role", True, lambda: decide_trust(call, {"identity": [[allow("sts:AssumeRole", ROLE_A)]]}, trust({"AWS": f"arn:aws:iam::{A}:root"}))) check("docs", "same account: trust names the account, user's policy allows a DIFFERENT role", False, lambda: decide_trust(call, {"identity": [[allow("sts:AssumeRole", f"arn:aws:iam::{A}:role/Other")]]}, trust({"AWS": f"arn:aws:iam::{A}:root"}))) check("docs", "same account: the trust policy does not name the user or account at all", False, lambda: decide_trust(call, {"identity": [[allow("sts:AssumeRole", "*")]]}, trust({"AWS": f"arn:aws:iam::{A}:user/someone-else"}))) check("docs", "same account: trust allows, but the user has an explicit Deny on sts:AssumeRole", False, lambda: decide_trust(call, {"identity": [[allow("sts:AssumeRole", "*"), deny("sts:AssumeRole")]]}, trust({"AWS": dev["arn"]}))) # ---- cross account (Cross-account page + AssumeRole page: the trust policy must trust the account AND the user needs permission delegated) callx = req(dev, "sts:AssumeRole", ROLE_B) trust_b = trust({"AWS": f"arn:aws:iam::{A}:root"}) check("docs", "cross account: trust in B names account A, user in A may AssumeRole on the role", True, lambda: decide_trust(callx, {}, trust_b, cross=True, requester_S={"identity": [[allow("sts:AssumeRole", ROLE_B)]]})) check("docs", "cross account: trust in B names account A, but the user has no policy in A", False, lambda: decide_trust(callx, {}, trust_b, cross=True, requester_S={"identity": []})) check("docs", "cross account: user may AssumeRole in A, but B's trust policy does not name A", False, lambda: decide_trust(callx, {}, trust({"AWS": f"arn:aws:iam::333333333333:root"}), cross=True, requester_S={"identity": [[allow("sts:AssumeRole", ROLE_B)]]})) check("docs", "cross account: even if B's trust names the user ARN, the user still needs a policy in A", False, lambda: decide_trust(callx, {}, trust({"AWS": dev["arn"]}), cross=True, requester_S={"identity": []})) # ---- external ID (Confused deputy page: sts:ExternalId condition in the trust policy) ext = trust({"AWS": f"arn:aws:iam::{A}:root"}, Condition={"StringEquals": {"sts:ExternalId": "12345"}}) pol_a = {"identity": [[allow("sts:AssumeRole", ROLE_B)]]} check("docs", "external ID: AssumeRole sent with ExternalId 12345 (the trust policy requires it)", True, lambda: decide_trust(req(dev, "sts:AssumeRole", ROLE_B, **{"sts:ExternalId": "12345"}), {}, ext, cross=True, requester_S=pol_a)) check("docs", "external ID: AssumeRole sent with the other customer's ExternalId 67890", False, lambda: decide_trust(req(dev, "sts:AssumeRole", ROLE_B, **{"sts:ExternalId": "67890"}), {}, ext, cross=True, requester_S=pol_a)) check("docs", "external ID: AssumeRole sent with no ExternalId", False, lambda: decide_trust(callx, {}, ext, cross=True, requester_S=pol_a)) no_ext = trust({"AWS": f"arn:aws:iam::{A}:root"}) check("docs", "no external ID condition: the SAME caller with ExternalId 67890 is let in (the confused deputy gap)", True, lambda: decide_trust(req(dev, "sts:AssumeRole", ROLE_B, **{"sts:ExternalId": "67890"}), {}, no_ext, cross=True, requester_S=pol_a)) # ---- service principal and cross-service confused deputy (Confused deputy page, CloudTrail and S3 example) cloudtrail = {"type": "service", "name": "cloudtrail.amazonaws.com", "arn": "cloudtrail.amazonaws.com", "account": "service"} bucket = "arn:aws:s3:::amzn-s3-demo-bucket1/AWSLogs/111122223333/x" loose = [dict(Effect="Allow", Principal={"Service": "cloudtrail.amazonaws.com"}, Action="s3:PutObject", Resource="arn:aws:s3:::amzn-s3-demo-bucket1/*")] tight = [dict(loose[0], Condition={"StringEquals": {"aws:SourceAccount": "111122223333"}})] check("docs", "service principal, no condition: CloudTrail acting for the owner's account 111122223333", True, lambda: decide(req(cloudtrail, "s3:PutObject", bucket, **{"aws:SourceAccount": "111122223333"}), {"resource": [loose]})) check("docs", "service principal, no condition: CloudTrail acting for an UNKNOWN account 999999999999", True, lambda: decide(req(cloudtrail, "s3:PutObject", bucket, **{"aws:SourceAccount": "999999999999"}), {"resource": [loose]})) check("docs", "with aws:SourceAccount=111122223333: CloudTrail acting for 111122223333", True, lambda: decide(req(cloudtrail, "s3:PutObject", bucket, **{"aws:SourceAccount": "111122223333"}), {"resource": [tight]})) check("docs", "with aws:SourceAccount=111122223333: CloudTrail acting for 999999999999", False, lambda: decide(req(cloudtrail, "s3:PutObject", bucket, **{"aws:SourceAccount": "999999999999"}), {"resource": [tight]})) check("docs", "with aws:SourceAccount=111122223333: a request that carries no SourceAccount at all", False, lambda: decide(req(cloudtrail, "s3:PutObject", bucket), {"resource": [tight]})) # ---- session policies (AssumeRole page: the session's permissions are the INTERSECTION of the role's policy and the session policy) sess = {"type": "role-session", "arn": f"arn:aws:sts::{A}:assumed-role/Deploy/s1", "account": A} role_pol = [[allow(["s3:*"])]] sp = [allow("s3:GetObject")] check("docs", "session policy allows s3:GetObject only; request s3:GetObject", True, lambda: decide(req(sess, "s3:GetObject"), {"identity": role_pol, "session": sp})) check("docs", "session policy allows s3:GetObject only; request s3:PutObject (role allows it)", False, lambda: decide(req(sess, "s3:PutObject"), {"identity": role_pol, "session": sp})) check("docs", "session policy allows ec2:*, the role does not -> a session policy cannot add permissions", False, lambda: decide(req(sess, "ec2:RunInstances"), {"identity": role_pol, "session": [allow("ec2:*")]})) check("docs", "no session policy passed -> the role's own permissions apply", True, lambda: decide(req(sess, "s3:PutObject"), {"identity": role_pol})) # ---- the MFA-age trap (derived: Numeric operators do not match a missing key + IfExists is true on a missing key) mfa_deny = {"NumericGreaterThan": {"aws:MultiFactorAuthAge": "3600"}} mfa_deny_fixed = {"NumericGreaterThanIfExists": {"aws:MultiFactorAuthAge": "3600"}} check("derived", "Deny if MFA age > 3600: caller used access keys (the key is absent) -> the Deny does NOT apply", False, lambda: (cond_ok(mfa_deny, {}), "the statement's condition is false, so the Deny is skipped")) check("derived", "same Deny written with NumericGreaterThanIfExists -> the Deny applies to the keys-only caller", True, lambda: (cond_ok(mfa_deny_fixed, {}), "IfExists is true when the key is absent")) check("derived", "same Deny written with NumericGreaterThan: caller with a 5000-second-old MFA session -> applies", True, lambda: (cond_ok(mfa_deny, {"aws:MultiFactorAuthAge": 5000}), "age 5000 > 3600")) if __name__ == "__main__": flip = "--flip" in sys.argv bad, w = 0, max(len(r[1]) for r in rows) print(f"{'case':<{w}} basis expect model result (derived rows: does the Deny statement apply?)") for i, (src, label, exp, ok, why) in enumerate(rows): if flip and i == 1: exp = not exp res = "match" if ok == exp else "MISMATCH" bad += res != "match" yes, no = ("Applies", "Skipped") if src == "derived" else ("Allow", "Deny") print(f"{label:<{w}} {src:<8} {yes if exp else no:<7} {yes if ok else no if ok is False else '?':<7} {res}") if "--why" in sys.argv: print(f"{'':<{w}} {why}") print(f"\n{len(rows)} cases, mismatches: {bad}") --- py -X utf8 t_trust.py --- case basis expect model result (derived rows: does the Deny statement apply?) same account: trust names the USER directly, user has no identity policy docs Allow Allow match same account: trust names the ACCOUNT, user has no identity policy docs Deny Deny match same account: trust names the ACCOUNT, user's policy allows sts:AssumeRole on this role docs Allow Allow match same account: trust names the account, user's policy allows a DIFFERENT role docs Deny Deny match same account: the trust policy does not name the user or account at all docs Deny Deny match same account: trust allows, but the user has an explicit Deny on sts:AssumeRole docs Deny Deny match cross account: trust in B names account A, user in A may AssumeRole on the role docs Allow Allow match cross account: trust in B names account A, but the user has no policy in A docs Deny Deny match cross account: user may AssumeRole in A, but B's trust policy does not name A docs Deny Deny match cross account: even if B's trust names the user ARN, the user still needs a policy in A docs Deny Deny match external ID: AssumeRole sent with ExternalId 12345 (the trust policy requires it) docs Allow Allow match external ID: AssumeRole sent with the other customer's ExternalId 67890 docs Deny Deny match external ID: AssumeRole sent with no ExternalId docs Deny Deny match no external ID condition: the SAME caller with ExternalId 67890 is let in (the confused deputy gap) docs Allow Allow match service principal, no condition: CloudTrail acting for the owner's account 111122223333 docs Allow Allow match service principal, no condition: CloudTrail acting for an UNKNOWN account 999999999999 docs Allow Allow match with aws:SourceAccount=111122223333: CloudTrail acting for 111122223333 docs Allow Allow match with aws:SourceAccount=111122223333: CloudTrail acting for 999999999999 docs Deny Deny match with aws:SourceAccount=111122223333: a request that carries no SourceAccount at all docs Deny Deny match session policy allows s3:GetObject only; request s3:GetObject docs Allow Allow match session policy allows s3:GetObject only; request s3:PutObject (role allows it) docs Deny Deny match session policy allows ec2:*, the role does not -> a session policy cannot add permissions docs Deny Deny match no session policy passed -> the role's own permissions apply docs Allow Allow match Deny if MFA age > 3600: caller used access keys (the key is absent) -> the Deny does NOT apply derived Skipped Skipped match same Deny written with NumericGreaterThanIfExists -> the Deny applies to the keys-only caller derived Applies Applies match same Deny written with NumericGreaterThan: caller with a 5000-second-old MFA session -> applies derived Applies Applies match 26 cases, mismatches: 0 --- py -X utf8 t_trust.py --flip (control) --- case basis expect model result (derived rows: does the Deny statement apply?) same account: trust names the USER directly, user has no identity policy docs Allow Allow match same account: trust names the ACCOUNT, user has no identity policy docs Allow Deny MISMATCH same account: trust names the ACCOUNT, user's policy allows sts:AssumeRole on this role docs Allow Allow match same account: trust names the account, user's policy allows a DIFFERENT role docs Deny Deny match same account: the trust policy does not name the user or account at all docs Deny Deny match same account: trust allows, but the user has an explicit Deny on sts:AssumeRole docs Deny Deny match cross account: trust in B names account A, user in A may AssumeRole on the role docs Allow Allow match cross account: trust in B names account A, but the user has no policy in A docs Deny Deny match cross account: user may AssumeRole in A, but B's trust policy does not name A docs Deny Deny match cross account: even if B's trust names the user ARN, the user still needs a policy in A docs Deny Deny match external ID: AssumeRole sent with ExternalId 12345 (the trust policy requires it) docs Allow Allow match external ID: AssumeRole sent with the other customer's ExternalId 67890 docs Deny Deny match external ID: AssumeRole sent with no ExternalId docs Deny Deny match no external ID condition: the SAME caller with ExternalId 67890 is let in (the confused deputy gap) docs Allow Allow match service principal, no condition: CloudTrail acting for the owner's account 111122223333 docs Allow Allow match service principal, no condition: CloudTrail acting for an UNKNOWN account 999999999999 docs Allow Allow match with aws:SourceAccount=111122223333: CloudTrail acting for 111122223333 docs Allow Allow match with aws:SourceAccount=111122223333: CloudTrail acting for 999999999999 docs Deny Deny match with aws:SourceAccount=111122223333: a request that carries no SourceAccount at all docs Deny Deny match session policy allows s3:GetObject only; request s3:GetObject docs Allow Allow match session policy allows s3:GetObject only; request s3:PutObject (role allows it) docs Deny Deny match session policy allows ec2:*, the role does not -> a session policy cannot add permissions docs Deny Deny match no session policy passed -> the role's own permissions apply docs Allow Allow match Deny if MFA age > 3600: caller used access keys (the key is absent) -> the Deny does NOT apply derived Skipped Skipped match same Deny written with NumericGreaterThanIfExists -> the Deny applies to the keys-only caller derived Applies Applies match same Deny written with NumericGreaterThan: caller with a 5000-second-old MFA session -> applies derived Applies Applies match 26 cases, mismatches: 1 --- duration.py --- """duration.py - the documented rules for the length of an AssumeRole session (AssumeRole API page, DurationSeconds), as a function. * the value may range from 900 seconds up to the maximum session duration set for the ROLE (1 to 12 hours) * default 3600 seconds when DurationSeconds is not given * asking for more than the role's maximum makes the operation fail * role chaining limits the session to one hour: asking for more than 3600 when chaining fails """ def assume(requested, role_max, chained=False): if requested is None: requested = 3600 if not 3600 <= role_max <= 43200: return "invalid role setting: the maximum session duration is 1 to 12 hours" if requested < 900: return "fails: below the 900-second minimum" if chained and requested > 3600: return "fails: role chaining limits the session to one hour" if requested > role_max: return "fails: above the role's maximum session duration (%d s)" % role_max return "ok: %d s (%.2f h)" % (requested, requested / 3600) CASES = [(None, 3600, False), (3600, 3600, False), (7200, 3600, False), (7200, 14400, False), (43200, 43200, False), (43200, 14400, False), (899, 3600, False), (900, 3600, False), (3600, 43200, True), (7200, 43200, True), (None, 43200, True), (None, 3600 * 12, False)] if __name__ == "__main__": print(f"{'DurationSeconds':<17}{'role max':<10}{'chained':<9}result") for r, m, c in CASES: print(f"{str(r) if r is not None else '(not given)':<17}{m:<10}{str(c):<9}{assume(r, m, c)}") --- py -X utf8 duration.py --- DurationSeconds role max chained result (not given) 3600 False ok: 3600 s (1.00 h) 3600 3600 False ok: 3600 s (1.00 h) 7200 3600 False fails: above the role's maximum session duration (3600 s) 7200 14400 False ok: 7200 s (2.00 h) 43200 43200 False ok: 43200 s (12.00 h) 43200 14400 False fails: above the role's maximum session duration (14400 s) 899 3600 False fails: below the 900-second minimum 900 3600 False ok: 900 s (0.25 h) 3600 43200 True ok: 3600 s (1.00 h) 7200 43200 True fails: role chaining limits the session to one hour (not given) 43200 True ok: 3600 s (1.00 h) (not given) 43200 False ok: 3600 s (1.00 h) --- sts_params.py --- """sts_params.py - which AssumeRole parameter limits does botocore (the AWS SDK for Python) enforce on the client, and which do only the AWS documentation state? No network is used: every call goes to botocore's Stubber. 'doc rule' = the constraint printed on the AssumeRole API page (read 2026-10-12). 'botocore' = what the installed SDK does. """ import re, datetime, warnings warnings.filterwarnings("ignore") import boto3, botocore from botocore.stub import Stubber from botocore.exceptions import ParamValidationError, ClientError FAKE = {"Credentials": {"AccessKeyId": "ASIAEXAMPLE0000000", "SecretAccessKey": "fakesecretfakesecret", "SessionToken": "faketokenfaketoken", "Expiration": datetime.datetime(2030, 1, 1)}, "AssumedRoleUser": {"AssumedRoleId": "AROAEXAMPLE:s", "Arn": "arn:aws:sts::123456789012:assumed-role/r/s"}} ROLE = "arn:aws:iam::123456789012:role/demo" BASE = dict(RoleArn=ROLE, RoleSessionName="session1") def sdk(params): c = boto3.client("sts", region_name="eu-west-1", aws_access_key_id="AKIAFAKE", aws_secret_access_key="fake") with Stubber(c) as st: st.add_response("assume_role", FAKE) try: c.assume_role(**params) return "accepted" except ParamValidationError as e: m = re.search(r"valid (min|max) (length|value): ([0-9]+)", str(e)) return "REJECTED (valid %s %s: %s)" % m.groups() if m else "REJECTED: " + str(e)[:50] # the documented rules (AssumeRole page), as code def doc_rule(p): errs = [] n = p.get("RoleSessionName", "") if not 2 <= len(n) <= 64: errs.append("RoleSessionName length 2-64") if not re.fullmatch(r"[\w+=,.@-]*", n): errs.append("RoleSessionName pattern [\\w+=,.@-]*") if "ExternalId" in p: e = p["ExternalId"] if not 2 <= len(e) <= 1224: errs.append("ExternalId length 2-1224") if not re.fullmatch(r"[\w+=,.@:/-]*", e): errs.append("ExternalId pattern [\\w+=,.@:/-]*") if "DurationSeconds" in p and not 900 <= p["DurationSeconds"] <= 43200: errs.append("DurationSeconds 900-43200") if len(p.get("Policy", "")) > 2048: errs.append("session policy at most 2,048 characters") if len(p.get("PolicyArns", [])) > 10: errs.append("at most 10 managed session policy ARNs") if "TokenCode" in p and not re.fullmatch(r"[0-9]{6}", p["TokenCode"]): errs.append("TokenCode exactly 6 digits") if "SerialNumber" in p and not 9 <= len(p["SerialNumber"]) <= 256: errs.append("SerialNumber length 9-256") if len(p["RoleArn"]) < 20: errs.append("RoleArn length at least 20") return "; ".join(errs) or "ok" cases = [ ("a valid call", {}), ("RoleSessionName of 1 character", {"RoleSessionName": "a"}), ("RoleSessionName of 65 characters", {"RoleSessionName": "a" * 65}), ("RoleSessionName with a space", {"RoleSessionName": "my session"}), ("DurationSeconds 899", {"DurationSeconds": 899}), ("DurationSeconds 43201", {"DurationSeconds": 43201}), ("DurationSeconds 3600", {"DurationSeconds": 3600}), ("ExternalId of 1 character", {"ExternalId": "x"}), ("ExternalId with a space", {"ExternalId": "has space"}), ("session Policy of 2,049 characters", {"Policy": "{" + " " * 2047 + "}"}), ("11 managed session policy ARNs", {"PolicyArns": [{"arn": "arn:aws:iam::aws:policy/ReadOnlyAccess"}] * 11}), ("TokenCode of 5 digits", {"SerialNumber": "arn:aws:iam::123456789012:mfa/me", "TokenCode": "12345"}), ("TokenCode of 6 letters", {"SerialNumber": "arn:aws:iam::123456789012:mfa/me", "TokenCode": "abcdef"}), ("RoleArn of 19 characters", {"RoleArn": "x" * 19}), ] if __name__ == "__main__": print(f"botocore {botocore.__version__}, boto3 {boto3.__version__}; no network (Stubber)\n") print(f"{'case':<40}{'botocore (client side)':<70}documented rule") only_doc = 0 for label, extra in cases: p = dict(BASE, **extra) s, d = sdk(p), doc_rule(p) if s == "accepted" and d != "ok": only_doc += 1 print(f"{label:<40}{s:<70}{d}") print(f"\ncases the documentation forbids but botocore accepted without complaint: {only_doc}") --- py -X utf8 sts_params.py --- botocore 1.43.112, boto3 1.43.112; no network (Stubber) case botocore (client side) documented rule a valid call accepted ok RoleSessionName of 1 character REJECTED (valid min length: 2) RoleSessionName length 2-64 RoleSessionName of 65 characters accepted RoleSessionName length 2-64 RoleSessionName with a space accepted RoleSessionName pattern [\w+=,.@-]* DurationSeconds 899 REJECTED (valid min value: 900) DurationSeconds 900-43200 DurationSeconds 43201 accepted DurationSeconds 900-43200 DurationSeconds 3600 accepted ok ExternalId of 1 character REJECTED (valid min length: 2) ExternalId length 2-1224 ExternalId with a space accepted ExternalId pattern [\w+=,.@:/-]* session Policy of 2,049 characters accepted session policy at most 2,048 characters 11 managed session policy ARNs accepted at most 10 managed session policy ARNs TokenCode of 5 digits REJECTED (valid min length: 6) TokenCode exactly 6 digits TokenCode of 6 letters accepted TokenCode exactly 6 digits RoleArn of 19 characters REJECTED (valid min length: 20) RoleArn length at least 20 cases the documentation forbids but botocore accepted without complaint: 7 --- mk_json.py (writes the small policy files shown in the chapter) --- import json, pathlib D = pathlib.Path(r"C:\lwaws\c2\pol"); D.mkdir(exist_ok=True) V = "2012-10-17" def w(n, o): (D / n).write_text(json.dumps(o, indent=2), encoding="utf-8") w("boundary_shirley.json", {"Version": V, "Statement": [{"Effect": "Allow", "Action": ["s3:*", "cloudwatch:*", "ec2:*"], "Resource": "*"}]}) w("boundary_create_user_only_with_boundary.json", {"Version": V, "Statement": [ {"Sid": "CreateOrChangeOnlyWithBoundary", "Effect": "Allow", "Action": ["iam:AttachUserPolicy", "iam:CreateUser", "iam:PutUserPermissionsBoundary", "iam:PutUserPolicy"], "Resource": "*", "Condition": {"StringEquals": {"iam:PermissionsBoundary": "arn:aws:iam::111111111111:policy/XCompanyBoundaries"}}}, {"Sid": "NoBoundaryUserDelete", "Effect": "Deny", "Action": "iam:DeleteUserPermissionsBoundary", "Resource": "*"}]}) w("session_policy.json", {"Version": V, "Statement": [{"Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::reports/*"}]}) w("trust_external_id.json", {"Version": V, "Statement": [{"Effect": "Allow", "Principal": {"AWS": "arn:aws:iam::111111111111:root"}, "Action": "sts:AssumeRole", "Condition": {"StringEquals": {"sts:ExternalId": "12345"}}}]}) w("trust_user_directly.json", {"Version": V, "Statement": [{"Effect": "Allow", "Principal": {"AWS": "arn:aws:iam::111111111111:user/dev"}, "Action": "sts:AssumeRole"}]}) w("bucket_policy_cloudtrail_tight.json", {"Version": V, "Statement": [{"Sid": "AWSCloudTrailWrite", "Effect": "Allow", "Principal": {"Service": "cloudtrail.amazonaws.com"}, "Action": "s3:PutObject", "Resource": "arn:aws:s3:::amzn-s3-demo-bucket1/*", "Condition": {"StringEquals": {"aws:SourceAccount": "111111111111"}}}]}) print("ok") --- lint_policies.py --- """lint_policies.py - lint IAM policies offline with the open-source 'parliament' package (version printed below). parliament carries its own copy of AWS's action and condition-key lists, so no AWS call is made. It treats every input as an IDENTITY policy: that is its limit, and it is shown below on a role trust policy. Community auditors are switched on for the last section. """ import warnings; warnings.filterwarnings("ignore") import json, importlib.metadata as md from parliament import analyze_policy_string S = lambda *st: {"Version": "2012-10-17", "Statement": list(st)} def lint(p, community=False): pol = analyze_policy_string(json.dumps(p), include_community_auditors=community) out = [] for f in pol.findings: d = str(f.detail) out.append(f.issue + (": " + (d[:60] + "..." if len(d) > 60 else d) if d and d != "None" else "")) return sorted(out) or ["(no findings)"] print("parliament", md.version("parliament"), "\n") core = { "correct": S({"Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::reports/*"}), "misspelt action (GetObjekt)": S({"Effect": "Allow", "Action": "s3:GetObjekt", "Resource": "arn:aws:s3:::reports/*"}), "bucket ARN where an object ARN is needed": S({"Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::reports"}), "Effect misspelt (Allowed)": S({"Effect": "Allowed", "Action": "s3:GetObject", "Resource": "*"}), "no Resource": S({"Effect": "Allow", "Action": "s3:GetObject"}), "condition operator misspelt": S({"Effect": "Allow", "Action": "s3:GetObject", "Resource": "*", "Condition": {"StringEqualz": {"aws:username": "a"}}}), "EC2 condition key on an S3 action": S({"Effect": "Allow", "Action": "s3:GetObject", "Resource": "*", "Condition": {"StringEquals": {"ec2:InstanceType": "t2.micro"}}}), "iam:* on every resource": S({"Effect": "Allow", "Action": "iam:*", "Resource": "*"}), "Action * on Resource *": S({"Effect": "Allow", "Action": "*", "Resource": "*"}), "Chapter 1's default-controls SCP (actions added to AWS in recent years)": S({"Effect": "Deny", "Action": ["organizations:LeaveOrganization", "account:CloseAccount"], "Resource": "*"}), } print("-- core checks") for k, p in core.items(): print(f" {k}") for line in lint(p): print(" ", line) print("\n-- community auditors (off by default), on a policy that can escalate privileges") esc = S({"Effect": "Allow", "Action": ["iam:PassRole", "lambda:CreateFunction", "lambda:InvokeFunction"], "Resource": "*"}) print(" core only: ", lint(esc)) print(" with community on:", lint(esc, community=True)) print("\n-- the limit: a role trust policy is not an identity policy") trust = {"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": {"AWS": "arn:aws:iam::111122223333:root"}, "Action": "sts:AssumeRole", "Condition": {"StringEquals": {"sts:ExternalId": "12345"}}}]} print(" ", lint(trust)) --- py -X utf8 lint_policies.py --- parliament 1.6.4 -- core checks correct (no findings) misspelt action (GetObjekt) UNKNOWN_ACTION: Unknown action s3:GetObjekt bucket ARN where an object ARN is needed RESOURCE_MISMATCH: [{'action': 's3:GetObject', 'required_format': 'arn:*:s3:::*... Effect misspelt (Allowed) MALFORMED: Unknown Effect used. Effect must be either Allow or Deny no Resource MALFORMED: Statement contains neither Resource nor NotResource condition operator misspelt RESOURCE_STAR: ['s3:GetObject'] UNKNOWN_OPERATOR: StringEqualz EC2 condition key on an S3 action RESOURCE_STAR: ['s3:GetObject'] UNKNOWN_CONDITION_FOR_ACTION: Unknown condition ec2:InstanceType for action s3:GetObject iam:* on every resource RESOURCE_STAR: ['iam:AddClientIDToOpenIDConnectProvider', 'iam:AddRoleToIns... Action * on Resource * (no findings) Chapter 1's default-controls SCP (actions added to AWS in recent years) (no findings) -- community auditors (off by default), on a policy that can escalate privileges core only: ["RESOURCE_STAR: ['iam:PassRole', 'lambda:CreateFunction', 'lambda:InvokeFunc..."] with community on: ['PERMISSIONS_MANAGEMENT_ACTIONS', 'PRIVILEGE_ESCALATION', "RESOURCE_STAR: ['iam:PassRole', 'lambda:CreateFunction', 'lambda:InvokeFunc..."] -- the limit: a role trust policy is not an identity policy ['MALFORMED: Statement contains neither Resource nor NotResource'] --- least_priv.py --- """least_priv.py - derive a least-privilege policy from a log of what a workload actually did, then check it. THE LOG IS INVENTED. gen_events() writes 'events.json' in the general shape of CloudTrail management and data events (eventSource, eventName, requestParameters). I have no CloudTrail to check that real events carry exactly these names; the point is the method. The real AWS feature that does this from CloudTrail is IAM Access Analyzer policy generation (see the chapter). Steps: 1 derive service:Action from each event 2 lint with parliament (finds names that are not IAM actions) 3 fix those by hand 4 scope resources from requestParameters 5 lint again 6 replay every logged request through iam_eval 7 try requests that were NOT logged 8 compare how much each policy allows, using parliament's own list of actions. """ import warnings; warnings.filterwarnings("ignore") import json, pathlib, collections, os import parliament from parliament import analyze_policy_string from iam_eval import decide B = pathlib.Path(r"C:\lwaws\c2") ACCT, REGION = "111111111111", "eu-west-1" def gen_events(): ev = [] def e(src, name, **p): ev.append({"eventSource": src + ".amazonaws.com", "eventName": name, "requestParameters": p}) for k in ("a.csv", "b.csv", "c.csv"): e("s3", "GetObject", bucketName="reports", key=k) e("s3", "ListObjectsV2", bucketName="reports") e("s3", "ListObjects", bucketName="reports") for k in ("out-1.json", "out-2.json"): e("s3", "PutObject", bucketName="reports-out", key=k) e("dynamodb", "GetItem", tableName="orders"); e("dynamodb", "PutItem", tableName="orders"); e("dynamodb", "Query", tableName="orders") e("sqs", "ReceiveMessage", queueUrl=f"https://sqs.{REGION}.amazonaws.com/{ACCT}/jobs"); e("sqs", "DeleteMessage", queueUrl=f"https://sqs.{REGION}.amazonaws.com/{ACCT}/jobs") (B / "events.json").write_text(json.dumps(ev, indent=1), encoding="utf-8") return ev def lint(policy, community=False): pol = analyze_policy_string(json.dumps(policy), include_community_auditors=community) return sorted((f.issue, str(f.detail)[:70]) for f in pol.findings) def action_of(ev): return ev["eventSource"].removesuffix(".amazonaws.com") + ":" + ev["eventName"] # names CloudTrail logs that are not IAM action names: MY knowledge, accepted only because parliament then recognises the action FIX = {"s3:ListObjectsV2": "s3:ListBucket", "s3:ListObjects": "s3:ListBucket"} def resource_for(ev, action): p = ev["requestParameters"] if action == "s3:ListBucket": return f"arn:aws:s3:::{p['bucketName']}" if action.startswith("s3:"): return f"arn:aws:s3:::{p['bucketName']}/*" if action.startswith("dynamodb:"): return f"arn:aws:dynamodb:{REGION}:{ACCT}:table/{p['tableName']}" if action.startswith("sqs:"): return f"arn:aws:sqs:{REGION}:{ACCT}:{p['queueUrl'].rsplit('/', 1)[-1]}" raise ValueError(action) def to_policy(pairs): by = collections.defaultdict(set) for a, r in pairs: by[a].add(r) grouped = collections.defaultdict(list) # one statement per distinct resource set for a, rs in sorted(by.items()): grouped[tuple(sorted(rs))].append(a) return {"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Action": sorted(acts), "Resource": list(rs)} for rs, acts in sorted(grouped.items())]} if __name__ == "__main__": events = gen_events() print(f"{len(events)} invented events written to events.json\n") # 1-2 naive derivation naive = sorted({action_of(e) for e in events}) print("1. naive actions (service:eventName):", naive) print("2. parliament on the naive policy:") for i in lint({"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Action": naive, "Resource": "*"}]}): if i[0] != "RESOURCE_STAR": print(" ", i) # 3-5 fix, scope, lint pairs = [] for e in events: a = FIX.get(action_of(e), action_of(e)); pairs.append((a, resource_for(e, a))) policy = to_policy(pairs) print("\n3-4. after the manual fix and resource scoping:") print(json.dumps(policy, indent=2)) (B / "derived_policy.json").write_text(json.dumps(policy, indent=2), encoding="utf-8") print("5. parliament on the derived policy:", lint(policy) or "(no findings)", "| community auditors:", lint(policy, True) or "(no findings)") # 6 replay who = {"type": "role-session", "arn": f"arn:aws:sts::{ACCT}:assumed-role/Worker/job1", "account": ACCT} S = {"identity": [policy]} ok = 0 for e in events: a = FIX.get(action_of(e), action_of(e)) ok += decide({"who": who, "action": a, "resource": resource_for(e, a), "ctx": {}}, S)[0] print(f"\n6. replay: {ok} of {len(events)} logged requests are allowed by the derived policy (model)") # 7 requests that were never logged print("7. requests that were NOT in the log:") for a, r in (("s3:DeleteObject", "arn:aws:s3:::reports/a.csv"), ("s3:GetObject", "arn:aws:s3:::other-bucket/a.csv"), ("s3:PutObject", "arn:aws:s3:::reports/new.csv"), ("dynamodb:DeleteTable", f"arn:aws:dynamodb:{REGION}:{ACCT}:table/orders"), ("dynamodb:GetItem", f"arn:aws:dynamodb:{REGION}:{ACCT}:table/users"), ("sqs:PurgeQueue", f"arn:aws:sqs:{REGION}:{ACCT}:jobs"), ("iam:CreateUser", "*")): print(f" {a:<22}{r:<58}{'ALLOWED' if decide({'who': who, 'action': a, 'resource': r, 'ctx': {}}, S)[0] else 'denied'}") # 8 how much each policy allows d = json.load(open(os.path.join(os.path.dirname(parliament.__file__), "iam_definition.json"), encoding="utf-8")) n = {x["prefix"]: len(x["privileges"]) for x in d} broad = ("s3", "dynamodb", "sqs") total = sum(n[p] for p in broad) used = len({a for a, _ in pairs}) print(f"\n8. the action lists inside parliament: " + ", ".join(f"{p}:* = {n[p]} actions" for p in broad)) print(f" a broad policy (s3:*, dynamodb:*, sqs:* on Resource *) allows {total} actions on every resource; the derived policy allows {used} actions on {len({r for _, r in pairs})} resources") print(f" {100 * used / total:.1f}% of the actions, and each limited to the resources actually used") --- py -X utf8 least_priv.py --- 12 invented events written to events.json 1. naive actions (service:eventName): ['dynamodb:GetItem', 'dynamodb:PutItem', 'dynamodb:Query', 's3:GetObject', 's3:ListObjects', 's3:ListObjectsV2', 's3:PutObject', 'sqs:DeleteMessage', 'sqs:ReceiveMessage'] 2. parliament on the naive policy: ('UNKNOWN_ACTION', 'Unknown action s3:ListObjects') ('UNKNOWN_ACTION', 'Unknown action s3:ListObjectsV2') 3-4. after the manual fix and resource scoping: { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:Query" ], "Resource": [ "arn:aws:dynamodb:eu-west-1:111111111111:table/orders" ] }, { "Effect": "Allow", "Action": [ "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::reports" ] }, { "Effect": "Allow", "Action": [ "s3:PutObject" ], "Resource": [ "arn:aws:s3:::reports-out/*" ] }, { "Effect": "Allow", "Action": [ "s3:GetObject" ], "Resource": [ "arn:aws:s3:::reports/*" ] }, { "Effect": "Allow", "Action": [ "sqs:DeleteMessage", "sqs:ReceiveMessage" ], "Resource": [ "arn:aws:sqs:eu-west-1:111111111111:jobs" ] } ] } 5. parliament on the derived policy: (no findings) | community auditors: (no findings) 6. replay: 12 of 12 logged requests are allowed by the derived policy (model) 7. requests that were NOT in the log: s3:DeleteObject arn:aws:s3:::reports/a.csv denied s3:GetObject arn:aws:s3:::other-bucket/a.csv denied s3:PutObject arn:aws:s3:::reports/new.csv denied dynamodb:DeleteTable arn:aws:dynamodb:eu-west-1:111111111111:table/orders denied dynamodb:GetItem arn:aws:dynamodb:eu-west-1:111111111111:table/users denied sqs:PurgeQueue arn:aws:sqs:eu-west-1:111111111111:jobs denied iam:CreateUser * denied 8. the action lists inside parliament: s3:* = 163 actions, dynamodb:* = 71 actions, sqs:* = 20 actions a broad policy (s3:*, dynamodb:*, sqs:* on Resource *) allows 254 actions on every resource; the derived policy allows 8 actions on 5 resources 3.1% of the actions, and each limited to the resources actually used WHY THIS WORKS AS AN ANSWER --------------------------- The role rows are the AssumeRole, Cross-account and Confused-deputy pages as tests: a trust policy that names the user is enough in the same account, one that names the account also needs an identity policy, another account always needs both, and an sts:ExternalId condition lets in only the caller that sends the right value (without it the same caller with the other customer's ID is let in: the confused deputy gap). The same shape protects a bucket from other accounts' CloudTrail via aws:SourceAccount. The last three rows are 'derived': they combine two documented rules (a numeric operator does not match a missing key; IfExists is true on a missing key) to show that a Deny on MFA age does not catch a caller with no MFA unless it is written with IfExists. All 26 match; the control reports one mismatch. The botocore table is a measurement of the installed SDK, offline: it enforced only minimum lengths and the minimum duration, and accepted 7 cases the documentation forbids (65-character session name, 43,201 seconds, a space in a session name or external ID, a 2,049-character session policy, 11 managed policy ARNs, a six-letter TokenCode). It says nothing about AWS's own response; I could not send a request. parliament catches misspelt actions, wrong resource ARN shapes, bad effects and operators, a condition key that does not belong to the action, and (with community auditors) a privilege-escalation combination. It does not flag Action * on Resource *, and it treats a trust policy as an identity policy and calls it malformed: both are limits, shown above. The least-privilege method works on any log of (service, action, resource): derive, lint (the lint caught that two event names are not IAM actions), fix by hand, scope resources, replay every logged request (12 of 12 allowed), and try unlogged requests (7 of 7 denied). 8 actions on 5 resources against 254 actions on every resource for the broad policy, using parliament's own action lists. Caveats stated in the chapter: the log shows only the window observed, the derived policy has no conditions, and the log is invented, so whether real CloudTrail events carry these names is not verified here. Not run: AWS Access Analyzer (policy generation, validation, custom checks, unused access), the IAM policy simulator, a real AssumeRole, any other SDK.