LANG-145: bind WebAuthn to server config and make passkeys OSS-only (#2744)

* LANG-145: bind WebAuthn to server config and make passkeys OSS-only

Two problems in the passkey implementation:

1. The RP ID and expected origin came from the request: a JSON 'origin' field,
   then the Origin/Referer headers, then the request URL. WebAuthn's
   anti-phishing guarantee depends on values the server chose; trusting the
   request lets a sibling host under the same registrable domain obtain a valid
   assertion for a victim's credential. Resolution now follows the same rule the
   OAuth and invitation links already use: api.webui_url, then
   api.webhook_prefix, then the configured port on loopback.

2. Passkeys are now an OSS-only feature. Cloud deployments authenticate through
   their own account service, so every passkey route (register options/verify,
   auth options/verify, list, rename, delete) refuses with
   edition_unsupported, and /account-info advertises neither passkey_login_enabled
   nor passkey_supported. Both cloud flavours are covered: system.edition ==
   'cloud' and deployment.mode == 'cloud'.

Tests:
- New unit suite for the edition matrix and for origin resolution (including
  that client-supplied origin/Origin/Referer are ignored).
- The three 'reverse proxy scenario' tests asserted the old, unsafe behaviour
  (server trusting the request). They now assert the server-config contract.
- test_smoke cloud capabilities expectation updated to passkey_* = false.

Co-authored-by: multica-agent <[email protected]>

* style(api): drop the typing import left unused by the WebAuthn origin change

Co-authored-by: multica-agent <[email protected]>

---------

Co-authored-by: multica-agent <[email protected]>

src/langbot/pkg/api/http/controller/groups/user.py

@@ -6,7 +6,6 @@ import asyncio
 import datetime
 import hmac
 import time
-import typing
 import uuid
 from urllib.parse import parse_qs, urlsplit
 
@@ -16,6 +15,7 @@ from ...context import RequestContext
 from .....cloud.launch import SpaceLaunchError
 from ...service.user import ControlPlaneDirectoryRequiredError, PublicRegistrationClosedError
 from ...service import totp as totp_module
+from .....utils import constants
 
 # Fixed-window admission quota for the unauthenticated reset-password endpoint (#2392).
 # The admission check and slot bump share ONE synchronous critical section with no await
@@ -68,21 +68,49 @@ class UserRouterGroup(group.RouterGroup):
 
         return redirect_uri
 
-    def _extract_origin_and_rp_id(self, json_data: dict[str, typing.Any] | None = None) -> tuple[str, str]:
-        origin = ''
-        if json_data and isinstance(json_data, dict):
-            origin = json_data.get('origin', '')
-        if not origin:
-            origin = quart.request.headers.get('Origin', '')
-        if not origin:
-            origin = quart.request.headers.get('Referer', '')
-        if not origin:
-            origin = quart.request.url_root.rstrip('/')
+    def _resolve_server_origin(self) -> tuple[str, str]:
+        """Return the server-configured browser origin and WebAuthn RP ID.
 
-        parsed = urlsplit(origin)
-        rp_id = parsed.hostname or 'localhost'
-        clean_origin = f'{parsed.scheme}://{parsed.netloc}' if parsed.scheme and parsed.netloc else origin.rstrip('/')
-        return clean_origin, rp_id
+        WebAuthn's anti-phishing guarantee depends on the RP ID and expected
+        origin being values the *server* chose. Deriving them from the request
+        (a JSON `origin` field, the Origin/Referer headers, or the Host header)
+        lets a sibling host under the same registrable domain obtain a valid
+        assertion for the victim's credential. Resolution order matches the
+        OAuth/invitation rule: api.webui_url, then api.webhook_prefix, then the
+        configured API port on loopback. Request data is never consulted.
+        """
+        data = getattr(getattr(self.ap, 'instance_config', None), 'data', {}) or {}
+        api = data.get('api', {}) if isinstance(data.get('api', {}), dict) else {}
+
+        for key in ('webui_url', 'webhook_prefix'):
+            candidate = str(api.get(key, '') or '').strip()
+            if candidate:
+                parsed = urlsplit(candidate)
+                if parsed.scheme and parsed.netloc and parsed.hostname:
+                    return f'{parsed.scheme}://{parsed.netloc}', parsed.hostname
+
+        port = api.get('port', 5300)
+        return f'http://127.0.0.1:{port}', 'localhost'
+
+    def _passkey_supported(self) -> bool:
+        """Passkeys are an OSS-only feature.
+
+        Cloud deployments authenticate through their own account service, so
+        passkey registration and login are refused there outright rather than
+        merely hidden in the UI. Both flavours count as cloud: the multi-tenant
+        Cloud (``system.edition == 'cloud'``) and a standalone cloud-edition
+        instance (``deployment.mode == 'cloud'``).
+        """
+        if constants.edition not in ('community', 'oss'):
+            return False
+        deployment_mode = getattr(getattr(self.ap, 'deployment', None), 'mode', None)
+        return deployment_mode != 'cloud'
+
+    def _passkey_disabled_response(self):
+        """Uniform refusal for every passkey route on a non-OSS edition."""
+        if self._passkey_supported():
+            return None
+        return self.http_status(403, 'edition_unsupported', 'Passkeys are only available in the OSS edition')
 
     async def initialize(self) -> None:
         @self.route('/init', methods=['GET', 'POST'], auth_type=group.AuthType.NONE)
@@ -507,8 +535,9 @@ class UserRouterGroup(group.RouterGroup):
                 capabilities['password_login_enabled'] = False
             capabilities['authenticated_invitation_acceptance_enabled'] = cloud_mode
             capabilities['invitation_registration_enabled'] = not cloud_mode
-            capabilities['passkey_login_enabled'] = True
-            capabilities['passkey_supported'] = True
+            passkey_supported = self._passkey_supported()
+            capabilities['passkey_login_enabled'] = passkey_supported
+            capabilities['passkey_supported'] = passkey_supported
             return self.success(data={'initialized': True, **capabilities})
 
         @self.route('/set-password', methods=['POST'], auth_type=group.AuthType.ACCOUNT_TOKEN)
@@ -602,14 +631,17 @@ class UserRouterGroup(group.RouterGroup):
         @self.route('/passkey/register/options', methods=['POST'], auth_type=group.AuthType.ACCOUNT_TOKEN)
         async def _(account) -> str:
             """Generate WebAuthn registration options for current account."""
+            denied = self._passkey_disabled_response()
+            if denied is not None:
+                return denied
+
             allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
                 'allow_modify_login_info', True
             )
             if not allow_modify_login_info:
                 return self.http_status(403, -1, 'Modifying login info is disabled')
 
