audit-labs/audit-tools

  1"""Unit tests for the new AWS security collectors, mocking boto3 clients."""
  2
  3from unittest.mock import MagicMock
  4
  5from applications.aws.collectors import iam as aws_iam
  6from applications.aws.collectors import monitoring
  7from applications.aws.collectors import security_groups as sg
  8
  9
 10class FakeSession:
 11    """Dispatch .client(service, region_name=...) to preconfigured mocks."""
 12
 13    def __init__(self, clients):
 14        self._clients = clients
 15
 16    def client(self, service, region_name=None):
 17        return self._clients[service]
 18
 19
 20def _cfg(clients):
 21    return {"session": FakeSession(clients), "region": "us-east-1"}
 22
 23
 24# --- account_security -------------------------------------------------------
 25
 26
 27def test_account_security():
 28    iam = MagicMock()
 29    iam.get_account_summary.return_value = {
 30        "SummaryMap": {
 31            "AccountMFAEnabled": 1,
 32            "AccountAccessKeysPresent": 0,
 33            "Users": 5,
 34            "Roles": 12,
 35        }
 36    }
 37    rows = aws_iam.account_security(_cfg({"iam": iam}))
 38    assert rows[0]["root_mfa_enabled"] is True
 39    assert rows[0]["root_access_keys_present"] is False
 40    assert rows[0]["users"] == 5
 41    assert rows[0]["roles"] == 12
 42
 43
 44# --- cloudtrail -------------------------------------------------------------
 45
 46
 47def test_cloudtrail():
 48    ct = MagicMock()
 49    ct.describe_trails.return_value = {
 50        "trailList": [
 51            {
 52                "Name": "org-trail",
 53                "TrailARN": "arn:aws:cloudtrail:...:trail/org-trail",
 54                "HomeRegion": "us-east-1",
 55                "IsMultiRegionTrail": True,
 56                "LogFileValidationEnabled": True,
 57                "S3BucketName": "logs",
 58            }
 59        ]
 60    }
 61    ct.get_trail_status.return_value = {"IsLogging": True}
 62    rows = monitoring.cloudtrail(_cfg({"cloudtrail": ct}))
 63    assert rows[0]["is_logging"] is True
 64    assert rows[0]["multi_region"] is True
 65    assert rows[0]["s3_bucket"] == "logs"
 66
 67
 68# --- config_recorders -------------------------------------------------------
 69
 70
 71def _ec2_one_region():
 72    ec2 = MagicMock()
 73    ec2.describe_regions.return_value = {"Regions": [{"RegionName": "us-east-1"}]}
 74    return ec2
 75
 76
 77def test_config_recorders_recording():
 78    config = MagicMock()
 79    config.describe_configuration_recorders.return_value = {
 80        "ConfigurationRecorders": [{"name": "default"}]
 81    }
 82    config.describe_configuration_recorder_status.return_value = {
 83        "ConfigurationRecordersStatus": [
 84            {"name": "default", "recording": True, "lastStatus": "SUCCESS"}
 85        ]
 86    }
 87    rows = monitoring.config_recorders(
 88        _cfg({"ec2": _ec2_one_region(), "config": config})
 89    )
 90    assert rows == [
 91        {
 92            "region": "us-east-1",
 93            "recorder": "default",
 94            "recording": True,
 95            "last_status": "SUCCESS",
 96        }
 97    ]
 98
 99
100def test_config_recorders_reports_gap():
101    config = MagicMock()
102    config.describe_configuration_recorders.return_value = {
103        "ConfigurationRecorders": []
104    }
105    config.describe_configuration_recorder_status.return_value = {
106        "ConfigurationRecordersStatus": []
107    }
108    rows = monitoring.config_recorders(
109        _cfg({"ec2": _ec2_one_region(), "config": config})
110    )
111    assert rows[0]["recorder"] == "(none)"
112    assert rows[0]["recording"] is False
113
114
115# --- security_groups --------------------------------------------------------
116
117
118def _ec2_with_groups(groups):
119    ec2 = _ec2_one_region()
120    paginator = MagicMock()
121    paginator.paginate.return_value = [{"SecurityGroups": groups}]
122    ec2.get_paginator.return_value = paginator
123    return ec2
124
125
126def test_security_groups_flags_open_ingress():
127    groups = [
128        {
129            "GroupId": "sg-1",
130            "GroupName": "web",
131            "IpPermissions": [
132                {
133                    "IpProtocol": "tcp",
134                    "FromPort": 22,
135                    "ToPort": 22,
136                    "IpRanges": [{"CidrIp": "0.0.0.0/0"}],
137                    "Ipv6Ranges": [],
138                }
139            ],
140        }
141    ]
142    rows = sg.security_groups(_cfg({"ec2": _ec2_with_groups(groups)}))
143    assert len(rows) == 1
144    assert rows[0]["group_id"] == "sg-1"
145    assert rows[0]["from_port"] == 22
146    assert rows[0]["open_to"] == "0.0.0.0/0"
147
148
149def test_security_groups_ignores_scoped_ingress():
150    groups = [
151        {
152            "GroupId": "sg-2",
153            "GroupName": "internal",
154            "IpPermissions": [
155                {
156                    "IpProtocol": "tcp",
157                    "FromPort": 5432,
158                    "ToPort": 5432,
159                    "IpRanges": [{"CidrIp": "10.0.0.0/8"}],
160                    "Ipv6Ranges": [],
161                }
162            ],
163        }
164    ]
165    rows = sg.security_groups(_cfg({"ec2": _ec2_with_groups(groups)}))
166    assert rows == []