"""Bind opaque Gradio session ids to authenticated Hugging Face identities. **The trap this module exists to close.** Gradio ships OAuth helpers, and off a Space they are mocked: `gradio.oauth._add_mocked_oauth_routes` signs the visitor in as a fake profile without a single call to hf.co, and warns in a way nobody reads. A server that puts `gr.LoginButton` on the page and trusts the resulting profile has, when self-hosted, no authentication at all while looking exactly like it has some. `refuse_mocked_oauth` is what stops this project shipping that, and it fails closed at start-up rather than per request, because a server that has already served one page has already leaked. Two halves live here: * the small, original one that turns a profile into a subject and stops one browser session being used by two identities, and * `OpenIdProvider`, which is the ordinary OAuth 2.0 authorisation code exchange written out, so that a self-hosted server has a real way to sign people in rather than the mocked one. **What is checked in the exchange, and why each one is there.** * `state` is generated per attempt and compared on return, so a callback the user did not start is refused. Without it a third party can walk somebody through a login to an account that is not theirs. * The token exchange happens server side with the client secret, so a code intercepted in a URL is not enough on its own. * The identity comes from the provider's userinfo response, never from anything the browser sent. """ from __future__ import annotations import base64 import hashlib import json import os import secrets import sys import threading import urllib.error import urllib.parse import urllib.request from collections.abc import Mapping as MappingABC from dataclasses import dataclass from typing import Any, Mapping, Optional class LoginRequired(PermissionError): pass def user_id_from_profile(profile: Optional[Mapping[str, Any]]) -> str: """The subject a profile names, or a refusal. Deliberately a pure mapping. Whether a profile is worth believing at all is a question about how the server was started, not about this dictionary, and it is answered once by `refuse_mocked_oauth` before anything is served. Putting it here as well would spread one decision across every request and still not stop the first one. """ if profile is not None: subject = profile.get("sub") or profile.get("preferred_username") if isinstance(subject, str) and subject.strip(): return subject.strip() # Development auth is never accepted when any real OAuth app is # configured. It used to check only `OAUTH_CLIENT_ID`, which is the # variable a *Space* injects, so a self-hosted server configured through # `DISTINCT_OAUTH_CLIENT_ID` with `DISTINCT_DEV_AUTH` still set in the # environment turned its real sign-in off and collapsed everybody into one # shared user. Both spellings are checked now, because both mean the same # thing: somebody configured a real login and meant it. if dev_auth_permitted(): return "local-development-user" raise LoginRequired("Sign in with Hugging Face to use Distinct") class SessionAuthorizer: """A process-local tenant boundary for otherwise client-held state.""" def __init__(self) -> None: self._owners: dict[str, str] = {} self._lock = threading.RLock() def bind(self, session_id: str, user_id: str) -> None: with self._lock: current = self._owners.get(session_id) if current is not None and current != user_id: raise LoginRequired("session belongs to another user") self._owners[session_id] = user_id def require(self, session_id: str, user_id: str) -> None: with self._lock: if self._owners.get(session_id) != user_id: raise LoginRequired("session is not authorized for this user") def release(self, session_id: str) -> None: with self._lock: self._owners.pop(session_id, None) #: Where the provider describes itself. Everything else is discovered from #: here rather than written down, because Hugging Face publishes #: `{OPENID_PROVIDER_URL}/.well-known/openid-configuration` and a hard-coded #: endpoint is a thing that goes stale without anybody noticing. DEFAULT_PROVIDER_URL = "https://huggingface.co" DISCOVERY_PATH = "/.well-known/openid-configuration" #: The endpoints used if discovery cannot be reached. Documented at #: https://huggingface.co/docs/hub/en/oauth and kept only as a fallback. HUGGINGFACE = { "authorization_endpoint": "https://huggingface.co/oauth/authorize", "token_endpoint": "https://huggingface.co/oauth/token", "userinfo_endpoint": "https://huggingface.co/oauth/userinfo", } #: `openid profile` is always included by Hugging Face and is all an identity #: needs: `openid` returns the ID token, `profile` the username. Every wider #: scope in their table grants access to somebody's repositories, billing or #: inference quota, and asking for one of those in order to learn a name would #: be asking for far more than this server has any use for. DEFAULT_SCOPES = "openid profile" #: What a Space injects, named exactly as Hugging Face names them. Read first, #: so the same code runs on a Space and self-hosted with no branch. ENV_SPACE_CLIENT_ID = "OAUTH_CLIENT_ID" ENV_SPACE_CLIENT_SECRET = "OAUTH_CLIENT_SECRET" ENV_SPACE_SCOPES = "OAUTH_SCOPES" ENV_SPACE_PROVIDER_URL = "OPENID_PROVIDER_URL" ENV_SPACE_HOST = "SPACE_HOST" #: And what a self-hosted deployment sets instead. ENV_CLIENT_ID = "DISTINCT_OAUTH_CLIENT_ID" ENV_CLIENT_SECRET = "DISTINCT_OAUTH_CLIENT_SECRET" ENV_REDIRECT_URI = "DISTINCT_OAUTH_REDIRECT_URI" ENV_PROVIDER_URL = "DISTINCT_OAUTH_PROVIDER_URL" ENV_SCOPES = "DISTINCT_OAUTH_SCOPES" ENV_PROVIDER_NAME = "DISTINCT_OAUTH_PROVIDER" ENV_DEV_AUTH = "DISTINCT_DEV_AUTH" #: The path a Space's OAuth app is registered against. Fixed by Hugging Face: #: https://huggingface.co/docs/hub/en/spaces-oauth SPACE_REDIRECT_PATH = "/login/callback" MAX_RESPONSE_BYTES = 256 * 1024 TIMEOUT_S = 20.0 class IdentityError(RuntimeError): """A sign-in did not produce an identity anybody should believe.""" class NotConfigured(IdentityError): """Sign-in is not set up, so the server must not pretend it is.""" @dataclass(frozen=True) class Identity: """One signed-in person, as the provider describes them.""" provider: str subject: str username: str name: str = "" def __post_init__(self) -> None: for label, value in (("provider", self.provider), ("subject", self.subject)): if not isinstance(value, str) or not value.strip(): raise IdentityError(f"{label} is required on an identity") if "\n" in value or "\r" in value: raise IdentityError(f"{label} must be a single line") @property def viewer_id(self) -> str: """The one string the rest of the server keys access on. Provider-qualified, so two providers cannot collide on a subject and quietly become the same person. """ return f"{self.provider}:{self.subject}" @property def label(self) -> str: """What to put on the screen. The username, never the subject.""" return self.username or self.subject @dataclass(frozen=True) class PendingSignIn: """The three secrets one sign-in attempt has to remember. All three are per attempt and none may be reused. They are returned together rather than generated inside the redirect because the caller has to put them in the user's session and hand them back at the callback: `state` proves the callback belongs to this attempt, `nonce` proves the ID token does, and `code_verifier` proves the code is being redeemed by whoever asked for it. """ url: str state: str nonce: str code_verifier: str def new_state() -> str: """A fresh, unguessable value to tie a callback to the attempt.""" return secrets.token_urlsafe(24) def _code_verifier() -> str: """A PKCE verifier, in the 43 to 128 character range RFC 7636 allows.""" return secrets.token_urlsafe(64) def _code_challenge(verifier: str) -> str: """The S256 challenge: base64url of the verifier's sha256, unpadded.""" digest = hashlib.sha256(verifier.encode("ascii")).digest() return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=") #: What Hugging Face sets on a real Space, beside `SPACE_ID`. Gradio requires #: it, and this project used to check only `SPACE_ID`, which is the bug the #: function below is written around. ENV_SYSTEM = "SYSTEM" SPACES_SYSTEM = "spaces" def dev_auth_permitted(environ: Optional[Mapping[str, str]] = None) -> bool: """Whether the development escape hatch is allowed to answer here. Only when it is asked for, and only when nothing real is configured. A server with a real OAuth app has an operator who meant to have sign-in, and an environment variable left over from a local experiment must not be able to turn that off. """ environ = os.environ if environ is None else environ if environ.get(ENV_DEV_AUTH) != "1": return False return not (environ.get(ENV_SPACE_CLIENT_ID) or environ.get(ENV_CLIENT_ID)) def gradio_oauth_is_real(environ: Optional[Mapping[str, str]] = None) -> bool: """Whether Gradio's own login routes will be real rather than mocked. **This exists because guessing the answer was wrong.** Gradio decides with `gradio.utils.get_space()`, which returns the Space id only when ``SYSTEM == "spaces"``; `SPACE_ID` alone is not enough. This project checked `SPACE_ID` alone, so an environment carrying `SPACE_ID` without `SYSTEM` slipped through the refusal below **and** was mocked by Gradio: every visitor signed in as a fake profile on a server that believed itself to be a real Space. That is precisely the state this refusal exists to prevent, reached by the one route the refusal did not look at. The condition is now the same one Gradio uses, plus the client id its `_add_oauth_routes` requires, so the two cannot disagree about what deployment this is. `tests/test_signin_boundary.py` checks this function against Gradio's own `get_space` rather than against a copy of it. """ environ = os.environ if environ is None else environ if environ.get(ENV_SYSTEM) != SPACES_SYSTEM: return False if not environ.get("SPACE_ID"): return False return bool(environ.get(ENV_SPACE_CLIENT_ID)) def refuse_mocked_oauth(environ: Optional[Mapping[str, str]] = None) -> None: """Fail closed where Gradio would sign everybody in as a fake user. Gradio mocks its login unless this is a real Space, and its mock signs every visitor in as a fake profile without contacting Hugging Face at all. That is a reasonable default for somebody debugging a demo and a catastrophic one for a server whose whole access model rests on knowing who is asking, so this project refuses to run in that state rather than inheriting it. """ environ = os.environ if environ is None else environ if gradio_oauth_is_real(environ): return # Either the Space variable or this project's own is enough: both mean a # real OAuth app exists and Gradio's mock is not what is answering. if environ.get(ENV_SPACE_CLIENT_ID) or environ.get(ENV_CLIENT_ID): return if environ.get(ENV_DEV_AUTH) == "1": # An explicit local escape hatch, and loud about what it costs. It is # only reachable when nothing real is configured, so it cannot be the # thing that quietly happens on a production box. print( "distinct: DISTINCT_DEV_AUTH is set, so this server has NO sign-in. " "Everyone who can reach it is the same user. Never do this where " "anything other than loopback can reach the port.", file=sys.stderr, flush=True, ) return raise NotConfigured( "Sign-in is not configured, and Gradio would mock it: off a Space it " "signs every visitor in as a fake profile without contacting Hugging " f"Face at all. Set {ENV_CLIENT_ID}, {ENV_CLIENT_SECRET} and " f"{ENV_REDIRECT_URI} from an OAuth app at " "https://huggingface.co/settings/applications, or run on a Space with " "hf_oauth: true in its README." ) class NoRedirects(urllib.request.HTTPRedirectHandler): """Refuse every redirect on the OAuth calls, because of what they carry. **This was a credential leak, demonstrated rather than suspected.** The token request sends the OAuth client secret in an HTTP Basic header and the userinfo request sends the access token in a Bearer header, and the default opener follows redirects. CPython's `urllib` forwards `Authorization` and `Cookie` across a cross-host 301, 302 or 303. So a provider that answered the token endpoint with a redirect, or a discovery document naming an endpoint that then redirects, would hand this server's client secret to whatever host it named. That needs a hostile `OPENID_PROVIDER_URL` or a compromised provider, which is not the common case. It is also a total loss when it happens: the client secret is the thing that lets somebody else be this application. A redirect these calls should follow does not exist. The endpoints come from the provider's own discovery document, which is where a provider says where its endpoints are; a provider that wants to move them edits that document rather than bouncing a request carrying a secret. """ def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102 return None class OpenIdProvider: """Hugging Face's OAuth, as https://huggingface.co/docs/hub/en/oauth has it. Four things here are theirs rather than mine, and each was wrong when this was written from the OAuth 2.0 specification instead of from their page: * **Endpoints are discovered**, from `{provider_url}/.well-known/openid-configuration`, which is what `OPENID_PROVIDER_URL` points at on a Space. Hard-coded URLs are kept only as a fallback for when discovery cannot be reached. * **The client authenticates with HTTP Basic** at the token endpoint, which their documentation recommends for apps holding a secret. Form parameters are the documented path for public apps with no secret, and that is when this uses them. * **PKCE**, which they support and recommend. The verifier never leaves this server, so an authorisation code seen in a redirect, a proxy log or a browser history cannot be redeemed by whoever found it. * **`nonce`**, which they put in the ID token. Checking it back is what binds that token to this attempt. The ID token's signature is not verified, and that is a deliberate reading of OpenID Connect Core 3.1.3.7 rather than an omission: it arrives in the response to a request this server made directly to the token endpoint over TLS, so the channel already establishes the issuer. The identity is taken from userinfo regardless; the ID token is only read to check the nonce. """ def __init__( self, *, client_id: str, client_secret: str = "", redirect_uri: str, provider_url: str = DEFAULT_PROVIDER_URL, scopes: str = DEFAULT_SCOPES, name: str = "huggingface", endpoints: Optional[Mapping[str, str]] = None, opener: Any = None, ) -> None: for label, value in (("client id", client_id), ("redirect URI", redirect_uri)): if not isinstance(value, str) or not value.strip(): raise NotConfigured(f"the OAuth {label} is required") self.client_id = client_id # Empty is legitimate: their documentation calls an app with no secret # a public app, and PKCE is what carries the proof in that case. self.client_secret = client_secret or "" self.redirect_uri = redirect_uri self.provider_url = provider_url.rstrip("/") self.scopes = scopes self.name = name self._opener = opener or urllib.request.build_opener(NoRedirects) self._endpoints = dict(endpoints) if endpoints else None # -- discovery ----------------------------------------------------- def endpoints(self) -> Mapping[str, str]: """The provider's own description of itself, fetched once. A provider that cannot be reached falls back to the documented Hugging Face URLs rather than refusing: discovery being down is not a reason to stop signing people in, and the fallback is the same host. """ if self._endpoints is None: try: discovered = self._json( urllib.request.Request( self.provider_url + DISCOVERY_PATH, headers={"Accept": "application/json"}, method="GET", ), "provider discovery", ) self._endpoints = { key: str(discovered[key]) for key in ( "authorization_endpoint", "token_endpoint", "userinfo_endpoint", ) if isinstance(discovered.get(key), str) } except IdentityError: self._endpoints = {} for key, fallback in HUGGINGFACE.items(): self._endpoints.setdefault(key, fallback) return self._endpoints # -- the two halves of the flow ------------------------------------ def begin(self) -> PendingSignIn: """Where to send the browser, and what to remember while it is gone.""" state = new_state() nonce = new_state() verifier = _code_verifier() query = urllib.parse.urlencode( { "client_id": self.client_id, "redirect_uri": self.redirect_uri, "response_type": "code", "scope": self.scopes, "state": state, "nonce": nonce, "code_challenge": _code_challenge(verifier), "code_challenge_method": "S256", } ) authorize = self.endpoints()["authorization_endpoint"] return PendingSignIn(f"{authorize}?{query}", state, nonce, verifier) def complete( self, code: str, *, state: str, pending: PendingSignIn, ) -> Identity: """Turn a callback into an identity, or refuse to. The state comparison happens before the network call, so a callback nobody started costs nothing and tells the caller nothing. """ if not isinstance(code, str) or not code.strip(): raise IdentityError("the provider returned no authorisation code") if not isinstance(state, str) or not isinstance(pending, PendingSignIn): raise IdentityError("a sign-in must carry a state value") if not secrets.compare_digest(state, pending.state): raise IdentityError( "that sign-in did not start here, so it is refused. Start again " "from the sign-in button on this page." ) payload = self._token(code, pending.code_verifier) self._check_nonce(payload, pending.nonce) token = payload.get("access_token") if not isinstance(token, str) or not token: raise IdentityError("the provider's token response carried no access token") return self._identity(token) # -- the calls ----------------------------------------------------- def _token(self, code: str, verifier: str) -> Mapping[str, Any]: form = { "grant_type": "authorization_code", "code": code, "redirect_uri": self.redirect_uri, "code_verifier": verifier, } headers = { "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json", } if self.client_secret: # Basic, which is what their documentation recommends for an app # that holds a secret. The secret never appears in a form body and # so never lands in a log that captured one. pair = f"{self.client_id}:{self.client_secret}".encode("utf-8") headers["Authorization"] = "Basic " + base64.b64encode(pair).decode("ascii") else: # A public app: the client id goes in the body and PKCE is what # proves the redemption is the one that asked. form["client_id"] = self.client_id return self._json( urllib.request.Request( self.endpoints()["token_endpoint"], data=urllib.parse.urlencode(form).encode("utf-8"), headers=headers, method="POST", ), "the token exchange", ) def _check_nonce(self, payload: Mapping[str, Any], nonce: str) -> None: """The ID token must answer this attempt, if there is one at all. Read, not verified: see the class docstring for why the signature is not checked and what is relied on instead. A missing ID token is not an error, because `openid` is a scope and not a guarantee, but one carrying somebody else's nonce is. """ raw = payload.get("id_token") if not isinstance(raw, str) or raw.count(".") != 2: return body = raw.split(".")[1] try: padded = body + "=" * (-len(body) % 4) claims = json.loads(base64.urlsafe_b64decode(padded).decode("utf-8")) except (ValueError, UnicodeDecodeError): raise IdentityError("the provider's ID token could not be read") from None if not isinstance(claims, MappingABC): raise IdentityError("the provider's ID token was not an object") found = claims.get("nonce") if isinstance(found, str) and not secrets.compare_digest(found, nonce): raise IdentityError( "the provider's ID token answers a different sign-in; refusing it" ) def _identity(self, token: str) -> Identity: payload = self._json( urllib.request.Request( self.endpoints()["userinfo_endpoint"], headers={"Authorization": f"Bearer {token}", "Accept": "application/json"}, method="GET", ), "the identity lookup", ) subject = payload.get("sub") if not isinstance(subject, str) or not subject: raise IdentityError("the provider did not say who this is") username = payload.get("preferred_username") or payload.get("name") or subject return Identity( provider=self.name, subject=subject, username=str(username)[:80], name=str(payload.get("name") or "")[:120], ) def _json(self, request: urllib.request.Request, what: str) -> Mapping[str, Any]: try: with self._opener.open(request, timeout=TIMEOUT_S) as response: raw = response.read(MAX_RESPONSE_BYTES + 1) except urllib.error.HTTPError as error: raise IdentityError(f"{what} was refused with HTTP {error.code}") from error except OSError as error: raise IdentityError( f"{what} could not reach the provider: {type(error).__name__}" ) from error if len(raw) > MAX_RESPONSE_BYTES: raise IdentityError(f"{what} returned an oversized response") try: payload = json.loads(raw.decode("utf-8")) except ValueError as error: raise IdentityError(f"{what} did not return JSON") from error if not isinstance(payload, MappingABC): raise IdentityError(f"{what} did not return an object") return payload def from_environment( environ: Optional[Mapping[str, str]] = None, *, opener: Any = None, ) -> OpenIdProvider: """Build the provider from whichever set of variables is present. A Space injects `OAUTH_CLIENT_ID`, `OAUTH_CLIENT_SECRET`, `OAUTH_SCOPES` and `OPENID_PROVIDER_URL`, and its redirect URI is fixed at `https://{SPACE_HOST}/login/callback`. Those are read first, so the same code runs there and self-hosted without a branch anywhere else. The refusal names the variables rather than describing them, so the fix is copy and paste. """ environ = os.environ if environ is None else environ client_id = environ.get(ENV_SPACE_CLIENT_ID) or environ.get(ENV_CLIENT_ID) or "" client_secret = ( environ.get(ENV_SPACE_CLIENT_SECRET) or environ.get(ENV_CLIENT_SECRET) or "" ) redirect_uri = environ.get(ENV_REDIRECT_URI) or "" if not redirect_uri and environ.get(ENV_SPACE_HOST): redirect_uri = f"https://{environ[ENV_SPACE_HOST]}{SPACE_REDIRECT_PATH}" if not client_id or not redirect_uri: raise NotConfigured( "Sign-in is not configured. On a Space, set `hf_oauth: true` in the " "README and Hugging Face injects " f"{ENV_SPACE_CLIENT_ID}, {ENV_SPACE_CLIENT_SECRET}, " f"{ENV_SPACE_SCOPES} and {ENV_SPACE_PROVIDER_URL}. Self-hosted, " "create an OAuth app at https://huggingface.co/settings/applications " f"and set {ENV_CLIENT_ID}, {ENV_CLIENT_SECRET} and " f"{ENV_REDIRECT_URI} to its values." ) return OpenIdProvider( client_id=client_id, client_secret=client_secret, redirect_uri=redirect_uri, provider_url=( environ.get(ENV_SPACE_PROVIDER_URL) or environ.get(ENV_PROVIDER_URL) or DEFAULT_PROVIDER_URL ), scopes=environ.get(ENV_SPACE_SCOPES) or environ.get(ENV_SCOPES) or DEFAULT_SCOPES, name=environ.get(ENV_PROVIDER_NAME, "huggingface"), opener=opener, )