fix(test-e2e-playwright): bypass the Tor proxy for every clearnet host the node serves

The onion pick only bypassed the SSO issuer, so a clearnet CDN, asset or dashboard host an onion page loads went through Tor, which the lab network cannot exit. browser_proxy_bypass derives the list from the deployed applications' domains plus the issuer; it is empty on a clearnet pick and on a tor-only node. Specs that passed their own proxy to newContext dropped the bypass and now inherit use.proxy.

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

plugins/lookup/browser_proxy_bypass.py

@@ -0,0 +1,55 @@
+"""Hosts a browser probe reaches without the proxy of the network it probes."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+from utils.networks.reachability import (
+    browser_proxy_bypass,
+    resolve_node_mode,
+    tor_serves_host,
+)
+from utils.tls_common import iter_domains
+
+
+class LookupModule(LookupBase):
+    """
+    Usage:
+      {{ lookup('browser_proxy_bypass', network_name, *extra_hosts) }}
+
+    Returns the hosts a browser probing a page of ``network_name`` on this node
+    reaches directly instead of through that network's proxy: the domains of
+    the applications deployed here plus ``extra_hosts`` (e.g. the SSO issuer),
+    each kept only when its network needs no browser proxy and the node serves
+    that network.
+    """
+
+    def run(self, terms, variables: dict[str, Any] | None = None, **kwargs):
+        if not terms or not str(terms[0] or "").strip():
+            raise AnsibleError(
+                "lookup('browser_proxy_bypass', network, *hosts) expects a network"
+            )
+        variables = variables or getattr(self._templar, "available_variables", {}) or {}
+        templar = getattr(self, "_templar", None)
+        raw_mode = variables.get("NETWORK_MODE") or ""
+        if templar is not None and isinstance(raw_mode, str):
+            raw_mode = templar.template(raw_mode)
+        try:
+            mode = resolve_node_mode(raw_mode, tor_provided=tor_serves_host(variables))
+        except ValueError as exc:
+            raise AnsibleError(f"browser_proxy_bypass: {exc}") from exc
+        domains = lookup_loader.get(
+            "domains", loader=self._loader, templar=templar
+        ).run([], variables=variables)[0]
+        deployed = set(variables.get("group_names") or [])
+        hosts = [
+            domain
+            for app, value in domains.items()
+            if app in deployed
+            for domain in iter_domains(value)
+        ]
+        return [browser_proxy_bypass(str(terms[0]), mode, [*hosts, *terms[1:]])]

roles/test-e2e-playwright/tasks/03_run_pass.yml

@@ -26,6 +26,10 @@
       PLAYWRIGHT_NETWORK_LEAK={{ (_playwright_passes | length > 1) | string | lower }}
       PLAYWRIGHT_NETWORK_SUFFIXES={{ _playwright_passes | map('network_of') | network_suffixes | to_json }}
       PLAYWRIGHT_NETWORK_ALLOWED_HOSTS={{ OIDC.URL | urlsplit('hostname') }}
+      {% set _bypass = lookup('browser_proxy_bypass', _playwright_pass_domain | network_of, OIDC.URL | urlsplit('hostname')) %}
+      {% if _bypass %}
+      PLAYWRIGHT_PROXY_BYPASS={{ _bypass | join(',') }}
+      {% endif %}
   vars:
     application_id: "{{ application_id }}"
     domains: "{{ lookup('domains') }}"
@@ -81,11 +85,6 @@
         + ['-e', 'INFINITO_PLAYWRIGHT_KEEP=' ~ test_e2e_playwright_keep]
         + ['-e', 'INFINITO_PLAYWRIGHT_GLOBAL_TIMEOUT_MS=' ~ TEST_E2E_PLAYWRIGHT_GLOBAL_TIMEOUT_MS]
         + ['-e', 'PLAYWRIGHT_PROXY=' ~ TEST_E2E_PLAYWRIGHT_PROXY]
-        + (
-            ['-e', 'PLAYWRIGHT_PROXY_BYPASS=' ~ (OIDC.URL | urlsplit('hostname'))]
-            if (OIDC.URL | urlsplit('hostname') | network_of) != (_playwright_pass_domain | network_of)
-            else []
-          )
         + ['-e', 'PLAYWRIGHT_WORKERS=' ~ test_e2e_playwright_workers]
         + ['-e', 'PLAYWRIGHT_FULLY_PARALLEL=' ~ TEST_E2E_PLAYWRIGHT_FULLY_PARALLEL]
         + ['-e', 'PLAYWRIGHT_TIMEOUT_FACTOR=' ~ TIMEOUT_FACTOR]

roles/web-app-stalwart/files/playwright/test-mailflow.js

@@ -48,10 +48,8 @@ test("stalwart: biber and the administrator exchange mail both ways", async ({ b
 
   const outbound = `Playwright stalwart ${Date.now()}`;
   const answer = `Playwright stalwart answer ${Date.now()}`;
-  const proxyServer = process.env.PLAYWRIGHT_PROXY;
   const contextOptions = {
     ignoreHTTPSErrors: true,
-    ...(proxyServer ? { proxy: { server: proxyServer } } : {}),
     ...recordVideoOptions(testInfo),
   };
   const biberContext = await browser.newContext(contextOptions);

roles/web-app-stalwart/files/playwright/test-onion-mailflow.js

@@ -25,10 +25,8 @@ test("biber: a .onion recipient is accepted and routed through the Tor gateway",
   expect(biberPassword, "BIBER_PASSWORD must be set").toBeTruthy();
 
   const testSubject = `Playwright onion route ${Date.now()}`;
-  const proxyServer = process.env.PLAYWRIGHT_PROXY;
   const context = await browser.newContext({
     ignoreHTTPSErrors: true,
-    ...(proxyServer ? { proxy: { server: proxyServer } } : {}),
     ...recordVideoOptions(testInfo),
   });
 

tests/unit/python/plugins/lookup/test_browser_proxy_bypass.py

@@ -0,0 +1,55 @@
+import unittest
+import unittest.mock as mock
+
+from ansible.errors import AnsibleError
+
+import plugins.lookup.browser_proxy_bypass as bypass_mod
+from utils.domains.default_primary import default_domain_primary
+
+PRIMARY = default_domain_primary()
+ONION = "abcdefghijklmnopqrstuvwxyz234567abcdefghijklmnopqrstuvw.onion"
+DOMAINS = {
+    "web-app-deployed": [f"app.{PRIMARY}", f"app.{ONION}"],
+    "web-app-elsewhere": [f"other.{PRIMARY}"],
+}
+
+
+def _loader():
+    loader = mock.MagicMock()
+    loader.get.return_value = mock.MagicMock(run=lambda terms, **_kw: [DOMAINS])
+    return mock.patch.object(bypass_mod, "lookup_loader", loader)
+
+
+class TestBrowserProxyBypassLookup(unittest.TestCase):
+    def _run(self, terms, **variables):
+        lookup = bypass_mod.LookupModule()
+        lookup._loader = mock.MagicMock()
+        with _loader():
+            return lookup.run(terms, variables=variables)[0]
+
+    def test_only_deployed_clearnet_hosts_and_the_issuer_go_direct(self):
+        self.assertEqual(
+            self._run(
+                ["tor", f"auth.{PRIMARY}"],
+                group_names=["svc-net-tor", "web-app-deployed"],
+            ),
+            sorted([f"app.{PRIMARY}", f"auth.{PRIMARY}"]),
+        )
+
+    def test_a_tor_only_node_bypasses_nothing(self):
+        self.assertEqual(
+            self._run(
+                ["tor", f"auth.{PRIMARY}"],
+                group_names=["svc-net-tor", "web-app-deployed"],
+                NETWORK_MODE="tor",
+            ),
+            [],
+        )
+
+    def test_a_mode_the_node_cannot_serve_fails_loudly(self):
+        with self.assertRaises(AnsibleError):
+            self._run(["tor"], group_names=["web-app-deployed"], NETWORK_MODE="multi")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/python/roles/test-e2e-playwright/test_env_canonical_domain.py

@@ -11,6 +11,7 @@ from plugins.filter.network_of import in_network, network_of, network_suffixes
 from plugins.lookup.network_siblings import served_siblings
 from utils.cache.yaml import load_yaml_any
 from utils.domains.default_primary import default_domain_primary
+from utils.networks.reachability import browser_proxy_bypass
 
 from . import PROJECT_ROOT
 
@@ -35,12 +36,14 @@ def render(pass_domain: str) -> str:
         urlsplit=lambda url, part: getattr(urlsplit(url), part),
     )
 
-    def lookup(name, term, template_vars=None):
+    def lookup(name, term, *extra, template_vars=None):
         if name == "template":
             return (
                 f'APP_BASE_URL="https://{template_vars["domain"]}/"\n'
                 f'CANONICAL_DOMAIN="{CLEARNET[1]}"\n'
             )
+        if name == "browser_proxy_bypass":
+            return browser_proxy_bypass(term, "multi", [*CLEARNET, *TOR, *extra])
         return served_siblings(term, {"web-app-minio": CLEARNET + TOR}, PRIMARY, NODE)
 
     return env.from_string(task["ansible.builtin.copy"]["content"]).render(
@@ -50,6 +53,7 @@ def render(pass_domain: str) -> str:
         _playwright_passes=[CLEARNET[0], TOR[0]],
         TEST_E2E_PLAYWRIGHT_CONFIG_TIMEOUTS={},
         OIDC={"URL": f"https://auth.{PRIMARY}/realms/main"},
+        DOMAIN_PRIMARY=PRIMARY,
         DEPLOYMENT_MODE="compose",
     )
 
@@ -60,6 +64,13 @@ class TestEnvCanonicalDomain(unittest.TestCase):
         self.assertIn(f'APP_BASE_URL="https://{TOR[0]}/"\n', rendered)
         self.assertIn(f'CANONICAL_DOMAIN="{TOR[1]}"\n', rendered)
 
+    def test_onion_pick_bypasses_the_proxy_for_the_issuer_and_the_clearnet_node(self):
+        self.assertIn(
+            f"PLAYWRIGHT_PROXY_BYPASS=api.s3.{PRIMARY},auth.{PRIMARY},console.s3.{PRIMARY}\n",
+            render(TOR[0]),
+        )
+        self.assertNotIn("PLAYWRIGHT_PROXY_BYPASS", render(CLEARNET[0]))
+
     def test_canonical_network_pick_keeps_the_rendered_env(self):
         self.assertTrue(
             render(CLEARNET[0]).startswith(

tests/unit/python/utils/networks/test_reachability.py

@@ -10,6 +10,7 @@ from utils.networks.reachability import (
     MULTI,
     TOR,
     allowed_networks,
+    browser_proxy_bypass,
     effective_networks,
     is_network,
     network,
@@ -27,6 +28,22 @@ def _cfg(**reachability) -> dict:
     return {"networks": {"reachability": reachability}} if reachability else {}
 
 
+class TestBrowserProxyBypass(unittest.TestCase):
+    HOSTS = (f"auth.{DOMAIN}", f"cdn.{DOMAIN}.", DOMAIN, f"cdn.{ONION}", "")
+
+    def test_an_onion_probe_on_a_multi_node_reaches_the_clearnet_hosts_directly(self):
+        self.assertEqual(
+            browser_proxy_bypass(TOR, MULTI, self.HOSTS),
+            sorted([f"auth.{DOMAIN}", f"cdn.{DOMAIN}", DOMAIN]),
+        )
+
+    def test_a_tor_only_node_sends_nothing_around_the_proxy(self):
+        self.assertEqual(browser_proxy_bypass(TOR, TOR, self.HOSTS), [])
+
+    def test_a_probe_without_a_proxy_bypasses_nothing(self):
+        self.assertEqual(browser_proxy_bypass(CLEARNET, MULTI, self.HOSTS), [])
+
+
 class TestNetworkOf(unittest.TestCase):
     def test_onion_suffix_is_tor(self) -> None:
         self.assertEqual(network_of(f"wazuh.{ONION}"), TOR)

utils/networks/reachability.py

@@ -184,6 +184,27 @@ def resolve_node_mode(raw: Any, *, tor_provided: bool) -> str:
     return value
 
 
+def browser_proxy_bypass(probe: str, node_mode: str, hosts: Any) -> list[str]:
+    """Hosts a browser probing a page of ``probe`` reaches without its proxy.
+
+    Args:
+        probe: network of the page the browser probes.
+        node_mode: the node's resolved ``NETWORK_MODE``.
+        hosts: candidate hosts, e.g. the deployed applications' domains and the
+            SSO issuer.
+
+    Returns:
+        The sorted, unique candidates whose network needs no browser proxy and
+        is served by the node; empty when ``probe`` needs no proxy itself.
+    """
+    if not network(probe).browser_proxy:
+        return []
+    direct = {name for name in NODE_MODES[node_mode] if not network(name).browser_proxy}
+    return sorted(
+        {host for host in map(_host, hosts) if host and network_of(host) in direct}
+    )
+
+
 def role_reachability(
     app_config: Any, *, app: str = ""
 ) -> tuple[tuple[str, ...], bool]: