Optionally allow noisy results to flag risky actions regardless of resource constraints or conditions usage (salesforce#257)

Add --flag-all-risky-actions to arguments for scan and scan-policy-file commands

Author: kmcquade

cloudsplaining/command/scan.py

@@ -25,6 +25,7 @@
 from cloudsplaining import set_log_level
 
 
+# fmt: off
 @click.command(
     short_help="Scan a single file containing AWS IAM account authorization details and generate report on "
     "IAM security posture. "
@@ -34,13 +35,16 @@
 @click.option("-o", "--output", required=False, type=click.Path(exists=True), default=os.getcwd(), help="Output directory.")
 @click.option("-s", "--skip-open-report", required=False, default=False, is_flag=True, help="Don't open the HTML report in the web browser after creating. This helps when running the report in automation.")
 @click.option("-m", "--minimize", required=False, default=False, is_flag=True, help="Reduce the size of the HTML Report by pulling the Cloudsplaining Javascript code over the internet.")
+@click.option("-aR", "--flag-all-risky-actions", is_flag=True, help="Flag all risky actions, regardless of whether resource ARN constraints or conditions are used.")
 @click.option("-v", "--verbose", "verbosity", help="Log verbosity level.", count=True)
+# fmt: on
 def scan(
     input_file: str,
     exclusions_file: str,
     output: str,
     skip_open_report: bool,
     minimize: bool,
+    flag_all_risky_actions: bool,
     verbosity: int,
 ) -> None:  # pragma: no cover
     """
@@ -60,6 +64,13 @@ def scan(
     else:
         exclusions = DEFAULT_EXCLUSIONS
 
+    if flag_all_risky_actions:
+        flag_conditional_statements = True
+        flag_resource_arn_statements = True
+    else:
+        flag_conditional_statements = False
+        flag_resource_arn_statements = False
+
     if os.path.isfile(input_file):
         account_name = os.path.basename(input_file).split(".")[0]
         with open(input_file) as f:
@@ -72,6 +83,8 @@ def scan(
             output,
             write_data_files=True,
             minimize=minimize,
+            flag_conditional_statements=flag_conditional_statements,
+            flag_resource_arn_statements=flag_resource_arn_statements,
         )
         html_output_file = os.path.join(output, f"iam-report-{account_name}.html")
         logger.info("Saving the report to %s", html_output_file)
@@ -137,7 +150,9 @@ def scan_account_authorization_details(
     output_directory: str = os.getcwd(),
     write_data_files: bool = False,
     minimize: bool = False,
-    return_json_results: bool = False
+    return_json_results: bool = False,
+    flag_conditional_statements: bool = False,
+    flag_resource_arn_statements: bool = False,
 ) -> Any:  # pragma: no cover
     """
     Given the path to account authorization details files and the exclusions config file, scan all inline and
@@ -150,7 +165,9 @@ def scan_account_authorization_details(
     )
     check_authorization_details_schema(account_authorization_details_cfg)
     authorization_details = AuthorizationDetails(
-        account_authorization_details_cfg, exclusions
+        account_authorization_details_cfg, exclusions=exclusions,
+        flag_conditional_statements=flag_conditional_statements,
+        flag_resource_arn_statements=flag_resource_arn_statements
     )
     results = authorization_details.results
 

cloudsplaining/command/scan_policy_file.py

@@ -31,10 +31,11 @@
 @click.option("-i", "--input-file", type=str, help="Path of the IAM policy file to evaluate.")
 @click.option("-e", "--exclusions-file", help="A yaml file containing a list of actions to ignore when scanning.", type=click.Path(exists=True), required=False, default=EXCLUSIONS_FILE)
 @click.option("--high-priority-only", required=False, default=False, is_flag=True, help="If issues are found, only print the high priority risks (Resource Exposure, Privilege Escalation, Data Exfiltration). This can help with prioritization.")
+@click.option("-aR", "--flag-all-risky-actions", is_flag=True, help="Flag all risky actions, regardless of whether resource ARN constraints or conditions are used.")
 @click.option("--verbose", "-v", "verbosity", count=True)
 # pylint: disable=redefined-builtin
 def scan_policy_file(
-    input_file: str, exclusions_file: str, high_priority_only: bool, verbosity: int
+    input_file: str, exclusions_file: str, high_priority_only: bool, flag_all_risky_actions: bool, verbosity: int
 ) -> None:  # pragma: no cover
     """Scan a single policy file to identify missing resource constraints."""
     set_log_level(verbosity)
@@ -57,10 +58,16 @@ def scan_policy_file(
             exclusions_cfg = yaml.safe_load(yaml_file)
         except yaml.YAMLError as exc:
             logger.critical(exc)
-    # exclusions = Exclusions(exclusions_cfg)
+
+    if flag_all_risky_actions:
+        flag_conditional_statements = True
+        flag_resource_arn_statements = True
+    else:
+        flag_conditional_statements = False
+        flag_resource_arn_statements = False
 
     # Run the scan and get the raw data.
-    results = scan_policy(policy, exclusions_cfg)
+    results = scan_policy(policy, exclusions_cfg, flag_resource_arn_statements=flag_resource_arn_statements, flag_conditional_statements=flag_conditional_statements)
 
     # There will only be one finding in the results but it is in a list.
     results_exist = 0
@@ -135,15 +142,19 @@ def scan_policy_file(
 def scan_policy(
     policy_json: Dict[str, Any],
     exclusions_config: Dict[str, List[str]] = DEFAULT_EXCLUSIONS_CONFIG,
+    flag_conditional_statements: bool = False,
+    flag_resource_arn_statements: bool = False,
 ) -> Dict[str, Any]:
     """
     Scan a policy document for missing resource constraints.
 
     :param policy_json: Dictionary containing the IAM policy.
     :param exclusions_config: Exclusions configuration. If none, just send an empty dictionary. Defaults to the contents of cloudsplaining.shared.default-exclusions.yml
+    :param flag_resource_arn_statements:
+    :param flag_conditional_statements:
     :return:
     """
-    policy_document = PolicyDocument(policy_json)
     exclusions = Exclusions(exclusions_config)
+    policy_document = PolicyDocument(policy_json, exclusions=exclusions, flag_resource_arn_statements=flag_resource_arn_statements, flag_conditional_statements=flag_conditional_statements)
     policy_finding = PolicyFinding(policy_document, exclusions)
     return policy_finding.results

cloudsplaining/scan/authorization_details.py

@@ -29,29 +29,43 @@ def __init__(
         self,
         auth_json: Dict[str, List[Dict[str, Any]]],
         exclusions: Exclusions = DEFAULT_EXCLUSIONS,
+        flag_conditional_statements: bool = False,
+        flag_resource_arn_statements: bool = False,
     ) -> None:
+        """
+        Object to hold and analyze Account Authorization details.
+
+        :param auth_json: the JSON response of the aws iam get-account-authorization-details command
+        :param exclusions: A list of exclusions to apply to the results
+        :param flag_conditional_statements: Flag IAM statements with conditions, not just wildcards
+        :param flag_resource_arn_statements: Flag IAM statements with resource ARN restrictions, not just wildcards
+        """
         self.auth_json = auth_json
 
         if not isinstance(exclusions, Exclusions):
             raise Exception(
                 "For exclusions, please provide an object of the Exclusions type"
             )
         self.exclusions = exclusions
+        self.flag_conditional_statements = flag_conditional_statements
+        self.flag_resource_arn_statements = flag_resource_arn_statements
 
-        self.policies = ManagedPolicyDetails(auth_json.get("Policies", []), exclusions)
+        self.policies = ManagedPolicyDetails(auth_json.get("Policies", []), exclusions, flag_conditional_statements=flag_conditional_statements, flag_resource_arn_statements=flag_resource_arn_statements)
 
         # New Authorization file stuff
         self.group_detail_list = GroupDetailList(
-            auth_json.get("GroupDetailList", []), self.policies, exclusions
+            auth_json.get("GroupDetailList", []), self.policies, exclusions, flag_conditional_statements=flag_conditional_statements, flag_resource_arn_statements=flag_resource_arn_statements
         )
         self.user_detail_list = UserDetailList(
             auth_json.get("UserDetailList", []),
             self.policies,
             self.group_detail_list,
             exclusions,
+            flag_conditional_statements=flag_conditional_statements,
+            flag_resource_arn_statements=flag_resource_arn_statements
         )
         self.role_detail_list = RoleDetailList(
-            auth_json.get("RoleDetailList", []), self.policies, exclusions
+            auth_json.get("RoleDetailList", []), self.policies, exclusions, flag_conditional_statements=flag_conditional_statements, flag_resource_arn_statements=flag_resource_arn_statements
         )
 
     @property

cloudsplaining/scan/group_details.py

@@ -22,16 +22,20 @@ def __init__(
         group_details: List[Dict[str, Any]],
         policy_details: ManagedPolicyDetails,
         exclusions: Exclusions = DEFAULT_EXCLUSIONS,
+        flag_conditional_statements: bool = False,
+        flag_resource_arn_statements: bool = False,
     ) -> None:
         if not isinstance(exclusions, Exclusions):
             raise Exception(
                 "The exclusions provided is not an Exclusions type object. "
                 "Please supply an Exclusions object and try again."
             )
         self.exclusions = exclusions
+        self.flag_conditional_statements = flag_conditional_statements
+        self.flag_resource_arn_statements = flag_resource_arn_statements
 
         self.groups = [
-            GroupDetail(group_detail, policy_details, exclusions)
+            GroupDetail(group_detail, policy_details, exclusions=exclusions, flag_conditional_statements=flag_conditional_statements, flag_resource_arn_statements=flag_resource_arn_statements)
             for group_detail in group_details
         ]
 
@@ -98,6 +102,8 @@ def __init__(
         group_detail: Dict[str, Any],
         policy_details: ManagedPolicyDetails,
         exclusions: Exclusions = DEFAULT_EXCLUSIONS,
+        flag_conditional_statements: bool = False,
+        flag_resource_arn_statements: bool = False,
     ):
         """
         Initialize the GroupDetail object.
@@ -111,6 +117,10 @@ def __init__(
         self.group_id = group_detail["GroupId"]
         self.group_name = group_detail["GroupName"]
 
+        # Fix Issue #254 - Allow flagging risky actions even when there are resource constraints
+        self.flag_conditional_statements = flag_conditional_statements
+        self.flag_resource_arn_statements = flag_resource_arn_statements
+
         if not isinstance(exclusions, Exclusions):
             raise Exception(
                 "The exclusions provided is not an Exclusions type object. "
@@ -130,7 +140,10 @@ def __init__(
                     exclusions.is_policy_excluded(policy_name)
                     or exclusions.is_policy_excluded(policy_id)
                 ):
-                    inline_policy = InlinePolicy(policy_detail)
+                    # NOTE: The Exclusions were not here before the #254 fix (which was an unfiled bug I just discovered) so the presence of this might break some older unit tests. Might need to fix that.
+                    inline_policy = InlinePolicy(
+                        policy_detail, exclusions=exclusions, flag_conditional_statements=flag_conditional_statements,
+                        flag_resource_arn_statements=flag_resource_arn_statements)
                     self.inline_policies.append(inline_policy)
 
         # Managed Policies (either AWS-managed or Customer managed)

cloudsplaining/scan/inline_policy.py

@@ -13,7 +13,9 @@ class InlinePolicy:
     """
 
     def __init__(
-        self, policy_detail: Dict[str, Any], exclusions: Exclusions = DEFAULT_EXCLUSIONS
+        self, policy_detail: Dict[str, Any], exclusions: Exclusions = DEFAULT_EXCLUSIONS,
+        flag_conditional_statements: bool = False,
+        flag_resource_arn_statements: bool = False,
     ) -> None:
         """
         Initialize the InlinePolicy object.
@@ -25,9 +27,15 @@ def __init__(
                 "The exclusions provided is not an Exclusions type object. "
                 "Please supply an Exclusions object and try again."
             )
+        # Fix Issue #254 - Allow flagging risky actions even when there are resource constraints
+        self.flag_conditional_statements = flag_conditional_statements
+        self.flag_resource_arn_statements = flag_resource_arn_statements
+
         self.policy_name = policy_detail.get("PolicyName", "")
         self.policy_document = PolicyDocument(
-            cast(Dict[str, Any], policy_detail.get("PolicyDocument")), exclusions
+            cast(Dict[str, Any], policy_detail.get("PolicyDocument")), exclusions=exclusions,
+            flag_conditional_statements=flag_conditional_statements,
+            flag_resource_arn_statements=flag_resource_arn_statements
         )
         # Generating the provider ID based on a string representation of the Policy Document,
         # to avoid collisions where there are inline policies with the same name but different contents

cloudsplaining/scan/managed_policy_detail.py

@@ -28,6 +28,8 @@ def __init__(
         self,
         policy_details: List[Dict[str, Any]],
         exclusions: Exclusions = DEFAULT_EXCLUSIONS,
+        flag_conditional_statements: bool = False,
+        flag_resource_arn_statements: bool = False,
     ) -> None:
         self.policy_details = []
         if not isinstance(exclusions, Exclusions):
@@ -36,6 +38,8 @@ def __init__(
                 "Please supply an Exclusions object and try again."
             )
         self.exclusions = exclusions
+        self.flag_conditional_statements = flag_conditional_statements
+        self.flag_resource_arn_statements = flag_resource_arn_statements
 
         for policy_detail in policy_details:
             this_policy_name = policy_detail["PolicyName"]
@@ -66,7 +70,7 @@ def __init__(
                     this_policy_path,
                 )
                 continue
-            self.policy_details.append(ManagedPolicy(policy_detail, exclusions))
+            self.policy_details.append(ManagedPolicy(policy_detail, exclusions=exclusions, flag_resource_arn_statements=self.flag_resource_arn_statements, flag_conditional_statements=self.flag_conditional_statements))
 
     def get_policy_detail(self, arn: str) -> "ManagedPolicy":
         """Get a ManagedPolicy object by providing the ARN. This is useful to PrincipalDetail objects"""
@@ -125,11 +129,16 @@ class ManagedPolicy:
     """
 
     def __init__(
-        self, policy_detail: Dict[str, Any], exclusions: Exclusions = DEFAULT_EXCLUSIONS
+        self, policy_detail: Dict[str, Any], exclusions: Exclusions = DEFAULT_EXCLUSIONS,
+        flag_conditional_statements: bool = False,
+        flag_resource_arn_statements: bool = False,
     ) -> None:
         # Store the Raw JSON data from this for safekeeping
         self.policy_detail = policy_detail
 
+        self.flag_conditional_statements = flag_conditional_statements
+        self.flag_resource_arn_statements = flag_resource_arn_statements
+
         # Store the attributes per Policy item
         self.policy_name = policy_detail["PolicyName"]
         self.policy_id = policy_detail["PolicyId"]
@@ -172,7 +181,7 @@ def _policy_document(self) -> PolicyDocument:
         for policy_version in self.policy_version_list:
             if policy_version.get("IsDefaultVersion") is True:
                 return PolicyDocument(
-                    policy_version.get("Document"), exclusions=self.exclusions
+                    policy_version.get("Document"), exclusions=self.exclusions, flag_resource_arn_statements=self.flag_resource_arn_statements, flag_conditional_statements=self.flag_conditional_statements
                 )
         raise Exception(
             "Managed Policy ARN %s has no default Policy Document version", self.arn

cloudsplaining/scan/policy_document.py

@@ -28,11 +28,15 @@ class PolicyDocument:
     """
 
     def __init__(
-        self, policy: Dict[str, Any], exclusions: Exclusions = DEFAULT_EXCLUSIONS
+        self, policy: Dict[str, Any], exclusions: Exclusions = DEFAULT_EXCLUSIONS,
+        flag_conditional_statements: bool = False,
+        flag_resource_arn_statements: bool = False,
     ) -> None:
         statement_structure = policy.get("Statement", [])
         self.policy = policy
         self.statements = []
+        self.flag_conditional_statements = flag_conditional_statements
+        self.flag_resource_arn_statements = flag_resource_arn_statements
 
         if not isinstance(exclusions, Exclusions):
             raise Exception(
@@ -46,7 +50,7 @@ def __init__(
             statement_structure = [statement_structure]  # pragma: no cover
 
         for statement in statement_structure:
-            self.statements.append(StatementDetail(statement))
+            self.statements.append(StatementDetail(statement, flag_conditional_statements=self.flag_conditional_statements, flag_resource_arn_statements=self.flag_resource_arn_statements))
 
     @property
     def json(self) -> Dict[str, Any]:
@@ -88,6 +92,9 @@ def all_allowed_unrestricted_actions(self) -> List[str]:
                 and statement.restrictable_actions
             ):
                 allowed_actions.update(statement.restrictable_actions)
+            # Fix Issue #254 - Allow flagging risky actions even when there are resource constraints
+            if self.flag_resource_arn_statements and statement.effect_allow:
+                allowed_actions.update(statement.restrictable_actions)
         allowed_actions = self.filter_deny_statements(allowed_actions)
         return list(allowed_actions)
 
@@ -102,6 +109,13 @@ def all_allowed_unrestrictable_actions(self) -> List[str]:
                 and statement.unrestrictable_actions
             ):
                 allowed_actions.update(statement.unrestrictable_actions)
+            # Fix Issue #254 - Allow flagging risky actions even when there are resource constraints
+            if (
+                statement.effect_allow
+                and not statement.has_condition
+                and self.flag_resource_arn_statements
+            ):
+                allowed_actions.update(statement.unrestrictable_actions)
         allowed_actions = self.filter_deny_statements(allowed_actions)
         return list(allowed_actions)
 
@@ -116,6 +130,7 @@ def infrastructure_modification(self) -> List[str]:
                 ):
                     if action.lower() not in self.exclusions.exclude_actions:
                         actions_missing_resource_constraints.append(action)
+        actions_missing_resource_constraints.sort()
         return actions_missing_resource_constraints
 
     @property
@@ -220,8 +235,9 @@ def allows_specific_actions_without_constraints(
                 allowed.add(unrestricted_actions_lower[specific_action_lower])
             elif specific_action_lower in unrestrictable_actions_lower:
                 allowed.add(unrestrictable_actions_lower[specific_action_lower])
-
-        return list(allowed)
+        results = list(allowed)
+        results.sort()
+        return results
 
     @property
     def allows_data_exfiltration_actions(self) -> List[str]: