v1.1.4 (#99)

* Added preset for service access

* Added preset for service access

* fix command line output in orgs subcommand

* added SCP support to admin-checks

* added SCP support to admin-checks

* cutting down on Lambda authorization simulations

* Initial work on resources and potential confused deputy risks

* implemented support for botocore client generation with custom arguments, added localstack endpoint support for "graph create" subcommand

* implemented a fix for #98 - cloudformation:UpdateStack risks

* implemented a fix for #97 - checking for Login Profile for full password check

* progress on datapipeline work

* hotfix

* pulling datapipeline until 1.1.5

Co-authored-by: Erik Steringer <erik.steringer@nccgroup.com>

Author: ncc-erik-steringer

principalmapper/__init__.py

@@ -15,4 +15,4 @@
 #      You should have received a copy of the GNU Affero General Public License
 #      along with Principal Mapper.  If not, see <https://www.gnu.org/licenses/>.
 
-__version__ = '1.1.3'
+__version__ = '1.1.4'

principalmapper/analysis/find_risks.py

@@ -28,13 +28,14 @@
 
 import datetime as dt
 import json
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
 import principalmapper
 from principalmapper.analysis.finding import Finding
 from principalmapper.analysis.report import Report
 from principalmapper.common import Graph, Node, Edge
 from principalmapper.querying import query_interface
+from principalmapper.querying.local_policy_simulation import resource_policy_authorization, ResourcePolicyEvalResult
 from principalmapper.querying.presets.privesc import can_privesc
 from principalmapper.util import arns
 
@@ -447,6 +448,74 @@ def gen_admin_users_without_mfa_finding(graph: Graph) -> List[Finding]:
     return result
 
 
+def gen_resources_with_potential_confused_deputies(graph: Graph) -> List[Finding]:
+    """Generates findings related to AWS resources that allow access to AWS services (via resource policy)
+    that may not correctly verify which AWS account is the true source of a request that
+    affects the given resource.
+
+    Primarily works by inspecting resource policies and making sure that access is guarded
+    with a condition using aws:SourceAccount."""
+
+    result = []
+
+    resource_service_action_map = {
+        's3': {
+            'serverlessrepo.amazonaws.com': [
+                's3:GetObject'
+            ]
+        }
+    }
+
+    affected_policies = []  # type: List[Tuple[str, str, str]]
+    for resource_type in resource_service_action_map.keys():
+        for policy in graph.policies:
+            if arns.get_service(policy.arn) == resource_type:
+                for service, action_list in resource_service_action_map[resource_type].items():
+                    available_actions = []
+                    for action in action_list:
+                        rpa_result = resource_policy_authorization(
+                            service,
+                            graph.metadata['account_id'],
+                            policy.policy_doc,
+                            action,
+                            policy.arn,
+                            {
+                                'aws:SourceAccount': '000000000000'
+                            }
+                        )
+                        if rpa_result.SERVICE_MATCH:
+                            available_actions.append(action)
+                    if len(available_actions) > 0:
+                        affected_policies.append(
+                            (policy.arn, service, ' | '.join(available_actions))
+                        )
+
+    if len(affected_policies) > 0:
+        desc_list_str = '\n'.join(['* With service {}, the resource {} for the action(s): {}'.format(y, x, z) for x, y, z in affected_policies])
+        result.append(
+            Finding(
+                'Resources With A Potential Confused-Deputy Risk',
+                'Medium',
+                'Depending on the affected resources and services, an attacker may be able to execute read or write '
+                'operations on the resources from another AWS account.',
+                'In AWS, certain services will create and use resources in the customer\'s own AWS account. This may '
+                'be controlled using a resource policy that grants access to the service that created the resource '
+                'in the customer\'s AWS account. However, some services require customers to use the '
+                '`${aws:SourceAccount}` condition context key to control access to the account resource from the '
+                'service. In other words, to prevent the service from accessing the resource on the behalf of '
+                'another customer, the resource needs a resource policy that allow-lists the true "source" of a '
+                'request.\n\n'
+                'The following AWS services and resources could allow an external account to potentially gain '
+                'read/write access to the resources:\n\n' + desc_list_str,
+                'Update the resource policy for all affected resources, and ensure that all statements granting '
+                'access to AWS services check against the `${aws:SourceAccount}` condition context key when '
+                'appropriate.'
+            )
+        )
+
+    return result
+
+
 def print_report(report: Report) -> None:
     """Given a report, uses print() to print out their contents in a Markdown format."""
 

principalmapper/graphing/autoscaling_edges.py

@@ -34,17 +34,20 @@ class AutoScalingEdgeChecker(EdgeChecker):
     """Class for identifying if EC2 Auto Scaling can be used by IAM principals to gain access to other IAM principals."""
 
     def return_edges(self, nodes: List[Node], region_allow_list: Optional[List[str]] = None,
-                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None) -> List[Edge]:
+                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None,
+                     client_args_map: Optional[dict] = None) -> List[Edge]:
         """Fulfills expected method return_edges."""
 
         logger.info('Generating Edges based on EC2 Auto Scaling.')
 
+        asargs = client_args_map.get('autoscaling', {})
+
         # Gather projects information for each region
         autoscaling_clients = []
         if self.session is not None:
             as_regions = botocore_tools.get_regions_to_search(self.session, 'autoscaling', region_allow_list, region_deny_list)
             for region in as_regions:
