drop Python 3.9 support

gruebel · gruebel · commit 680155d5c2ec · 2025-10-11T00:24:13.000+02:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -13,7 +13,7 @@ permissions:
   contents: read
 
 env:
-  MIN_PYTHON_VERSION: "3.9"
+  MIN_PYTHON_VERSION: "3.10"
   TERRAFORM_VERSION: "1.10"
 
 jobs:
@@ -80,7 +80,7 @@ jobs:
     strategy:
       fail-fast: true
       matrix:
-        python: ['3.10', '3.11', '3.12', '3.13', '3.14']
+        python: ['3.11', '3.12', '3.13', '3.14']
     steps:
       - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0
       - uses: actions/setup-python@e797f83bcb11b83ae66e0230d6156d7c80228e7c # v6.0.0
diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml
@@ -11,7 +11,7 @@ permissions:
   contents: read
 
 env:
-  MIN_PYTHON_VERSION: "3.9"
+  MIN_PYTHON_VERSION: "3.10"
 
 jobs:
   test:
diff --git a/.github/workflows/python-dependency-updater.yml b/.github/workflows/python-dependency-updater.yml
@@ -10,7 +10,7 @@ permissions:
   contents: read
 
 env:
-  MIN_PYTHON_VERSION: "3.9"
+  MIN_PYTHON_VERSION: "3.10"
 
 jobs:
   python-dependency-updater:
diff --git a/.github/workflows/update.yml b/.github/workflows/update.yml
@@ -10,7 +10,7 @@ permissions:
   contents: read
 
 env:
-  MIN_PYTHON_VERSION: "3.9"
+  MIN_PYTHON_VERSION: "3.10"
 
 jobs:
   update-actions:
diff --git a/.readthedocs.yml b/.readthedocs.yml
@@ -7,7 +7,7 @@ version: 2
 build:
   os: ubuntu-24.04
   tools:
-    python: 3.9
+    python: 3.10
 
 mkdocs:
   configuration: mkdocs.yml
diff --git a/policy_sentry/analysis/analyze.py b/policy_sentry/analysis/analyze.py
@@ -34,8 +34,7 @@ def analyze_by_access_level(policy_json: dict[str, Any], access_level: str) -> l
     expanded_policy = get_expanded_policy(policy_json)
     requested_actions = get_actions_from_policy(expanded_policy)
     # expanded_actions = determine_actions_to_expand(requested_actions)
-    actions_by_level = remove_actions_not_matching_access_level(requested_actions, access_level)
-    return actions_by_level
+    return remove_actions_not_matching_access_level(requested_actions, access_level)
 
 
 def analyze_statement_by_access_level(statement_json: dict[str, Any], access_level: str) -> list[str]:
