fix(keycloak): resolve each consumer's redirect scheme, not Keycloak's

KEYCLOAK_REDIRECT_URIS derived the scheme from Keycloak's OWN TLS state and
stamped that single scalar onto every SSO consumer. With svc-net-tor
deployed Keycloak is onion-primary, so resolve_enabled returns False and
lookup('tls', ...) yields "http" -- which then went onto clearnet apps too.
Keycloak registered http://matrix.infinito.example/* while Synapse sends
https://.../_synapse/client/oidc/callback, and its redirect-URI match is
scheme-exact: 90 invalid_redirect_uri events, an error page instead of a
login form, and a 60s selector timeout that looked like a Playwright bug.
jitsi hit the same wall on https://meet.infinito.example/oauth2/callback.

Resolve per consumer instead, via resolve_enabled: an .onion is always
plaintext, a clearnet app follows its own server.tls.enabled, and only
then the global TLS_ENABLED. That override existed and was silently
discarded -- passing a pre-computed scheme string threw it away, which is
the same "what I am" instead of "what the consumer is" mistake one layer
up. resolve_enabled handles the onion case itself, so the filter's own
is_onion_domain branch is gone rather than duplicated.

Move it from a filter to a lookup while at it: the thing needs domains,
applications and TLS_ENABLED anyway, so resolving them inside collapses
the call site to lookup('redirect_uris') and drops
KEYCLOAK_REDIRECT_FEATURES, which listed the same path twice. A lookup
under plugins/lookup/ is also plainly importable, so the test loses the
sys.modules stubbing apparatus a role-local filter needed -- 60 lines net
smaller.

KEYCLOAK_WEB_ORIGINS is regex-derived from these URIs and inherits the
fix. Verified against the real applications view: 69 URIs, all https with
TLS_ENABLED true and all http with it false -- the same 69 the failing run
registered as plaintext.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

