From e92fe0d7fbafa031180adff500f9c7a1cbcaf925 Mon Sep 17 00:00:00 2001
From: 1445043649 <>
Date: Sun, 28 Jun 2026 19:07:55 +0800
Subject: [PATCH] feat(authz): require_workspace_admin dependency (Stage 1 PR4)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 backend/app/gateway/authz.py                  | 15 +++++++
 backend/tests/test_require_workspace_admin.py | 41 +++++++++++++++++++
 2 files changed, 56 insertions(+)
 create mode 100644 backend/tests/test_require_workspace_admin.py

diff --git a/backend/app/gateway/authz.py b/backend/app/gateway/authz.py
index 1830f90c..2b7c958a 100644
--- a/backend/app/gateway/authz.py
+++ b/backend/app/gateway/authz.py
@@ -307,3 +307,18 @@ def require_permission(
         return wrapper
 
     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")
diff --git a/backend/tests/test_require_workspace_admin.py b/backend/tests/test_require_workspace_admin.py
new file mode 100644
index 00000000..009b2b80
--- /dev/null
+++ b/backend/tests/test_require_workspace_admin.py
@@ -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