Suppress findings when conditions exist, except when --flag-all-risky-actions flag is included

cloudsplaining/command/scan.py

@@ -35,7 +35,7 @@
 @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("-aR", "--flag-all-risky-actions", required=False, default=False, 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)
 @click.option("-f", "--filter-severity", "severity", help="Filter the severity of findings to be reported.", multiple=True,type=click.Choice(['CRITICAL','HIGH', 'MEDIUM','LOW','NONE'], case_sensitive=False))
 

cloudsplaining/command/scan_multi_account.py

@@ -50,6 +50,7 @@ def _accounts(self) -> Dict[str, str]:
 @optgroup.option("-b", "--output-bucket", "output_bucket", type=str, help="The S3 bucket to save the results. Supply this and/or --output-directory.")
 @optgroup.group("Other Options", help="")
 @optgroup.option("-w", "--write-data-file", is_flag=True, required=False, default=False, help="Save the cloudsplaining JSON-formatted data results.")
+@click.option("-aR", "--flag-all-risky-actions", required=False, default=False, 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)
 @click.option("-f", "--filter-severity", "severity", help="Filter the severity of findings to be reported.", multiple=True,type=click.Choice(['CRITICAL','HIGH', 'MEDIUM','LOW','NONE'], case_sensitive=False))
 
@@ -61,6 +62,7 @@ def scan_multi_account(
     output_directory: str,
     output_bucket: str,
     write_data_file: bool,
+    flag_all_risky_actions: bool,
     verbosity: int,
     severity: List[str],
 ) -> None:
