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

main: applications/aws/collectors/s3.py · raw

 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
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 rather than an insecure request.
13ALL_USERS = "http://acs.amazonaws.com/groups/global/AllUsers"  # NOSONAR
14
15
16def s3_public_access(cfg):
17    s3 = cfg["session"].client("s3")
18    rows = []
19    for b in s3.list_buckets().get("Buckets", []):
20        name = b["Name"]
21        rows.append(
22            {
23                "bucket": name,
24                "region": _bucket_region(s3, name),
25                "public_access_block": _pab_status(s3, name),
26                "policy_public": _policy_public(s3, name),
27                "acl_public": _acl_public(s3, name),
28            }
29        )
30    return rows
31
32
33def _bucket_region(s3, name):
34    try:
35        loc = s3.get_bucket_location(Bucket=name).get("LocationConstraint")
36        return loc or "us-east-1"
37    except ClientError:
38        return "unknown"
39
40
41def _pab_status(s3, name):
42    try:
43        pab = s3.get_public_access_block(Bucket=name)["PublicAccessBlockConfiguration"]
44    except ClientError as e:
45        if e.response["Error"]["Code"] == "NoSuchPublicAccessBlockConfiguration":
46            return "MISSING"
47        return "error"
48    all_on = all(
49        [
50            pab.get("BlockPublicAcls"),
51            pab.get("IgnorePublicAcls"),
52            pab.get("BlockPublicPolicy"),
53            pab.get("RestrictPublicBuckets"),
54        ]
55    )
56    return "fully-restricted" if all_on else "partial"
57
58
59def _policy_public(s3, name):
60    try:
61        return s3.get_bucket_policy_status(Bucket=name)["PolicyStatus"]["IsPublic"]
62    except ClientError as e:
63        if e.response["Error"]["Code"] == "NoSuchBucketPolicy":
64            return False
65        return "error"
66
67
68def _acl_public(s3, name):
69    try:
70        grants = s3.get_bucket_acl(Bucket=name).get("Grants", [])
71    except ClientError:
72        return "error"
73    return any(g.get("Grantee", {}).get("URI") == ALL_USERS for g in grants)