Round 6 revoke cleanups: constants, atomic-path helper, deterministic output

- RESOURCE_NAME_SEP / NON_RESOURCE_URL_PREFIX module-level constants —
  six callsites across provider-kubeconfig.py used bare string literals;
  a typo in either file silently broke revoke matching.
- _rule_matches_revoke: check nonResourceURLs before resourceNames so a
  malformed revoke rule with both fields isn't rejected on a resourceNames
  gate that doesn't apply to URL rules.
- _atomic_rule_after_revoke helper unifies the wildcard-scope and
  no-resources branches; both paths strip the union of matched revoke
  verbs — RBAC can't express 'everything except X'.
- _bucket_cells_by_verbs helper shared by URL and resource paths.
- Sort bucket iteration by verbs_key so output rule ordering is stable
  across permutations of semantically equivalent input.
- Hoist set(mr['verbs']) once per matched-revoke in both split paths.
- Derive has_names from existing_resource_names directly (was cells[0][2]).
- 3 new unit tests: resource key containing the substring 'resourceName'
  but no '/resourceName::' separator; _dim_matches with revoke-side '*';
  multi-revoke composition against a wildcard-scope rule.

Author: anniegracehu

.gitignore

@@ -8,3 +8,4 @@ operator-manager/artifacts/deployment/operator-manager
 operator-manager/operator-manager
 platform-operator/platform-operator
 platform-operator/artifacts/deployment/platform-operator
+__pycache__/

provider-kubeconfig.py

@@ -8,6 +8,9 @@
 
 from logging.config import dictConfig
 