-                autoscaling_clients.append(self.session.create_client('autoscaling', region_name=region))
+                autoscaling_clients.append(self.session.create_client('autoscaling', region_name=region, **asargs))
 
         launch_configs = []
         for as_client in autoscaling_clients:

principalmapper/graphing/cloudformation_edges.py

@@ -36,17 +36,20 @@ class CloudFormationEdgeChecker(EdgeChecker):
     """Class for identifying if CloudFormation can be used by IAM principals to gain access to other IAM principals."""
 
     def return_edges(self, nodes: List[Node], region_allow_list: Optional[List[str]] = None,
-                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None) -> List[Edge]:
+                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None,
+                     client_args_map: Optional[dict] = None) -> List[Edge]:
         """Fulfills expected method return_edges."""
 
         logger.info('Pulling data on CloudFormation stacks.')
 
+        cfargs = client_args_map.get('cloudformation', {})
+
         # Grab existing stacks in each region
         cloudformation_clients = []
         if self.session is not None:
             cf_regions = botocore_tools.get_regions_to_search(self.session, 'cloudformation', region_allow_list, region_deny_list)
             for region in cf_regions:
-                cloudformation_clients.append(self.session.create_client('cloudformation', region_name=region))
+                cloudformation_clients.append(self.session.create_client('cloudformation', region_name=region, **cfargs))
 
         # grab existing cloudformation stacks
         stack_list = []
@@ -136,7 +139,7 @@ def generate_edges_locally(nodes: List[Node], stack_list: List[dict], scps: Opti
 
             relevant_stacks = []  # we'll reuse this for *ChangeSet
             for stack in stack_list:
-                if 'RoleArn' in stack:
+                if 'RoleARN' in stack:
                     if stack['RoleARN'] == node_destination.arn:
                         relevant_stacks.append(stack)
 

principalmapper/graphing/codebuild_edges.py

@@ -34,18 +34,21 @@ class CodeBuildEdgeChecker(EdgeChecker):
     """Class for identifying if CodeBuild can be used by IAM principals to gain access to other IAM principals."""
 
     def return_edges(self, nodes: List[Node], region_allow_list: Optional[List[str]] = None,
-                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None) -> List[Edge]:
+                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None,
+                     client_args_map: Optional[dict] = None) -> List[Edge]:
         """Fulfills expected method return_edges."""
 
         logger.info('Generating Edges based on CodeBuild.')
 
         # Gather projects information for each region
 
+        cbargs = client_args_map.get('codebuild', {})
+
         codebuild_clients = []
         if self.session is not None:
             cf_regions = botocore_tools.get_regions_to_search(self.session, 'codebuild', region_allow_list, region_deny_list)
             for region in cf_regions:
-                codebuild_clients.append(self.session.create_client('codebuild', region_name=region))
+                codebuild_clients.append(self.session.create_client('codebuild', region_name=region, **cbargs))
 
         codebuild_projects = []
         for cb_client in codebuild_clients:

principalmapper/graphing/ec2_edges.py

@@ -35,7 +35,8 @@ class EC2EdgeChecker(EdgeChecker):
     """Class for identifying if EC2 can be used by IAM principals to gain access to other IAM principals."""
 
     def return_edges(self, nodes: List[Node], region_allow_list: Optional[List[str]] = None,
-                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None) -> List[Edge]:
+                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None,
+                     client_args_map: Optional[dict] = None) -> List[Edge]:
         """Fulfills expected method return_edges."""
 
         logger.info('Generating Edges based on EC2.')

principalmapper/graphing/edge_checker.py

@@ -32,7 +32,8 @@ def __init__(self, session: botocore.session.Session):
         self.session = session
 
     def return_edges(self, nodes: List[Node], region_allow_list: Optional[List[str]] = None,
-                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None) -> List[Edge]:
+                     region_deny_list: Optional[List[str]] = None, scps: Optional[List[List[dict]]] = None,
+                     client_args_map: Optional[dict] = None) -> List[Edge]:
         """Subclasses shall override this method. Given a list of nodes, the EdgeChecker should be able to use its session
         object in order to make clients and call the AWS API to resolve information about the account. Then,
         with this information, it should return a list of edges between the passed nodes.

principalmapper/graphing/edge_identification.py

@@ -51,7 +51,7 @@
 
 def obtain_edges(session: Optional[botocore.session.Session], checker_list: List[str], nodes: List[Node],
                  region_allow_list: Optional[List[str]] = None, region_deny_list: Optional[List[str]] = None,
-                 scps: Optional[List[List[dict]]] = None) -> List[Edge]:
+                 scps: Optional[List[List[dict]]] = None, client_args_map: Optional[dict] = None) -> List[Edge]:
     """Given a list of nodes and a botocore Session, return a list of edges between those nodes. Only checks
     against services passed in the checker_list param. """
     result = []
@@ -60,5 +60,5 @@ def obtain_edges(session: Optional[botocore.session.Session], checker_list: List
     for check in checker_list:
         if check in checker_map:
             checker_obj = checker_map[check](session)
-            result.extend(checker_obj.return_edges(nodes, region_allow_list, region_deny_list, scps))
+            result.extend(checker_obj.return_edges(nodes, region_allow_list, region_deny_list, scps, client_args_map))
     return result