OpenShell/e2e/python/oidc/workspace_authz_test.py

# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""End-to-end tests for workspace-scoped authorization enforcement.

Validates that every workspace-scoped RPC enforces membership and role
checks when OIDC is configured. Uses two Keycloak users:

- admin@test  — openshell-admin role → Platform Admin (bypasses membership)
- user@test   — openshell-user role  → must be an explicit workspace member

Skip condition: set OPENSHELL_E2E_OIDC=1 to enable these tests.
"""

from __future__ import annotations

import contextlib
import os
from typing import TYPE_CHECKING, Any

import grpc
import pytest

if TYPE_CHECKING:
    from collections.abc import Callable

from openshell._proto import (
    datamodel_pb2,
    openshell_pb2,
    openshell_pb2_grpc,
)

from .helpers import extract_sub, get_token, stub_with_token

WS = "e2e-authz-test"

pytestmark = pytest.mark.skipif(
    os.environ.get("OPENSHELL_E2E_OIDC") != "1",
    reason="OIDC e2e tests disabled (set OPENSHELL_E2E_OIDC=1)",
)


# ── Helpers ──────────────────────────────────────────────────────────────


def _admin_token() -> str:
    return get_token("admin@test", "admin", scopes="openid openshell:all")


def _user_token() -> str:
    return get_token("user@test", "user", scopes="openid openshell:all")


def _add_member(
    stub: openshell_pb2_grpc.OpenShellStub,
    metadata: list[tuple[str, str]],
    workspace: str,
    subject: str,
    role: int,
) -> None:
    stub.AddWorkspaceMember(
        openshell_pb2.AddWorkspaceMemberRequest(
            workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=workspace),
            principal_subject=subject,
            role=role,
        ),
        metadata=metadata,
    )


def _remove_member(
    stub: openshell_pb2_grpc.OpenShellStub,
    metadata: list[tuple[str, str]],
    workspace: str,
    subject: str,
) -> None:
    with contextlib.suppress(grpc.RpcError):
        stub.RemoveWorkspaceMember(
            openshell_pb2.RemoveWorkspaceMemberRequest(
                workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=workspace),
                principal_subject=subject,
            ),
            metadata=metadata,
        )


# ── RPC call builders for parametrized non-member rejection tests ────────
#
# Each entry is (test_id, callable(stub, metadata) -> response).
# The callable constructs a minimal valid request for the given RPC.


def _assert_non_member_denial(
    error: grpc.RpcError,
    workspace: str,
    subject: str,
    rpc_name: str,
) -> None:
    assert error.code() == grpc.StatusCode.PERMISSION_DENIED, (
        f"{rpc_name}: expected PERMISSION_DENIED, got {error.code()}"
    )
    details = error.details()
    assert f"not a member of workspace '{workspace}'" in details, (
        f"{rpc_name}: denial came from the wrong authorization layer: {details}"
    )
    command = (
        "openshell workspace member add "
        f"--workspace '{workspace}' --subject '{subject}' --role user"
    )
    assert command in details, (
        f"{rpc_name}: denial omitted the actionable membership command: {details}"
    )


def _assert_hidden_sandbox(error: grpc.RpcError, rpc_name: str) -> None:
    assert error.code() == grpc.StatusCode.NOT_FOUND, (
        f"{rpc_name}: expected NOT_FOUND, got {error.code()}"
    )


def _assert_workspace_admin_denial(
    error: grpc.RpcError,
    workspace: str,
    subject: str,
    rpc_name: str,
) -> None:
    assert error.code() == grpc.StatusCode.PERMISSION_DENIED, (
        f"{rpc_name}: expected PERMISSION_DENIED, got {error.code()}"
    )
    details = error.details()
    assert f"workspace role 'admin' required in workspace '{workspace}'" in details, (
        f"{rpc_name}: denial came from the wrong authorization layer: {details}"
    )
    command = (
        "openshell workspace member add "
        f"--workspace '{workspace}' --subject '{subject}' --role admin"
    )
    assert command in details, (
        f"{rpc_name}: denial omitted the admin remediation command: {details}"
    )


def _workspace_rpcs() -> list[tuple[str, Callable]]:
    """All workspace-scoped RPCs that accept a workspace field."""
    return [
        # ── Workspace domain ──
        (
            "GetWorkspace",
            lambda s, m: s.GetWorkspace(
                openshell_pb2.GetWorkspaceRequest(name=WS), metadata=m
            ),
        ),
        (
            "ListWorkspaceMembers",
            lambda s, m: s.ListWorkspaceMembers(
                openshell_pb2.ListWorkspaceMembersRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=m,
            ),
        ),
        (
            "AddWorkspaceMember",
            lambda s, m: s.AddWorkspaceMember(
                openshell_pb2.AddWorkspaceMemberRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    principal_subject="fake",
                    role=openshell_pb2.WORKSPACE_ROLE_USER,
                ),
                metadata=m,
            ),
        ),
        (
            "RemoveWorkspaceMember",
            lambda s, m: s.RemoveWorkspaceMember(
                openshell_pb2.RemoveWorkspaceMemberRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    principal_subject="fake",
                ),
                metadata=m,
            ),
        ),
        # ── Sandbox domain ──
        (
            "CreateSandbox",
            lambda s, m: s.CreateSandbox(
                openshell_pb2.CreateSandboxRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    spec=openshell_pb2.SandboxSpec(
                        template=openshell_pb2.SandboxTemplate(image="ubuntu:24.04")
                    ),
                ),
                metadata=m,
            ),
        ),
        (
            "GetSandbox",
            lambda s, m: s.GetSandbox(
                openshell_pb2.GetSandboxRequest(
                    name="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ListSandboxes",
            lambda s, m: s.ListSandboxes(
                openshell_pb2.ListSandboxesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=m,
            ),
        ),
        (
            "DeleteSandbox",
            lambda s, m: s.DeleteSandbox(
                openshell_pb2.DeleteSandboxRequest(
                    name="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ListSandboxProviders",
            lambda s, m: s.ListSandboxProviders(
                openshell_pb2.ListSandboxProvidersRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "AttachSandboxProvider",
            lambda s, m: s.AttachSandboxProvider(
                openshell_pb2.AttachSandboxProviderRequest(
                    sandbox="nonexistent",
                    provider="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "DetachSandboxProvider",
            lambda s, m: s.DetachSandboxProvider(
                openshell_pb2.DetachSandboxProviderRequest(
                    sandbox="nonexistent",
                    provider="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        # ── Provider domain ──
        (
            "CreateProvider",
            lambda s, m: s.CreateProvider(
                openshell_pb2.CreateProviderRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    provider=datamodel_pb2.Provider(
                        metadata=datamodel_pb2.ObjectMeta(
                            name="authz-test", workspace=WS
                        ),
                        type="claude-code",
                        credentials={"ANTHROPIC_API_KEY": "v"},
                    ),
                ),
                metadata=m,
            ),
        ),
        (
            "GetProvider",
            lambda s, m: s.GetProvider(
                openshell_pb2.GetProviderRequest(
                    name="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ListProviders",
            lambda s, m: s.ListProviders(
                openshell_pb2.ListProvidersRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=m,
            ),
        ),
        (
            "UpdateProvider",
            lambda s, m: s.UpdateProvider(
                openshell_pb2.UpdateProviderRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    provider=datamodel_pb2.Provider(
                        metadata=datamodel_pb2.ObjectMeta(
                            name="nonexistent", workspace=WS
                        ),
                        type="claude-code",
                        credentials={"ANTHROPIC_API_KEY": "v"},
                    ),
                ),
                metadata=m,
            ),
        ),
        (
            "DeleteProvider",
            lambda s, m: s.DeleteProvider(
                openshell_pb2.DeleteProviderRequest(
                    name="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ListProviderProfiles",
            lambda s, m: s.ListProviderProfiles(
                openshell_pb2.ListProviderProfilesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=m,
            ),
        ),
        (
            "GetProviderProfile",
            lambda s, m: s.GetProviderProfile(
                openshell_pb2.GetProviderProfileRequest(
                    id="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ImportProviderProfiles",
            lambda s, m: s.ImportProviderProfiles(
                openshell_pb2.ImportProviderProfilesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    profiles=[],
                ),
                metadata=m,
            ),
        ),
        (
            "UpdateProviderProfiles",
            lambda s, m: s.UpdateProviderProfiles(
                openshell_pb2.UpdateProviderProfilesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    id="nonexistent",
                ),
                metadata=m,
            ),
        ),
        (
            "LintProviderProfiles",
            lambda s, m: s.LintProviderProfiles(
                openshell_pb2.LintProviderProfilesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    profiles=[],
                ),
                metadata=m,
            ),
        ),
        (
            "DeleteProviderProfile",
            lambda s, m: s.DeleteProviderProfile(
                openshell_pb2.DeleteProviderProfileRequest(
                    id="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "GetProviderRefreshStatus",
            lambda s, m: s.GetProviderRefreshStatus(
                openshell_pb2.GetProviderRefreshStatusRequest(
                    provider="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ConfigureProviderRefresh",
            lambda s, m: s.ConfigureProviderRefresh(
                openshell_pb2.ConfigureProviderRefreshRequest(
                    provider="nonexistent",
                    credential_key="k",
                    strategy=openshell_pb2.PROVIDER_CREDENTIAL_REFRESH_STRATEGY_STATIC,
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "RotateProviderCredential",
            lambda s, m: s.RotateProviderCredential(
                openshell_pb2.RotateProviderCredentialRequest(
                    provider="nonexistent",
                    credential_key="k",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "DeleteProviderRefresh",
            lambda s, m: s.DeleteProviderRefresh(
                openshell_pb2.DeleteProviderRefreshRequest(
                    provider="nonexistent",
                    credential_key="k",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        # ── Service domain ──
        (
            "ExposeService",
            lambda s, m: s.ExposeService(
                openshell_pb2.ExposeServiceRequest(
                    sandbox="nonexistent",
                    name="svc",
                    target_port=8080,
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "GetService",
            lambda s, m: s.GetService(
                openshell_pb2.GetServiceRequest(
                    sandbox="nonexistent",
                    name="svc",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ListServices",
            lambda s, m: s.ListServices(
                openshell_pb2.ListServicesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=m,
            ),
        ),
        (
            "DeleteService",
            lambda s, m: s.DeleteService(
                openshell_pb2.DeleteServiceRequest(
                    sandbox="nonexistent",
                    name="svc",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        # ── Policy domain ──
        (
            "GetSandboxPolicyStatus",
            lambda s, m: s.GetSandboxPolicyStatus(
                openshell_pb2.GetSandboxPolicyStatusRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ListSandboxPolicies",
            lambda s, m: s.ListSandboxPolicies(
                openshell_pb2.ListSandboxPoliciesRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "GetDraftPolicy",
            lambda s, m: s.GetDraftPolicy(
                openshell_pb2.GetDraftPolicyRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ApproveDraftChunk",
            lambda s, m: s.ApproveDraftChunk(
                openshell_pb2.ApproveDraftChunkRequest(
                    sandbox="nonexistent",
                    chunk_id="x",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "RejectDraftChunk",
            lambda s, m: s.RejectDraftChunk(
                openshell_pb2.RejectDraftChunkRequest(
                    sandbox="nonexistent",
                    chunk_id="x",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ApproveAllDraftChunks",
            lambda s, m: s.ApproveAllDraftChunks(
                openshell_pb2.ApproveAllDraftChunksRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "EditDraftChunk",
            lambda s, m: s.EditDraftChunk(
                openshell_pb2.EditDraftChunkRequest(
                    sandbox="nonexistent",
                    chunk_id="x",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "UndoDraftChunk",
            lambda s, m: s.UndoDraftChunk(
                openshell_pb2.UndoDraftChunkRequest(
                    sandbox="nonexistent",
                    chunk_id="x",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "ClearDraftChunks",
            lambda s, m: s.ClearDraftChunks(
                openshell_pb2.ClearDraftChunksRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
        (
            "GetDraftHistory",
            lambda s, m: s.GetDraftHistory(
                openshell_pb2.GetDraftHistoryRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=m,
            ),
        ),
    ]


def _platform_profile_rpcs() -> list[tuple[str, Callable]]:
    """Provider-profile RPCs targeting the explicit platform scope."""
    return [
        (
            "ListProviderProfiles",
            lambda s, m: s.ListProviderProfiles(
                openshell_pb2.ListProviderProfilesRequest(), metadata=m
            ),
        ),
        (
            "GetProviderProfile",
            lambda s, m: s.GetProviderProfile(
                openshell_pb2.GetProviderProfileRequest(id="nonexistent"),
                metadata=m,
            ),
        ),
        (
            "ImportProviderProfiles",
            lambda s, m: s.ImportProviderProfiles(
                openshell_pb2.ImportProviderProfilesRequest(profiles=[]),
                metadata=m,
            ),
        ),
        (
            "UpdateProviderProfiles",
            lambda s, m: s.UpdateProviderProfiles(
                openshell_pb2.UpdateProviderProfilesRequest(id="nonexistent"),
                metadata=m,
            ),
        ),
        (
            "LintProviderProfiles",
            lambda s, m: s.LintProviderProfiles(
                openshell_pb2.LintProviderProfilesRequest(profiles=[]),
                metadata=m,
            ),
        ),
        (
            "DeleteProviderProfile",
            lambda s, m: s.DeleteProviderProfile(
                openshell_pb2.DeleteProviderProfileRequest(id="nonexistent"),
                metadata=m,
            ),
        ),
    ]


def _global_policy_read_rpcs() -> list[tuple[str, Callable]]:
    """Policy history reads targeting the explicit global scope."""
    return [
        (
            "GetSandboxPolicyStatus",
            lambda s, m: s.GetSandboxPolicyStatus(
                openshell_pb2.GetSandboxPolicyStatusRequest(**{"global": True}),
                metadata=m,
            ),
        ),
        (
            "ListSandboxPolicies",
            lambda s, m: s.ListSandboxPolicies(
                openshell_pb2.ListSandboxPoliciesRequest(**{"global": True}),
                metadata=m,
            ),
        ),
    ]


# ── Test class ───────────────────────────────────────────────────────────


class TestWorkspaceAuthorization:
    """Workspace-scoped authorization enforcement tests."""

    @pytest.fixture(autouse=True, scope="class")
    def workspace(self) -> Any:
        """Create a test workspace and tear it down after all tests."""
        token = _admin_token()
        stub, metadata = stub_with_token(token)

        with contextlib.suppress(grpc.RpcError):
            stub.CreateWorkspace(
                openshell_pb2.CreateWorkspaceRequest(name=WS),
                metadata=metadata,
            )

        yield WS

        with contextlib.suppress(grpc.RpcError):
            stub.DeleteWorkspace(
                openshell_pb2.DeleteWorkspaceRequest(name=WS),
                metadata=metadata,
            )

    @pytest.fixture(scope="class")
    def admin_ctx(
        self,
    ) -> tuple[openshell_pb2_grpc.OpenShellStub, list[tuple[str, str]]]:
        token = _admin_token()
        return stub_with_token(token)

    @pytest.fixture(scope="class")
    def user_ctx(
        self,
    ) -> tuple[openshell_pb2_grpc.OpenShellStub, list[tuple[str, str]], str]:
        token = _user_token()
        stub, metadata = stub_with_token(token)
        sub = extract_sub(token)
        return stub, metadata, sub

    @pytest.fixture(scope="class")
    def seed_provider(self, admin_ctx: Any, workspace: str) -> Any:
        """Create a provider so read RPCs have data to return."""
        stub, metadata = admin_ctx
        prov_name = "e2e-authz-provider"
        with contextlib.suppress(grpc.RpcError):
            stub.CreateProvider(
                openshell_pb2.CreateProviderRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(
                        workspace=workspace
                    ),
                    provider=datamodel_pb2.Provider(
                        metadata=datamodel_pb2.ObjectMeta(
                            name=prov_name, workspace=workspace
                        ),
                        type="claude-code",
                        credentials={"ANTHROPIC_API_KEY": "test"},
                    ),
                ),
                metadata=metadata,
            )
        yield prov_name
        with contextlib.suppress(grpc.RpcError):
            stub.DeleteProvider(
                openshell_pb2.DeleteProviderRequest(
                    name=prov_name,
                    workspace_scope=datamodel_pb2.WorkspaceSelector(
                        workspace=workspace
                    ),
                ),
                metadata=metadata,
            )

    # ── Test 1: Non-member rejection — workspace-field RPCs ──────────

    @pytest.mark.parametrize(
        "rpc_name,call",
        _workspace_rpcs(),
        ids=[r[0] for r in _workspace_rpcs()],
    )
    def test_non_member_rejected(
        self,
        rpc_name: str,
        call: Callable,
        user_ctx: Any,
    ) -> None:
        stub, metadata, user_sub = user_ctx
        with pytest.raises(grpc.RpcError) as exc_info:
            call(stub, metadata)
        if rpc_name in {
            "GetSandbox",
            "DeleteSandbox",
            "ListSandboxProviders",
            "AttachSandboxProvider",
            "DetachSandboxProvider",
            "ExposeService",
            "GetService",
            "DeleteService",
            "GetSandboxPolicyStatus",
            "ListSandboxPolicies",
            "GetDraftPolicy",
            "ApproveDraftChunk",
            "RejectDraftChunk",
            "ApproveAllDraftChunks",
            "EditDraftChunk",
            "UndoDraftChunk",
            "ClearDraftChunks",
            "GetDraftHistory",
        }:
            _assert_hidden_sandbox(exc_info.value, rpc_name)
        else:
            _assert_non_member_denial(exc_info.value, WS, user_sub, rpc_name)

    # ── Test 2: Non-member rejection — dual-mode RPCs ────────────────

    def test_non_member_rejected_update_config(
        self,
        user_ctx: Any,
    ) -> None:
        stub, metadata, _user_sub = user_ctx
        with pytest.raises(grpc.RpcError) as exc_info:
            stub.UpdateConfig(
                openshell_pb2.UpdateConfigRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=metadata,
            )
        _assert_hidden_sandbox(exc_info.value, "UpdateConfig")

    # ── Test 3: Sandbox log authorization uses persisted workspace ───

    def test_get_sandbox_logs_rejects_spoofed_workspace(
        self,
        admin_ctx: Any,
        user_ctx: Any,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx
        other_workspace = "e2e-authz-log-b"
        sandbox_name = "e2e-log-target"

        with contextlib.suppress(grpc.RpcError):
            admin_stub.DeleteWorkspace(
                openshell_pb2.DeleteWorkspaceRequest(name=other_workspace),
                metadata=admin_md,
            )
        admin_stub.CreateWorkspace(
            openshell_pb2.CreateWorkspaceRequest(name=other_workspace),
            metadata=admin_md,
        )
        _add_member(
            admin_stub, admin_md, WS, user_sub, openshell_pb2.WORKSPACE_ROLE_USER
        )

        sandbox_created = False
        try:
            admin_stub.CreateSandbox(
                openshell_pb2.CreateSandboxRequest(
                    name=sandbox_name,
                    workspace_scope=datamodel_pb2.WorkspaceSelector(
                        workspace=other_workspace
                    ),
                    spec=openshell_pb2.SandboxSpec(),
                ),
                metadata=admin_md,
            )
            sandbox_created = True

            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.GetSandboxLogs(
                    openshell_pb2.GetSandboxLogsRequest(
                        sandbox=sandbox_name,
                        workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    ),
                    metadata=user_md,
                )
            # Name-based handlers normalize unauthorized responses to NOT_FOUND
            # so cross-workspace sandbox existence cannot be inferred (CWE-203).
            assert exc_info.value.code() == grpc.StatusCode.NOT_FOUND, (
                "GetSandboxLogs: expected NOT_FOUND for cross-workspace sandbox, "
                f"got {exc_info.value.code()}"
            )
        finally:
            if sandbox_created:
                with contextlib.suppress(grpc.RpcError):
                    admin_stub.DeleteSandbox(
                        openshell_pb2.DeleteSandboxRequest(
                            name=sandbox_name,
                            workspace_scope=datamodel_pb2.WorkspaceSelector(
                                workspace=other_workspace
                            ),
                        ),
                        metadata=admin_md,
                    )
            _remove_member(admin_stub, admin_md, WS, user_sub)
            with contextlib.suppress(grpc.RpcError):
                admin_stub.DeleteWorkspace(
                    openshell_pb2.DeleteWorkspaceRequest(name=other_workspace),
                    metadata=admin_md,
                )

    # ── Test 4: Global config requires Platform Admin ────────────────

    def test_global_update_rejected_for_default_workspace_admin(
        self,
        admin_ctx: Any,
        user_ctx: Any,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx

        _remove_member(admin_stub, admin_md, "default", user_sub)
        _add_member(
            admin_stub,
            admin_md,
            "default",
            user_sub,
            openshell_pb2.WORKSPACE_ROLE_ADMIN,
        )
        try:
            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.UpdateConfig(
                    openshell_pb2.UpdateConfigRequest(
                        setting_key="log_level",
                        delete_setting=True,
                        **{"global": True},
                    ),
                    metadata=user_md,
                )
            assert exc_info.value.code() == grpc.StatusCode.PERMISSION_DENIED
            assert "platform admin role required" in exc_info.value.details(), (
                "UpdateConfig: denial came from the wrong authorization layer: "
                f"{exc_info.value.details()}"
            )
        finally:
            _remove_member(admin_stub, admin_md, "default", user_sub)

    # ── Test 5: Non-member ListWorkspaces returns filtered results ───

    def test_non_member_list_workspaces_filtered(
        self,
        user_ctx: Any,
    ) -> None:
        stub, metadata, _ = user_ctx
        resp = stub.ListWorkspaces(
            openshell_pb2.ListWorkspacesRequest(),
            metadata=metadata,
        )
        ws_names = [w.metadata.name for w in resp.workspaces]
        assert WS not in ws_names, (
            f"non-member should not see workspace {WS} in ListWorkspaces"
        )

    # ── Test 6: Platform Admin bypass ────────────────────────────────

    @pytest.mark.usefixtures("seed_provider")
    def test_platform_admin_get_workspace(
        self,
        admin_ctx: Any,
    ) -> None:
        stub, metadata = admin_ctx
        resp = stub.GetWorkspace(
            openshell_pb2.GetWorkspaceRequest(name=WS),
            metadata=metadata,
        )
        assert resp.workspace.metadata.name == WS

    def test_platform_admin_list_sandboxes(
        self,
        admin_ctx: Any,
    ) -> None:
        stub, metadata = admin_ctx
        stub.ListSandboxes(
            openshell_pb2.ListSandboxesRequest(
                workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
            ),
            metadata=metadata,
        )

    def test_platform_admin_get_provider(
        self,
        admin_ctx: Any,
        seed_provider: str,
    ) -> None:
        stub, metadata = admin_ctx
        resp = stub.GetProvider(
            openshell_pb2.GetProviderRequest(
                name=seed_provider,
                workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
            ),
            metadata=metadata,
        )
        assert resp.provider.metadata.name == seed_provider

    def test_platform_admin_list_services(
        self,
        admin_ctx: Any,
    ) -> None:
        stub, metadata = admin_ctx
        stub.ListServices(
            openshell_pb2.ListServicesRequest(
                workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
            ),
            metadata=metadata,
        )

    def test_platform_admin_get_draft_history(
        self,
        admin_ctx: Any,
    ) -> None:
        stub, metadata = admin_ctx
        # May fail with NOT_FOUND for the sandbox name, but should not fail with PERMISSION_DENIED
        try:
            stub.GetDraftHistory(
                openshell_pb2.GetDraftHistoryRequest(
                    sandbox="nonexistent",
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=metadata,
            )
        except grpc.RpcError as e:
            assert e.code() != grpc.StatusCode.PERMISSION_DENIED

    # ── Test 7: User member — read operations succeed ────────────────

    def test_user_member_read_operations(
        self,
        admin_ctx: Any,
        user_ctx: Any,
        seed_provider: str,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx

        _add_member(
            admin_stub, admin_md, WS, user_sub, openshell_pb2.WORKSPACE_ROLE_USER
        )
        try:
            # GetWorkspace
            resp = user_stub.GetWorkspace(
                openshell_pb2.GetWorkspaceRequest(name=WS),
                metadata=user_md,
            )
            assert resp.workspace.metadata.name == WS

            # ListSandboxes
            user_stub.ListSandboxes(
                openshell_pb2.ListSandboxesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=user_md,
            )

            # GetProvider
            resp = user_stub.GetProvider(
                openshell_pb2.GetProviderRequest(
                    name=seed_provider,
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                ),
                metadata=user_md,
            )
            assert resp.provider.metadata.name == seed_provider

            # ListProviders
            user_stub.ListProviders(
                openshell_pb2.ListProvidersRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=user_md,
            )

            # ListServices
            user_stub.ListServices(
                openshell_pb2.ListServicesRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=user_md,
            )

            # ListWorkspaceMembers
            user_stub.ListWorkspaceMembers(
                openshell_pb2.ListWorkspaceMembersRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS)
                ),
                metadata=user_md,
            )
        finally:
            _remove_member(admin_stub, admin_md, WS, user_sub)

    # ── Test 8: User member — admin operations denied ────────────────

    def test_user_member_admin_operations_denied(
        self,
        admin_ctx: Any,
        user_ctx: Any,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx

        _add_member(
            admin_stub, admin_md, WS, user_sub, openshell_pb2.WORKSPACE_ROLE_USER
        )
        try:
            # CreateProvider requires workspace admin
            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.CreateProvider(
                    openshell_pb2.CreateProviderRequest(
                        workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                        provider=datamodel_pb2.Provider(
                            metadata=datamodel_pb2.ObjectMeta(
                                name="user-blocked", workspace=WS
                            ),
                            type="claude-code",
                            credentials={"ANTHROPIC_API_KEY": "v"},
                        ),
                    ),
                    metadata=user_md,
                )
            _assert_workspace_admin_denial(
                exc_info.value,
                WS,
                user_sub,
                "CreateProvider",
            )

            # AddWorkspaceMember requires workspace admin
            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.AddWorkspaceMember(
                    openshell_pb2.AddWorkspaceMemberRequest(
                        workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                        principal_subject="fake-subject",
                        role=openshell_pb2.WORKSPACE_ROLE_USER,
                    ),
                    metadata=user_md,
                )
            _assert_workspace_admin_denial(
                exc_info.value,
                WS,
                user_sub,
                "AddWorkspaceMember",
            )

            # ApproveDraftChunk requires workspace admin
            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.ApproveDraftChunk(
                    openshell_pb2.ApproveDraftChunkRequest(
                        sandbox="nonexistent",
                        chunk_id="x",
                        workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    ),
                    metadata=user_md,
                )
            # Name-scoped sandbox operations hide whether an unknown sandbox
            # exists before evaluating workspace-admin authorization.
            _assert_hidden_sandbox(exc_info.value, "ApproveDraftChunk")
        finally:
            _remove_member(admin_stub, admin_md, WS, user_sub)

    # ── Test 9: Workspace Admin — admin operations succeed ───────────

    def test_workspace_admin_can_create_provider(
        self,
        admin_ctx: Any,
        user_ctx: Any,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx

        _add_member(
            admin_stub, admin_md, WS, user_sub, openshell_pb2.WORKSPACE_ROLE_ADMIN
        )
        prov_name = "e2e-authz-ws-admin-prov"
        try:
            user_stub.CreateProvider(
                openshell_pb2.CreateProviderRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    provider=datamodel_pb2.Provider(
                        metadata=datamodel_pb2.ObjectMeta(name=prov_name, workspace=WS),
                        type="claude-code",
                        credentials={"ANTHROPIC_API_KEY": "v"},
                    ),
                ),
                metadata=user_md,
            )

            # Also test AddWorkspaceMember with User role
            user_stub.AddWorkspaceMember(
                openshell_pb2.AddWorkspaceMemberRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    principal_subject="fake-member-subject",
                    role=openshell_pb2.WORKSPACE_ROLE_USER,
                ),
                metadata=user_md,
            )
            _remove_member(admin_stub, admin_md, WS, "fake-member-subject")
        finally:
            with contextlib.suppress(grpc.RpcError):
                admin_stub.DeleteProvider(
                    openshell_pb2.DeleteProviderRequest(
                        name=prov_name,
                        workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    ),
                    metadata=admin_md,
                )
            _remove_member(admin_stub, admin_md, WS, user_sub)

    # ── Test 10: all_workspaces rejected for non-Platform-Admin ──────

    def test_all_workspaces_rejected_for_workspace_admin(
        self,
        admin_ctx: Any,
        user_ctx: Any,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx

        _add_member(
            admin_stub, admin_md, WS, user_sub, openshell_pb2.WORKSPACE_ROLE_ADMIN
        )
        try:
            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.ListSandboxes(
                    openshell_pb2.ListSandboxesRequest(
                        workspace_scope=datamodel_pb2.WorkspaceSelector(
                            all_workspaces=datamodel_pb2.AllWorkspaces()
                        )
                    ),
                    metadata=user_md,
                )
            assert exc_info.value.code() == grpc.StatusCode.PERMISSION_DENIED

            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.ListProviders(
                    openshell_pb2.ListProvidersRequest(
                        workspace_scope=datamodel_pb2.WorkspaceSelector(
                            all_workspaces=datamodel_pb2.AllWorkspaces()
                        )
                    ),
                    metadata=user_md,
                )
            assert exc_info.value.code() == grpc.StatusCode.PERMISSION_DENIED

            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.ListServices(
                    openshell_pb2.ListServicesRequest(
                        workspace_scope=datamodel_pb2.WorkspaceSelector(
                            all_workspaces=datamodel_pb2.AllWorkspaces()
                        )
                    ),
                    metadata=user_md,
                )
            assert exc_info.value.code() == grpc.StatusCode.PERMISSION_DENIED
        finally:
            _remove_member(admin_stub, admin_md, WS, user_sub)

    # ── Test 11: Workspace Admin cannot assign Admin role ────────────

    def test_workspace_admin_cannot_assign_admin_role(
        self,
        admin_ctx: Any,
        user_ctx: Any,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx

        _add_member(
            admin_stub, admin_md, WS, user_sub, openshell_pb2.WORKSPACE_ROLE_ADMIN
        )
        try:
            # Workspace Admin cannot assign Admin role
            with pytest.raises(grpc.RpcError) as exc_info:
                user_stub.AddWorkspaceMember(
                    openshell_pb2.AddWorkspaceMemberRequest(
                        workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                        principal_subject="another-subject",
                        role=openshell_pb2.WORKSPACE_ROLE_ADMIN,
                    ),
                    metadata=user_md,
                )
            assert exc_info.value.code() == grpc.StatusCode.PERMISSION_DENIED

            # But User role assignment succeeds
            user_stub.AddWorkspaceMember(
                openshell_pb2.AddWorkspaceMemberRequest(
                    workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=WS),
                    principal_subject="another-subject",
                    role=openshell_pb2.WORKSPACE_ROLE_USER,
                ),
                metadata=user_md,
            )
            _remove_member(admin_stub, admin_md, WS, "another-subject")
        finally:
            _remove_member(admin_stub, admin_md, WS, user_sub)

    # ── Test 12: ListWorkspaces filtered by membership ───────────────

    def test_list_workspaces_filtered_by_membership(
        self,
        admin_ctx: Any,
        user_ctx: Any,
    ) -> None:
        admin_stub, admin_md = admin_ctx
        user_stub, user_md, user_sub = user_ctx

        ws2 = "e2e-authz-test-2"
        with contextlib.suppress(grpc.RpcError):
            admin_stub.CreateWorkspace(
                openshell_pb2.CreateWorkspaceRequest(name=ws2),
                metadata=admin_md,
            )

        _add_member(
            admin_stub, admin_md, WS, user_sub, openshell_pb2.WORKSPACE_ROLE_USER
        )
        try:
            resp = user_stub.ListWorkspaces(
                openshell_pb2.ListWorkspacesRequest(),
                metadata=user_md,
            )
            ws_names = [w.metadata.name for w in resp.workspaces]
            assert WS in ws_names, f"member should see {WS}"
            assert ws2 not in ws_names, f"non-member should not see {ws2}"
            assert "default" not in ws_names, "non-member should not see default"
        finally:
            _remove_member(admin_stub, admin_md, WS, user_sub)
            with contextlib.suppress(grpc.RpcError):
                admin_stub.DeleteWorkspace(
                    openshell_pb2.DeleteWorkspaceRequest(name=ws2),
                    metadata=admin_md,
                )

    # ── Test 13: Platform provider profiles require Platform Admin ───

    @pytest.mark.parametrize(
        "rpc_name,call",
        _platform_profile_rpcs(),
        ids=[r[0] for r in _platform_profile_rpcs()],
    )
    def test_platform_provider_profile_operations_require_platform_admin(
        self,
        rpc_name: str,
        call: Callable,
        user_ctx: Any,
    ) -> None:
        stub, metadata, _ = user_ctx

        with pytest.raises(grpc.RpcError) as exc_info:
            call(stub, metadata)

        assert exc_info.value.code() == grpc.StatusCode.PERMISSION_DENIED, (
            f"{rpc_name}: expected PERMISSION_DENIED, got {exc_info.value.code()}"
        )
        assert "platform admin role required" in exc_info.value.details(), (
            f"{rpc_name}: denial came from the wrong authorization layer: "
            f"{exc_info.value.details()}"
        )

    @pytest.mark.parametrize(
        "rpc_name,call",
        _platform_profile_rpcs(),
        ids=[r[0] for r in _platform_profile_rpcs()],
    )
    def test_platform_admin_can_access_platform_provider_profile_operations(
        self,
        rpc_name: str,
        call: Callable,
        admin_ctx: Any,
    ) -> None:
        stub, metadata = admin_ctx

        try:
            call(stub, metadata)
        except grpc.RpcError as error:
            assert error.code() != grpc.StatusCode.PERMISSION_DENIED, (
                f"{rpc_name}: Platform Admin was denied: {error.details()}"
            )

    # ── Test 14: Global policy reads require Platform Admin ──────────

    @pytest.mark.parametrize(
        "rpc_name,call",
        _global_policy_read_rpcs(),
        ids=[r[0] for r in _global_policy_read_rpcs()],
    )
    def test_global_policy_reads_require_platform_admin(
        self,
        rpc_name: str,
        call: Callable,
        user_ctx: Any,
    ) -> None:
        stub, metadata, _ = user_ctx

        with pytest.raises(grpc.RpcError) as exc_info:
            call(stub, metadata)

        assert exc_info.value.code() == grpc.StatusCode.PERMISSION_DENIED, (
            f"{rpc_name}: expected PERMISSION_DENIED, got {exc_info.value.code()}"
        )
        assert "platform admin role required" in exc_info.value.details(), (
            f"{rpc_name}: denial came from the wrong authorization layer: "
            f"{exc_info.value.details()}"
        )

    @pytest.mark.parametrize(
        "rpc_name,call",
        _global_policy_read_rpcs(),
        ids=[r[0] for r in _global_policy_read_rpcs()],
    )
    def test_platform_admin_can_access_global_policy_reads(
        self,
        rpc_name: str,
        call: Callable,
        admin_ctx: Any,
    ) -> None:
        stub, metadata = admin_ctx

        try:
            call(stub, metadata)
        except grpc.RpcError as error:
            assert error.code() != grpc.StatusCode.PERMISSION_DENIED, (
                f"{rpc_name}: Platform Admin was denied: {error.details()}"
            )