-            json_data = (await quart.request.json) or {}
-            origin, rp_id = self._extract_origin_and_rp_id(json_data)
+            origin, rp_id = self._resolve_server_origin()
 
             try:
                 options, challenge_token = await self.ap.user_service.generate_passkey_registration_options(
@@ -625,6 +657,10 @@ class UserRouterGroup(group.RouterGroup):
         @self.route('/passkey/register/verify', methods=['POST'], auth_type=group.AuthType.ACCOUNT_TOKEN)
         async def _(account) -> str:
             """Verify WebAuthn registration response and save credential."""
+            denied = self._passkey_disabled_response()
+            if denied is not None:
+                return denied
+
             allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
                 'allow_modify_login_info', True
             )
@@ -658,9 +694,13 @@ class UserRouterGroup(group.RouterGroup):
         @self.route('/passkey/auth/options', methods=['POST'], auth_type=group.AuthType.NONE)
         async def _() -> str:
             """Generate WebAuthn authentication options for passkey login."""
+            denied = self._passkey_disabled_response()
+            if denied is not None:
+                return denied
+
             json_data = (await quart.request.json) or {}
             email = json_data.get('email')
-            origin, rp_id = self._extract_origin_and_rp_id(json_data)
+            origin, rp_id = self._resolve_server_origin()
 
             try:
                 options, challenge_token = await self.ap.user_service.generate_passkey_authentication_options(
@@ -675,6 +715,10 @@ class UserRouterGroup(group.RouterGroup):
         @self.route('/passkey/auth/verify', methods=['POST'], auth_type=group.AuthType.NONE)
         async def _() -> str:
             """Verify WebAuthn authentication response and log in."""
+            denied = self._passkey_disabled_response()
+            if denied is not None:
+                return denied
+
             json_data = await quart.request.json
             challenge_token = json_data.get('challenge_token')
             credential = json_data.get('credential') or json_data.get('response')
@@ -699,6 +743,10 @@ class UserRouterGroup(group.RouterGroup):
         @self.route('/passkeys', methods=['GET'], auth_type=group.AuthType.ACCOUNT_TOKEN)
         async def _(account) -> str:
             """List registered passkeys for the current user."""
+            denied = self._passkey_disabled_response()
+            if denied is not None:
+                return denied
+
             passkeys = await self.ap.user_service.get_user_passkeys(account.uuid)
             return self.success(
                 data=[
@@ -718,6 +766,10 @@ class UserRouterGroup(group.RouterGroup):
         @self.route('/passkey/<passkey_uuid>', methods=['PATCH'], auth_type=group.AuthType.ACCOUNT_TOKEN)
         async def _(account, passkey_uuid: str) -> str:
             """Rename a registered passkey."""
+            denied = self._passkey_disabled_response()
+            if denied is not None:
+                return denied
+
             allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
                 'allow_modify_login_info', True
             )
@@ -741,6 +793,10 @@ class UserRouterGroup(group.RouterGroup):
         @self.route('/passkey/<passkey_uuid>', methods=['DELETE'], auth_type=group.AuthType.ACCOUNT_TOKEN)
         async def _(account, passkey_uuid: str) -> str:
             """Delete/revoke a registered passkey."""
+            denied = self._passkey_disabled_response()
+            if denied is not None:
+                return denied
+
             allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
                 'allow_modify_login_info', True
             )

tests/integration/api/test_smoke.py

@@ -335,8 +335,10 @@ class TestUserInitEndpoint:
             'invitation_registration_enabled': False,
             'password_login_enabled': False,
             'space_login_enabled': True,
-            'passkey_login_enabled': True,
-            'passkey_supported': True,
+            # Passkeys are OSS-only: a cloud deployment must advertise neither
+            # passkey login nor passkey support, and the routes refuse it.
+            'passkey_login_enabled': False,
+            'passkey_supported': False,
         }
 
     @pytest.mark.asyncio

