Per-connector resource allowlist (OAuth access control)

enlace_auth/auth/oauth_server.py

@@ -39,7 +39,7 @@
 import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, MutableMapping, Optional
+from typing import Any, Mapping, MutableMapping, Optional
 from urllib.parse import urlencode
 
 from fastapi import APIRouter, Form, Request
@@ -143,14 +143,31 @@ def make_oauth_server_router(
     code_ttl: int = 120,
     scopes_supported: tuple[str, ...] = ("mcp:read",),
     require_consent: bool = True,
+    resource_allowlist: Mapping[str, list[str]] | None = None,
 ) -> APIRouter:
     """Build the OAuth 2.1 authorization-server router (see the module docstring).
 
     *issuer* pins the token ``iss`` and the discovery URLs; when ``None`` it is
     derived from each request's base URL (so the same code serves any domain). The
     consent step reuses the platform session — an unauthenticated ``/authorize``
     redirects to ``/auth/login`` and returns.
+
+    *resource_allowlist* maps a connector resource URL to the emails permitted to
+    authorize for it. A resource **not** in the map is open to any authenticated
+    user (back-compatible); a resource that **is** in the map denies everyone
+    else — per-connector access control (e.g. restrict a private connector to its
+    own staff). Secure to key on the resource: the connector only accepts tokens
+    whose ``aud`` is that exact resource, so a denied user can't get a usable token
+    another way.
     """
+    _allowlist = {
+        r.rstrip("/"): {e.lower() for e in emails}
+        for r, emails in (resource_allowlist or {}).items()
+    }
+
+    def _resource_allowed(resource: str, email: str) -> bool:
+        allowed = _allowlist.get((resource or "").rstrip("/"))
+        return allowed is None or email.lower() in allowed
     router = APIRouter(tags=["oauth-server"])
 
     def _issuer(request: Request) -> str:
@@ -317,6 +334,9 @@ async def authorize(request: Request):
                 f"/auth/login?{urlencode({'next': here})}", status_code=302
             )
 
+        if not _resource_allowed(auth.resource, email):
+            return HTMLResponse(_denied_page(email), status_code=403)
+
         if not require_consent:
             code = _issue_code(auth, email)
             return RedirectResponse(
@@ -351,6 +371,8 @@ async def authorize_consent(
         auth = _Authorized(
             client_id, redirect_uri, code_challenge, state, scope, resource
         )
+        if not _resource_allowed(resource, email):
+            return _redirect_error(redirect_uri, "access_denied", state)
         if decision != "approve":
             return _redirect_error(redirect_uri, "access_denied", state)
         code = _issue_code(auth, email)
@@ -413,6 +435,16 @@ async def token(
     return router
 
 
+def _denied_page(email: str) -> str:
+    """Render the 'not authorized for this connector' page (allowlist denial)."""
+    return pages._page(
+        "Access denied",
+        f"<h1>Access denied</h1><p><strong>{email}</strong> is not authorized to "
+        "use this connector. Contact the connector owner if you believe this is "
+        "a mistake.</p>",
+    )
+
+
 def _consent_page(
     request: Request, auth: _Authorized, email: str, signing_key: str
 ) -> str:

enlace_auth/config.py

@@ -61,6 +61,14 @@ class OAuthServerConfig(BaseModel):
     code_ttl_seconds: int = 120
     scopes_supported: list[str] = Field(default_factory=lambda: ["mcp:read"])
     require_consent: bool = True
+    resource_allowlist: dict[str, list[str]] = Field(
+        default_factory=dict,
+        description=(
+            "Per-connector access control: maps a connector resource URL to the "
+            "emails allowed to authorize for it. A resource not listed is open to "
+            "any authenticated user; a listed resource denies everyone else."
+        ),
+    )
 
 
 class AuthConfig(BaseModel):

enlace_auth/plugin.py

@@ -299,6 +299,7 @@ def wire(parent: "FastAPI", config) -> None:
                 code_ttl=osc.code_ttl_seconds,
                 scopes_supported=tuple(osc.scopes_supported),
                 require_consent=osc.require_consent,
+                resource_allowlist=osc.resource_allowlist,
             )
             parent.include_router(oauth_server_router)
         except ImportError:

tests/test_oauth_server.py

@@ -38,7 +38,7 @@ def _pkce():
     return verifier, challenge
 
 
-def _build(tmp_path, *, require_consent=True):
+def _build(tmp_path, *, require_consent=True, resource_allowlist=None):
     session_store = SessionStore({})
     sid = session_store.create(user_id=EMAIL, email=EMAIL)
     router = make_oauth_server_router(
@@ -51,6 +51,7 @@ def _build(tmp_path, *, require_consent=True):
         keys=OAuthKeys(tmp_path / "keys"),
         issuer=None,  # derive from request → http://testserver
         require_consent=require_consent,
+        resource_allowlist=resource_allowlist,
     )
     app = FastAPI()
     app.include_router(router)
@@ -237,3 +238,34 @@ def test_token_rejects_wrong_pkce_verifier(tmp_path):
     )
     assert tok.status_code == 400
     assert tok.json()["error"] == "invalid_grant"
+
+
+def _authorize_params(cid, challenge):
+    return {
+        "response_type": "code", "client_id": cid, "redirect_uri": REDIRECT,
+        "code_challenge": challenge, "code_challenge_method": "S256",
+        "resource": RESOURCE,
+    }
+
+
+def test_resource_allowlist_denies_unlisted_user(tmp_path):
+    client, cookie = _build(tmp_path, resource_allowlist={RESOURCE: ["other@x.com"]})
+    cid = _register(client)
+    _, challenge = _pkce()
+    r = client.get(
+        "/auth/oauth/authorize", params=_authorize_params(cid, challenge),
+        cookies={COOKIE: cookie},
+    )
+    assert r.status_code == 403
+    assert "Access denied" in r.text
+
+
+def test_resource_allowlist_allows_listed_user(tmp_path):
+    client, cookie = _build(tmp_path, resource_allowlist={RESOURCE: [EMAIL]})
+    cid = _register(client)
+    _, challenge = _pkce()
+    r = client.get(
+        "/auth/oauth/authorize", params=_authorize_params(cid, challenge),
+        cookies={COOKIE: cookie},
+    )
+    assert r.status_code == 200 and "Approve" in r.text