feat: authorize services by OIDC groups

Author: luketainton

.env.example

@@ -17,7 +17,14 @@ MCP_AUTH_OIDC_CLIENT_ID=
 MCP_AUTH_OIDC_CLIENT_SECRET=
 MCP_AUTH_OIDC_REDIRECT_PATH=/auth/callback
 MCP_AUTH_OIDC_JWT_SIGNING_KEY=
-MCP_AUTH_REQUIRED_SCOPES=
+# Request groups from Pocket ID to use service authorization.
+MCP_AUTH_REQUIRED_SCOPES=openid,profile,groups
+# Claim holding a user's groups or roles. Pocket ID uses groups by default.
+MCP_AUTH_GROUP_CLAIM=groups
+# JSON mapping: service keys (gitea, pocket_id) to Pocket ID OIDC group/role names.
+# A mapped service requires membership of any listed Pocket ID group or role.
+# MCP_SERVICE_GROUPS={"gitea":["mcp-gitea"],"pocket_id":["mcp-pocket-id"]}
+MCP_SERVICE_GROUPS={}
 
 # Use jwt when clients already have a Pocket ID JWT and only need token verification.
 # For Pocket ID, issuer defaults to POCKET_ID_URL.

README.md

@@ -47,7 +47,7 @@ MCP_AUTH_BASE_URL=https://labmcp.example.com
 MCP_AUTH_OIDC_CLIENT_ID=<pocket-id-client-id>
 MCP_AUTH_OIDC_CLIENT_SECRET=<pocket-id-client-secret>
 MCP_AUTH_OIDC_JWT_SIGNING_KEY=<stable-random-secret>