tests/integration/api/test_user_passkey_api.py

@@ -130,15 +130,32 @@ class TestPasskeyProtectedEndpoints:
 
 
 class TestPasskeyReverseProxyScenarios:
+    """The RP ID and expected origin must come from server config, never the request.
+
+    WebAuthn's anti-phishing guarantee depends on values the *server* chose. If
+    the server trusts a JSON ``origin`` field, the ``Origin``/``Referer`` header,
+    or the ``Host`` header, then a sibling host under the same registrable domain
+    can obtain a valid assertion for a victim's credential.
+    """
+
+    @staticmethod
+    def _set_webui_url(fake_api_app, url: str) -> None:
+        config = getattr(fake_api_app.instance_config, 'data', None)
+        if not isinstance(config, dict):
+            config = {}
+            fake_api_app.instance_config.data = config
+        config.setdefault('api', {})['webui_url'] = url
+
     @pytest.mark.asyncio
-    async def test_auth_options_respects_custom_origin_body_behind_proxy(self, quart_test_client, fake_api_app):
+    async def test_auth_options_ignores_client_supplied_origin_body(self, quart_test_client, fake_api_app):
         fake_api_app.user_service.generate_passkey_authentication_options = AsyncMock(
-            return_value=({'challenge': 'test_chal', 'rpId': 'proxy.company.com'}, 'token_proxy')
+            return_value=({'challenge': 'test_chal', 'rpId': 'bot.langbot.app'}, 'token_cfg')
         )
+        self._set_webui_url(fake_api_app, 'https://bot.langbot.app')
 
         response = await quart_test_client.post(
             '/api/v1/user/passkey/auth/options',
-            json={'origin': 'https://proxy.company.com:8443'},
+            json={'origin': 'https://proxy.attacker.test:8443'},
             headers={'Host': '127.0.0.1:5300'},
         )
 
