Update setup.py to include cached-property; Fix linting (#160)

* Update setup.py to include cached-property

* Bump to 0.3.0

* black

Author: kmcquade

cloudsplaining/__init__.py

@@ -1,24 +1,26 @@
 # pylint: disable=missing-module-docstring
 import logging
 import sys
+
 # Set default logging handler to avoid "No handler found" warnings.
 from logging import NullHandler
 
-#logging.getLogger(__name__).addHandler(NullHandler())
+# logging.getLogger(__name__).addHandler(NullHandler())
 # Uncomment to get the full debug logs.
 # 2020-10-06 10:04:17,200 - root - DEBUG - Leveraging the bundled IAM Definition.
 # Need to figure out how to get click_log to do this for me.
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.WARNING)
 handler = logging.StreamHandler(sys.stdout)
-formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
 handler.setFormatter(formatter)
 logger.addHandler(handler)
 
-#logging.getLogger().addHandler(NullHandler())
+# logging.getLogger().addHandler(NullHandler())
 
 name = "cloudsplaining"  # pylint: disable=invalid-name
 
+
 def change_log_level(log_level):
     """"Change log level of module logger"""
     logger.setLevel(log_level)

cloudsplaining/bin/version.py

@@ -1,2 +1,2 @@
 # pylint: disable=missing-module-docstring
-__version__ = "0.2.5"
+__version__ = "0.3.0"

cloudsplaining/command/create_exclusions_file.py

@@ -29,9 +29,12 @@
     help="Relative path to output file where we want to store the exclusions template.",
 )
 @click.option(
-    '--verbose','-v',
-    type=click.Choice(['critical', 'error', 'warning', 'info', 'debug'],
-    case_sensitive=False))
+    "--verbose",
+    "-v",
+    type=click.Choice(
+        ["critical", "error", "warning", "info", "debug"], case_sensitive=False
+    ),
+)
 def create_exclusions_file(output_file, verbose):
     """
     Creates a YML file to be used as a custom exclusions template,
@@ -51,4 +54,6 @@ def create_exclusions_file(output_file, verbose):
     )
     print("\tcloudsplaining download")
     print("You can use this with the scan command as shown below: ")
-    print("\tcloudsplaining scan --exclusions-file exclusions.yml --input-file default.json")
+    print(
+        "\tcloudsplaining scan --exclusions-file exclusions.yml --input-file default.json"
+    )

cloudsplaining/command/download.py

@@ -42,9 +42,12 @@
     " Note that this will dramatically increase the size of the downloaded file.",
 )
 @click.option(
-    '--verbose','-v',
-    type=click.Choice(['critical', 'error', 'warning', 'info', 'debug'],
-    case_sensitive=False))
+    "--verbose",
+    "-v",
+    type=click.Choice(
+        ["critical", "error", "warning", "info", "debug"], case_sensitive=False
+    ),
+)
 def download(profile, output, include_non_default_policy_versions, verbose):
     """
     Runs aws iam get-authorization-details on all accounts specified in the aws credentials file, and stores them in
@@ -62,7 +65,9 @@ def download(profile, output, include_non_default_policy_versions, verbose):
     else:
         output_filename = "default.json"
 
-    results = get_account_authorization_details(session_data, include_non_default_policy_versions)
+    results = get_account_authorization_details(
+        session_data, include_non_default_policy_versions
+    )
 
     if os.path.exists(output_filename):
         os.remove(output_filename)
@@ -72,7 +77,9 @@ def download(profile, output, include_non_default_policy_versions, verbose):
     return 1
 
 
-def get_account_authorization_details(session_data, include_non_default_policy_versions):
+def get_account_authorization_details(
+    session_data, include_non_default_policy_versions
+):
     """Runs aws-iam-get-account-authorization-details"""
     session = boto3.Session(**session_data)
     config = Config(connect_timeout=5, retries={"max_attempts": 10})

cloudsplaining/command/expand_policy.py

