xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-08-09 04:40:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(cloud): support scoped admin owner sessions

Browse Source
This commit is contained in:
dadachann
2026-07-31 04:26:15 +00:00
parent ae9769c81e
commit 6f4200ee36
7 changed files with 205 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit_tests/api/service/test_user_service.py
+47
View File
@@ -418,6 +418,53 @@ class TestUserServiceGenerateJwtToken:
assert token is not None
async def test_admin_owner_token_is_scoped_and_capped_at_five_minutes(self):
ap = SimpleNamespace()
ap.instance_config = SimpleNamespace()
ap.instance_config.data = {'system': {'jwt': {'secret': 'test_secret', 'expire': 7200}}}
service = UserService(ap)
before = datetime.datetime.now(datetime.timezone.utc).timestamp()
token = await service.generate_jwt_token(
'owner@example.com',
admin_owner_scope={
'actor_account_uuid': 'admin-uuid',
'workspace_uuid': 'workspace-uuid',
'effective_role': 'owner',
},
)
payload = jwt.decode(
token,
'test_secret',
algorithms=['HS256'],
options={'verify_aud': False},
)
assert payload['exp'] <= before + 301
assert service.get_admin_owner_scope(token) == {
'actor_account_uuid': 'admin-uuid',
'workspace_uuid': 'workspace-uuid',
'effective_role': 'owner',
}
async def test_admin_owner_token_rejects_invalid_scope(self):
ap = SimpleNamespace()
ap.instance_config = SimpleNamespace()
ap.instance_config.data = {'system': {'jwt': {'secret': 'test_secret', 'expire': 7200}}}
service = UserService(ap)
with pytest.raises(ValueError, match='Invalid admin owner token scope'):
await service.generate_jwt_token(
'owner@example.com',
admin_owner_scope={
'actor_account_uuid': 'admin-uuid',
'workspace_uuid': 'workspace-uuid',
'effective_role': 'member',
},
)
class TestUserServiceVerifyJwtToken:
"""Tests for verify_jwt_token method."""
tests/unit_tests/cloud/test_space_launch.py
+53
View File
@@ -60,6 +60,7 @@ def _service(private_key: Ed25519PrivateKey, *, now: int) -> SpaceLaunchService:
app = SimpleNamespace(
deployment=SimpleNamespace(multi_workspace_enabled=True, verification_key_id=KEY_ID),
workspace_service=SimpleNamespace(instance_uuid=INSTANCE_UUID),
logger=SimpleNamespace(info=lambda *args, **kwargs: None),
instance_config=SimpleNamespace(
data={
'space': {
@@ -86,6 +87,58 @@ async def test_consumes_valid_workspace_launch_assertion_once():
await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
async def test_consumes_admin_owner_launch_once_and_validates_claims():
private_key = Ed25519PrivateKey.generate()
now = int(time.time())
service = _service(private_key, now=now)
claims = _claims(now=now)
claims['payload'].update(
{
'launch_mode': 'admin_owner',
'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
'effective_role': 'owner',
}
)
token = _sign(private_key, claims)
launch = await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
assert launch == {
'account_uuid': ACCOUNT_UUID,
'workspace_uuid': WORKSPACE_UUID,
'launch_mode': 'admin_owner',
'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
'effective_role': 'owner',
}
with pytest.raises(SpaceLaunchError, match='already been consumed'):
await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
invalid = _claims(now=now)
invalid['payload'].update(
{
'launch_mode': 'admin_owner',
'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
'effective_role': 'member',
}
)
with pytest.raises(SpaceLaunchError, match='effective role'):
await service.consume_assertion(_sign(private_key, invalid), expected_workspace_uuid=WORKSPACE_UUID)
too_long = _claims(now=now)
too_long['exp'] = now + 91
too_long['payload'].update(
{
'launch_mode': 'admin_owner',
'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
'effective_role': 'owner',
}
)
with pytest.raises(SpaceLaunchError, match='lifetime exceeds 90 seconds'):
await service.consume_assertion(_sign(private_key, too_long), expected_workspace_uuid=WORKSPACE_UUID)
async def test_replay_cache_does_not_scan_all_live_assertions(monkeypatch):
private_key = Ed25519PrivateKey.generate()
now = int(time.time())