Merge pull request #37 from kmcquade/refactor/granular-exclusions-v4

Granular Exclusions, UI Uplift, track group membership

Author: kmcquade

CHANGELOG.md

@@ -3,6 +3,10 @@
 ## Unreleased
 * Docker
 
+## 0.1.0 (2020-05-11)
+* Granular exclusions: Fixed issue where exclusions file was including dangling policies in the results (Fixes #33)
+* Changed IAM Principals table so that the principals can be sorted according to their risks. This will really help with pentesting
+
 ## 0.0.14 (2020-05-07)
 * Fix issue where Data Exposure tallies were not showing up in the AWS Managed table correctly.
 

Pipfile

@@ -17,7 +17,7 @@ bandit = "==1.6.2"
 mkdocs = "==1.1"
 
 [packages]
-policy_sentry = "==0.8.0.5"
+policy_sentry = "==0.8.0.6"
 click = "==7.0"
 schema = "==0.7.1"
 boto3 = "*"

cloudsplaining/bin/cli.py

@@ -7,7 +7,7 @@
 """
     Cloudsplaining is an AWS IAM Assessment tool that identifies violations of least privilege and generates a risk-prioritized HTML report with a triage worksheet.
 """
-__version__ = "0.0.14"
+__version__ = "0.1.0"
 import click
 from cloudsplaining import command
 

cloudsplaining/command/scan.py

@@ -14,10 +14,8 @@
 import click
 import click_log
 from cloudsplaining.shared.constants import EXCLUSIONS_FILE
-from cloudsplaining.shared.validation import (
-    check_exclusions_schema,
-    check_authorization_details_schema,
-)
+from cloudsplaining.shared.validation import check_authorization_details_schema
+from cloudsplaining.shared.exclusions import Exclusions, DEFAULT_EXCLUSIONS
 from cloudsplaining.scan.authorization_details import AuthorizationDetails
 from cloudsplaining.output.html_report import generate_html_report
 from cloudsplaining.output.data_file import write_results_data_file
@@ -67,29 +65,43 @@
 )
 @click_log.simple_verbosity_option()
 # pylint: disable=redefined-builtin
-def scan(input, exclusions_file, output, all_access_levels, skip_open_report):  # pragma: no cover
+def scan(
+    input, exclusions_file, output, all_access_levels, skip_open_report
+):  # pragma: no cover
     """
     Given the path to account authorization details files and the exclusions config file, scan all inline and
     managed policies in the account to identify actions that do not leverage resource constraints.
     """
     if exclusions_file:
+        # Get the exclusions configuration
         with open(exclusions_file, "r") as yaml_file:
             try:
                 exclusions_cfg = yaml.safe_load(yaml_file)
             except yaml.YAMLError as exc:
                 logger.critical(exc)
-        check_exclusions_schema(exclusions_cfg)
+        exclusions = Exclusions(exclusions_cfg)
+    else:
+        exclusions = DEFAULT_EXCLUSIONS
+
     if os.path.isfile(input):
-        scan_account_authorization_file(input, exclusions_cfg, output, all_access_levels, skip_open_report)
+        scan_account_authorization_file(
+            input, exclusions, output, all_access_levels, skip_open_report
+        )
     if os.path.isdir(input):
-        logger.info("The path given is a directory. Scanning for account authorization files and generating report.")
+        logger.info(
+            "The path given is a directory. Scanning for account authorization files and generating report."
+        )
         input_files = get_authorization_files_in_directory(input)
         for file in input_files:
             logger.info(f"Scanning file: {file}")