@@ -25,9 +25,12 @@
     help="Path to the JSON policy file.",
 )
 @click.option(
-    '--verbose','-v',
-    type=click.Choice(['critical', 'error', 'warning', 'info', 'debug'],
-    case_sensitive=False))
+    "--verbose",
+    "-v",
+    type=click.Choice(
+        ["critical", "error", "warning", "info", "debug"], case_sensitive=False
+    ),
+)
 def expand_policy(input_file, verbose):  # pylint: disable=redefined-builtin
     """
     Expand the * Actions in IAM policy files to improve readability

cloudsplaining/command/scan.py

@@ -62,15 +62,19 @@
     required=False,
     default=False,
     is_flag=True,
-    help="Reduce the size of the HTML Report by pulling the Cloudsplaining Javascript code over the internet."
+    help="Reduce the size of the HTML Report by pulling the Cloudsplaining Javascript code over the internet.",
 )
 @click.option(
-    '--verbose','-v',
-    type=click.Choice(['critical', 'error', 'warning', 'info', 'debug'],
-    case_sensitive=False))
+    "--verbose",
+    "-v",
+    type=click.Choice(
+        ["critical", "error", "warning", "info", "debug"], case_sensitive=False
+    ),
+)
 # pylint: disable=redefined-builtin
 def scan(
-    input_file, exclusions_file, output, skip_open_report, minimize, verbose):  # pragma: no cover
+    input_file, exclusions_file, output, skip_open_report, minimize, verbose
+):  # 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.
@@ -95,8 +99,12 @@ def scan(
             contents = f.read()
             account_authorization_details_cfg = json.loads(contents)
         rendered_html_report = scan_account_authorization_details(
-            account_authorization_details_cfg, exclusions, account_name, output, write_data_files=True,
-            minimize=minimize
+            account_authorization_details_cfg,
+            exclusions,
+            account_name,
+            output,
+            write_data_files=True,
+            minimize=minimize,
         )
         html_output_file = os.path.join(output, f"iam-report-{account_name}.html")
         logger.info("Saving the report to %s", html_output_file)
@@ -128,8 +136,12 @@ def scan(
             account_name = Path(file).stem
             # Scan the Account Authorization Details config
             rendered_html_report = scan_account_authorization_details(
-                account_authorization_details_cfg, exclusions, account_name, output, write_data_files=True,
-                minimize=minimize
+                account_authorization_details_cfg,
+                exclusions,
+                account_name,
+                output,
+                write_data_files=True,
+                minimize=minimize,
             )
             html_output_file = os.path.join(output, f"iam-report-{account_name}.html")
             logger.info("Saving the report to %s", html_output_file)
@@ -149,8 +161,12 @@ def scan(
 
 
 def scan_account_authorization_details(
-    account_authorization_details_cfg, exclusions, account_name="default", output_directory=os.getcwd(),
-    write_data_files=False, minimize=False
+    account_authorization_details_cfg,
+    exclusions,
+    account_name="default",
+    output_directory=os.getcwd(),
+    write_data_files=False,
+    minimize=False,
 ):  # pragma: no cover
     """
     Given the path to account authorization details files and the exclusions config file, scan all inline and
@@ -162,7 +178,9 @@ def scan_account_authorization_details(
         "resource constraints..."
     )
     check_authorization_details_schema(account_authorization_details_cfg)
-    authorization_details = AuthorizationDetails(account_authorization_details_cfg, exclusions)
+    authorization_details = AuthorizationDetails(
+        account_authorization_details_cfg, exclusions
+    )
     results = authorization_details.results
 
     # Lazy method to get an account ID
@@ -176,7 +194,7 @@ def scan_account_authorization_details(
         account_id=account_id,
         account_name=account_name,
         results=results,
-        minimize=minimize
+        minimize=minimize,
     )
     rendered_report = html_report.get_html_report()
 