@@ -71,6 +73,13 @@ def scan_multi_account(
     with open(config_file, "r") as yaml_file:
         config = yaml.safe_load(yaml_file)
 
+    if flag_all_risky_actions:
+        flag_conditional_statements = True
+        flag_resource_arn_statements = True
+    else:
+        flag_conditional_statements = False
+        flag_resource_arn_statements = False
+
     # Use the following lines to run this in a library
     multi_account_config = MultiAccountConfig(config=config, role_name=role_name)
     exclusions = get_exclusions(exclusions_file=exclusions_file)
@@ -83,6 +92,8 @@ def scan_multi_account(
         output_bucket=output_bucket,
         write_data_file=write_data_file,
         severity=severity,
+        flag_conditional_statements=flag_conditional_statements,
+        flag_resource_arn_statements=flag_resource_arn_statements
     )
 
 
@@ -95,6 +106,8 @@ def scan_accounts(
     output_directory: Optional[str] = None,
     output_bucket: Optional[str] = None,
     severity: List[str] = [],
+    flag_conditional_statements: bool = False,
+    flag_resource_arn_statements: bool = False
 ) -> None:
     """Use this method as a library to scan multiple accounts"""
     # TODO: Speed improvements? Multithreading? This currently runs sequentially.
@@ -108,6 +121,8 @@ def scan_accounts(
             exclusions=exclusions,
             profile=profile,
             severity=severity,
+            flag_conditional_statements=flag_conditional_statements,
+            flag_resource_arn_statements=flag_resource_arn_statements
         )
         html_report = HTMLReport(
             account_id=target_account_id,
@@ -167,6 +182,8 @@ def scan_account(
     exclusions: Exclusions,
     profile: Optional[str] = None,
     severity: List[str] = [],
+    flag_conditional_statements: bool = False,
+    flag_resource_arn_statements: bool = False
 ) -> Dict[str, Dict[str, Any]]:
     """Scan a target account in one shot"""
     account_authorization_details = download_account_authorization_details(
@@ -175,7 +192,13 @@ def scan_account(
         profile=profile,
     )
     check_authorization_details_schema(account_authorization_details)
-    authorization_details = AuthorizationDetails(account_authorization_details, exclusions=exclusions,severity=severity)
+    authorization_details = AuthorizationDetails(
+        auth_json = account_authorization_details,
+        exclusions = exclusions,
+        severity = severity,
+        flag_conditional_statements = flag_conditional_statements,
+        flag_resource_arn_statements = flag_resource_arn_statements
+    )
     results = authorization_details.results
     return results
 

cloudsplaining/command/scan_policy_file.py

@@ -49,6 +49,8 @@
 @click.option(
     "-aR",
     "--flag-all-risky-actions",
+    required=False,
+    default=False,
     is_flag=True,
     help="Flag all risky actions, regardless of whether resource ARN constraints or conditions are used.",
 )

cloudsplaining/output/policy_finding.py

@@ -52,7 +52,7 @@ def _missing_resource_constraints_for_modify_actions(self) -> List[str]:
         actions_missing_resource_constraints = set()
         for statement in self.policy_document.statements:
             logger.debug("Evaluating statement: %s", statement.json)
-            if statement.effect == "Allow":
+            if statement.effect == "Allow" and not statement.has_condition:
                 actions_missing_resource_constraints.update(
                     statement.missing_resource_constraints_for_modify_actions(
                         self.exclusions

cloudsplaining/scan/inline_policy.py

@@ -66,8 +66,8 @@ def _is_excluded(self, exclusions: Exclusions) -> bool:
             or exclusions.is_policy_excluded(self.policy_id)
         )
 
-    def getFindingLinks(self, findings: Any) -> List[Any]:
-        links: List[Any] = []
+    def getFindingLinks(self, findings: List[Dict[str, Any]]) -> Dict[str, str]:
+        links = {}
         for finding in findings:
             links[
                 finding["type"]

cloudsplaining/scan/policy_document.py

@@ -125,7 +125,7 @@ def infrastructure_modification(self) -> List[str]:
         """Return a list of modify only missing resource constraints"""
         actions_missing_resource_constraints = []
         for statement in self.statements:
-            if statement.effect == "Allow":
+            if statement.effect == "Allow" and not statement.has_condition:
                 for action in statement.missing_resource_constraints_for_modify_actions(
                     self.exclusions
                 ):
@@ -272,7 +272,7 @@ def service_wildcard(self) -> List[str]:
 
         for statement in self.statements:
             logger.debug("Evaluating statement: %s", statement.json)
-            if statement.effect_allow:
+            if statement.effect_allow and not statement.has_condition:
                 if isinstance(statement.actions, list):
                     for action in statement.actions:
                         # If the action is a straight up *

cloudsplaining/scan/statement_detail.py

@@ -198,9 +198,8 @@ def permissions_management_actions_without_constraints(self) -> List[str]:
         do not have resource constraints"""
         result = []
         if (
-            not self.has_resource_constraints
-            # Fix Issue #254 - Allow flagging risky actions even when there are resource constraints
-            or self.flag_resource_arn_statements
+            (not self.has_resource_constraints or self.flag_resource_arn_statements) and
+             not self.has_condition
         ):
             result = remove_actions_not_matching_access_level(
                 self.restrictable_actions, "Permissions management"
@@ -214,9 +213,8 @@ def write_actions_without_constraints(self) -> List[str]:
         do not have resource constraints"""
         result = []
         if (
-            not self.has_resource_constraints
-            # Fix Issue #254 - Allow flagging risky actions even when there are resource constraints
-            or self.flag_resource_arn_statements
+            (not self.has_resource_constraints or self.flag_resource_arn_statements) and
+             not self.has_condition
         ):
             result = remove_actions_not_matching_access_level(
                 self.restrictable_actions, "Write"

test/files/scanning/test_inline_policy_results.json

@@ -24,7 +24,7 @@
     "severity": "high",
     "description": "<p>These policies allow a combination of IAM actions that allow a principal with these permissions to escalate their privileges - for example, by creating an access key for another IAM user, or modifying their own permissions. This research was pioneered by Spencer Gietzen at Rhino Security Labs.  Remediation guidance can be found <a href=\"https://rhinosecuritylabs.com/aws/aws-privilege-escalation-methods-mitigation/\">here</a>.</p>",
     "findings":[],
-    "links":[]
+    "links":{}
   },
   "DataExfiltration":{
     "severity": "medium",

test/scanning/test_policy_document.py

@@ -539,3 +539,101 @@ def test_gh_254_flag_risky_actions_with_resource_constraints_infrastructure_modi
         self.assertListEqual(policy_document.infrastructure_modification, ['ec2:AuthorizeSecurityGroupIngress'])
         policy_document = PolicyDocument(test_policy, flag_resource_arn_statements=False)
         self.assertListEqual(policy_document.infrastructure_modification, [])
+
+    def test_gh_278_all_issue_types_respect_conditions_on_policy(self):
+        test_policy_all_issues_no_conditions = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Action": [
+                        # Credentials Exposure
+                        "ec2:GetPasswordData",
+                        # Data Exfiltration
+                        "s3:GetObject",
+                        # Infrastructure Modification
+                        "ec2:AuthorizeSecurityGroupIngress",
+                        # Privilege Escalation
+                        "ec2:RunInstances",
+                        "iam:PassRole",
+                        # Resource Exposure
+                        "ec2:CreateNetworkInterfacePermission"
+                    ],
+                    "Resource": "*"
+                }
+            ]
+        }
+        policy_document = PolicyDocument(test_policy_all_issues_no_conditions)
+        self.assertListEqual(policy_document.credentials_exposure, ["ec2:GetPasswordData"])
+        self.assertListEqual(policy_document.allows_data_exfiltration_actions, ["s3:GetObject"])
+        self.assertListEqual(policy_document.infrastructure_modification, [
+            "ec2:AuthorizeSecurityGroupIngress", 
+            "ec2:CreateNetworkInterfacePermission", 
+            "ec2:RunInstances",
+            "iam:PassRole",
+            "s3:GetObject"
+        ])
+        self.assertListEqual(policy_document.allows_privilege_escalation, [{
+            "actions": [
+                "iam:passrole",
+                "ec2:runinstances"
+            ],
+            "type": "CreateEC2WithExistingIP"
+        }])
+        self.assertListEqual(policy_document.permissions_management_without_constraints, [
+            "ec2:CreateNetworkInterfacePermission",
+            "iam:PassRole"
+        ])
+
+        test_policy_service_wildcard = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Action": [
+                        # Service Wildcard
+                        "kinesis:*"
+                    ],
+                    "Resource": "*"
+                }
+            ]
+        }
+        policy_document_service_wildcard = PolicyDocument(test_policy_service_wildcard)
+        self.assertListEqual(policy_document_service_wildcard.service_wildcard, ["kinesis"])
+
+        test_policy_all_issues_conditions = {
+            "Version": "2012-10-17",
+            "Statement": [
+                {
+                    "Effect": "Allow",
+                    "Action": [
+                        # Credentials Exposure
+                        "ec2:GetPasswordData",
+                        # Data Exfiltration
+                        "s3:GetObject",
+                        # Infrastructure Modification
+                        "ec2:AuthorizeSecurityGroupIngress",
+                        # Privilege Escalation
+                        "ec2:RunInstances",
+                        "iam:PassRole",
+                        # Resource Exposure
+                        "ec2:CreateNetworkInterfacePermission",
+                        # Service Wildcard
+                        "kinesis:*"
+                    ],
+                    "Resource": "*",
+                    "Condition": {
+                        "StringEquals": {
+                            "aws:PrincipalAccount": "123456789012"
+                        }
+                    }
+                }
+            ]
+        }
+        policy_document_condition = PolicyDocument(test_policy_all_issues_conditions)
+        self.assertListEqual(policy_document_condition.credentials_exposure, [])
+        self.assertListEqual(policy_document_condition.allows_data_exfiltration_actions, [])
+        self.assertListEqual(policy_document_condition.infrastructure_modification, [])
+        self.assertListEqual(policy_document_condition.allows_privilege_escalation, [])
+        self.assertListEqual(policy_document_condition.permissions_management_without_constraints, [])
+        self.assertListEqual(policy_document.service_wildcard, [])