@@ -146,45 +163,67 @@ class TestPasskeyReverseProxyScenarios:
         data = await response.get_json()
         assert data['code'] == 0
         fake_api_app.user_service.generate_passkey_authentication_options.assert_awaited_once_with(
-            rp_id='proxy.company.com',
-            origin='https://proxy.company.com:8443',
+            rp_id='bot.langbot.app',
+            origin='https://bot.langbot.app',
             email=None,
         )
 
     @pytest.mark.asyncio
-    async def test_auth_options_falls_back_to_origin_header(self, quart_test_client, fake_api_app):
+    async def test_auth_options_ignores_origin_header(self, quart_test_client, fake_api_app):
         fake_api_app.user_service.generate_passkey_authentication_options = AsyncMock(
-            return_value=({'challenge': 'test_chal', 'rpId': 'bot.example.com'}, 'token_header')
+            return_value=({'challenge': 'test_chal', 'rpId': 'bot.langbot.app'}, 'token_hdr')
         )
+        self._set_webui_url(fake_api_app, 'https://bot.langbot.app')
 
         response = await quart_test_client.post(
             '/api/v1/user/passkey/auth/options',
             json={},
-            headers={'Origin': 'https://bot.example.com'},
+            headers={'Origin': 'https://evil.attacker.test'},
         )
 
+        assert response.status_code == 200
+        fake_api_app.user_service.generate_passkey_authentication_options.assert_awaited_once_with(
+            rp_id='bot.langbot.app',
+            origin='https://bot.langbot.app',
+            email=None,
+        )
+
+    @pytest.mark.asyncio
+    async def test_auth_options_ignores_referer_header(self, quart_test_client, fake_api_app):
+        fake_api_app.user_service.generate_passkey_authentication_options = AsyncMock(
+            return_value=({'challenge': 'test_chal', 'rpId': 'bot.langbot.app'}, 'token_ref')
+        )
+        self._set_webui_url(fake_api_app, 'https://bot.langbot.app')
+
+        response = await quart_test_client.post(
+            '/api/v1/user/passkey/auth/options',
+            json={},
+            headers={'Referer': 'https://evil.attacker.test:9000/login'},
+        )
+
+        assert response.status_code == 200
+        fake_api_app.user_service.generate_passkey_authentication_options.assert_awaited_once_with(
+            rp_id='bot.langbot.app',
+            origin='https://bot.langbot.app',
+            email=None,
+        )
+
+    @pytest.mark.asyncio
+    async def test_auth_options_falls_back_to_webhook_prefix(self, quart_test_client, fake_api_app):
+        fake_api_app.user_service.generate_passkey_authentication_options = AsyncMock(
+            return_value=({'challenge': 'test_chal', 'rpId': 'bot.example.com'}, 'token_prefix')
+        )
+        config = getattr(fake_api_app.instance_config, 'data', None)
+        if not isinstance(config, dict):
+            config = {}
+            fake_api_app.instance_config.data = config
+        config['api'] = {'webui_url': '', 'webhook_prefix': 'https://bot.example.com'}
+
+        response = await quart_test_client.post('/api/v1/user/passkey/auth/options', json={})
+
         assert response.status_code == 200
         fake_api_app.user_service.generate_passkey_authentication_options.assert_awaited_once_with(
             rp_id='bot.example.com',
             origin='https://bot.example.com',
             email=None,
         )
-
-    @pytest.mark.asyncio
-    async def test_auth_options_falls_back_to_referer_header(self, quart_test_client, fake_api_app):
-        fake_api_app.user_service.generate_passkey_authentication_options = AsyncMock(
-            return_value=({'challenge': 'test_chal', 'rpId': 'bot.example.com'}, 'token_referer')
-        )
-
-        response = await quart_test_client.post(
-            '/api/v1/user/passkey/auth/options',
-            json={},
-            headers={'Referer': 'https://bot.example.com:9000/login'},
-        )
-
-        assert response.status_code == 200
-        fake_api_app.user_service.generate_passkey_authentication_options.assert_awaited_once_with(
-            rp_id='bot.example.com',
-            origin='https://bot.example.com:9000',
-            email=None,
-        )

tests/unit_tests/api/test_passkey_edition_gate.py