@@ -185,11 +203,17 @@ def scan_account_authorization_details(
         if output_directory is None:
             output_directory = os.getcwd()
 
-        results_data_file = os.path.join(output_directory, f"iam-results-{account_name}.json")
-        results_data_filepath = write_results_data_file(authorization_details.results, results_data_file)
+        results_data_file = os.path.join(
+            output_directory, f"iam-results-{account_name}.json"
+        )
+        results_data_filepath = write_results_data_file(
+            authorization_details.results, results_data_file
+        )
         print(f"Results data saved: {str(results_data_filepath)}")
 
-        findings_data_file = os.path.join(output_directory, f"iam-findings-{account_name}.json")
+        findings_data_file = os.path.join(
+            output_directory, f"iam-findings-{account_name}.json"
+        )
         findings_data_filepath = write_results_data_file(results, findings_data_file)
         print(f"Findings data file saved: {str(findings_data_filepath)}")
 

cloudsplaining/command/scan_policy_file.py

@@ -16,6 +16,7 @@
 from cloudsplaining.shared.exclusions import Exclusions
 from cloudsplaining.output.policy_finding import PolicyFinding
 from cloudsplaining import change_log_level
+
 logger = logging.getLogger(__name__)
 BOLD = "\033[1m"
 RED = "\033[91m"
@@ -47,11 +48,16 @@
     " (Resource Exposure, Privilege Escalation, Data Exfiltration). This can help with prioritization.",
 )
 @click.option(
-    '--verbose','-v',
-    type=click.Choice(['critical', 'error', 'warning', 'info', 'debug'],
-    case_sensitive=False))
+    "--verbose",
+    "-v",
+    type=click.Choice(
+        ["critical", "error", "warning", "info", "debug"], case_sensitive=False
+    ),
+)
 # pylint: disable=redefined-builtin
-def scan_policy_file(input_file, exclusions_file, high_priority_only, verbose):  # pragma: no cover
+def scan_policy_file(
+    input_file, exclusions_file, high_priority_only, verbose
+):  # pragma: no cover
     """Scan a single policy file to identify missing resource constraints."""
     if verbose:
         log_level = getattr(logging, verbose.upper())
@@ -85,7 +91,9 @@ def scan_policy_file(input_file, exclusions_file, high_priority_only, verbose):
     if results:
         # Privilege Escalation
         if results.get("PrivilegeEscalation"):
-            print(f"{RED}Potential Issue found: Policy is capable of Privilege Escalation{END}")
+            print(
+                f"{RED}Potential Issue found: Policy is capable of Privilege Escalation{END}"
+            )
             results_exist += 1
             for item in results.get("PrivilegeEscalation"):
                 print(f"- Method: {item.get('type')}")
@@ -94,31 +102,33 @@ def scan_policy_file(input_file, exclusions_file, high_priority_only, verbose):
         # Data Exfiltration
         if results.get("DataExfiltration"):
             results_exist += 1
-            print(f"{RED}Potential Issue found: Policy is capable of Data Exfiltration{END}")
             print(
-                f"{BOLD}Actions{END}: {', '.join(results.get('DataExfiltration'))}\n"
+                f"{RED}Potential Issue found: Policy is capable of Data Exfiltration{END}"
             )
+            print(f"{BOLD}Actions{END}: {', '.join(results.get('DataExfiltration'))}\n")
 
         # Resource Exposure
         if results.get("ResourceExposure"):
             results_exist += 1
-            print(f"{RED}Potential Issue found: Policy is capable of Resource Exposure{END}")
             print(
-                f"{BOLD}Actions{END}: {', '.join(results.get('ResourceExposure'))}\n"
+                f"{RED}Potential Issue found: Policy is capable of Resource Exposure{END}"
             )
+            print(f"{BOLD}Actions{END}: {', '.join(results.get('ResourceExposure'))}\n")
 
         # Service Wildcard
         if results.get("ServiceWildcard"):
             results_exist += 1
-            print(f"{RED}Potential Issue found: Policy allows ALL Actions from a service (like service:*){END}")
             print(
-                f"{BOLD}Actions{END}: {', '.join(results.get('ServiceWildcard'))}\n"
+                f"{RED}Potential Issue found: Policy allows ALL Actions from a service (like service:*){END}"
             )
+            print(f"{BOLD}Actions{END}: {', '.join(results.get('ServiceWildcard'))}\n")
 
         # Credentials Exposure
         if results.get("CredentialsExposure"):
             results_exist += 1
