ci/images/authentik-gate-sync/controller.py

@@ -566,7 +566,8 @@ def ensure_oidc(ak, spec, flows, refs):
         raise GateError(f"application {slug} exists and is not managed by {MANAGER}")
 
     want = oauth2_body(spec, flows, refs)
-    provider = next((p for p in ak.list("providers/oauth2/") if p.get("name", "").lower() == name), None)
+    provider = next((p for p in ak.list("providers/oauth2/")
+                     if is_plain_oauth2(p) and p.get("name", "").lower() == name), None)
     if app is not None and (provider is None or app.get("provider") != provider["pk"]):
         # E.g. a gate's application: never re-point it at an OAuth2 provider.
         raise GateError(f"application {slug} exists without OAuth2 provider {name}")
@@ -597,6 +598,12 @@ def ensure_oidc(ak, spec, flows, refs):
 
 
 PROVIDER_PATHS = ("providers/proxy/", "providers/oauth2/")
+OAUTH2_MODEL = "authentik_providers_oauth2.oauth2provider"
+
+
+def is_plain_oauth2(p):
+    # ProxyProvider subclasses OAuth2Provider, so providers/oauth2/ lists proxy providers too.
+    return p.get("meta_model_name") == OAUTH2_MODEL
 
 
 def policy_name(app):
@@ -628,6 +635,8 @@ def owned_applications(ak):
     and OAuth2 provider lists and confirmed one by one."""
     for path in PROVIDER_PATHS:
         for p in ak.list(path):
+            if path == "providers/oauth2/" and not is_plain_oauth2(p):
+                continue
             slug = p.get("assigned_application_slug")
             if not slug:
                 continue

ci/images/authentik-gate-sync/test_controller.py

@@ -206,7 +206,13 @@ class FakeAuthentik:
 
     def list(self, path, **q):
         out = []
-        for o in self.db[path].values():
+        objs = list(self.db[path].values())
+        if path == "providers/oauth2/":
+            # ProxyProvider subclasses OAuth2Provider: the real list endpoint returns both.
+            objs = [dict(o, meta_model_name=c.OAUTH2_MODEL) for o in objs] + \
+                [dict(o, meta_model_name="authentik_providers_proxy.proxyprovider")
+                 for o in self.db["providers/proxy/"].values()]
+        for o in objs:
             ok = True
             for k, v in q.items():
                 field = k.replace("__iexact", "")
@@ -482,7 +488,7 @@ class ReconcileOidc(unittest.TestCase):
         self.ak.db["core/groups/"].clear()
         with self.assertRaises(c.GateError):
             self.run_once()
-        self.assertEqual(self.ak.list("providers/oauth2/"), [])
+        self.assertEqual([p for p in self.ak.list("providers/oauth2/") if c.is_plain_oauth2(p)], [])
 
     def test_refuses_unowned_application(self):
         self.ak.add("core/applications/", {"slug": "youtrack", "meta_publisher": "", "provider": None})
@@ -520,7 +526,7 @@ class ReconcileOidc(unittest.TestCase):
                                            "provider": gate_provider["pk"], "pbm_uuid": "pbm-x"})
         with self.assertRaises(c.GateError):
             self.run_once()
-        self.assertEqual(self.ak.list("providers/oauth2/"), [])
+        self.assertEqual([p for p in self.ak.list("providers/oauth2/") if c.is_plain_oauth2(p)], [])
         self.assertEqual(self.ak.get("core/applications/youtrack/")["provider"], gate_provider["pk"])
 
 
@@ -602,7 +608,26 @@ class ReconcileLoop(unittest.TestCase):
         k8s.clients = []
         c.reconcile(k8s, self.ak)
         self.assertIsNone(self.ak.get("core/applications/youtrack/"))
-        self.assertEqual(self.ak.list("providers/oauth2/"), [])
+        self.assertEqual([p for p in self.ak.list("providers/oauth2/") if c.is_plain_oauth2(p)], [])
+
+    def test_gates_survive_alongside_oidc(self):
+        k8s = FakeK8s(ingresses=[LLAMA], services={"llama-server": SERVICE}, clients=[self.client()])
+        c.reconcile(k8s, self.ak)
+        gate = c.gate_name(LLAMA)
+        app = self.ak.get(f"core/applications/{gate}/")
+        before = len(self.ak.calls)
+        c.reconcile(k8s, self.ak)
+        self.assertEqual(self.ak.calls[before:], [])
+        self.assertEqual(self.ak.get(f"core/applications/{gate}/")["provider"], app["provider"])
+        self.assertIsNotNone(self.ak.get("core/applications/youtrack/"))
+
+    def test_gate_only_is_stable(self):
+        k8s = FakeK8s(ingresses=[LLAMA], services={"llama-server": SERVICE})
+        c.reconcile(k8s, self.ak)
+        before = len(self.ak.calls)
+        c.reconcile(k8s, self.ak)
+        self.assertEqual([m for m, _ in self.ak.calls[before:] if m != "GET"], [])
+        self.assertIsNotNone(self.ak.get(f"core/applications/{c.gate_name(LLAMA)}/"))
 
     def test_slug_equal_to_gate_name_leaves_gate_alone(self):
         k8s = FakeK8s(ingresses=[LLAMA], services={"llama-server": SERVICE})
@@ -612,7 +637,7 @@ class ReconcileLoop(unittest.TestCase):
         k8s.clients = [self.client("evil", slug=gate)]
         c.reconcile(k8s, self.ak)
         self.assertEqual(self.ak.get(f"core/applications/{gate}/")["provider"], app["provider"])
-        self.assertEqual(self.ak.list("providers/oauth2/"), [])
+        self.assertEqual([p for p in self.ak.list("providers/oauth2/") if c.is_plain_oauth2(p)], [])
         self.assertEqual(k8s.status["evil"]["conditions"][0]["status"], "False")