-            scan_account_authorization_file(file, exclusions_cfg, output, all_access_levels, skip_open_report)
+            scan_account_authorization_file(
+                file, exclusions, output, all_access_levels, skip_open_report
+            )
 
 
-def scan_account_authorization_file(input_file, exclusions_cfg, output, all_access_levels, skip_open_report):  # pragma: no cover
+def scan_account_authorization_file(
+    input_file, exclusions, output, all_access_levels, skip_open_report
+):  # pragma: no cover
     """
     Given the path to account authorization details files and the exclusions config file, scan all inline and
     managed policies in the account to identify actions that do not leverage resource constraints.
@@ -107,7 +119,7 @@ def scan_account_authorization_file(input_file, exclusions_cfg, output, all_acce
         )
         authorization_details = AuthorizationDetails(account_authorization_details_cfg)
         results = authorization_details.missing_resource_constraints(
-            exclusions_cfg, modify_only=False
+            exclusions, modify_only=False
         )
     else:
         logger.debug(
@@ -116,10 +128,26 @@ def scan_account_authorization_file(input_file, exclusions_cfg, output, all_acce
         )
         authorization_details = AuthorizationDetails(account_authorization_details_cfg)
         results = authorization_details.missing_resource_constraints(
-            exclusions_cfg, modify_only=True
+            exclusions, modify_only=True
         )
 
     principal_policy_mapping = authorization_details.principal_policy_mapping
+    for principal_policy_entry in principal_policy_mapping:
+        for finding_result in results:
+            if (
+                principal_policy_entry.get("PolicyName").lower()
+                == finding_result.get("PolicyName").lower()
+            ):
+                principal_policy_entry["Actions"] = len(finding_result["Actions"])
+                principal_policy_entry["PrivilegeEscalation"] = len(
+                    finding_result["PrivilegeEscalation"]
+                )
+                principal_policy_entry["DataExfiltrationActions"] = len(
+                    finding_result["DataExfiltrationActions"]
+                )
+                principal_policy_entry["PermissionsManagementActions"] = len(
+                    finding_result["PermissionsManagementActions"]
+                )
 
     account_name = Path(input_file).stem
 
@@ -149,8 +177,12 @@ def scan_account_authorization_file(input_file, exclusions_cfg, output, all_acce
     print(f"Raw data file saved: {str(raw_data_filepath)}")
 
     # Principal policy mapping
-    principal_policy_mapping_file = os.path.join(output, f"iam-principals-{account_name}.json")
-    principal_policy_mapping_filepath = write_results_data_file(principal_policy_mapping, principal_policy_mapping_file)
+    principal_policy_mapping_file = os.path.join(
+        output, f"iam-principals-{account_name}.json"
+    )
+    principal_policy_mapping_filepath = write_results_data_file(
+        principal_policy_mapping, principal_policy_mapping_file
+    )
     print(f"Principals data file saved: {str(principal_policy_mapping_filepath)}")
 
     print("Creating the HTML Report")
@@ -159,24 +191,30 @@ def scan_account_authorization_file(input_file, exclusions_cfg, output, all_acce
         results,
         principal_policy_mapping,
         output_directory,
-        exclusions_cfg,
+        exclusions.config,
         skip_open_report=skip_open_report,
     )
 
 
 def get_authorization_files_in_directory(directory):  # pragma: no cover
     """Get a list of download-account-authorization-files in a directory"""
-    file_list = [f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))]
+    file_list = [
+        f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f))
+    ]
     file_list_with_full_path = []
     for file in file_list:
         if file.endswith(".json"):
-            file_list_with_full_path.append(os.path.abspath(os.path.join(directory, file)))
+            file_list_with_full_path.append(
+                os.path.abspath(os.path.join(directory, file))
+            )
     new_file_list = []
     for file in file_list_with_full_path:
         with open(file) as f:
             contents = f.read()
         account_authorization_details_cfg = json.loads(contents)
-        valid_schema = check_authorization_details_schema(account_authorization_details_cfg)
+        valid_schema = check_authorization_details_schema(
+            account_authorization_details_cfg
+        )
         if valid_schema:
             new_file_list.append(file)
     return new_file_list

cloudsplaining/command/scan_policy_file.py

@@ -12,11 +12,13 @@
 import yaml
 import click
 import click_log
-from cloudsplaining.output.findings import Findings, Finding
-from cloudsplaining.shared.constants import EXCLUSIONS_FILE, DEFAULT_EXCLUSIONS_CONFIG
+from cloudsplaining.output.findings import Findings, PolicyFinding
+from cloudsplaining.shared.constants import EXCLUSIONS_FILE
 from cloudsplaining.scan.policy_document import PolicyDocument
-from cloudsplaining.shared.validation import check_exclusions_schema
-from cloudsplaining.shared.exclusions import is_name_excluded
+from cloudsplaining.shared.exclusions import (
+    Exclusions,
+    DEFAULT_EXCLUSIONS,
+)
 
 logger = logging.getLogger(__name__)
 click_log.basic_config(logger)
@@ -60,7 +62,7 @@ def scan_policy_file(input, exclusions_file, high_priority_only):  # pragma: no
             exclusions_cfg = yaml.safe_load(yaml_file)
         except yaml.YAMLError as exc:
             logger.critical(exc)
-    check_exclusions_schema(exclusions_cfg)
+    exclusions = Exclusions(exclusions_cfg)
 
     # Get the Policy
     with open(file) as json_file:
@@ -70,79 +72,72 @@ def scan_policy_file(input, exclusions_file, high_priority_only):  # pragma: no
     policy_name = Path(file).stem
 
     # Run the scan and get the raw data.
-    results = scan_policy(policy, policy_name, exclusions_cfg)
+    results = scan_policy(policy, policy_name, exclusions)
 
     # There will only be one finding in the results but it is in a list.
+    results_exist = 0
     for finding in results:
         if finding["PrivilegeEscalation"]:
             print(f"{RED}Issue found: Privilege Escalation{END}")
+            results_exist += 1
             for item in finding["PrivilegeEscalation"]:
                 print(f"- Method: {item['type']}")
                 print(f"  Actions: {', '.join(item['PrivilegeEscalation'])}\n")
         if finding["DataExfiltrationActions"]:
+            results_exist += 1
             print(f"{RED}Issue found: Data Exfiltration{END}")
             print(
                 f"{BOLD}Actions{END}: {', '.join(finding['DataExfiltrationActions'])}\n"
             )
         if finding["PermissionsManagementActions"]:
+            results_exist += 1
             print(f"{RED}Issue found: Resource Exposure{END}")
             print(
                 f"{BOLD}Actions{END}: {', '.join(finding['PermissionsManagementActions'])}\n"
             )
         if not high_priority_only:
+            results_exist += 1
             print(f"{RED}Issue found: Unrestricted Infrastructure Modification{END}")
             print(f"{BOLD}Actions{END}: {', '.join(finding['Actions'])}")
+    if results_exist == 0:
+        print("There were no results found.")
 
 
-def scan_policy(policy_json, policy_name, exclusions_cfg=DEFAULT_EXCLUSIONS_CONFIG):
+def scan_policy(policy_json, policy_name, exclusions=DEFAULT_EXCLUSIONS):
     """
     Scan a policy document for missing resource constraints.
 