@@ -50,5 +49,4 @@ def analyze_statement_by_access_level(statement_json: dict[str, Any], access_lev
     """
     requested_actions = get_actions_from_statement(statement_json)
     expanded_actions = determine_actions_to_expand(requested_actions)
-    actions_by_level = remove_actions_not_matching_access_level(expanded_actions, access_level)
-    return actions_by_level
+    return remove_actions_not_matching_access_level(expanded_actions, access_level)
diff --git a/policy_sentry/analysis/expand.py b/policy_sentry/analysis/expand.py
@@ -32,7 +32,7 @@ def expand(action: str | list[str]) -> list[str]:
 
     if action == "*":
         return list(get_all_actions())
-    elif "*" in action:
+    if "*" in action:
         service_prefix = action.split(":", maxsplit=1)[0]
         service_actions = get_actions_for_service(service_prefix=service_prefix)
         expanded = [
diff --git a/policy_sentry/command/create_template.py b/policy_sentry/command/create_template.py
@@ -59,7 +59,7 @@ def create_template(output_file: str | Path, template_type: str, verbose: str) -
         sys.exit()
 
     filename = Path(output_file).resolve()
-    with open(filename, "a", encoding="utf-8") as file_obj:
+    with filename.open("a", encoding="utf-8") as file_obj:
         file_obj.write(template)
 
     print(f"write-policy template file written to: {filename}")
diff --git a/policy_sentry/command/query.py b/policy_sentry/command/query.py
@@ -54,7 +54,7 @@ def print_dict(output: list[Any] | dict[Any, Any], fmt: str = "json") -> None:
     """Common method on how to print a dict, depending on whether the user requests JSON, YAML or CSV output"""
     if fmt == "csv":
         if not output:
-            return None
+            return
         print(",".join(output[0].keys()))
         for entry in output:
             print(",".join(entry.values()))
diff --git a/policy_sentry/command/write_policy.py b/policy_sentry/command/write_policy.py
@@ -154,5 +154,4 @@ def write_policy_with_template(cfg: dict[str, Any], minimize: int | None = None)
         Dictionary: The JSON policy
     """
     sid_group = SidGroup()
-    policy = sid_group.process_template(cfg, minimize)
-    return policy
+    return sid_group.process_template(cfg, minimize)
diff --git a/policy_sentry/querying/actions.py b/policy_sentry/querying/actions.py
@@ -77,17 +77,17 @@ def get_action_data(service: str, action_name: str) -> dict[str, list[dict[str,
                     results.extend(entries)
             action_data_results[service] = results
             return action_data_results
-        else:
-            this_action_name = service_prefix_data["privileges_lower_name"].get(action_name.lower())
-            if this_action_name:
-                this_action_data = service_prefix_data["privileges"][this_action_name]
-                entries = create_action_data_entries(
-                    service_prefix_data=service_prefix_data,
-                    action_name=this_action_name,
-                    action_data=this_action_data,
-                )
-                action_data_results[service] = entries
-                return action_data_results
+
+        this_action_name = service_prefix_data["privileges_lower_name"].get(action_name.lower())
+        if this_action_name:
+            this_action_data = service_prefix_data["privileges"][this_action_name]
+            entries = create_action_data_entries(
+                service_prefix_data=service_prefix_data,
+                action_name=this_action_name,
+                action_data=this_action_data,
+            )
+            action_data_results[service] = entries
+            return action_data_results
     except TypeError as t_e:
         logger.debug(t_e)
 
diff --git a/policy_sentry/querying/conditions.py b/policy_sentry/querying/conditions.py
@@ -27,8 +27,7 @@ def get_condition_keys_for_service(service_prefix: str) -> list[str]:
         List: A list of condition keys
     """
     service_prefix_data = get_service_prefix_data(service_prefix)
-    results = list(service_prefix_data["conditions"])
-    return results
+    return list(service_prefix_data["conditions"])
 
 
 # Per condition key name
@@ -46,12 +45,11 @@ def get_condition_key_details(service_prefix: str, condition_key_name: str) -> d
     service_prefix_data = get_service_prefix_data(service_prefix)
     for condition_name, condition_data in service_prefix_data["conditions"].items():
         if is_condition_key_match(condition_name, condition_key_name):
-            output = {
+            return {
                 "name": condition_name,
                 "description": condition_data["description"],
                 "condition_value_type": condition_data["type"].lower(),
             }
-            return output
 
     return {}
 
diff --git a/policy_sentry/shared/awsdocs.py b/policy_sentry/shared/awsdocs.py
@@ -119,7 +119,7 @@ def update_html_docs_directory(html_docs_destination: Path) -> None:
                 logger.warning(a_e)
                 logger.warning(script)
 
-        with open(html_docs_destination / page, "w", encoding="utf-8") as file:
+        with (html_docs_destination / page).open("w", encoding="utf-8") as file:
             # file.write(str(soup.html))
             file.write(str(soup.prettify()))
             file.close()
@@ -153,23 +153,24 @@ def create_database(destination_directory: str | Path, access_level_overrides_fi
     """
 
     # Create the docs directory if it doesn't exist
-    Path(os.path.join(destination_directory, "docs")).mkdir(parents=True, exist_ok=True)
+    destination_directory = Path(destination_directory)
+    (destination_directory / "docs").mkdir(parents=True, exist_ok=True)
 
     # This holds the entire IAM definition
     schema: dict[str, Any] = {POLICY_SENTRY_SCHEMA_VERSION_NAME: POLICY_SENTRY_SCHEMA_VERSION_LATEST}
 
     # for filename in ['list_amazonathena.partial.html']:
     file_list = []
-    for filename in os.listdir(LOCAL_HTML_DIRECTORY_PATH):
-        if os.path.isfile(os.path.join(LOCAL_HTML_DIRECTORY_PATH, filename)) and filename not in file_list:
+    for filename in os.listdir(LOCAL_HTML_DIRECTORY_PATH):  # noqa: PTH208
+        if (LOCAL_HTML_DIRECTORY_PATH / filename).is_file() and filename not in file_list:
             file_list.append(filename)
 
     file_list.sort()
     for filename in file_list:
         if not filename.startswith("list_"):
             continue
 
-        content = (Path(LOCAL_HTML_DIRECTORY_PATH) / filename).read_text()
+        content = (LOCAL_HTML_DIRECTORY_PATH / filename).read_text()
         soup = BeautifulSoup(content, "html.parser")
         main_content = soup.find(id="main-content")
         if not isinstance(main_content, Tag):
@@ -256,9 +257,9 @@ def create_database(destination_directory: str | Path, access_level_overrides_fi
                         # Skip the <a id='...'> tags
                         api_documentation_link = None
                         continue
-                    else:
-                        api_documentation_link = link.attrs.get("href")
-                        logger.debug(api_documentation_link)
+
+                    api_documentation_link = link.attrs.get("href")
+                    logger.debug(api_documentation_link)
                     priv = chomp(link.text)
                 if priv == "":
                     priv = chomp(cells[0].text)
@@ -408,7 +409,7 @@ def create_database(destination_directory: str | Path, access_level_overrides_fi
         # }
         # schema.update(this_service_schema)
 
-    iam_definition_file = os.path.join(destination_directory, "iam-definition.json")
-    with open(iam_definition_file, "w", encoding="utf-8") as file:
+    iam_definition_file = destination_directory / "iam-definition.json"
+    with iam_definition_file.open("w", encoding="utf-8") as file:
         json.dump(schema, file, indent=2)
     logger.info(f"Wrote IAM definition file to path: {iam_definition_file}")
diff --git a/policy_sentry/util/arns.py b/policy_sentry/util/arns.py
@@ -70,8 +70,7 @@ def _resource_string(self) -> str:
         #     Case 5: resourcetype:resource
         #     Case 6: resourcetype:resource:qualifier
         split_arn = self.arn.split(":")
-        resource_string = ":".join(split_arn[5:])
-        return resource_string
+        return ":".join(split_arn[5:])
 
     def same_resource_type(self, arn_in_database: str) -> bool:
         """Given an arn, see if it has the same resource type"""
@@ -146,14 +145,13 @@ def same_resource_type(self, arn_in_database: str) -> bool:
             # If we've made it this far, then it is a special type
             # return True
             # Presence of / would mean it's an object in both so it matches
-            elif "/" in self.resource_string and "/" in elements[5]:  # noqa: SIM114
+            if "/" in self.resource_string and "/" in elements[5]:
                 return True
             # / not being present in either means it's a bucket in both so it matches
-            elif "/" not in self.resource_string and "/" not in elements[5]:  # noqa: SIM103
+            if "/" not in self.resource_string and "/" not in elements[5]:  # noqa: SIM103
                 return True
             # If there is a / in one but not in the other, it does not match
-            else:
-                return False
+            return False
 
         # 5. If we've made it this far, then it should pass
         return True
@@ -225,8 +223,7 @@ def get_resource_string(arn: str) -> str:
         String: The resource string, like `resourcetype/resource`
     """
     split_arn = arn.split(":")
-    resource_string = ":".join(split_arn[5:])
-    return resource_string
+    return ":".join(split_arn[5:])
 
 
 # In the meantime, we have to skip this pylint check (consider this as tech debt)
diff --git a/policy_sentry/util/conditions.py b/policy_sentry/util/conditions.py
@@ -14,18 +14,18 @@ def translate_condition_key_data_types(condition_str: str) -> str:
     condition_lowercase = condition_str.lower()
     if condition_lowercase in ("arn", "arn"):
         return "Arn"
-    elif condition_lowercase in ("bool", "boolean"):
+    if condition_lowercase in ("bool", "boolean"):
         return "Bool"
-    elif condition_lowercase == "date":
+    if condition_lowercase == "date":
         return "Date"
-    elif condition_lowercase in ("long", "numeric"):
+    if condition_lowercase in ("long", "numeric"):
         return "Number"
-    elif condition_lowercase in ("string", "string", "arrayofstring"):
+    if condition_lowercase in ("string", "string", "arrayofstring"):
         return "String"
-    elif condition_lowercase == "ip":
+    if condition_lowercase == "ip":
         return "Ip"
-    else:
-        raise Exception(f"Unknown data format: {condition_lowercase}")
+
+    raise Exception(f"Unknown data format: {condition_lowercase}")
 
 
 def get_service_from_condition_key(condition_key: str) -> str:
diff --git a/policy_sentry/util/file.py b/policy_sentry/util/file.py
@@ -5,14 +5,11 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Any, cast
+from pathlib import Path
+from typing import Any, cast
 
 import yaml
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
-
 logger = logging.getLogger(__name__)
 
 
@@ -23,7 +20,7 @@ def read_yaml_file(filename: str | Path) -> dict[str, Any]:
     :param filename: name of the yaml file
     :return: dictionary of YAML file contents
     """
-    with open(filename, encoding="utf-8") as yaml_file:
+    with Path(filename).open(encoding="utf-8") as yaml_file:
         try:
             cfg = cast("dict[str, Any]", yaml.safe_load(yaml_file))
         except yaml.YAMLError as exc:
diff --git a/policy_sentry/util/policy_files.py b/policy_sentry/util/policy_files.py
@@ -19,19 +19,20 @@ def get_actions_from_statement(statement: dict[str, Any]) -> list[str]:
     # We only want to evaluate policies that have Effect = "Allow"
     if statement.get("Effect") == "Deny":
         return actions_list
+
+    action_clause = statement.get("Action")
+    if not action_clause:
+        logger.debug("No actions contained in statement")
+        return actions_list
+    # Action = "s3:GetObject"
+    if isinstance(action_clause, str):
+        actions_list.append(action_clause)
+    # Action = ["s3:GetObject", "s3:ListBuckets"]
+    elif isinstance(action_clause, list):
+        actions_list.extend(action_clause)
     else:
-        action_clause = statement.get("Action")
-        if not action_clause:
-            logger.debug("No actions contained in statement")
-            return actions_list
-        # Action = "s3:GetObject"
-        if isinstance(action_clause, str):
-            actions_list.append(action_clause)
-        # Action = ["s3:GetObject", "s3:ListBuckets"]
-        elif isinstance(action_clause, list):
-            actions_list.extend(action_clause)
-        else:
-            logger.debug("Unknown error: The 'Action' is neither a list nor a string")
+        logger.debug("Unknown error: The 'Action' is neither a list nor a string")
+
     return actions_list
 
 
@@ -84,13 +85,11 @@ def get_sid_names_from_policy(policy_json: dict[str, Any]) -> list[str]:
     """
     Given a Policy JSON, get a list of the Statement IDs. This is helpful in unit tests.
     """
-    sid_names = [statement["Sid"] for statement in policy_json.get("Statement", []) if "Sid" in statement]
-    return sid_names
+    return [statement["Sid"] for statement in policy_json.get("Statement", []) if "Sid" in statement]
 
 
 def get_statement_from_policy_using_sid(policy_json: dict[str, Any], sid: str) -> dict[str, Any] | None:
     """
     Helper function to get a statement just by providing the policy JSON and the Statement ID
     """
-    res = next((sub for sub in policy_json["Statement"] if sub.get("Sid") == sid), None)
-    return res
+    return next((sub for sub in policy_json["Statement"] if sub.get("Sid") == sid), None)
diff --git a/policy_sentry/writing/minimize.py b/policy_sentry/writing/minimize.py
@@ -39,9 +39,7 @@ def _get_prefixes_for_action(action: str) -> list[str]:
     :return: [ "iam:", "iam:c", "iam:ca", "iam:cat" ]
     """
     technology, permission = action.split(":")
-    retval = [f"{technology}:{permission[:i]}" for i in range(len(permission) + 1)]
-
-    return retval
+    return [f"{technology}:{permission[:i]}" for i in range(len(permission) + 1)]
 
 
 # Adapted version of policyuniverse's _get_denied_prefixes_from_desired, here:
@@ -51,12 +49,10 @@ def get_denied_prefixes_from_desired(desired_actions: list[str], all_actions: se
     Adapted version of policyuniverse's _get_denied_prefixes_from_desired, here: https://github.com/Netflix-Skunkworks/policyuniverse/blob/master/policyuniverse/expander_minimizer.py#L101
     """
     denied_actions = all_actions.difference(desired_actions)
-    denied_prefixes = {
+    return {
         denied_prefix for denied_action in denied_actions for denied_prefix in _get_prefixes_for_action(denied_action)
     }
 
-    return denied_prefixes
-
 
 # Adapted version of policyuniverse's _check_permission_length. We are commenting out the skipping prefix message
 # https://github.com/Netflix-Skunkworks/policyuniverse/blob/master/policyuniverse/expander_minimizer.py#L111
@@ -108,6 +104,4 @@ def minimize_statement_actions(
             logger.debug(f"Could not find suitable prefix. Defaulting to {prefixes[-1]}")
             minimized_actions.add(prefixes[-1])
     # sort the actions
-    minimized_actions_list = sorted(minimized_actions)
-
-    return minimized_actions_list
+    return sorted(minimized_actions)
diff --git a/policy_sentry/writing/sid_group.py b/policy_sentry/writing/sid_group.py
diff --git a/policy_sentry/writing/validate.py b/policy_sentry/writing/validate.py
diff --git a/pyproject.toml b/pyproject.toml
diff --git a/setup.py b/setup.py
diff --git a/utils/download_docs.py b/utils/download_docs.py