plugins/lookup/redirect_uris.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+from utils.get_url import get_url
+from utils.roles.applications.config import (
+    AppConfigKeyError,
+    ConfigEntryNotSetError,
+    get,
+)
+from utils.tls_common import resolve_enabled
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable, Sequence
+
+DEFAULT_FEATURES = ("services.sso.enabled",)
+DEFAULT_WILDCARD = "/*"
+
+
+def _stable_dedup(items: Sequence[str]) -> list[str]:
+    seen = set()
+    out: list[str] = []
+    for x in items:
+        if x not in seen:
+            seen.add(x)
+            out.append(x)
+    return out
+
+
+def _iter_domains(value: Any) -> Iterable[str]:
+    """Yield domains from str | list/tuple[str] | dict[*, str|list|tuple]."""
+    if value is None:
+        return
+    if isinstance(value, str):
+        yield value
+    elif isinstance(value, dict):
+        for val in value.values():
+            yield from _iter_domains(val)
+    elif isinstance(value, (list, tuple)):
+        for val in value:
+            yield from _iter_domains(val)
+    else:
+        raise AnsibleError(
+            "redirect_uris: domain value must be str, list/tuple[str], or dict mapping to those"
+        )
+
+
+def build_redirect_uris(
+    domains: dict,
+    applications: dict,
+    tls_enabled: bool,
+    *,
+    wildcard: str = DEFAULT_WILDCARD,
+    features: Iterable[str] = DEFAULT_FEATURES,
+    dedup: bool = True,
+) -> list[str]:
+    """Registered redirect URIs for every SSO consumer, one per domain.
+
+    Args:
+        domains: app_id -> domain, list of domains, or nested mapping of those.
+        applications: merged applications view, used for feature gating and for
+            each consumer's ``server.tls.enabled`` override.
+        tls_enabled: global TLS default, overridden per consumer by
+            ``resolve_enabled``; an .onion domain always resolves to plaintext.
+        wildcard: suffix appended to every URI.
+        features: config paths ORed together to decide whether an app gets a URI.
+        dedup: drop repeated URIs, preserving first-seen order.
+    """
+    if not isinstance(domains, dict):
+        raise AnsibleError(
+            "redirect_uris: 'domains' must be a dict mapping app_id -> domain or list of domains"
+        )
+
+    uris: list[str] = []
+    for app_id, domain_value in domains.items():
+        try:
+            has_feature = any(
+                bool(get(applications, app_id, f, False)) for f in features
+            )
+        except (AppConfigKeyError, ConfigEntryNotSetError):
+            has_feature = False
+        if not has_feature:
+            continue
+
+        for domain in _iter_domains(domain_value):
+            proto = (
+                "https"
+                if resolve_enabled(
+                    applications.get(app_id) or {}, tls_enabled, primary_domain=domain
+                )
+                else "http"
+            )
+            try:
+                url = get_url({app_id: domain}, app_id, proto)
+            except Exception as exc:
+                raise AnsibleError(
+                    f"redirect_uris: get_url failed for app '{app_id}' "
+                    f"with domain '{domain}': {exc}"
+                ) from exc
+            uris.append(f"{url}{wildcard}")
+
+    return _stable_dedup(uris) if dedup else uris
+
+
+class LookupModule(LookupBase):
+    """
+    Usage:
+        {{ lookup('redirect_uris') }}
+        {{ lookup('redirect_uris', wildcard='/cb') }}
+
+    Returns the redirect URIs Keycloak must whitelist: one per domain of every
+    application whose config enables SSO. Each URI's scheme is resolved for the
+    CONSUMER -- its own ``server.tls.enabled`` over the global ``TLS_ENABLED``,
+    and always plaintext for an .onion -- never from Keycloak's own TLS state.
+
+    - parameters:
+        wildcard: suffix appended to every URI (default '/*')
+        features: config paths ORed to gate an application (default services.sso.enabled)
+        dedup: drop repeated URIs (default true)
+    """
+
+    def run(
+        self,
+        terms: list[Any] | None,
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[list[str]]:
+        if terms:
+            raise AnsibleError("lookup('redirect_uris') expects no terms.")
+
+        variables = variables or getattr(self._templar, "available_variables", {}) or {}
+        if "TLS_ENABLED" not in variables:
+            raise AnsibleError("redirect_uris: TLS_ENABLED is not defined.")
+
+        loader = getattr(self, "_loader", None)
+        templar = getattr(self, "_templar", None)
+        domains = lookup_loader.get("domains", loader=loader, templar=templar).run(
+            [], variables=variables
+        )[0]
+        applications = lookup_loader.get(
+            "applications", loader=loader, templar=templar
+        ).run([], variables=variables)[0]
+
+        features = kwargs.get("features") or DEFAULT_FEATURES
+        return [
+            build_redirect_uris(
+                domains,
+                applications,
+                bool(variables["TLS_ENABLED"]),
+                wildcard=kwargs.get("wildcard", DEFAULT_WILDCARD),
+                features=features,
+                dedup=bool(kwargs.get("dedup", True)),
+            )
+        ]

roles/web-app-keycloak/filter_plugins/redirect_uris.py

@@ -1,127 +0,0 @@
-from __future__ import annotations
-
-import sys
-from pathlib import Path
-from typing import TYPE_CHECKING
-
-from ansible.errors import AnsibleFilterError
-
-from utils.get_url import get_url  # returns "<protocol>://<domain>"
-from utils.roles.applications.config import (
-    AppConfigKeyError,
-    ConfigEntryNotSetError,
-    get,
-)
-from utils.tls_common import is_onion_domain
-
-if TYPE_CHECKING:
-    from collections.abc import Iterable, Sequence
-
-
-# --- Locate project root that contains `utils/` dynamically (up to 5 levels) ---
-def _ensure_utils_on_path():
-    here = Path(__file__).resolve().parent
-    for depth in range(1, 6):
-        candidate = here.parents[depth - 1] if depth - 1 < len(here.parents) else here
-        if (candidate / "utils").is_dir():
-            candidate_str = str(candidate)
-            if candidate_str not in sys.path:
-                sys.path.insert(0, candidate_str)
-            return
-    # If not found, imports below will raise a clear error
-
-
-_ensure_utils_on_path()
-
-
-def _stable_dedup(items: Sequence[str]) -> list[str]:
-    seen = set()
-    out: list[str] = []
-    for x in items:
-        if x not in seen:
-            seen.add(x)
-            out.append(x)
-    return out
-
-
-def _iter_domains(v) -> Iterable[str]:
-    """Yield domains from str | list/tuple[str] | dict[*, str|list|tuple]."""
-    if v is None:
-        return
-    if isinstance(v, str):
-        yield v
-    elif isinstance(v, dict):
-        for val in v.values():
-            yield from _iter_domains(val)
-    elif isinstance(v, (list, tuple)):
-        for val in v:
-            yield from _iter_domains(val)
-    else:
-        raise AnsibleFilterError(
-            "redirect_uris: domain_value must be str, list/tuple[str], or dict mapping to those"
-        )
-
-
-def redirect_uris(
-    domains: dict,
-    applications: dict,
-    web_protocol: str = "https",
-    wildcard: str = "/*",
-    features: Iterable[str] = ("services.sso.enabled",),
-    dedup: bool = True,
-) -> list[str]:
-    """
-    Build redirect URIs using:
-      - get(applications, app_id, dotted_key, default) for feature gating
-      - get_url(domains_subset, app_id, web_protocol) to form "<proto>://<domain>"
-
-    For domain lists, we call get_url() once per domain by passing a minimal
-    per-app subset like {app_id: "example.org"} to preserve your original
-    'one entry per domain' behavior.
-    """
-    if not isinstance(domains, dict):
-        raise AnsibleFilterError(
-            "redirect_uris: 'domains' must be a dict mapping app_id -> domain or list of domains"
-        )
-
-    uris: list[str] = []
-
-    for app_id, domain_value in domains.items():
-        # Feature check via get
-        try:
-            has_feature = any(
-                bool(get(applications, app_id, f, False)) for f in features
-            )
-        except (AppConfigKeyError, ConfigEntryNotSetError):
-            has_feature = False
-
-        if not has_feature:
-            continue
-
-        # Normalize to iterable of domains
-        doms = list(_iter_domains(domain_value))
-
-        for d in doms:
-            # Use get_url() to produce "<proto>://<domain>"
-            # Pass a minimal per-app mapping so get_domain() resolves to 'd'.
-            # .onion targets are plaintext (Tor encrypts) so they need http, or
-            # the callback scheme mismatches Keycloak's whitelist.
-            proto = "http" if is_onion_domain(d) else web_protocol
-            try:
-                url = get_url({app_id: d}, app_id, proto)
-            except Exception as e:
-                raise AnsibleFilterError(
-                    f"redirect_uris: get_url failed for app '{app_id}' with domain '{d}': {e}"
-                ) from e
-            uris.append(f"{url}{wildcard}")
-
-    return _stable_dedup(uris) if dedup else uris
-
-
-class FilterModule:
-    """Infinito.Nexus redirect URI builder (uses get + get_url)"""
-
-    def filters(self):
-        return {
-            "redirect_uris": redirect_uris,
-        }

roles/web-app-keycloak/vars/main.yml

@@ -82,10 +82,8 @@ KEYCLOAK_REPLICAS_WANT: "{{ KEYCLOAK_REPLICAS }}/{{ KEYCLOAK_REPLICAS }}"
 KEYCLOAK_DB_NETWORK_ARG: "{% if lookup('database', application_id, 'shared') | bool %}--network {{ lookup('database', application_id, 'network') | quote }}{% endif
   %}"
 
