feat(authz): require_workspace_admin dependency (Stage 1 PR4)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -307,3 +307,18 @@ def require_permission(
|
|||||||
return wrapper
|
return wrapper
|
||||||
|
|
||||||
return decorator
|
return decorator
|
||||||
|
|
||||||
|
|
||||||
|
def require_workspace_admin() -> None:
|
||||||
|
"""FastAPI dependency: require the caller's workspace role to be
|
||||||
|
owner or admin. Reads the role from the workspace contextvar that
|
||||||
|
AuthMiddleware stamps per request.
|
||||||
|
|
||||||
|
Raises HTTPException 403 if no workspace is in context or the role is
|
||||||
|
below admin. Use on management endpoints (service accounts, API keys).
|
||||||
|
"""
|
||||||
|
from deerflow.runtime.workspace_context import get_current_workspace
|
||||||
|
|
||||||
|
workspace = get_current_workspace()
|
||||||
|
if workspace is None or getattr(workspace, "role", None) not in ("owner", "admin"):
|
||||||
|
raise HTTPException(status_code=403, detail="workspace owner/admin role required")
|
||||||
|
|||||||
@@ -0,0 +1,41 @@
|
|||||||
|
"""Tests for require_workspace_admin dependency (Stage 1 PR4)."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from fastapi import HTTPException
|
||||||
|
|
||||||
|
from app.gateway.authz import require_workspace_admin
|
||||||
|
from deerflow.runtime.workspace_context import reset_current_workspace, set_current_workspace
|
||||||
|
|
||||||
|
|
||||||
|
class _WS:
|
||||||
|
def __init__(self, role):
|
||||||
|
self.id = "w-1"
|
||||||
|
self.role = role
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("role", ["owner", "admin"])
|
||||||
|
def test_allows_owner_admin(role):
|
||||||
|
token = set_current_workspace(_WS(role))
|
||||||
|
try:
|
||||||
|
require_workspace_admin() # no raise
|
||||||
|
finally:
|
||||||
|
reset_current_workspace(token)
|
||||||
|
|
||||||
|
|
||||||
|
def test_rejects_member():
|
||||||
|
token = set_current_workspace(_WS("member"))
|
||||||
|
try:
|
||||||
|
with pytest.raises(HTTPException) as exc:
|
||||||
|
require_workspace_admin()
|
||||||
|
assert exc.value.status_code == 403
|
||||||
|
finally:
|
||||||
|
reset_current_workspace(token)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.no_auto_workspace
|
||||||
|
def test_rejects_no_workspace():
|
||||||
|
with pytest.raises(HTTPException) as exc:
|
||||||
|
require_workspace_admin()
|
||||||
|
assert exc.value.status_code == 403
|
||||||
Reference in New Issue
Block a user