mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-09-16 14:57:15 +00:00
feat(auth): add webauthn authentication support
This commit is contained in:
BiFangKNT
@@ -310,6 +310,8 @@ class TestUserInitEndpoint:
|
||||
'invitation_registration_enabled': True,
|
||||
'password_login_enabled': True,
|
||||
'space_login_enabled': False,
|
||||
'passkey_login_enabled': True,
|
||||
'passkey_supported': True,
|
||||
}
|
||||
fake_api_app.user_service.get_login_capabilities.assert_awaited_once_with()
|
||||
fake_api_app.user_service.get_first_user.assert_not_awaited()
|
||||
@@ -334,6 +336,8 @@ class TestUserInitEndpoint:
|
||||
'invitation_registration_enabled': False,
|
||||
'password_login_enabled': False,
|
||||
'space_login_enabled': True,
|
||||
'passkey_login_enabled': True,
|
||||
'passkey_supported': True,
|
||||
}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -355,6 +359,8 @@ class TestUserInitEndpoint:
|
||||
'invitation_registration_enabled': True,
|
||||
'password_login_enabled': False,
|
||||
'space_login_enabled': True,
|
||||
'passkey_login_enabled': True,
|
||||
'passkey_supported': True,
|
||||
}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
"""
|
||||
Integration smoke tests for Passkey API endpoints.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from tests.integration.api.test_smoke import (
|
||||
fake_api_app,
|
||||
mock_circular_import_chain,
|
||||
quart_test_client,
|
||||
)
|
||||
|
||||
|
||||
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures('mock_circular_import_chain')]
|
||||
|
||||
|
||||
class TestPasskeyPublicEndpoints:
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_options_endpoint(self, quart_test_client, fake_api_app):
|
||||
fake_api_app.user_service.generate_passkey_authentication_options = AsyncMock(
|
||||
return_value=({'challenge': 'test_chal', 'rpId': 'localhost'}, 'token_123')
|
||||
)
|
||||
|
||||
response = await quart_test_client.post(
|
||||
'/api/v1/user/passkey/auth/options',
|
||||
json={'origin': 'http://localhost:3000'},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = await response.get_json()
|
||||
assert data['code'] == 0
|
||||
assert data['data']['challenge_token'] == 'token_123'
|
||||
assert data['data']['options']['rpId'] == 'localhost'
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_verify_missing_payload(self, quart_test_client, fake_api_app):
|
||||
response = await quart_test_client.post(
|
||||
'/api/v1/user/passkey/auth/verify',
|
||||
json={},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = await response.get_json()
|
||||
assert data['code'] != 0
|
||||
assert 'Missing challenge_token or credential' in data['msg']
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_verify_success(self, quart_test_client, fake_api_app):
|
||||
fake_api_app.user_service.verify_passkey_authentication = AsyncMock(
|
||||
return_value=('jwt_token_abc', Mock(user='user@example.com'))
|
||||
)
|
||||
|
||||
response = await quart_test_client.post(
|
||||
'/api/v1/user/passkey/auth/verify',
|
||||
json={'challenge_token': 'token_123', 'credential': {'id': 'cred_id'}},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = await response.get_json()
|
||||
assert data['code'] == 0
|
||||
assert data['data']['token'] == 'jwt_token_abc'
|
||||
assert data['data']['user'] == 'user@example.com'
|
||||
|
||||
|
||||
class TestPasskeyProtectedEndpoints:
|
||||
@pytest.mark.asyncio
|
||||
async def test_register_options_requires_auth(self, quart_test_client):
|
||||
response = await quart_test_client.post('/api/v1/user/passkey/register/options', json={})
|
||||
assert response.status_code == 401
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_passkeys_list_requires_auth(self, quart_test_client):
|
||||
response = await quart_test_client.get('/api/v1/user/passkeys')
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
class TestPasskeyReverseProxyScenarios:
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_options_respects_custom_origin_body_behind_proxy(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')
|
||||
)
|
||||
|
||||
response = await quart_test_client.post(
|
||||
'/api/v1/user/passkey/auth/options',
|
||||
json={'origin': 'https://proxy.company.com:8443'},
|
||||
headers={'Host': '127.0.0.1:5300'},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
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',
|
||||
email=None,
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auth_options_falls_back_to_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')
|
||||
)
|
||||
|
||||
response = await quart_test_client.post(
|
||||
'/api/v1/user/passkey/auth/options',
|
||||
json={},
|
||||
headers={'Origin': 'https://bot.example.com'},
|
||||
)
|
||||
|
||||
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,
|
||||
)
|
||||
@@ -0,0 +1,104 @@
|
||||
"""
|
||||
Unit tests for Passkey WebAuthn service operations in UserService.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from langbot.pkg.api.http.service.user import UserService
|
||||
from langbot.pkg.entity.persistence.user import AccountStatus, User
|
||||
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
class TestPasskeyChallengeLifecycle:
|
||||
async def test_challenge_issuance_and_consumption(self):
|
||||
service = UserService(SimpleNamespace())
|
||||
token, challenge_bytes = await service.issue_passkey_challenge(
|
||||
purpose='register',
|
||||
rp_id='localhost',
|
||||
origin='http://localhost:3000',
|
||||
account_uuid='acc-123',
|
||||
user_email='user@example.com',
|
||||
)
|
||||
|
||||
assert len(token) > 20
|
||||
assert len(challenge_bytes) == 32
|
||||
|
||||
data = await service.consume_passkey_challenge(token, 'register')
|
||||
assert data.challenge == challenge_bytes
|
||||
assert data.rp_id == 'localhost'
|
||||
assert data.origin == 'http://localhost:3000'
|
||||
assert data.account_uuid == 'acc-123'
|
||||
assert data.user_email == 'user@example.com'
|
||||
|
||||
# Replay should fail
|
||||
with pytest.raises(ValueError, match='Invalid or expired passkey challenge'):
|
||||
await service.consume_passkey_challenge(token, 'register')
|
||||
|
||||
async def test_challenge_purpose_mismatch_fails(self):
|
||||
service = UserService(SimpleNamespace())
|
||||
token, _ = await service.issue_passkey_challenge(
|
||||
purpose='register',
|
||||
rp_id='localhost',
|
||||
origin='http://localhost:3000',
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match='Passkey challenge purpose mismatch'):
|
||||
await service.consume_passkey_challenge(token, 'auth')
|
||||
|
||||
async def test_challenge_expiration(self):
|
||||
service = UserService(SimpleNamespace())
|
||||
token, _ = await service.issue_passkey_challenge(
|
||||
purpose='auth',
|
||||
rp_id='localhost',
|
||||
origin='http://localhost:3000',
|
||||
ttl_seconds=0,
|
||||
)
|
||||
|
||||
with pytest.raises(ValueError, match='Invalid or expired passkey challenge'):
|
||||
await service.consume_passkey_challenge(token, 'auth')
|
||||
|
||||
|
||||
class TestPasskeyOptionsGeneration:
|
||||
async def test_generate_registration_options(self):
|
||||
service = UserService(SimpleNamespace())
|
||||
mock_user = Mock(spec=User)
|
||||
mock_user.uuid = 'acc-test-uuid'
|
||||
mock_user.user = 'test@example.com'
|
||||
mock_user.status = AccountStatus.ACTIVE.value
|
||||
service.get_user_by_uuid = AsyncMock(return_value=mock_user)
|
||||
service.get_user_passkeys = AsyncMock(return_value=[])
|
||||
|
||||
options, token = await service.generate_passkey_registration_options(
|
||||
account_uuid='acc-test-uuid',
|
||||
rp_id='localhost',
|
||||
origin='http://localhost:3000',
|
||||
rp_name='LangBot Test',
|
||||
)
|
||||
|
||||
assert isinstance(options, dict)
|
||||
assert options['rp']['name'] == 'LangBot Test'
|
||||
assert options['rp']['id'] == 'localhost'
|
||||
assert options['user']['name'] == 'test@example.com'
|
||||
assert 'challenge' in options
|
||||
assert len(token) > 0
|
||||
|
||||
async def test_generate_authentication_options_discoverable(self):
|
||||
service = UserService(SimpleNamespace())
|
||||
|
||||
options, token = await service.generate_passkey_authentication_options(
|
||||
rp_id='localhost',
|
||||
origin='http://localhost:3000',
|
||||
)
|
||||
|
||||
assert isinstance(options, dict)
|
||||
assert options['rpId'] == 'localhost'
|
||||
assert 'challenge' in options
|
||||
assert len(token) > 0
|
||||
Reference in New Issue
Block a user