+    :param exclusions: Exclusions object
     :param policy_json: The AWS IAM policy document.
-    :param exclusions_cfg: Defaults to the embedded exclusions file, which has no effect here.
     :param policy_name: The name of the IAM policy. Defaults to the filename when used from command line.
     :return:
     """
-    policy_document = PolicyDocument(policy_json)
     actions_missing_resource_constraints = []
 
-    # EXCLUDED ACTIONS - actions to exclude if they are false positives
-    excluded_actions = exclusions_cfg.get("exclude-actions", None)
-    if excluded_actions == [""]:
-        excluded_actions = None
-
-    # convert to lowercase for comparison purposes
-    # some weird if/else logic to reduce loops and improve performance slightly
-    if excluded_actions:
-        excluded_actions = [x.lower() for x in excluded_actions]
+    policy_document = PolicyDocument(policy_json)
 
-    always_include_actions = exclusions_cfg.get("include-actions")
-    findings = Findings()
+    findings = Findings(exclusions)
 
     for statement in policy_document.statements:
+        logger.debug("Evaluating statement: %s", statement.json)
         if statement.effect == "Allow":
             actions_missing_resource_constraints.extend(
-                statement.missing_resource_constraints_for_modify_actions(
-                    always_include_actions
-                )
+                statement.missing_resource_constraints_for_modify_actions(exclusions)
             )
     if actions_missing_resource_constraints:
-        results_placeholder = []
-        for action in actions_missing_resource_constraints:
-            if excluded_actions:
-                if not is_name_excluded(action.lower(), excluded_actions):
-                    results_placeholder.append(action)  # pragma: no cover
-            else:
-                results_placeholder.append(action)
-        actions_missing_resource_constraints = list(
-            dict.fromkeys(results_placeholder)
+        these_results = list(
+            dict.fromkeys(actions_missing_resource_constraints)
         )  # remove duplicates
-        actions_missing_resource_constraints.sort()
-        finding = Finding(
+        these_results.sort()
+        finding = PolicyFinding(
             policy_name=policy_name,
             arn=policy_name,
-            actions=actions_missing_resource_constraints,
+            actions=these_results,
             policy_document=policy_document,
+            exclusions=exclusions,
         )
-        findings.add(finding)
-    return findings.json
+        findings.add_policy_finding(finding)
+        findings.single_use = True
+        return finding.json
+    else:
+        return []