+RESOURCE_NAME_SEP = "/resourceName::"
+NON_RESOURCE_URL_PREFIX = "nonResourceURL::"
+
 dictConfig({
     "version": 1,
     "formatters": {"default": {"format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s"}},
@@ -70,16 +73,16 @@ def _parse_permission_rules(self, perms):
                                 resources.append(resource.strip())
                             rule_group = {}
                             if api_group == "non-apigroup":
-                                if "nonResourceURL" in resource:
-                                    parts = resource.split("nonResourceURL::")
+                                if NON_RESOURCE_URL_PREFIX in resource:
+                                    parts = resource.split(NON_RESOURCE_URL_PREFIX)
                                     non_res = parts[1].strip() if len(parts) > 1 else parts[0].strip()
                                     rule_group["nonResourceURLs"] = [non_res]
                                     rule_group["verbs"] = verbs
                             else:
                                 rule_group["apiGroups"] = [api_group]
                                 rule_group["verbs"] = verbs
-                                if "/resourceName::" in resource:
-                                    parts = resource.split("/resourceName::")
+                                if RESOURCE_NAME_SEP in resource:
+                                    parts = resource.split(RESOURCE_NAME_SEP)
                                     rule_group["resources"] = [parts[0].strip()]
                                     rule_group["resourceNames"] = [parts[1].strip()]
                                 else:
@@ -208,14 +211,14 @@ def _strip_verbs(self, existing_verbs, revoke_verbs):
         def _permission_keys(self, resource, resource_names):
                 """Permission-file key form used by _parse_permission_rules / the perms configmap."""
                 if resource_names:
-                        return [resource + "/resourceName::" + rn for rn in resource_names]
+                        return [resource + RESOURCE_NAME_SEP + rn for rn in resource_names]
                 return [resource]
 
         def _retained_keys_for_rule(self, rule):
                 """All permission-file keys covered by an (unmodified) k8s rule."""
                 keys = []
                 for url in rule.get("nonResourceURLs") or []:
-                        keys.append("nonResourceURL::" + url)
+                        keys.append(NON_RESOURCE_URL_PREFIX + url)
                 resource_names = rule.get("resourceNames") or []
                 for res in rule.get("resources") or []:
                         keys.extend(self._permission_keys(res, resource_names))
@@ -236,13 +239,13 @@ def _rule_matches_revoke(self, existing_norm, revoke_norm_rule):
                 RBAC has no 'deny', so we cannot trim a named cell out of a rule that
                 grants all names.
                 """
+                if revoke_norm_rule["nonResourceURLs"]:
+                        return bool(set(revoke_norm_rule["nonResourceURLs"]).intersection(existing_norm["nonResourceURLs"]))
                 if revoke_norm_rule["resourceNames"]:
                         if not existing_norm["resourceNames"]:
                                 return False
                         if not set(revoke_norm_rule["resourceNames"]).intersection(existing_norm["resourceNames"]):
                                 return False
-                if revoke_norm_rule["nonResourceURLs"]:
-                        return bool(set(revoke_norm_rule["nonResourceURLs"]).intersection(existing_norm["nonResourceURLs"]))
                 if not self._dim_matches(revoke_norm_rule["apiGroups"], existing_norm["apiGroups"]):
                         return False
                 if not self._dim_matches(revoke_norm_rule["resources"], existing_norm["resources"]):
@@ -824,51 +827,31 @@ def _split_rule_after_revoke(self, rule, matched_revoke_rules, retained_keys):
                         cell_verbs = {url: set(existing_verbs) for url in existing_urls}
                         for mr in matched_revoke_rules:
                                 mr_urls = set(mr["nonResourceURLs"])
+                                mr_verbs = mr["verbs"]
                                 for url in cell_verbs:
                                         if url not in mr_urls:
                                                 continue
-                                        cell_verbs[url] = self._strip_to_set(cell_verbs[url], mr["verbs"])
-                        by_verbs = {}
-                        for url, cv in cell_verbs.items():
-                                if not cv:
-                                        continue
-                                by_verbs.setdefault(tuple(sorted(cv)), []).append(url)
+                                        cell_verbs[url] = self._strip_to_set(cell_verbs[url], mr_verbs)
                         out_rules = []
-                        for verbs_key, urls in by_verbs.items():
+                        for verbs_key, urls in sorted(self._bucket_cells_by_verbs(cell_verbs).items()):
                                 updated = dict(rule)
                                 updated["nonResourceURLs"] = urls
                                 updated["verbs"] = list(verbs_key)
                                 out_rules.append(updated)
                                 for url in urls:
-                                        retained_keys.add("nonResourceURL::" + url)
+                                        retained_keys.add(NON_RESOURCE_URL_PREFIX + url)
                         return out_rules
 
                 existing_api_groups = rule.get("apiGroups") or []
                 existing_resources = rule.get("resources") or []
                 existing_resource_names = rule.get("resourceNames") or []
 
-                # Wildcard scope: we cannot enumerate "everything except X". Treat atomic.
-                if "*" in existing_api_groups or "*" in existing_resources:
-                        combined_verbs = [v for mr in matched_revoke_rules for v in mr["verbs"]]
-                        stripped = self._strip_verbs(existing_verbs, combined_verbs)
-                        if not stripped:
-                                return []
-                        updated = dict(rule)
-                        updated["verbs"] = stripped
-                        for key in self._retained_keys_for_rule(rule):
-                                retained_keys.add(key)
-                        return [updated]
-
-                if not existing_resources:
-                        combined_verbs = [v for mr in matched_revoke_rules for v in mr["verbs"]]
-                        stripped = self._strip_verbs(existing_verbs, combined_verbs)
-                        if not stripped:
-                                return []
-                        updated = dict(rule)
-                        updated["verbs"] = stripped
-                        return [updated]
+                # Wildcard scope or no resources: can't enumerate "all except X" — treat atomic.
+                if "*" in existing_api_groups or "*" in existing_resources or not existing_resources:
+                        return self._atomic_rule_after_revoke(rule, matched_revoke_rules, retained_keys)
 
                 names_axis = existing_resource_names or [None]
+                has_names = bool(existing_resource_names)
                 cell_verbs = {}
                 for ag in existing_api_groups:
                         for res in existing_resources:
@@ -878,6 +861,7 @@ def _split_rule_after_revoke(self, rule, matched_revoke_rules, retained_keys):
                         mr_apis = set(mr["apiGroups"])
                         mr_ress = set(mr["resources"])
                         mr_names = set(mr["resourceNames"]) if mr["resourceNames"] else None
+                        mr_verbs = mr["verbs"]
                         if not mr_apis or not mr_ress:
                                 continue
                         for cell in cell_verbs:
@@ -886,19 +870,12 @@ def _split_rule_after_revoke(self, rule, matched_revoke_rules, retained_keys):
                                         continue
                                 if mr_names is not None and rn not in mr_names:
                                         continue
-                                cell_verbs[cell] = self._strip_to_set(cell_verbs[cell], mr["verbs"])
-
-                by_verbs = {}
-                for cell, cv in cell_verbs.items():
-                        if not cv:
-                                continue
-                        by_verbs.setdefault(tuple(sorted(cv)), []).append(cell)
+                                cell_verbs[cell] = self._strip_to_set(cell_verbs[cell], mr_verbs)
 
                 out_rules = []
-                for verbs_key, cells in by_verbs.items():
+                for verbs_key, cells in sorted(self._bucket_cells_by_verbs(cell_verbs).items()):
                         apis = sorted({c[0] for c in cells})
                         ress = sorted({c[1] for c in cells})
-                        has_names = cells[0][2] is not None
                         names = sorted({c[2] for c in cells}) if has_names else []
                         expected = len(apis) * len(ress) * (len(names) if has_names else 1)
                         if len(cells) == expected:
@@ -930,6 +907,30 @@ def _split_rule_after_revoke(self, rule, matched_revoke_rules, retained_keys):
                                         for key in self._permission_keys(res, cell_names):
                                                 retained_keys.add(key)
                 return out_rules
+
+        def _atomic_rule_after_revoke(self, rule, matched_revoke_rules, retained_keys):
+                """Atomic path: strip the union of matched revoke verbs from the whole rule.
+                Used when the existing rule has '*' scope or no resource dimension —
+                RBAC cannot express 'everything except X', so we trim globally or drop.
+                """
+                combined_verbs = [v for mr in matched_revoke_rules for v in mr["verbs"]]
+                stripped = self._strip_verbs(rule.get("verbs", []), combined_verbs)
+                if not stripped:
+                        return []
+                updated = dict(rule)
+                updated["verbs"] = stripped
+                for key in self._retained_keys_for_rule(rule):
+                        retained_keys.add(key)
+                return [updated]
+
+        def _bucket_cells_by_verbs(self, cell_verbs):
+                """Group non-empty cells by their sorted-verb tuple for regrouping."""
+                buckets = {}
+                for cell, cv in cell_verbs.items():
+                        if not cv:
+                                continue
+                        buckets.setdefault(tuple(sorted(cv)), []).append(cell)
+                return buckets
 
 
         def _apply_rbac(self, sa, namespace, entity='', kubeconfig=''):

tests/test_provider_kubeconfig.py

@@ -342,6 +342,40 @@ def test_split_rule_non_rectangular_emits_per_cell(self):
             covered, {("", "services"), ("apps", "pods"), ("apps", "services")}
         )
 
+    def test_parse_permission_rules_resource_name_substring_without_separator(self):
+        """A resource key containing 'resourceName' but no '/resourceName::' separator
+        must not be split — it's a literal resource name."""
+        perms = {"": [{"myresourceNames": ["get"]}]}
+        rule_list, resources = self.generator._parse_permission_rules(perms)
+        self.assertEqual(len(rule_list), 1)
+        self.assertEqual(rule_list[0]["resources"], ["myresourceNames"])
+        self.assertNotIn("resourceNames", rule_list[0])
+        self.assertEqual(resources, ["myresourceNames"])
+
+    def test_dim_matches_wildcard_on_revoke_side(self):
+        """A revoke-side '*' matches any specific existing apiGroup/resource."""
+        self.assertTrue(self.generator._dim_matches(("*",), ("",)))
+        self.assertTrue(self.generator._dim_matches(("*",), ("apps",)))
+        self.assertTrue(self.generator._dim_matches(("",), ("*",)))
+        self.assertFalse(self.generator._dim_matches(("",), ("apps",)))
+
+    def test_split_rule_wildcard_scope_multi_revoke_composition(self):
+        """When multiple revokes target distinct specific cells of a wildcard rule,
+        verbs accumulate — we can't express exclusion, so the combined set is stripped."""
+        rule = {"apiGroups": ["*"], "resources": ["*"], "verbs": ["get", "delete", "list"]}
+        matched = [
+                self.generator._normalize_rule(
+                        {"apiGroups": [""], "resources": ["pods"], "verbs": ["delete"]}
+                ),
+                self.generator._normalize_rule(
+                        {"apiGroups": [""], "resources": ["services"], "verbs": ["get"]}
+                ),
+        ]
+        retained = set()
+        result = self.generator._split_rule_after_revoke(rule, matched, retained)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(sorted(result[0]["verbs"]), ["list"])
+
     def test_permission_fixture_files_parse_json_and_yaml(self):
         """Fixture files in tests/permission_files should load and parse via script helpers."""
         fixture_dir = os.path.join(ROOT, "tests", "permission_files")