-MCP_AUTH_REQUIRED_SCOPES=openid,profile
+MCP_AUTH_REQUIRED_SCOPES=openid,profile,groups
 ```
 
 By default, `MCP_AUTH_OIDC_CONFIG_URL` is derived as `<POCKET_ID_URL>/.well-known/openid-configuration`, and the callback path is `/auth/callback`. Set them explicitly if your Pocket ID issuer or reverse proxy path differs from `POCKET_ID_URL`.
@@ -67,6 +67,20 @@ MCP_AUTH_JWT_AUDIENCE=<pocket-id-client-id>
 
 In `jwt` mode, `MCP_AUTH_JWT_ISSUER` defaults to `POCKET_ID_URL`, and `MCP_AUTH_JWT_JWKS_URI` defaults to `<issuer>/.well-known/jwks.json`. If you configure `MCP_AUTH_REQUIRED_SCOPES`, provide a comma-separated list of scopes that must be present in the token.
 
+### Service access by Pocket ID group or role
+
+Tools can be made visible only to users in specific Pocket ID groups. Request Pocket ID's `groups` scope and map each service to one or more permitted groups:
+
+```sh
+MCP_AUTH_REQUIRED_SCOPES=openid,profile,groups
+MCP_AUTH_GROUP_CLAIM=groups
+MCP_SERVICE_GROUPS={"gitea":["mcp-gitea"],"pocket_id":["mcp-pocket-id"]}
+```
+
+Membership of any group listed for a service permits access to that service's tools. The Gitea tools use the `gitea` service key; all Pocket ID tools use `pocket_id`. Once `MCP_SERVICE_GROUPS` is non-empty, any omitted service is denied. Leave it as `{}` only when every authenticated user should have access to all services.
+
+Pocket ID normally provides groups in the `groups` claim. To authorize from a custom role claim instead, configure that custom claim on the relevant Pocket ID groups, set `MCP_AUTH_GROUP_CLAIM` to its name, and use its values in `MCP_SERVICE_GROUPS`.
+
 ## CI and code quality
 
 Pull requests run `.gitea/workflows/ci.yml`, which installs dependencies with `uv`, runs Ruff, executes tests through Coverage.py, writes `testresults.xml` and `coverage.xml`, prints the coverage summary, compiles the Python sources, and builds a Docker image.

pyproject.toml

@@ -5,7 +5,7 @@ description = "A unified Model Context Protocol server for a home lab"
 readme = "README.md"
 requires-python = ">=3.14"
 dependencies = [
-    "fastmcp>=3.0,<4.0",
+    "fastmcp>=3.2.1,<4.0",
     "httpx>=0.27,<1.0",
     "pydantic-settings>=2.0,<3.0",
 ]

src/labmcp/authorization.py

@@ -0,0 +1,35 @@
+from collections.abc import Callable, Mapping, Sequence
+from typing import Any
+
+from .config import Settings
+
+AuthCheck = Callable[[Any], bool]
+
+
+def require_service_access(service: str, settings: Settings) -> AuthCheck | None:
+    """Create a per-service authorization check from the configured claim mapping.
+
+    Without a mapping, every authenticated user may use every service. Once a mapping
+    is configured, each service requires membership of at least one listed group.
+    """
+    if not settings.mcp_service_groups:
+        return None
+    allowed_groups = settings.mcp_service_groups.get(service, [])
+
+    def check(context: Any) -> bool:
+        token = context.token
+        if token is None:
+            return False
+        user_groups = _claim_values(token.claims, settings.mcp_auth_group_claim)
+        return bool(user_groups.intersection(allowed_groups))
+
+    return check
+
+
+def _claim_values(claims: Mapping[str, Any], claim_name: str) -> set[str]:
+    value = claims.get(claim_name)
+    if isinstance(value, str):
+        return {value}
+    if isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray)):
+        return {item for item in value if isinstance(item, str)}
+    return set()

src/labmcp/config.py

@@ -1,7 +1,7 @@
 from functools import lru_cache
 from typing import Literal
 
-from pydantic import SecretStr
+from pydantic import Field, SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -26,6 +26,8 @@ class Settings(BaseSettings):
     mcp_auth_jwt_audience: str | None = None
     mcp_auth_required_scopes: str | None = None
     mcp_auth_jwt_required_scopes: str | None = None
+    mcp_auth_group_claim: str = "groups"
+    mcp_service_groups: dict[str, list[str]] = Field(default_factory=dict)
     mcp_auth_oidc_config_url: str | None = None
     mcp_auth_oidc_client_id: str | None = None
     mcp_auth_oidc_client_secret: SecretStr | None = None

src/labmcp/server.py

@@ -3,6 +3,7 @@
 from fastmcp import FastMCP
 
 from .auth import create_auth_provider, ensure_network_transport_is_authenticated
+from .authorization import require_service_access
 from .clients import gitea_client, pocket_id_client
 from .config import get_settings
 from .version import get_version
@@ -11,13 +12,17 @@
 mcp = FastMCP(f"Home Lab ({get_version()})", auth=create_auth_provider(_settings))
 
 
+def _service_auth(service: str):
+    return require_service_access(service, _settings)
+
+
 @mcp.tool()
 def labmcp_get_version() -> str:
     """Return the running labmcp package version."""
     return get_version()
 
 
-@mcp.tool()
+@mcp.tool(auth=_service_auth("gitea"))
 async def gitea_list_repositories(
     page: int = 1,
     limit: int = 50,
@@ -34,13 +39,13 @@ async def gitea_list_repositories(
     return result
 
 
-@mcp.tool()
+@mcp.tool(auth=_service_auth("gitea"))
 async def gitea_get_repository(owner: str, repo: str) -> dict[str, Any]:
     """Get metadata for one Gitea repository."""
     return await gitea_client(get_settings()).request("GET", f"/api/v1/repos/{owner}/{repo}")
 
 
-@mcp.tool()
+@mcp.tool(auth=_service_auth("gitea"))
 async def gitea_list_issues(
     owner: str,
     repo: str,
@@ -60,7 +65,7 @@ async def gitea_list_issues(
     )
 
 
-@mcp.tool()
+@mcp.tool(auth=_service_auth("gitea"))
 async def gitea_create_issue(owner: str, repo: str, title: str, body: str = "") -> dict[str, Any]:
     """Create an issue in a Gitea repository."""
     if not title.strip():
@@ -70,22 +75,22 @@ async def gitea_create_issue(owner: str, repo: str, title: str, body: str = "")
     )
 
 
-@mcp.tool()
+@mcp.tool(auth=_service_auth("pocket_id"))
 async def pocket_id_openid_configuration() -> dict[str, Any]:
     """Read Pocket ID's OpenID Connect discovery document."""
     return await pocket_id_client(get_settings()).request(
         "GET", "/.well-known/openid-configuration"
     )
 
 
-@mcp.tool()
+@mcp.tool(auth=_service_auth("pocket_id"))
 async def pocket_id_health() -> Any:
     """Check Pocket ID health using POCKET_ID_HEALTH_PATH (default: /api/health)."""
     settings = get_settings()
     return await pocket_id_client(settings).request("GET", settings.pocket_id_health_path)
 
 
-@mcp.tool()
+@mcp.tool(auth=_service_auth("pocket_id"))
 async def pocket_id_get_json(path: str) -> Any:
     """GET a JSON endpoint on Pocket ID, useful for deployments with an enabled API."""
     if not path.startswith("/") or path.startswith("//"):

tests/test_authorization.py

@@ -0,0 +1,43 @@
+from types import SimpleNamespace
+
+from labmcp.authorization import require_service_access
+from labmcp.config import Settings
+
+
+def _context(claims: dict[str, object]) -> SimpleNamespace:
+    return SimpleNamespace(token=SimpleNamespace(claims=claims))
+
+
+def test_empty_mapping_does_not_add_a_group_restriction() -> None:
+    settings = Settings()
+
+    assert require_service_access("pocket_id", settings) is None
+
+
+def test_unmapped_service_is_denied_when_group_mapping_is_enabled() -> None:
+    settings = Settings(mcp_service_groups={"gitea": ["mcp-gitea"]})
+    check = require_service_access("pocket_id", settings)
+
+    assert check is not None
+    assert check(_context({"groups": ["mcp-pocket-id"]})) is False
+
+
+def test_service_group_access_requires_an_allowed_group() -> None:
+    settings = Settings(mcp_service_groups={"gitea": ["mcp-gitea", "mcp-admin"]})
+    check = require_service_access("gitea", settings)
+
+    assert check is not None
+    assert check(_context({"groups": ["mcp-gitea"]})) is True
+    assert check(_context({"groups": ["mcp-pocket-id"]})) is False
+    assert check(SimpleNamespace(token=None)) is False
+
+
+def test_service_group_access_supports_a_custom_role_claim() -> None:
+    settings = Settings(
+        mcp_auth_group_claim="roles",
+        mcp_service_groups={"pocket_id": ["administrator"]},
+    )
+    check = require_service_access("pocket_id", settings)
+
+    assert check is not None
+    assert check(_context({"roles": "administrator"})) is True