@@ -0,0 +1,104 @@
+"""Passkey gating (OSS-only) and server-side WebAuthn origin resolution.
+
+Regression coverage for LANG-145:
+- WebAuthn RP ID / origin must come from server config, never from the request,
+  otherwise a sibling host can obtain a valid assertion for a victim credential.
+- Passkeys are an OSS-only feature: every passkey route must refuse on a cloud
+  deployment (multi-tenant Cloud or standalone cloud edition).
+"""
+
+from types import SimpleNamespace
+
+import pytest
+
+from langbot.pkg.api.http.controller.groups import user as user_module
+from langbot.pkg.api.http.controller.groups.user import UserRouterGroup
+from langbot.pkg.utils import constants
+
+
+def _router(instance_config=None, deployment_mode=None):
+    """Build just enough of the router group to call its helpers."""
+    group = UserRouterGroup.__new__(UserRouterGroup)
+    group.ap = SimpleNamespace(
+        instance_config=SimpleNamespace(data=instance_config or {}),
+        deployment=SimpleNamespace(mode=deployment_mode) if deployment_mode else None,
+    )
+    return group
+
+
+def test_origin_and_rp_id_come_from_server_config(monkeypatch):
+    monkeypatch.setattr(constants, 'edition', 'community')
+    group = _router({'api': {'webui_url': 'https://bot.langbot.app', 'webhook_prefix': 'https://example.test'}})
+
+    origin, rp_id = group._resolve_server_origin()
+
+    assert origin == 'https://bot.langbot.app'
+    assert rp_id == 'bot.langbot.app'
+
+
+def test_origin_falls_back_to_webhook_prefix(monkeypatch):
+    monkeypatch.setattr(constants, 'edition', 'community')
+    group = _router({'api': {'webui_url': '', 'webhook_prefix': 'https://bot2.langbot.app'}})
+
+    origin, rp_id = group._resolve_server_origin()
+
+    assert origin == 'https://bot2.langbot.app'
+    assert rp_id == 'bot2.langbot.app'
+
+
+def test_origin_never_uses_loopback_of_attacker_supplied_host(monkeypatch):
+    """With nothing configured the RP falls back to loopback, not the request."""
+    monkeypatch.setattr(constants, 'edition', 'community')
+    group = _router({'api': {'webui_url': '', 'webhook_prefix': '', 'port': 5300}})
+
+    origin, rp_id = group._resolve_server_origin()
+
+    assert origin == 'http://127.0.0.1:5300'
+    assert rp_id == 'localhost'
+
+
+def test_client_supplied_origin_helper_no_longer_exists():
+    """The request-derived helper must be gone, not merely unused."""
+    assert not hasattr(UserRouterGroup, '_extract_origin_and_rp_id')
+
+
+@pytest.mark.parametrize(
+    'edition,deployment_mode,expected',
+    [
+        ('community', None, True),
+        ('oss', None, True),
+        ('community', 'oss', True),
+        ('cloud', None, False),
+        ('community', 'cloud', False),
+        ('cloud', 'cloud', False),
+    ],
+)
+def test_passkey_supported_only_on_oss(monkeypatch, edition, deployment_mode, expected):
+    monkeypatch.setattr(constants, 'edition', edition)
+    group = _router(deployment_mode=deployment_mode)
+
+    assert group._passkey_supported() is expected
+
+
+def test_refusal_builds_only_when_unsupported(monkeypatch):
+    """The denial path is taken on cloud and skipped on OSS.
+
+    ``_passkey_disabled_response`` builds a Quart response, which needs an app
+    context, so the decision itself is asserted here through the gate that
+    drives it; the wiring of every route to that denial is covered below.
+    """
+    monkeypatch.setattr(constants, 'edition', 'cloud')
+    assert _router()._passkey_supported() is False
+
+    monkeypatch.setattr(constants, 'edition', 'community')
+    assert _router()._passkey_supported() is True
+
+
+def test_every_passkey_route_is_gated():
+    """Guards against a new passkey route being added without the edition gate."""
+    import inspect
+
+    source = inspect.getsource(user_module)
+    passkey_routes = source.count("@self.route('/passkey")
+    assert passkey_routes >= 7, f'expected at least 7 passkey routes, found {passkey_routes}'
+    assert source.count('_passkey_disabled_response()') >= passkey_routes