-KEYCLOAK_REDIRECT_FEATURES: ["services.sso.enabled", "services.sso.enabled"]
 KEYCLOAK_FRONTCHANNEL_LOGOUT_URL: "{{ lookup('tls', 'web-svc-logout', 'url.base') }}"
-KEYCLOAK_REDIRECT_URIS: "{{ lookup('domains') | redirect_uris(lookup('applications'), lookup('tls', application_id, 'protocols.web'), '/*', KEYCLOAK_REDIRECT_FEATURES)
-  }}"
+KEYCLOAK_REDIRECT_URIS: "{{ lookup('redirect_uris') }}"
 KEYCLOAK_WEB_ORIGINS: >-
   {{ KEYCLOAK_REDIRECT_URIS
       | map('regex_replace','/\\*$','')

tests/unit/plugins/lookup/test_redirect_uris.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import unittest
+
+from ansible.errors import AnsibleError
+
+from plugins.lookup.redirect_uris import LookupModule, build_redirect_uris
+
+SSO = {"services": {"sso": {"enabled": True}}}
+
+
+class TestBuildRedirectUris(unittest.TestCase):
+    def test_single_domain(self):
+        result = build_redirect_uris({"app1": "example.org"}, {"app1": SSO}, True)
+        self.assertEqual(result, ["https://example.org/*"])
+
+    def test_multiple_domains(self):
+        result = build_redirect_uris(
+            {"appX": ["a.example.org", "b.example.org"]}, {"appX": SSO}, True
+        )
+        self.assertCountEqual(
+            result, ["https://a.example.org/*", "https://b.example.org/*"]
+        )
+
+    def test_onion_domain_stays_plaintext(self):
+        domains = {"clear": "app.example.org", "hidden": "abcdefghijklmnop.onion"}
+        result = build_redirect_uris(domains, {"clear": SSO, "hidden": SSO}, True)
+        self.assertCountEqual(
+            result, ["https://app.example.org/*", "http://abcdefghijklmnop.onion/*"]
+        )
+
+    def test_per_app_tls_override_wins_over_the_global_default(self):
+        domains = {"secure": "a.example.org", "plain": "b.example.org"}
+        applications = {
+            "secure": SSO,
+            "plain": {**SSO, "server": {"tls": {"enabled": False}}},
+        }
+        result = build_redirect_uris(domains, applications, True)
+        self.assertCountEqual(
+            result, ["https://a.example.org/*", "http://b.example.org/*"]
+        )
+
+    def test_global_default_off_yields_plaintext(self):
+        result = build_redirect_uris({"app1": "example.org"}, {"app1": SSO}, False)
+        self.assertEqual(result, ["http://example.org/*"])
+
+    def test_feature_missing_is_skipped(self):
+        applications = {"app1": {"features": {"oauth2": False, "oidc": False}}}
+        self.assertEqual(
+            build_redirect_uris({"app1": "example.org"}, applications, True), []
+        )
+
+    def test_wildcard_customization(self):
+        result = build_redirect_uris(
+            {"app1": "x.test"}, {"app1": SSO}, False, wildcard="/cb"
+        )
+        self.assertEqual(result, ["http://x.test/cb"])
+
+    def test_dedup_default_true(self):
+        domains = {"app1": ["dup.test", "dup.test", "other.test"]}
+        result = build_redirect_uris(domains, {"app1": SSO}, True)
+        self.assertEqual(result, ["https://dup.test/*", "https://other.test/*"])
+
+    def test_dedup_false_keeps_duplicates(self):
+        domains = {"app1": ["dup.test", "dup.test"]}
+        result = build_redirect_uris(domains, {"app1": SSO}, True, dedup=False)
+        self.assertEqual(result, ["https://dup.test/*", "https://dup.test/*"])
+
+    def test_nested_domain_mapping_is_flattened(self):
+        domains = {
+            "app1": {
+                "primary": "a.example.org",
+                "nested": {"x": "b.example.org", "y": ["c.example.org"]},
+            }
+        }
+        result = build_redirect_uris(domains, {"app1": SSO}, True)
+        self.assertEqual(
+            result,
+            [
+                "https://a.example.org/*",
+                "https://b.example.org/*",
+                "https://c.example.org/*",
+            ],
+        )
+
+    def test_invalid_domains_type_raises(self):
+        with self.assertRaises(AnsibleError):
+            build_redirect_uris(["not-a-dict"], {}, True)  # type: ignore[arg-type]
+
+    def test_invalid_domain_value_raises(self):
+        with self.assertRaises(AnsibleError):
+            build_redirect_uris({"app1": 42}, {"app1": SSO}, True)
+
+
+class TestRedirectUrisLookup(unittest.TestCase):
+    def setUp(self) -> None:
+        self.lookup = LookupModule()
+        self.lookup._templar = None
+
+    def test_terms_raise(self):
+        with self.assertRaises(AnsibleError):
+            self.lookup.run(["x"], variables={"TLS_ENABLED": True})
+
+    def test_missing_tls_enabled_raises(self):
+        with self.assertRaises(AnsibleError):
+            self.lookup.run([], variables={})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/roles/web-app-keycloak/filter_plugins/test_redirect_uris.py

@@ -1,203 +0,0 @@
-import importlib.util
-import sys
-import types
-import unittest
-
-from utils import tls_common as _real_tls_common
-from utils.cache import PROJECT_ROOT
-
-PLUGIN_PATH = (
-    PROJECT_ROOT / "roles" / "web-app-keycloak" / "filter_plugins" / "redirect_uris.py"
-)
-
-
-def _load_module_from_path(name, file_path):
-    spec = importlib.util.spec_from_file_location(name, file_path)
-    module = importlib.util.module_from_spec(spec)
-    assert spec and spec.loader, f"Cannot load spec for {file_path}"
-    spec.loader.exec_module(module)
-    return module
-
-
-_STUBBED_MODULE_NAMES = (
-    "utils",
-    "utils.roles.applications",
-    "utils.roles.applications.config",
-    "utils.get_url",
-    "utils.tls_common",
-)
-
-
-class RedirectUrisTest(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        cls._orig_sys_modules = {
-            name: sys.modules[name]
-            for name in _STUBBED_MODULE_NAMES
-            if name in sys.modules
-        }
-
-        mu = types.ModuleType("utils")
-        mu_apps = types.ModuleType("utils.roles.applications")
-        mu_config = types.ModuleType("utils.roles.applications.config")
-        mu_geturl = types.ModuleType("utils.get_url")
-
-        # Define stub exceptions
-        class AppConfigKeyError(Exception):
-            pass
-
-        class ConfigEntryNotSetError(Exception):
-            pass
-
-        # Define a practical get that understands dotted keys
-        def get(applications, app_id, dotted, default=None):
-            data = applications.get(app_id, {})
-            cur = data
-            for part in dotted.split("."):
-                if isinstance(cur, dict) and part in cur:
-                    cur = cur[part]
-                else:
-                    return default
-            return cur
-
-        # Define a simple get_url matching your utils/get_url contract
-        # get_url(domains, application_id, protocol) -> "<protocol>://<domain>"
-        def get_url(domains, application_id, protocol):
-            domain = domains[application_id]
-            return f"{protocol}://{domain}"
-
-        # Attach to stub modules
-        mu_config.get = staticmethod(get)
-        mu_config.AppConfigKeyError = AppConfigKeyError
-        mu_config.ConfigEntryNotSetError = ConfigEntryNotSetError
-
-        mu_geturl.get_url = staticmethod(get_url)
-
-        # Register in sys.modules so plugin imports succeed
-        sys.modules["utils"] = mu
-        sys.modules["utils.roles.applications"] = mu_apps
-        sys.modules["utils.roles.applications.config"] = mu_config
-        sys.modules["utils.get_url"] = mu_geturl
-        sys.modules["utils.tls_common"] = _real_tls_common
-
-        # Load the plugin by path
-        cls.plugin = _load_module_from_path(
-            "test_target.redirect_uris", str(PLUGIN_PATH)
-        )
-
-        # Keep originals for per-test monkeypatching
-        cls._orig_get = cls.plugin.get
-        cls._orig_get_url = cls.plugin.get_url
-
-    def tearDown(self):
-        # Restore plugin functions if a test monkeypatched them
-        self.plugin.get = self._orig_get
-        self.plugin.get_url = self._orig_get_url
-
-    @classmethod
-    def tearDownClass(cls):
-        for name in _STUBBED_MODULE_NAMES:
-            if name in cls._orig_sys_modules:
-                sys.modules[name] = cls._orig_sys_modules[name]
-            else:
-                sys.modules.pop(name, None)
-
-    def test_single_domain_oauth2_enabled(self):
-        domains = {"app1": "example.org"}
-        applications = {"app1": {"services": {"sso": {"enabled": True}}}}
-        result = self.plugin.redirect_uris(domains, applications, web_protocol="https")
-        self.assertEqual(result, ["https://example.org/*"])
-
-    def test_multiple_domains_oidc_enabled(self):
-        domains = {"appX": ["a.example.org", "b.example.org"]}
-        applications = {"appX": {"services": {"sso": {"enabled": True}}}}
-        result = self.plugin.redirect_uris(domains, applications, web_protocol="https")
-        self.assertCountEqual(
-            result, ["https://a.example.org/*", "https://b.example.org/*"]
-        )
-
-    def test_feature_missing_is_skipped(self):
-        domains = {"app1": "example.org"}
-        applications = {"app1": {"features": {"oauth2": False, "oidc": False}}}
-        result = self.plugin.redirect_uris(domains, applications)
-        self.assertEqual(result, [])
-
-    def test_protocol_and_wildcard_customization(self):
-        domains = {"app1": "x.test"}
-        applications = {"app1": {"services": {"sso": {"enabled": True}}}}
-        result = self.plugin.redirect_uris(
-            domains, applications, web_protocol="http", wildcard="/cb"
-        )
-        self.assertEqual(result, ["http://x.test/cb"])
-
-    def test_dedup_default_true(self):
-        domains = {"app1": ["dup.test", "dup.test", "other.test"]}
-        applications = {"app1": {"services": {"sso": {"enabled": True}}}}
-        result = self.plugin.redirect_uris(domains, applications)
-        self.assertEqual(result, ["https://dup.test/*", "https://other.test/*"])
-
-    def test_dedup_false_keeps_duplicates(self):
-        domains = {"app1": ["dup.test", "dup.test"]}
-        applications = {"app1": {"services": {"sso": {"enabled": True}}}}
-        result = self.plugin.redirect_uris(domains, applications, dedup=False)
-        self.assertEqual(result, ["https://dup.test/*", "https://dup.test/*"])
-
-    def test_invalid_domains_type_raises(self):
-        with self.assertRaises(self.plugin.AnsibleFilterError):
-            self.plugin.redirect_uris(["not-a-dict"], {})  # type: ignore[arg-type]
-
-    def test_get_url_failure_is_wrapped(self):
-        # Make get_url raise an arbitrary error; plugin should re-raise AnsibleFilterError
-        def boom(*args, **kwargs):
-            raise RuntimeError("boom")
-
-        self.plugin.get_url = boom
-
-        domains = {"app1": "example.org"}
-        applications = {"app1": {"services": {"sso": {"enabled": True}}}}
-
-        with self.assertRaises(self.plugin.AnsibleFilterError) as ctx:
-            self.plugin.redirect_uris(domains, applications)
-        self.assertIn("get_url failed", str(ctx.exception))
-
-    def test_get_exception_is_handled_as_no_feature(self):
-        # Make get raise AppConfigKeyError; plugin should treat as not enabled and skip
-        def raising_get(*args, **kwargs):
-            raise self.plugin.AppConfigKeyError("missing key")
-
-        self.plugin.get = raising_get
-
-        domains = {"app1": "example.org"}
-        applications = {
-            "app1": {"services": {"sso": {"enabled": True}}}
-        }  # value won't be read due to exception
-
-        result = self.plugin.redirect_uris(domains, applications)
-        self.assertEqual(result, [])
-
-    def test_domain_value_dict_is_flattened_in_order(self):
-        # Dict with mixed value types and a duplicate to verify stable dedup
-        domains = {
-            "app1": {
-                "primary": "a.example.org",
-                "alt": ["b.example.org", "b.example.org"],
-                "nested": {"x": "c.example.org", "y": ["d.example.org"]},
-            }
-        }
-        applications = {"app1": {"services": {"sso": {"enabled": True}}}}
-
-        result = self.plugin.redirect_uris(domains, applications)
-
-        self.assertEqual(
-            result,
-            [
-                "https://a.example.org/*",
-                "https://b.example.org/*",  # duplicate trimmed
-                "https://c.example.org/*",
-                "https://d.example.org/*",
-            ],
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()