-            print(f"{RED}Potential Issue found: Policy allows actions that return credentials{END}")
+            print(
+                f"{RED}Potential Issue found: Policy allows actions that return credentials{END}"
+            )
             print(
                 f"{BOLD}Actions{END}: {', '.join(results.get('CredentialsExposure'))}\n"
             )
@@ -127,8 +137,12 @@ def scan_policy_file(input_file, exclusions_file, high_priority_only, verbose):
 
             # Infrastructure Modification
             results_exist += 1
-            print(f"{RED}Potential Issue found: Policy is capable of Unrestricted Infrastructure Modification{END}")
-            print(f"{BOLD}Actions{END}: {', '.join(results.get('InfrastructureModification'))}")
+            print(
+                f"{RED}Potential Issue found: Policy is capable of Unrestricted Infrastructure Modification{END}"
+            )
+            print(
+                f"{BOLD}Actions{END}: {', '.join(results.get('InfrastructureModification'))}"
+            )
 
         if results_exist == 0:
             print("There were no results found.")

cloudsplaining/output/policy_finding.py

@@ -7,12 +7,12 @@
 import logging
 from cloudsplaining.shared.constants import (
     READ_ONLY_DATA_EXFILTRATION_ACTIONS,
-    ACTIONS_THAT_RETURN_CREDENTIALS
+    ACTIONS_THAT_RETURN_CREDENTIALS,
 )
 from cloudsplaining.shared.exclusions import (
     Exclusions,
     DEFAULT_EXCLUSIONS,
-    is_name_excluded
+    is_name_excluded,
 )
 
 logger = logging.getLogger(__name__)
@@ -31,7 +31,9 @@ def __init__(self, policy_document, exclusions=DEFAULT_EXCLUSIONS):
         self.exclusions = exclusions
         self.always_exclude_actions = exclusions.exclude_actions
 
-        self.missing_resource_constraints_for_modify_actions = self._missing_resource_constraints_for_modify_actions()
+        self.missing_resource_constraints_for_modify_actions = (
+            self._missing_resource_constraints_for_modify_actions()
+        )
 
     def _missing_resource_constraints_for_modify_actions(self):
         """Find modify actions that lack resource ARN constraints"""
@@ -40,7 +42,9 @@ def _missing_resource_constraints_for_modify_actions(self):
             logger.debug("Evaluating statement: %s", statement.json)
             if statement.effect == "Allow":
                 actions_missing_resource_constraints.extend(
-                    statement.missing_resource_constraints_for_modify_actions(self.exclusions)
+                    statement.missing_resource_constraints_for_modify_actions(
+                        self.exclusions
+                    )
                 )
         if actions_missing_resource_constraints:
             these_results = list(
@@ -101,7 +105,9 @@ def privilege_escalation(self):
     def data_exfiltration(self):
         """Returns data exfiltration actions in the policy, if present"""
         result = []
-        for action in self.policy_document.allows_specific_actions_without_constraints(READ_ONLY_DATA_EXFILTRATION_ACTIONS):
+        for action in self.policy_document.allows_specific_actions_without_constraints(
+            READ_ONLY_DATA_EXFILTRATION_ACTIONS
+        ):
             if action.lower() not in self.exclusions.exclude_actions:
                 result.append(action)
         return result
@@ -116,7 +122,9 @@ def credentials_exposure(self):
         """Determine if the action returns credentials"""
         # https://gist.github.com/kmcquade/33860a617e651104d243c324ddf7992a
         results = []
-        for action in self.policy_document.allows_specific_actions_without_constraints(ACTIONS_THAT_RETURN_CREDENTIALS):
+        for action in self.policy_document.allows_specific_actions_without_constraints(
+            ACTIONS_THAT_RETURN_CREDENTIALS
+        ):
             if action.lower() not in self.exclusions.exclude_actions:
                 results.append(action)
         return results
@@ -131,6 +139,6 @@ def results(self):
             ResourceExposure=self.resource_exposure,
             DataExfiltration=self.data_exfiltration,
             CredentialsExposure=self.credentials_exposure,
-            InfrastructureModification=self.missing_resource_constraints_for_modify_actions
+            InfrastructureModification=self.missing_resource_constraints_for_modify_actions,
         )
         return findings