Collect SCPs for access edges (#2235)

Author: meln1k

plugins/aws/fix_plugin_aws/access_edges.py

@@ -1,6 +1,6 @@
 from functools import lru_cache
 from attr import frozen, define
-from fix_plugin_aws.resource.base import AwsResource, GraphBuilder
+from fix_plugin_aws.resource.base import AwsAccount, AwsResource, GraphBuilder
 
 from typing import List, Literal, Set, Optional, Tuple, Union, Pattern
 
@@ -632,6 +632,15 @@ def __init__(self, builder: GraphBuilder):
         self._init_principals()
 
     def _init_principals(self) -> None:
+
+        account_id = self.builder.account.id
+        service_control_policy_levels: List[List[PolicyDocument]] = []
+        account = next(self.builder.nodes(clazz=AwsAccount, filter=lambda a: a.id == account_id), None)
+        if account and account._service_control_policies:
+            service_control_policy_levels = [
+                [PolicyDocument(json) for json in level] for level in account._service_control_policies
+            ]
+
         for node in self.builder.nodes(clazz=AwsResource):
             if isinstance(node, AwsIamUser):
 
@@ -643,9 +652,6 @@ def _init_principals(self) -> None:
                         if pdj := pb_policy.policy_document_json():
                             permission_boundaries.append(PolicyDocument(pdj))
 
-                # todo: collect these resources
-                service_control_policy_levels: List[List[PolicyDocument]] = []
-
                 request_context = IamRequestContext(
                     principal=node,
                     identity_policies=identity_based_policies,
@@ -657,8 +663,6 @@ def _init_principals(self) -> None:
 
             if isinstance(node, AwsIamGroup):
                 identity_based_policies = self._get_group_based_policies(node)
-                # todo: collect these resources
-                service_control_policy_levels = []
 
                 request_context = IamRequestContext(
                     principal=node,
@@ -677,8 +681,6 @@ def _init_principals(self) -> None:
                     for pb_policy in self.builder.nodes(clazz=AwsIamPolicy, filter=lambda p: p.arn == pb_arn):
                         if pdj := pb_policy.policy_document_json():
                             permission_boundaries.append(PolicyDocument(pdj))
-                # todo: collect these resources
-                service_control_policy_levels = []
 
                 request_context = IamRequestContext(
                     principal=node,

plugins/aws/fix_plugin_aws/collector.py

@@ -48,6 +48,7 @@
     waf,
     backup,
     bedrock,
+    scp,
 )
 from fix_plugin_aws.resource.base import (
     AwsAccount,
@@ -469,3 +470,6 @@ def add_accounts(parent: Union[AwsOrganizationalRoot, AwsOrganizationalUnit]) ->
                 create_org_graph()
             except Exception as e:
                 log.exception(f"Error creating organization graph: {e}")
+
+        if account_scps := scp.collect_account_scps(self.account.id, self.config.scrape_org_role_arn, self.client):
+            self.account._service_control_policies = account_scps

plugins/aws/fix_plugin_aws/resource/base.py

@@ -310,6 +310,7 @@ class AwsAccount(BaseAccount, AwsResource, BaseIamPrincipal):
     is_organization_master: bool = False
     organization_id: Optional[str] = None
     organization_arn: Optional[str] = None
+    _service_control_policies: Optional[List[List[Json]]] = None
 
 
 default_ctime = datetime(2006, 3, 19, tzinfo=timezone.utc)  # AWS public launch date

plugins/aws/fix_plugin_aws/resource/scp.py

@@ -0,0 +1,133 @@
+from fix_plugin_aws.aws_client import AwsClient
+from typing import List, Optional
+from json import loads as json_loads
+from fixlib.types import Json
+
+
+_expected_errors = ["AccessDeniedException", "AWSOrganizationsNotInUseException"]
+
+
+def get_scps(target_id: str, client: AwsClient) -> Optional[List[Json]]:
+    policies: List[Json] = client.list(
+        "organizations",
+        "list_policies_for_target",
+        "Policies",
+        TargetId=target_id,
+        Filter="SERVICE_CONTROL_POLICY",
+        expected_errors=_expected_errors,
+    )
+
+    if not policies:
+        return None
+
+    policy_documents = []
+
+    for policy in policies:
+        policy_details = client.get(
+            "organizations", "describe_policy", "Policy", PolicyId=policy["Id"], expected_errors=_expected_errors
+        )
+        if not policy_details:
+            continue
+        policy_document = json_loads(policy_details["Content"])
+        policy_documents.append(policy_document)
+
+    return policy_documents
+
+
+def find_account_scps(client: AwsClient, account_id: str) -> List[List[Json]]:
+
+    def process_children(parent_id: str, parent_scps: List[List[Json]]) -> List[List[Json]]:
+        child_ous = client.list(
+            "organizations", "list_organizational_units_for_parent", "OrganizationalUnits", ParentId=parent_id
+        )
+        for child_ou in child_ous:
+            # copy the list to avoid modifying the parent list
+            parent_scps = list(parent_scps)
+            org_scps = get_scps(child_ou["Id"], client)
+            if org_scps:
+                parent_scps.append(org_scps)
+            accounts = client.list(
+                "organizations",
+                "list_accounts_for_parent",
+                "Accounts",
+                ParentId=child_ou["Id"],
+                expected_errors=_expected_errors,
+            )
+            for account in accounts:
+                if account["Id"] == account_id:
+                    account_scps = get_scps(account_id, client)
+                    if account_scps:
+                        parent_scps.append(account_scps)
+                    return parent_scps
+
+            if result := process_children(child_ou["Id"], list(parent_scps)):
+                return result
+
+        return []
+
+    roots = client.list("organizations", "list_roots", "Roots", expected_errors=_expected_errors)
+    for root in roots:
+        root_id = root["Id"]
+        parent_scps = []
+        root_scps = get_scps(root_id, client)
+        if root_scps:
+            parent_scps.append(root_scps)
+        accounts = client.list(
+            "organizations",
+            "list_accounts_for_parent",
+            "Accounts",
+            ParentId=root_id,
+            expected_errors=_expected_errors,
+        )
+        # if an account is attached to the root, return early
+        for account in accounts:
+            if account["Id"] == account_id:
+                account_scps = get_scps(account_id, client)
+                if account_scps:
+                    parent_scps.append(account_scps)
+                return parent_scps
+
+        if result := process_children(root["Id"], list(parent_scps)):
+            return result
+
+    return []
+
+
+def is_allow_all_scp(scp: Json) -> bool:
+    for statement in scp.get("Statement", []):
+        if all(
+            [
+                statement.get("Effect") == "Allow",
+                statement.get("Action") == "*",
+                statement.get("Resource") == "*",
+            ]
+        ):
+            return True
+
+    return False
+
+
+def filter_allow_all(levels: List[List[Json]]) -> List[List[Json]]:
+    return [[scp for scp in level if not is_allow_all_scp(scp)] for level in levels]
+
+
+def collect_account_scps(account_id: str, scrape_org_role_arn: Optional[str], client: AwsClient) -> List[List[Json]]:
+
+    if scrape_org_role_arn:
+        scp_client = AwsClient(
+            client.config,
+            client.account_id,
+            role=scrape_org_role_arn,
+            profile=client.profile,
+            region=client.region,
+            partition=client.partition,
+            error_accumulator=client.error_accumulator,
+        )
+    else:
+        scp_client = client
+
+    account_scps = find_account_scps(scp_client, account_id)
+    account_scps = filter_allow_all(account_scps)
+    account_scps = [level for level in account_scps if level]
+
+    return account_scps