Fix fail-closed backend authorization

backend/auth.py

@@ -1,12 +1,22 @@
+"""Authentication and the legacy role-based authorization boundary.
+
+Backend tools must not infer authorization from a Streamlit session.  The
+Gateway validates a tool/role pair first, then binds that role only for the
+duration of the tool call.  Direct calls therefore fail closed.
 """
-backend/auth.py
-使用者驗證與角色型存取控制 (RBAC)
-"""
 
-import streamlit as st
+from contextlib import contextmanager
+from contextvars import ContextVar
+from collections.abc import Iterator
+
 from .database import run_query
 
 
+_AUTHORIZED_ROLE: ContextVar[str | None] = ContextVar(
+    "erp_authorized_role", default=None
+)
+
+
 def check_login(username: str, password: str) -> dict | None:
     """驗證帳號密碼，成功回傳 {role, name}，失敗回傳 None。
     （N3）密碼以 salted hash 比對；遇到 legacy 明文則於登入成功時就地升級。"""
@@ -27,20 +37,29 @@ def check_login(username: str, password: str) -> dict | None:
     return {"role": role, "name": name}
 
 
-def check_permission(allowed_roles: list) -> bool:
-    """依目前 session 角色判斷是否有權限；admin 永遠通過"""
-    try:
-        from streamlit.runtime.scriptrunner import get_script_run_ctx
-        if not get_script_run_ctx():
-            return True
-    except Exception:
-        pass
-
+@contextmanager
+def authorized_role(role: str) -> Iterator[None]:
+    """Bind a role that has already been authorized by the Tool Gateway.
+
+    This helper is an internal execution mechanism, not an authorization
+    decision.  Callers must validate the tool/role pair before entering it.
+    ContextVar keeps concurrent requests and async tasks isolated.
+    """
+    normalized_role = str(role or "").strip()
+    if not normalized_role:
+        raise PermissionError("A verified role is required")
+    token = _AUTHORIZED_ROLE.set(normalized_role)
     try:
-        current_role = st.session_state.get("role", "")
-    except Exception:
-        return True
+        yield
+    finally:
+        _AUTHORIZED_ROLE.reset(token)
+
 
+def check_permission(allowed_roles: list[str] | tuple[str, ...] | set[str]) -> bool:
+    """Check the Gateway-bound role; missing identity always denies access."""
+    current_role = _AUTHORIZED_ROLE.get()
+    if not current_role:
+        return False
     if current_role == "admin":
         return True
     return current_role in allowed_roles

backend/tool_gateway.py

@@ -613,14 +613,20 @@ def _execute(self, tool_name: str, args: dict, role: str) -> GatewayResult:
         """實際執行工具函式"""
         try:
             import inspect
+            from backend.auth import authorized_role
+
             func   = tools_mapping[tool_name]
             sig = inspect.signature(func)
             has_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values())
             if has_kwargs:
                 filtered_args = args or {}
             else:
                 filtered_args = {k: v for k, v in (args or {}).items() if k in sig.parameters}
-            result = func(**filtered_args)
+            # ``call`` has already checked the registry's role allowlist.
+            # Bind that decision only around the backend function invocation;
+            # direct function calls have no context and fail closed.
+            with authorized_role(role):
+                result = func(**filtered_args)
             _write_log(tool_name, args, role, result, success=True)
             return GatewayResult(status="ok", data=result)
         except Exception as e:

tests/test_auth_fail_closed.py

@@ -0,0 +1,29 @@
+"""Regression tests for the backend authorization execution boundary."""
+
+from backend.auth import authorized_role, check_permission
+
+
+def test_permission_without_gateway_context_is_denied():
+    assert check_permission(["warehouse"]) is False
+    assert check_permission(["admin"]) is False
+
+
+def test_verified_role_is_scoped_to_context():
+    with authorized_role("warehouse"):
+        assert check_permission(["warehouse"]) is True
+        assert check_permission(["sales"]) is False
+
+    assert check_permission(["warehouse"]) is False
+
+
+def test_admin_override_requires_verified_context():
+    with authorized_role("admin"):
+        assert check_permission(["warehouse"]) is True
+
+
+def test_nested_context_restores_outer_role():
+    with authorized_role("warehouse"):
+        with authorized_role("sales"):
+            assert check_permission(["sales"]) is True
+            assert check_permission(["warehouse"]) is False
+        assert check_permission(["warehouse"]) is True