mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-10-24 12:07:58 +09:00
Juan Denis
d67e5d07d0
Centralize backup schedule calculations and shared request/form flows, encode query values, and move repository finalization into its service. Add browser CI coverage, repair stale test fixtures, and provide opt-in request and build measurement tools.
278 lines
13 KiB
Python
278 lines
13 KiB
Python
"""GHSA-6w78-q5vm-rfmh-class — shared variable groups and the queue bus must
|
|
enforce scope/ownership-level authorization, not just authentication.
|
|
|
|
shared_resources (/api/v1/shared/variable-groups...):
|
|
- group reads gate on the group's scope (workspace membership / app access)
|
|
- group + variable mutations gate on workspace write roles (can_edit_app for
|
|
application scope)
|
|
- attach/detach additionally gate on the TARGET resource — attaching a group
|
|
injects its variables into the target at deploy time
|
|
- an unscoped group listing spans every workspace -> panel admins only
|
|
|
|
queue_bus (/api/v1/queue/groups/<g>/queues/<q>/messages...):
|
|
- receive/complete/fail/list/get on system-owned groups are admin-only
|
|
(payload theft / internal-processing stalls otherwise)
|
|
- user-owned groups are limited to the owning user (and panel admins)
|
|
"""
|
|
import pytest
|
|
|
|
G = '/api/v1/shared/variable-groups'
|
|
Q = '/api/v1/queue/groups'
|
|
|
|
|
|
@pytest.fixture
|
|
def sr_rbac(app, scoping_rbac):
|
|
"""A workspace-scoped variable group (+ one secret var) in scoping_rbac's
|
|
workspace, plus a second application owned by the foreign user."""
|
|
from types import SimpleNamespace
|
|
from app import db
|
|
from app.models import Application, User
|
|
from app.services.workspace_service import WorkspaceService
|
|
from app.services.shared_resource_service import SharedResourceService as S
|
|
|
|
# The scoping fixture's creator has no WorkspaceMember row (production adds
|
|
# one via WorkspaceService.create_workspace); add it so the owner persona
|
|
# resolves to the 'owner' workspace role.
|
|
owner = User.query.filter_by(username='scope_owner').first()
|
|
WorkspaceService.add_member(scoping_rbac.ws_id, owner.id, role='owner')
|
|
|
|
foreign = User.query.filter_by(username='scope_foreign').first()
|
|
foreign_app = Application(name='foreign-app', app_type='php',
|
|
user_id=foreign.id, root_path='/srv/foreign')
|
|
db.session.add(foreign_app)
|
|
db.session.commit()
|
|
|
|
group = S.create_group('workspace', str(scoping_rbac.ws_id), 'Shared Creds')
|
|
var = S.set_variable(group.id, 'API_TOKEN', 'secret-value', is_secret=True)
|
|
|
|
return SimpleNamespace(
|
|
s=scoping_rbac, group_id=group.id, var_id=var.id,
|
|
foreign_app_id=foreign_app.id,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def qb(app, scoping_rbac):
|
|
"""A system-owned queue with one internal message, plus a user-owned queue
|
|
owned by the scoping 'owner' persona with one message."""
|
|
from types import SimpleNamespace
|
|
from app.models import User
|
|
from app.queue_bus.service import QueueBusService
|
|
|
|
QueueBusService.reset_broker()
|
|
|
|
QueueBusService.ensure_queue('sys-group', 'sys-queue')
|
|
QueueBusService.send('sys-group', 'sys-queue', {'job': 'internal'})
|
|
|
|
owner = User.query.filter_by(username='scope_owner').first()
|
|
QueueBusService.create_group('user-group', name='User Group',
|
|
owner_type='user', owner_id=str(owner.id))
|
|
QueueBusService.create_queue('user-group', 'user-queue', name='User Queue')
|
|
QueueBusService.send('user-group', 'user-queue', {'hello': 'world'})
|
|
|
|
return SimpleNamespace(s=scoping_rbac)
|
|
|
|
|
|
# ----------------------------------------------------------- shared resources
|
|
|
|
def test_get_group_read_personas(client, sr_rbac):
|
|
"""Anyone with a path to the workspace reads the (masked) group; a foreign
|
|
caller is denied."""
|
|
s = sr_rbac.s
|
|
url = f'{G}/{sr_rbac.group_id}'
|
|
for persona in ('owner', 'member', 'viewer', 'admin'):
|
|
assert client.get(url, headers=getattr(s, persona)).status_code == 200, persona
|
|
assert client.get(url, headers=s.foreign).status_code == 403
|
|
|
|
|
|
def test_get_group_missing_404(client, sr_rbac):
|
|
assert client.get(f'{G}/999999', headers=sr_rbac.s.admin).status_code == 404
|
|
|
|
|
|
def test_update_group_requires_write_role(client, sr_rbac):
|
|
s = sr_rbac.s
|
|
url = f'{G}/{sr_rbac.group_id}'
|
|
for persona in ('owner', 'member', 'admin'):
|
|
assert client.put(url, json={'description': 'x'},
|
|
headers=getattr(s, persona)).status_code == 200, persona
|
|
for persona in ('viewer', 'foreign'):
|
|
assert client.put(url, json={'description': 'x'},
|
|
headers=getattr(s, persona)).status_code == 403, persona
|
|
|
|
|
|
def test_delete_group_requires_write_role(client, sr_rbac):
|
|
s = sr_rbac.s
|
|
url = f'{G}/{sr_rbac.group_id}'
|
|
for persona in ('viewer', 'foreign'):
|
|
assert client.delete(url, headers=getattr(s, persona)).status_code == 403, persona
|
|
assert client.delete(url, headers=s.owner).status_code == 200
|
|
|
|
|
|
def test_add_variable_requires_write_role(client, sr_rbac):
|
|
s = sr_rbac.s
|
|
url = f'{G}/{sr_rbac.group_id}/variables'
|
|
assert client.post(url, json={'key': 'NEW', 'value': '1'},
|
|
headers=s.member).status_code == 201
|
|
for persona in ('viewer', 'foreign'):
|
|
assert client.post(url, json={'key': 'INJECTED', 'value': 'x'},
|
|
headers=getattr(s, persona)).status_code == 403, persona
|
|
|
|
|
|
def test_update_and_delete_variable_require_write_role(client, sr_rbac):
|
|
s = sr_rbac.s
|
|
url = f'{G}/{sr_rbac.group_id}/variables/{sr_rbac.var_id}'
|
|
assert client.put(url, json={'value': 'pwned'}, headers=s.foreign).status_code == 403
|
|
assert client.delete(url, headers=s.viewer).status_code == 403
|
|
assert client.put(url, json={'value': 'changed'}, headers=s.owner).status_code == 200
|
|
assert client.delete(url, headers=s.owner).status_code == 200
|
|
|
|
|
|
def test_attach_to_workspace_app(client, sr_rbac):
|
|
"""Workspace write personas attach a workspace group to an app in the same
|
|
workspace; viewer/foreign cannot (group write or target write fails)."""
|
|
s = sr_rbac.s
|
|
url = f'{G}/{sr_rbac.group_id}/attach'
|
|
body = {'resource_type': 'application', 'resource_id': s.app_id}
|
|
assert client.post(url, json=body, headers=s.member).status_code == 201
|
|
assert client.post(url, json=body, headers=s.viewer).status_code == 403
|
|
assert client.post(url, json=body, headers=s.foreign).status_code == 403
|
|
|
|
|
|
def test_attach_to_foreign_app_denied(client, sr_rbac):
|
|
"""Even a workspace owner cannot attach their group to a foreign user's app
|
|
(deploy-time env-var injection into someone else's resource)."""
|
|
s = sr_rbac.s
|
|
url = f'{G}/{sr_rbac.group_id}/attach'
|
|
body = {'resource_type': 'application', 'resource_id': sr_rbac.foreign_app_id}
|
|
for persona in ('owner', 'member'):
|
|
assert client.post(url, json=body,
|
|
headers=getattr(s, persona)).status_code == 403, persona
|
|
# The panel admin may still wire it up deliberately.
|
|
assert client.post(url, json=body, headers=s.admin).status_code == 201
|
|
|
|
|
|
def test_detach_requires_group_and_target_write(client, sr_rbac):
|
|
s = sr_rbac.s
|
|
body = {'resource_type': 'application', 'resource_id': s.app_id}
|
|
client.post(f'{G}/{sr_rbac.group_id}/attach', json=body, headers=s.owner)
|
|
url = f'{G}/{sr_rbac.group_id}/detach'
|
|
assert client.post(url, json=body, headers=s.foreign).status_code == 403
|
|
resp = client.post(url, json=body, headers=s.owner)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['detached'] is True
|
|
|
|
|
|
def test_create_group_requires_scope_write(client, sr_rbac):
|
|
s = sr_rbac.s
|
|
body = {'scope_type': 'workspace', 'scope_id': str(s.ws_id), 'name': 'New'}
|
|
for persona, status in [('member', 201), ('viewer', 403), ('foreign', 403)]:
|
|
response = client.post(G, json=body, headers=getattr(s, persona))
|
|
assert response.status_code == status, (persona, response.get_json())
|
|
|
|
|
|
def test_create_group_application_scope_foreign_denied(client, sr_rbac):
|
|
"""Application-scoped groups gate on can_edit_app; a foreign caller is
|
|
denied before the service validates the scope."""
|
|
s = sr_rbac.s
|
|
body = {'scope_type': 'application', 'scope_id': str(s.app_id), 'name': 'App'}
|
|
assert client.post(G, json=body, headers=s.foreign).status_code == 403
|
|
assert client.post(G, json=body, headers=s.viewer).status_code == 403
|
|
|
|
|
|
def test_list_groups_scoped_requires_membership(client, sr_rbac):
|
|
s = sr_rbac.s
|
|
url = f'{G}?scope_type=workspace&scope_id={s.ws_id}'
|
|
for persona in ('owner', 'member', 'viewer', 'admin'):
|
|
assert client.get(url, headers=getattr(s, persona)).status_code == 200, persona
|
|
assert client.get(url, headers=s.foreign).status_code == 403
|
|
|
|
|
|
def test_list_groups_unscoped_admin_only(client, sr_rbac):
|
|
"""An unscoped listing spans every workspace — panel admins only (mirrors
|
|
the global deploy-history gate)."""
|
|
s = sr_rbac.s
|
|
assert client.get(G, headers=s.admin).status_code == 200
|
|
for persona in ('owner', 'member', 'viewer', 'foreign'):
|
|
assert client.get(G, headers=getattr(s, persona)).status_code == 403, persona
|
|
|
|
|
|
# ------------------------------------------------------------------ queue bus
|
|
|
|
def test_system_queue_receive_admin_only(client, qb):
|
|
"""Non-admins must not pop messages from system queues (visibility-timeout
|
|
theft stalls jobs/notifications/webhook deliveries)."""
|
|
s = qb.s
|
|
url = f'{Q}/sys-group/queues/sys-queue/messages/receive'
|
|
for persona in ('owner', 'member', 'viewer', 'foreign'):
|
|
assert client.post(url, json={},
|
|
headers=getattr(s, persona)).status_code == 403, persona
|
|
resp = client.post(url, json={}, headers=s.admin)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()['messages'][0]['payload'] == {'job': 'internal'}
|
|
|
|
|
|
def test_system_queue_list_and_get_admin_only(client, qb):
|
|
s = qb.s
|
|
base = f'{Q}/sys-group/queues/sys-queue'
|
|
assert client.get(f'{base}/messages', headers=s.admin).status_code == 200
|
|
for persona in ('owner', 'member', 'viewer', 'foreign'):
|
|
assert client.get(f'{base}/messages',
|
|
headers=getattr(s, persona)).status_code == 403, persona
|
|
msg_id = client.get(f'{base}/messages', headers=s.admin).get_json()['messages'][0]['id']
|
|
assert client.get(f'{base}/messages/{msg_id}', headers=s.admin).status_code == 200
|
|
assert client.get(f'{base}/messages/{msg_id}', headers=s.foreign).status_code == 403
|
|
|
|
|
|
def test_system_queue_complete_and_fail_admin_only(client, qb):
|
|
s = qb.s
|
|
base = f'{Q}/sys-group/queues/sys-queue'
|
|
msg = client.post(f'{base}/messages/receive', json={},
|
|
headers=s.admin).get_json()['messages'][0]
|
|
assert client.post(f'{base}/messages/{msg["id"]}/complete', json={},
|
|
headers=s.foreign).status_code == 403
|
|
assert client.post(f'{base}/messages/{msg["id"]}/fail', json={},
|
|
headers=s.owner).status_code == 403
|
|
assert client.post(f'{base}/messages/{msg["id"]}/complete', json={},
|
|
headers=s.admin).status_code == 200
|
|
|
|
|
|
def test_user_queue_owner_full_access(client, qb):
|
|
"""The owning user keeps the pre-fix behavior on their own groups."""
|
|
s = qb.s
|
|
base = f'{Q}/user-group/queues/user-queue'
|
|
assert client.get(f'{base}/messages', headers=s.owner).status_code == 200
|
|
resp = client.post(f'{base}/messages/receive', json={}, headers=s.owner)
|
|
assert resp.status_code == 200
|
|
msg_id = resp.get_json()['messages'][0]['id']
|
|
assert client.get(f'{base}/messages/{msg_id}', headers=s.owner).status_code == 200
|
|
assert client.post(f'{base}/messages/{msg_id}/complete', json={},
|
|
headers=s.owner).status_code == 200
|
|
|
|
|
|
def test_user_queue_other_users_denied(client, qb):
|
|
"""A user-owned group's messages are not readable/ackable by other users."""
|
|
s = qb.s
|
|
base = f'{Q}/user-group/queues/user-queue'
|
|
for persona in ('member', 'viewer', 'foreign'):
|
|
headers = getattr(s, persona)
|
|
assert client.get(f'{base}/messages', headers=headers).status_code == 403, persona
|
|
assert client.post(f'{base}/messages/receive', json={},
|
|
headers=headers).status_code == 403, persona
|
|
msg_id = client.get(f'{base}/messages', headers=s.owner).get_json()['messages'][0]['id']
|
|
for persona in ('member', 'foreign'):
|
|
headers = getattr(s, persona)
|
|
assert client.get(f'{base}/messages/{msg_id}', headers=headers).status_code == 403, persona
|
|
assert client.post(f'{base}/messages/{msg_id}/complete', json={},
|
|
headers=headers).status_code == 403, persona
|
|
assert client.post(f'{base}/messages/{msg_id}/fail', json={},
|
|
headers=headers).status_code == 403, persona
|
|
# The panel admin still can.
|
|
assert client.get(f'{base}/messages', headers=s.admin).status_code == 200
|
|
|
|
|
|
def test_missing_group_still_404(client, qb):
|
|
"""The access guard defers to the service's 404 for unknown groups."""
|
|
resp = client.post(f'{Q}/nope/queues/nope/messages/receive', json={},
|
|
headers=qb.s.admin)
|
|
assert resp.status_code == 404
|