From 8adaa79f76337b88a3aad66125799422b2e4665e Mon Sep 17 00:00:00 2001 From: David Pastl Date: Mon, 6 Jul 2026 14:47:41 -0600 Subject: [PATCH 01/11] Updating authenticators --- bluesky_httpserver/authenticators.py | 336 ++++++++++++++++++++++++--- 1 file changed, 302 insertions(+), 34 deletions(-) diff --git a/bluesky_httpserver/authenticators.py b/bluesky_httpserver/authenticators.py index a58fedf..7448d0c 100644 --- a/bluesky_httpserver/authenticators.py +++ b/bluesky_httpserver/authenticators.py @@ -4,9 +4,10 @@ import logging import re import secrets +import uuid from collections.abc import Iterable from datetime import timedelta -from typing import Any, List, Mapping, Optional, cast +from typing import Any, Dict, List, Mapping, Optional, cast import httpx from cachetools import TTLCache, cached @@ -63,11 +64,15 @@ class DictionaryAuthenticator(InternalAuthenticator): description: May be displayed by client after successful login. """ - def __init__(self, users_to_passwords: Mapping[str, str], confirmation_message: str = ""): + def __init__( + self, users_to_passwords: Mapping[str, str], confirmation_message: str = "" + ): self._users_to_passwords = users_to_passwords self.confirmation_message = confirmation_message - async def authenticate(self, username: str, password: str) -> Optional[UserSessionState]: + async def authenticate( + self, username: str, password: str + ) -> Optional[UserSessionState]: true_password = self._users_to_passwords.get(username) if not true_password: # Username is not valid. @@ -92,12 +97,16 @@ class PAMAuthenticator(InternalAuthenticator): def __init__(self, service: str = "login", confirmation_message: str = ""): if not modules_available("pamela"): - raise ModuleNotFoundError("This PAMAuthenticator requires the module 'pamela' to be installed.") + raise ModuleNotFoundError( + "This PAMAuthenticator requires the module 'pamela' to be installed." + ) self.service = service self.confirmation_message = confirmation_message # TODO Try to open a PAM session. - async def authenticate(self, username: str, password: str) -> Optional[UserSessionState]: + async def authenticate( + self, username: str, password: str + ) -> Optional[UserSessionState]: import pamela try: @@ -179,17 +188,21 @@ def token_endpoint(self) -> str: @functools.cached_property def authorization_endpoint(self) -> httpx.URL: - return httpx.URL(cast(str, self._config_from_oidc_url.get("authorization_endpoint"))) + return httpx.URL( + cast(str, self._config_from_oidc_url.get("authorization_endpoint")) + ) @functools.cached_property def device_authorization_endpoint(self) -> str: - return cast(str, self._config_from_oidc_url.get("device_authorization_endpoint")) + return cast( + str, self._config_from_oidc_url.get("device_authorization_endpoint") + ) @functools.cached_property def end_session_endpoint(self) -> str: return cast(str, self._config_from_oidc_url.get("end_session_endpoint")) - @cached(TTLCache(maxsize=1, ttl=timedelta(days=7).total_seconds())) + @cached(TTLCache(maxsize=1, ttl=timedelta(hours=1).total_seconds())) def keys(self) -> List[str]: return httpx.get(self.jwks_uri).raise_for_status().json().get("keys", []) @@ -205,7 +218,9 @@ def decode_token(self, token: str) -> dict[str, Any]: async def authenticate(self, request: Request) -> Optional[UserSessionState]: code = request.query_params.get("code") if not code: - logger.warning("Authentication failed: No authorization code parameter provided.") + logger.warning( + "Authentication failed: No authorization code parameter provided." + )and return None # A proxy in the middle may make the request into something like # 'http://localhost:8000/...' so we fix the first part but keep @@ -310,18 +325,227 @@ def oauth2_schema(self) -> OAuth2: return self._oidc_bearer +class EntraAuthenticator(ProxiedOIDCAuthenticator): + def __init__( + self, + audience: str, + client_id: str, + well_known_uri: str, + device_flow_client_id: str, + extra_scopes: Optional[List[str]] = None, + confirmation_message: str = "", + scopes_map: Optional[Dict[str, list[str]]] = None, + client_secret: str = "", + redirect_on_success: Optional[str] = None, + ): + self.scopes_map = scopes_map if scopes_map is not None else {} + self.extra_scopes = extra_scopes or [] + super().__init__( + audience, + client_id, + well_known_uri, + device_flow_client_id, + scopes=None, # not used by Entra; enforcement is via scopes_map + confirmation_message=confirmation_message, + ) + # Override the empty secret from ProxiedOIDCAuthenticator if provided. + if client_secret: + self._client_secret = Secret(client_secret) + self.redirect_on_success = redirect_on_success + + @property + def scopes(self): + mapped = set() + for tiled_scopes in self.scopes_map.values(): + mapped.update(tiled_scopes) + return list(mapped) + + @scopes.setter + def scopes(self, value): + pass # ignored; scopes are derived from scopes_map + + def decode_token( + self, id_token: str, access_token: Optional[str] = None + ) -> dict[str, Any]: + claims = super().decode_token(id_token, access_token) + + # sub generated by Entra is an opaque string; generate a stable UUID + # for Tiled based on "iss|sub" for uniqueness across tenants. + # Preserve the original Entra sub separately so it can be used as a + # fallback display name — it is more human-readable than the UUID5 hex. + original_sub = claims.get("sub") + issuer = claims.get("iss", "") + claims["sub"] = uuid.uuid5(uuid.NAMESPACE_URL, f"{issuer}|{original_sub}").hex + claims["entra_sub"] = original_sub + + # Derive a human-readable username from the token claims. + # Priority: nameID (explicit app config) → preferred_username (v2 tokens) + # → upn (v1 tokens) → email → original Entra sub (opaque but stable and + # meaningful, unlike the UUID5 hex stored in claims["sub"]). + # + # Note: preferred_username / upn are often absent from *access* tokens + # unless explicitly added as optional claims in the Entra app registration. + # They are typically present in id_tokens. If none are found, the + # original_sub is used and a warning is emitted so operators know to add + # the optional claim. + claims["entra_username"] = ( + claims.get("nameID") + or claims.get("preferred_username") + or claims.get("upn") + or claims.get("email") + ) + + if user := claims.get("entra_username"): + user = user.strip() + if "\\" in user: + user = user.rsplit("\\", 1)[-1] + elif "@" in user: + user = user.split("@", 1)[0] + else: + # No human-readable claim was found. Fall back to the original + # Entra sub (opaque but at least stable and not a UUID5 hex). + # This produces a workable identity but authz policies that match + # on username will need to use the Entra sub value. + user = original_sub + logger.warning( + "EntraAuthenticator: no human-readable username claim found in token " + "(checked nameID, preferred_username, upn, email). " + "Falling back to Entra sub=%r. " + "To fix: add 'preferred_username' as an optional claim in the " + "Entra app registration → Token configuration → Optional claims → Access token.", + original_sub, + ) + claims["user"] = user + + # Translate Entra scopes to tiled scopes. + # The "scp" claim is present in access tokens but may be absent from + # id_tokens (e.g. during the authorization code flow). When absent, + # assume all mapped scopes were granted (Entra would not have issued + # the tokens if the user lacked the requested scopes). + scp_raw = claims.get("scp", "") + tiled_scope_set = set() + if scp_raw: + for scope in scp_raw.split(" "): + mapped_scopes = self.scopes_map.get(scope) + if mapped_scopes is None: + logger.warning("Unmapped Entra scope in 'scp': %s", scope) + continue + tiled_scope_set.update(mapped_scopes) + else: + # No scp claim — grant all tiled scopes from the map. + for mapped_scopes in self.scopes_map.values(): + tiled_scope_set.update(mapped_scopes) + claims["scope"] = " ".join(tiled_scope_set) + + return claims + + async def authenticate(self, request: Request) -> Optional[UserSessionState]: + """Complete the Entra OIDC authorization-code flow and return a session. + + After a successful code exchange the Entra ``access_token`` and + ``refresh_token`` are stored in ``UserSessionState.state`` under the + keys ``entra_access_token`` and ``entra_refresh_token`` respectively. + Tiled persists this state in the session DB and embeds it verbatim in + every Tiled HMAC access token, making the tokens available to downstream + services that rely on Tiled authentication via ``get_session_state()``. + + Security note: the Entra access token is therefore visible inside the + Tiled JWT (base64-encoded, not encrypted). The Tiled access token is + short-lived (default 15 min) and only transmitted over HTTPS, which + limits the exposure window. + + The ``refresh_token`` enables silent renewal: when the Entra access + token expires (~1 h), a downstream service can call the Entra token + endpoint with ``grant_type=refresh_token`` to obtain a fresh pair and + write it back to the session DB so subsequent Tiled ``slide_session`` + calls propagate the update automatically. + """ + code = request.query_params.get("code") + if not code: + logger.warning( + "Authentication failed: No authorization code parameter provided." + ) + return None + redirect_uri = f"{get_root_url(request)}{request.url.path}" + response = await exchange_code( + self.token_endpoint, + code, + self._client_id, + self._client_secret.get_secret_value(), + redirect_uri, + extra_scopes=self.extra_scopes, + ) + response_body = response.json() + if response.is_error: + logger.error("Authentication error: %r", response_body) + return None + id_token = response_body["id_token"] + access_token = response_body.get("access_token") + refresh_token = response_body.get("refresh_token") + try: + verified_body = self.decode_token(id_token, access_token) + except JWTError: + logger.exception( + "Authentication error. Unverified token: %r", + jwt.get_unverified_claims(id_token), + ) + return None + # Log the id_token claims available for username resolution so + # misconfigurations (missing optional claims) are easy to diagnose. + # Logged at DEBUG to avoid leaking PII in production logs by default. + logger.debug( + "EntraAuthenticator.authenticate: id_token claims present: %s", + sorted(verified_body.keys()), + ) + logger.debug( + "EntraAuthenticator.authenticate: entra_username=%r user=%r entra_sub=%r", + verified_body.get("entra_username"), + verified_body.get("user"), + verified_body.get("entra_sub"), + ) + # Use the human-readable username (e.g. "dallan") instead of the + # opaque UUID-based "sub" that OIDCAuthenticator would use. + username = verified_body.get("user") or verified_body["sub"] + # Store the Entra access and refresh tokens so that downstream + # services that rely on Tiled authentication can perform an OBO exchange + # to obtain per-user tokens for other services. The refresh token + # allows silent renewal without requiring the user to re-authenticate. + state: dict = {} + if access_token: + state["entra_access_token"] = access_token + if refresh_token: + state["entra_refresh_token"] = refresh_token + return UserSessionState(username, state) + + async def exchange_code( token_uri: str, auth_code: str, client_id: str, client_secret: str, redirect_uri: str, + extra_scopes: Optional[List[str]] = None, ) -> httpx.Response: - """Method that talks to an IdP to exchange a code for an access_token and/or id_token - Args: - token_url ([type]): [description] - auth_code ([type]): [description] + """Exchange an authorization code for tokens at the IdP token endpoint. + + Explicitly requests ``openid offline_access`` scopes in the token POST body + so that the IdP returns a ``refresh_token`` unconditionally. This is safe + even when ``offline_access`` was already included in the authorization URL + scope — the IdP simply ignores duplicates. Including it here makes the + refresh token reliable regardless of how the authorization URL was + constructed, which is important for downstream OBO refresh flows. + + ``extra_scopes`` (e.g. ``["api:///access_as_user"]``) are + appended to the scope string. Entra only issues an ``access_token`` whose + ``aud`` matches the requested resource scope, so any scope that a downstream + OBO exchange will use as the ``assertion`` audience **must** be included + here — requesting it only on the authorization URL redirect is not + sufficient, because Entra does not carry scopes from the redirect into the + token POST implicitly. """ + scopes = {"openid", "offline_access"} + if extra_scopes: + scopes.update(extra_scopes) auth_value = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() response = httpx.post( url=token_uri, @@ -331,6 +555,7 @@ async def exchange_code( "redirect_uri": redirect_uri, "code": auth_code, "client_secret": client_secret, + "scope": " ".join(sorted(scopes)), }, headers={"Authorization": f"Basic {auth_value}"}, ) @@ -338,7 +563,6 @@ async def exchange_code( class SAMLAuthenticator(ExternalAuthenticator): - def __init__( self, saml_settings, # See EXAMPLE_SAML_SETTINGS below. @@ -356,7 +580,9 @@ def __init__( # The PyPI package name is 'python3-saml' # but it imports as 'onelogin'. # https://github.com/onelogin/python3-saml - raise ModuleNotFoundError("This SAMLAuthenticator requires 'python3-saml' to be installed.") + raise ModuleNotFoundError( + "This SAMLAuthenticator requires 'python3-saml' to be installed." + ) from onelogin.saml2.auth import OneLogin_Saml2_Auth @@ -371,7 +597,9 @@ async def saml_login(request: Request) -> RedirectResponse: async def authenticate(self, request: Request) -> Optional[UserSessionState]: if not modules_available("onelogin"): - raise ModuleNotFoundError("This SAMLAuthenticator requires the module 'oneline' to be installed.") + raise ModuleNotFoundError( + "This SAMLAuthenticator requires the module 'oneline' to be installed." + ) from onelogin.saml2.auth import OneLogin_Saml2_Auth req = await prepare_saml_from_fastapi_request(request, True) @@ -380,7 +608,8 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]: errors = auth.get_errors() # This method receives an array with the errors if errors: raise Exception( - "Error when processing SAML Response: %s %s" % (", ".join(errors), auth.get_last_error_reason()) + "Error when processing SAML Response: %s %s" + % (", ".join(errors), auth.get_last_error_reason()) ) if auth.is_authenticated(): # Return a string that the Identity can use as id. @@ -399,7 +628,7 @@ async def prepare_saml_from_fastapi_request(request: Request) -> Mapping[str, st "server_port": request.url.port, "script_name": request.url.path, "post_data": {}, - "get_data": {}, + "get_data": {} # Advanced request options # "https": "", # "request_uri": "", @@ -420,7 +649,6 @@ async def prepare_saml_from_fastapi_request(request: Request) -> Mapping[str, st class LDAPAuthenticator(InternalAuthenticator): """ - LDAP authenticator. The authenticator code is based on https://github.com/jupyterhub/ldapauthenticator The parameter ``use_tls`` was added for convenience of testing. @@ -645,7 +873,9 @@ def __init__( self.escape_userdn = escape_userdn self.search_filter = search_filter self.attributes = attributes if attributes else [] - self.auth_state_attributes = auth_state_attributes if auth_state_attributes else [] + self.auth_state_attributes = ( + auth_state_attributes if auth_state_attributes else [] + ) self.use_lookup_dn_username = use_lookup_dn_username if isinstance(server_address, str): @@ -658,10 +888,14 @@ def __init__( f"type(server_address)={type(server_address)}" ) if not server_address_list: - raise ValueError("No servers are specified: 'server_address' is an empty list") + raise ValueError( + "No servers are specified: 'server_address' is an empty list" + ) self.server_address_list = server_address_list - self.server_port = server_port if server_port is not None else self._server_port_default() + self.server_port = ( + server_port if server_port is not None else self._server_port_default() + ) self.confirmation_message = confirmation_message def _server_port_default(self): @@ -715,8 +949,15 @@ async def resolve_username(self, username_supplied_by_user): response = conn.response if len(response) == 0 or "attributes" not in response[0].keys(): - msg = "No entry found for user '{username}' " "when looking up attribute '{attribute}'" - logger.warning(msg.format(username=username_supplied_by_user, attribute=self.user_attribute)) + msg = ( + "No entry found for user '{username}' " + "when looking up attribute '{attribute}'" + ) + logger.warning( + msg.format( + username=username_supplied_by_user, attribute=self.user_attribute + ) + ) return (None, None) user_dn = response[0]["attributes"][self.lookup_dn_user_dn_attribute] @@ -774,7 +1015,9 @@ def get_connection(self, userdn, password): ) server_pool.add(server) - auto_bind_no_ssl = ldap3.AUTO_BIND_TLS_BEFORE_BIND if self.use_tls else ldap3.AUTO_BIND_NO_TLS + auto_bind_no_ssl = ( + ldap3.AUTO_BIND_TLS_BEFORE_BIND if self.use_tls else ldap3.AUTO_BIND_NO_TLS + ) auto_bind = ldap3.AUTO_BIND_NO_TLS if self.use_ssl else auto_bind_no_ssl conn = ldap3.Connection( server_pool, @@ -799,7 +1042,9 @@ async def get_user_attributes(self, conn, userdn): attrs = conn.entries[0].entry_attributes_as_dict return attrs - async def authenticate(self, username: str, password: str) -> Optional[UserSessionState]: + async def authenticate( + self, username: str, password: str + ) -> Optional[UserSessionState]: import ldap3 username_saved = username # Save the user name passed as a parameter @@ -825,7 +1070,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi # sanity check if not self.lookup_dn and not bind_dn_template: - logger.warning("Login not allowed, please configure 'lookup_dn' or 'bind_dn_template'.") + logger.warning( + "Login not allowed, please configure 'lookup_dn' or 'bind_dn_template'." + ) return None if self.lookup_dn: @@ -863,7 +1110,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi if conn.bound: is_bound = True else: - is_bound = await asyncio.get_running_loop().run_in_executor(None, conn.bind) + is_bound = await asyncio.get_running_loop().run_in_executor( + None, conn.bind + ) msg = msg.format(username=username, userdn=userdn, is_bound=is_bound) logger.debug(msg) @@ -876,7 +1125,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi return None if self.search_filter: - search_filter = self.search_filter.format(userattr=self.user_attribute, username=username) + search_filter = self.search_filter.format( + userattr=self.user_attribute, username=username + ) search_func = functools.partial( conn.search, @@ -890,18 +1141,33 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi n_users = len(conn.response) if n_users == 0: msg = "User with '{userattr}={username}' not found in directory" - logger.warning(msg.format(userattr=self.user_attribute, username=username)) + logger.warning( + msg.format(userattr=self.user_attribute, username=username) + ) return None if n_users > 1: - msg = "Duplicate users found! " "{n_users} users found with '{userattr}={username}'" - logger.warning(msg.format(userattr=self.user_attribute, username=username, n_users=n_users)) + msg = ( + "Duplicate users found! " + "{n_users} users found with '{userattr}={username}'" + ) + logger.warning( + msg.format( + userattr=self.user_attribute, username=username, n_users=n_users + ) + ) return None if self.allowed_groups: logger.debug("username:%s Using dn %s", username, userdn) found = False for group in self.allowed_groups: - group_filter = "(|" "(member={userdn})" "(uniqueMember={userdn})" "(memberUid={uid})" ")" + group_filter = ( + "(|" + "(member={userdn})" + "(uniqueMember={userdn})" + "(memberUid={uid})" + ")" + ) group_filter = group_filter.format(userdn=userdn, uid=username) group_attributes = ["member", "uniqueMember", "memberUid"] @@ -912,7 +1178,9 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi search_filter=group_filter, attributes=group_attributes, ) - found = await asyncio.get_running_loop().run_in_executor(None, search_func) + found = await asyncio.get_running_loop().run_in_executor( + None, search_func + ) if found: break From a3bbe20fa08cf2ee5a8101d60109b2de96a67e4a Mon Sep 17 00:00:00 2001 From: David Pastl Date: Mon, 6 Jul 2026 15:24:27 -0600 Subject: [PATCH 02/11] Migrating changes from 0.2.12 tiled in --- bluesky_httpserver/_authentication.py | 297 ++++++++++++++++-- bluesky_httpserver/app.py | 54 +++- bluesky_httpserver/authentication/__init__.py | 4 + bluesky_httpserver/authenticators.py | 36 ++- bluesky_httpserver/database/core.py | 31 +- .../a1b2c3d4e5f6_add_pending_sessions.py | 59 ++++ .../b2c3d4e5f6a7_add_session_state.py | 40 +++ bluesky_httpserver/database/orm.py | 7 +- bluesky_httpserver/routers/core_api.py | 74 ++++- bluesky_httpserver/schemas.py | 10 + 10 files changed, 547 insertions(+), 65 deletions(-) create mode 100644 bluesky_httpserver/database/migrations/versions/a1b2c3d4e5f6_add_pending_sessions.py create mode 100644 bluesky_httpserver/database/migrations/versions/b2c3d4e5f6a7_add_session_state.py diff --git a/bluesky_httpserver/_authentication.py b/bluesky_httpserver/_authentication.py index c1144f5..19404ca 100644 --- a/bluesky_httpserver/_authentication.py +++ b/bluesky_httpserver/_authentication.py @@ -1,5 +1,6 @@ import asyncio import hashlib +import logging import secrets import uuid as uuid_module import warnings @@ -36,6 +37,7 @@ from .database import orm from .database.core import ( create_user, + get_or_create_principal, latest_principal_activity, lookup_valid_api_key, lookup_valid_pending_session_by_device_code, @@ -60,6 +62,8 @@ DEVICE_CODE_MAX_AGE = timedelta(minutes=10) DEVICE_CODE_POLLING_INTERVAL = 5 # seconds +logger = logging.getLogger(__name__) + def utcnow(): "UTC now with second resolution" @@ -140,7 +144,19 @@ def create_refresh_token(session_id, secret_key, expires_delta): return encoded_jwt -def decode_token(token, secret_keys): +def decode_token(token, secret_keys, proxied_authenticator=None): + """Decode a JWT. + + Tries every bluesky-httpserver HMAC ``secret_key`` first (supports key + rotation). If none succeed and a ``proxied_authenticator`` (typically a + :class:`bluesky_httpserver.authenticators.ProxiedOIDCAuthenticator`) is + provided, delegate to its ``decode_token`` so that access tokens minted + by an external OIDC provider (e.g. via device-code flow) are accepted. + + Raises :class:`fastapi.HTTPException` with status 401 if the token cannot + be validated by either mechanism. Propagates :class:`ExpiredSignatureError` + so the caller can surface a distinct "expired token" error. + """ credentials_exception = HTTPException( status_code=401, detail="Could not validate credentials", @@ -149,19 +165,53 @@ def decode_token(token, secret_keys): # The first key in settings.secret_keys is used for *encoding*. # All keys are tried for *decoding* until one works or they all # fail. They supports key rotation. + last_error = None for secret_key in secret_keys: try: - payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM]) - break + return jwt.decode(token, secret_key, algorithms=[ALGORITHM]) except ExpiredSignatureError: # Do not let this be caught below with the other JWTError types. raise - except JWTError: + except JWTError as err: + last_error = err # Try the next key in the key rotation. continue - else: - raise credentials_exception - return payload + # None of the bluesky-httpserver HMAC keys accepted the token. + # Fall back to the proxied OIDC authenticator (if any) so that JWTs minted + # by an upstream provider — e.g. via device-code flow — are honored. + if proxied_authenticator is not None: + try: + return proxied_authenticator.decode_token(token) + except ExpiredSignatureError: + raise + except JWTError: + pass + raise credentials_exception from last_error + + +def _extract_scopes(decoded_access_token, authenticator=None): + """Normalize the ``scopes`` set from an OIDC-decoded JWT. + + OIDC providers spell the scope claim inconsistently: + + * ``scp`` may be a space-separated string (Microsoft Entra) or a list. + * ``scope`` is typically a space-separated string. + + Additionally, an authenticator with a ``scopes_map`` attribute (e.g. + :class:`EntraAuthenticator`) may already have translated raw provider + scopes into local scopes and stored the result in ``claims["scope"]``. + In that case, prefer the pre-translated value. + """ + scopes: set[str] = set() + scp = decoded_access_token.get("scp") + if isinstance(scp, list): + scopes.update(scp) + elif isinstance(scp, str) and scp: + scopes.update(scp.split()) + scope = decoded_access_token.get("scope") + if isinstance(scope, str) and scope: + scopes.update(scope.split()) + return scopes async def get_api_key( @@ -175,6 +225,52 @@ async def get_api_key( return None +def get_session_state( + access_token: str = Depends(oauth2_scheme), + settings: BaseSettings = Depends(get_settings), + authenticators=Depends(get_authenticators), +): + """FastAPI dependency: return the ``state`` dict embedded in the current + access token, if any, else ``{}``. + + This mirrors tiled's ``get_session_state`` and lets downstream routes + retrieve authenticator-supplied state (e.g. upstream OIDC tokens for OBO + exchange) without a database round-trip. + """ + if not access_token: + return {} + proxied_auth = _find_proxied_authenticator(authenticators) + try: + payload = decode_token(access_token, settings.secret_keys, proxied_auth) + except HTTPException: + return {} + except ExpiredSignatureError: + return {} + return payload.get("state") or {} + + +def _find_proxied_authenticator(authenticators): + """Return the first authenticator that is a ``ProxiedOIDCAuthenticator``. + + Imported lazily to avoid a top-level circular import between + ``_authentication.py`` and ``authenticators.py``. + """ + if not authenticators: + return None + try: + from .authenticators import ProxiedOIDCAuthenticator + except ImportError: + return None + if isinstance(authenticators, dict): + candidates = authenticators.values() + else: + candidates = authenticators + for a in candidates: + if isinstance(a, ProxiedOIDCAuthenticator): + return a + return None + + def get_current_principal( request: Request, security_scopes: SecurityScopes, @@ -279,30 +375,85 @@ def get_current_principal( if ("api_key" in request.query_params) and (request.cookies.get(API_KEY_COOKIE_NAME) != api_key): request.state.cookies_to_set.append({"key": API_KEY_COOKIE_NAME, "value": api_key}) elif access_token is not None: + proxied_authenticator = _find_proxied_authenticator(authenticators) try: - payload = decode_token(access_token, settings.secret_keys) + payload = decode_token(access_token, settings.secret_keys, proxied_authenticator) except ExpiredSignatureError: raise HTTPException( status_code=401, detail="Access token has expired. Refresh token.", headers=headers_for_401, ) - principal = schemas.Principal( - uuid=uuid_module.UUID(hex=payload["sub"]), - type=payload["sub_typ"], - identities=[ - schemas.Identity(id=identity["id"], provider=identity["idp"]) for identity in payload["ids"] - ], - ) - # scopes = payload["scp"] + if "sub_typ" in payload: + # Token minted by bluesky-httpserver (login / refresh / device-code + # flow). ``payload`` carries everything needed to reconstruct the + # Principal without a database hit. + principal = schemas.Principal( + uuid=uuid_module.UUID(hex=payload["sub"]), + type=payload["sub_typ"], + identities=[ + schemas.Identity(id=identity["id"], provider=identity["idp"]) + for identity in payload["ids"] + ], + ) + + # scopes = payload["scp"] + + # Combine scopes for all identities (it is expected to be only one identity). + ids = [_["id"] for _ in payload["ids"] if _["idp"] in settings.authentication_provider_names] + scopes = set.union(*[api_access_manager.get_user_scopes(_) for _ in ids]) - # Combine scopes for all identities (it is expected to be only one identity). - ids = [_["id"] for _ in payload["ids"] if _["idp"] in settings.authentication_provider_names] - scopes = set.union(*[api_access_manager.get_user_scopes(_) for _ in ids]) + roles_sets = [api_access_manager.get_user_roles(_) for _ in ids] + roles = set.union(*roles_sets) if roles_sets else set() - roles_sets = [api_access_manager.get_user_roles(_) for _ in ids] - roles = set.union(*roles_sets) if roles_sets else set() + else: + # Token minted by an external OIDC provider (proxied OIDC flow). + # Resolve the Principal via the auth DB, using the app-scoped + # provider name that was set at startup. + if proxied_authenticator is None: + raise HTTPException( + status_code=401, + detail="Access token is not recognized", + headers=headers_for_401, + ) + identity_id = payload.get("user") or payload.get("sub") + if identity_id is None: + raise HTTPException( + status_code=401, + detail="Access token missing subject", + headers=headers_for_401, + ) + provider = getattr(request.app.state, "provider", None) + if provider is None: + # Fall back to the first configured provider name. In practice + # ``app.state.provider`` is set by build_app() whenever a + # ProxiedOIDCAuthenticator is configured, but we do not want + # to fail if someone bootstraps the app differently. + provider = ( + settings.authentication_provider_names[0] + if settings.authentication_provider_names + else _DEFAULT_ANONYMOUS_PROVIDER_NAME + ) + with get_sessionmaker(settings.database_settings)() as db: + principal_orm = get_or_create_principal(db, provider, identity_id) + principal = schemas.Principal( + uuid=principal_orm.uuid, + type=schemas.PrincipalType.user, + identities=[schemas.Identity(id=identity_id, provider=provider)], + access_token=access_token, + ) + # Combine scopes carried in the token itself with any additional + # scopes granted to this user by the api_access_manager (which is + # the fork's replacement for tiled's DB-role machinery). + token_scopes = _extract_scopes(payload, proxied_authenticator) + if api_access_manager.is_user_known(identity_id): + extra_scopes = api_access_manager.get_user_scopes(identity_id) + roles = api_access_manager.get_user_roles(identity_id) + else: + extra_scopes = set() + roles = set() + scopes = set(token_scopes) | set(extra_scopes) else: # No form of authentication is present. @@ -373,12 +524,24 @@ def get_current_principal_websocket( auth_header = websocket.headers.get("Authorization", "") access_token, api_key = None, None - # Currently we do not support authentication with tokens - # if auth_header.startswith("Bearer "): - # access_token = auth_header[len("Bearer") :].strip() - if auth_header.startswith("ApiKey "): + if auth_header.startswith("Bearer "): + access_token = auth_header[len("Bearer") :].strip() + elif auth_header.startswith("ApiKey "): api_key = auth_header[len("ApiKey") :].strip() + # Also honor an ``access_token`` query parameter so browsers that cannot + # set Authorization headers on a WebSocket handshake still authenticate. + if access_token is None and api_key is None: + access_token = websocket.query_params.get("access_token") + if access_token is None: + api_key = websocket.query_params.get("api_key") + + # If nothing was supplied on the initial handshake, return None instead of + # raising 401. The socket route can then attempt the first-message + # protocol handled by ``authenticate_websocket_first_message``. + if not access_token and not api_key: + return None + principal = None try: principal = get_current_principal( @@ -391,12 +554,58 @@ def get_current_principal_websocket( api_access_manager=api_access_manager, ) except HTTPException as ex: - print(f"WebSocket connection failed: {ex}") + logger.info("WebSocket authentication failed: %s", ex.detail) return principal -def create_session(settings, identity_provider, id, scopes): +def authenticate_websocket_first_message(websocket, message): + """Handle a ``{"type": "auth", ...}`` handshake message on a WebSocket. + + The socket route awaits this only when the standard header/query + handshake produced no principal (i.e. ``get_current_principal_websocket`` + returned ``None``). It accepts either an API key or an access token in + the message body: + + {"type": "auth", "api_key": ""} + {"type": "auth", "access_token": ""} + + Returns the resolved :class:`schemas.Principal`, or ``None`` if the + message is malformed or the credentials are invalid. The socket route + is expected to close the connection on failure. + """ + if not isinstance(message, dict): + return None + if message.get("type") != "auth": + return None + + app = websocket.app + settings = app.dependency_overrides[get_settings]() + authenticators = app.dependency_overrides[get_authenticators]() + api_access_manager = app.dependency_overrides[get_api_access_manager]() + + api_key = message.get("api_key") + access_token = message.get("access_token") + if not api_key and not access_token: + return None + + security_scopes = SecurityScopes(scopes=[]) + try: + return get_current_principal( + request=websocket, + security_scopes=security_scopes, + access_token=access_token, + api_key=api_key, + settings=settings, + authenticators=authenticators, + api_access_manager=api_access_manager, + ) + except HTTPException as ex: + logger.info("WebSocket first-message authentication failed: %s", ex.detail) + return None + + +def create_session(settings, identity_provider, id, scopes, state=None): with get_sessionmaker(settings.database_settings)() as db: # Have we seen this Identity before? identity = ( @@ -420,6 +629,7 @@ def create_session(settings, identity_provider, id, scopes): session = orm.Session( principal_id=principal.id, expiration_time=utcnow() + settings.session_max_age, + state=state or {}, ) db.add(session) db.commit() @@ -432,6 +642,7 @@ def create_session(settings, identity_provider, id, scopes): "sub_typ": principal.type.value, "scp": list(scopes), "ids": [{"id": identity.id, "idp": identity.provider} for identity in principal.identities], + "state": session.state or {}, } access_token = create_access_token( data=data, @@ -463,6 +674,7 @@ async def auth_code( request.state.endpoint = "auth" user_session_state = await authenticator.authenticate(request) username = user_session_state.user_name if user_session_state else None + session_state = (user_session_state.state or {}) if user_session_state else {} if username and api_access_manager.is_user_known(username): scopes = api_access_manager.get_user_scopes(username) @@ -470,7 +682,7 @@ async def auth_code( raise HTTPException(status_code=401, detail="Authentication failure") tokens = await asyncio.get_running_loop().run_in_executor( - None, create_session, settings, provider, username, scopes + None, create_session, settings, provider, username, scopes, session_state ) # Show only the refresh_token, which is what the user should # paste into a terminal-based client. @@ -495,6 +707,7 @@ async def handle_credentials( username=form_data.username, password=form_data.password ) username = user_session_state.user_name if user_session_state else None + session_state = (user_session_state.state or {}) if user_session_state else {} err_msg = None if not username: @@ -511,7 +724,7 @@ async def handle_credentials( headers={"WWW-Authenticate": "Bearer"}, ) return await asyncio.get_running_loop().run_in_executor( - None, create_session, settings, provider, username, scopes + None, create_session, settings, provider, username, scopes, session_state ) return handle_credentials @@ -558,11 +771,22 @@ async def authorize_redirect( """Redirect browser to OAuth provider for authentication.""" redirect_uri = f"{get_base_url(request)}/auth/provider/{provider}/code" + # Always request ``openid`` and ``offline_access`` so the IdP returns a + # refresh_token in the code exchange. Authenticators (e.g. Entra) may + # advertise extra scopes via an ``extra_scopes`` attribute to obtain + # per-resource access tokens. ``prompt=login`` forces the IdP to + # prompt the user rather than silently reissue a session — this + # prevents surprising SSO behavior when a user explicitly navigates + # to the login flow. + scopes = {"openid", "profile", "email", "offline_access"} + scopes.update(getattr(authenticator, "extra_scopes", []) or []) + params = { "client_id": authenticator.client_id, "response_type": "code", - "scope": "openid profile email", + "scope": " ".join(sorted(scopes)), "redirect_uri": redirect_uri, + "prompt": "login", } if state: params["state"] = state @@ -591,13 +815,16 @@ async def device_code_authorize( pending_session = create_pending_session(db) verification_uri = f"{get_base_url(request)}/auth/provider/{provider}/token" + scopes = {"openid", "profile", "email", "offline_access"} + scopes.update(getattr(authenticator, "extra_scopes", []) or []) authorization_uri = authenticator.authorization_endpoint.copy_with( params={ "client_id": authenticator.client_id, "response_type": "code", - "scope": "openid profile email", + "scope": " ".join(sorted(scopes)), "redirect_uri": f"{get_base_url(request)}/auth/provider/{provider}/device_code", "state": pending_session["user_code"].replace("-", ""), + "prompt": "login", } ) return { @@ -684,6 +911,7 @@ async def _complete_device_code_authorization( return HTMLResponse(content=error_html, status_code=401) username = user_session_state.user_name + session_state = user_session_state.state or {} if not api_access_manager.is_user_known(username): error_html = f""" @@ -710,7 +938,7 @@ async def _complete_device_code_authorization( # Create the session session = await asyncio.get_running_loop().run_in_executor( - None, _create_session_orm, settings, provider, username, db + None, _create_session_orm, settings, provider, username, db, session_state ) # Link the pending session to the real session @@ -837,7 +1065,7 @@ async def device_code_submit( return device_code_submit -def _create_session_orm(settings, identity_provider, id, db): +def _create_session_orm(settings, identity_provider, id, db, state=None): """ Create a session and return the ORM object (for device code flow). @@ -864,6 +1092,7 @@ def _create_session_orm(settings, identity_provider, id, db): session = orm.Session( principal_id=principal.id, expiration_time=utcnow() + settings.session_max_age, + state=state or {}, ) db.add(session) db.commit() @@ -925,6 +1154,7 @@ async def device_code_token( "sub_typ": principal.type.value, "scp": list(scopes), "ids": [{"id": ident.id, "idp": ident.provider} for ident in principal.identities], + "state": session.state or {}, } access_token = create_access_token( data=data, @@ -1153,6 +1383,7 @@ def slide_session(refresh_token, settings, db, api_access_manager): "sub_typ": principal.type.value, "scp": list(scopes), "ids": [{"id": identity.id, "idp": identity.provider} for identity in principal.identities], + "state": session.state or {}, } access_token = create_access_token( data=data, diff --git a/bluesky_httpserver/app.py b/bluesky_httpserver/app.py index 0d96667..6936a32 100644 --- a/bluesky_httpserver/app.py +++ b/bluesky_httpserver/app.py @@ -181,6 +181,11 @@ def build_app(authentication=None, api_access=None, resource_access=None, server # Authenticators provide Router(s) for their particular flow. # Collect them in the authentication_router. + # Lazy import: avoid importing authenticators at module import time so + # that lightweight test doubles that pass fake authenticators don't + # need heavy optional dependencies (e.g. python-jose, httpx). + from .authenticators import ProxiedOIDCAuthenticator + for spec in authentication["providers"]: provider = spec["provider"] authenticator = spec["authenticator"] @@ -217,6 +222,24 @@ def build_app(authentication=None, api_access=None, resource_access=None, server authentication_router.post(f"/provider/{provider}/token")( build_device_code_token_route(authenticator, provider) ) + # Warn if the operator forgot to configure a redirect target + # for successful browser-based logins. Without it the user + # will get a page of raw JSON instead of being sent to the UI. + if not getattr(authenticator, "redirect_on_success", None): + logger.warning( + "External authenticator %r has no 'redirect_on_success' " + "configured. Browser-based login will return raw JSON " + "tokens instead of redirecting to a UI landing page. " + "Set 'redirect_on_success' in the authenticator " + "configuration to a UI callback URL to silence this " + "warning.", + provider, + ) + # Expose the OIDC provider name on app.state so that + # get_current_principal / websocket handshakes can resolve + # externally-minted JWTs to a Principal in the auth DB. + if isinstance(authenticator, ProxiedOIDCAuthenticator): + app.state.provider = provider else: raise ValueError(f"unknown authenticator type {type(authenticator)}") for custom_router in getattr(authenticator, "include_routers", []): @@ -416,10 +439,31 @@ async def purge_expired_sessions_and_api_keys(): @app.on_event("shutdown") async def shutdown_event(): - await SR.RM.close() - await SR.console_output_loader.stop() - await SR.console_output_stream.stop() - await SR.system_info_stream.stop() + # Cancel any background tasks started by startup_event, even if + # startup itself failed part-way through and never reached + # ``app.state.tasks = []``. Iterating with a getattr default keeps + # the shutdown handler idempotent under partial-startup failures. + for task in getattr(app.state, "tasks", []): + task.cancel() + # Best-effort teardown of REManager / console streams; each guarded + # so that a failure in one does not skip the others. + for closer_name in ( + "console_output_loader", + "console_output_stream", + "system_info_stream", + ): + closer = getattr(SR, closer_name, None) + if closer is not None: + try: + await closer.stop() + except Exception: + logger.exception("Error stopping %s", closer_name) + rm = getattr(SR, "RM", None) + if rm is not None: + try: + await rm.close() + except Exception: + logger.exception("Error closing REManagerAPI connection") @lru_cache(1) def override_get_authenticators(): @@ -506,7 +550,7 @@ async def double_submit_cookie_csrf_protection(request: Request, call_next): csrf_cookie = request.cookies.get(CSRF_COOKIE_NAME) if (request.method not in SAFE_METHODS) and set(request.cookies).intersection(SENSITIVE_COOKIES): if not csrf_cookie: - return Response(status_code=403, content="Expected tiled_csrf_token cookie") + return Response(status_code=403, content=f"Expected {CSRF_COOKIE_NAME} cookie") # Get the token from the Header or (if not there) the query parameter. csrf_token = request.headers.get(CSRF_HEADER_NAME) if csrf_token is None: diff --git a/bluesky_httpserver/authentication/__init__.py b/bluesky_httpserver/authentication/__init__.py index 85d835e..13043c5 100644 --- a/bluesky_httpserver/authentication/__init__.py +++ b/bluesky_httpserver/authentication/__init__.py @@ -1,4 +1,5 @@ from .._authentication import ( + authenticate_websocket_first_message, base_authentication_router, build_auth_code_route, build_authorize_route, @@ -9,6 +10,7 @@ build_handle_credentials_route, get_current_principal, get_current_principal_websocket, + get_session_state, oauth2_scheme, ) from .authenticator_base import ( @@ -21,8 +23,10 @@ "ExternalAuthenticator", "InternalAuthenticator", "UserSessionState", + "authenticate_websocket_first_message", "get_current_principal", "get_current_principal_websocket", + "get_session_state", "base_authentication_router", "build_auth_code_route", "build_authorize_route", diff --git a/bluesky_httpserver/authenticators.py b/bluesky_httpserver/authenticators.py index 7448d0c..1c2bf6f 100644 --- a/bluesky_httpserver/authenticators.py +++ b/bluesky_httpserver/authenticators.py @@ -206,13 +206,20 @@ def end_session_endpoint(self) -> str: def keys(self) -> List[str]: return httpx.get(self.jwks_uri).raise_for_status().json().get("keys", []) - def decode_token(self, token: str) -> dict[str, Any]: + def decode_token( + self, id_token: str, access_token: Optional[str] = None + ) -> dict[str, Any]: + # Passing ``access_token`` allows python-jose to validate the ``at_hash`` + # claim (present in id_tokens when the token endpoint returns an + # access_token alongside). Callers that only have an id_token (or + # only an access_token they want to decode as a plain JWT) may omit it. return jwt.decode( - token, + id_token, key=self.keys(), algorithms=self.id_token_signing_alg_values_supported, audience=self._audience, issuer=self.issuer, + access_token=access_token, ) async def authenticate(self, request: Request) -> Optional[UserSessionState]: @@ -220,7 +227,7 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]: if not code: logger.warning( "Authentication failed: No authorization code parameter provided." - )and + ) return None # A proxy in the middle may make the request into something like # 'http://localhost:8000/...' so we fix the first part but keep @@ -243,8 +250,9 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]: # 2. Some providers (like Microsoft Entra) return opaque access_tokens # that cannot be decoded with the JWKS keys when the resource is # a first-party Microsoft API (e.g., Graph API with User.Read scope) + access_token = response_body.get("access_token") try: - verified_body = self.decode_token(id_token) + verified_body = self.decode_token(id_token, access_token) except JWTError: logger.exception( "Authentication error. Unverified token: %r", @@ -353,16 +361,16 @@ def __init__( self._client_secret = Secret(client_secret) self.redirect_on_success = redirect_on_success - @property - def scopes(self): - mapped = set() - for tiled_scopes in self.scopes_map.values(): - mapped.update(tiled_scopes) - return list(mapped) - - @scopes.setter - def scopes(self, value): - pass # ignored; scopes are derived from scopes_map + @property + def scopes(self): + mapped = set() + for tiled_scopes in self.scopes_map.values(): + mapped.update(tiled_scopes) + return list(mapped) + + @scopes.setter + def scopes(self, value): + pass # ignored; scopes are derived from scopes_map def decode_token( self, id_token: str, access_token: Optional[str] = None diff --git a/bluesky_httpserver/database/core.py b/bluesky_httpserver/database/core.py index 52d102f..88f7ffa 100644 --- a/bluesky_httpserver/database/core.py +++ b/bluesky_httpserver/database/core.py @@ -15,9 +15,9 @@ # This is the alembic revision ID of the database revision # required by this version of Tiled. -REQUIRED_REVISION = "a1b2c3d4e5f6" +REQUIRED_REVISION = "b2c3d4e5f6a7" # This is list of all valid revisions (from current to oldest). -ALL_REVISIONS = ["a1b2c3d4e5f6", "722ff4e4fcc7", "481830dd6c11"] +ALL_REVISIONS = ["b2c3d4e5f6a7", "a1b2c3d4e5f6", "722ff4e4fcc7", "481830dd6c11"] # def create_default_roles(engine): @@ -209,6 +209,33 @@ def create_user(db, identity_provider, id): return principal +def get_or_create_principal(db, identity_provider, id): + """Return a Principal for (identity_provider, id), creating it if needed. + + Mirrors tiled's ``authn_database.core.get_or_create_principal``. Unlike + :func:`create_session`, this helper only touches the Principal/Identity + tables — it never creates a Session row. It is intended for principals + that authenticate with a token minted by an external OIDC provider (i.e. + :class:`bluesky_httpserver.authenticators.ProxiedOIDCAuthenticator` + subclasses) where the JWT itself is authoritative and no bluesky-httpserver + session lifetime is required. + + On successful lookup the matching ``Identity.latest_login`` is updated to + now. + """ + identity = ( + db.query(Identity) + .filter(Identity.id == id) + .filter(Identity.provider == identity_provider) + .first() + ) + if identity is not None: + identity.latest_login = datetime.utcnow() + db.commit() + return identity.principal + return create_user(db, identity_provider, id) + + def lookup_valid_session(db, session_id): if isinstance(session_id, int): # Old versions of tiled used an integer sid. diff --git a/bluesky_httpserver/database/migrations/versions/a1b2c3d4e5f6_add_pending_sessions.py b/bluesky_httpserver/database/migrations/versions/a1b2c3d4e5f6_add_pending_sessions.py new file mode 100644 index 0000000..8876219 --- /dev/null +++ b/bluesky_httpserver/database/migrations/versions/a1b2c3d4e5f6_add_pending_sessions.py @@ -0,0 +1,59 @@ +"""Add PendingSession table for device code flow. + +Revision ID: a1b2c3d4e5f6 +Revises: 722ff4e4fcc7 +Create Date: 2026-02-13 12:00:00.000000 + +""" + +from alembic import op +from sqlalchemy import Column, DateTime, ForeignKey, Integer, LargeBinary, Unicode +from sqlalchemy.sql import func + +# revision identifiers, used by Alembic. +revision = "a1b2c3d4e5f6" +down_revision = "722ff4e4fcc7" +branch_labels = None +depends_on = None + + +def upgrade(): + """ + Add pending_sessions table for device code flow authentication. + """ + op.create_table( + "pending_sessions", + Column("time_created", DateTime(timezone=False), server_default=func.now()), + Column("time_updated", DateTime(timezone=False), onupdate=func.now()), + Column( + "hashed_device_code", + LargeBinary(32), + primary_key=True, + index=True, + nullable=False, + ), + Column( + "user_code", + Unicode(8), + index=True, + nullable=False, + ), + Column( + "expiration_time", + DateTime(timezone=False), + nullable=False, + ), + Column( + "session_id", + Integer, + ForeignKey("sessions.id"), + nullable=True, + ), + ) + + +def downgrade(): + """ + Remove pending_sessions table. + """ + op.drop_table("pending_sessions") diff --git a/bluesky_httpserver/database/migrations/versions/b2c3d4e5f6a7_add_session_state.py b/bluesky_httpserver/database/migrations/versions/b2c3d4e5f6a7_add_session_state.py new file mode 100644 index 0000000..3fa5e27 --- /dev/null +++ b/bluesky_httpserver/database/migrations/versions/b2c3d4e5f6a7_add_session_state.py @@ -0,0 +1,40 @@ +"""Add ``state`` column to sessions. + +Revision ID: b2c3d4e5f6a7 +Revises: a1b2c3d4e5f6 +Create Date: 2026-07-06 12:00:00.000000 + +Ports the ``Session.state`` column introduced upstream in tiled 0.2.10. The +column stores a JSON dict populated by an authenticator's +``UserSessionState.state`` (e.g. upstream OIDC access/refresh tokens carried +across refresh_session calls) so downstream services that share +bluesky-httpserver authentication can retrieve them via Tiled access tokens. +""" + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = "b2c3d4e5f6a7" +down_revision = "a1b2c3d4e5f6" +branch_labels = None +depends_on = None + + +def upgrade(): + """Add sessions.state JSON column with default ``{}``.""" + with op.batch_alter_table("sessions") as batch_op: + batch_op.add_column( + sa.Column( + "state", + sa.JSON(), + nullable=False, + server_default="{}", + ) + ) + + +def downgrade(): + """Drop sessions.state column.""" + with op.batch_alter_table("sessions") as batch_op: + batch_op.drop_column("state") diff --git a/bluesky_httpserver/database/orm.py b/bluesky_httpserver/database/orm.py index 7611824..9d173ef 100644 --- a/bluesky_httpserver/database/orm.py +++ b/bluesky_httpserver/database/orm.py @@ -1,7 +1,7 @@ import json import uuid as uuid_module -from sqlalchemy import Boolean, Column, DateTime, Enum, ForeignKey, Integer, LargeBinary, Unicode # Table, +from sqlalchemy import JSON, Boolean, Column, DateTime, Enum, ForeignKey, Integer, LargeBinary, Unicode # Table, from sqlalchemy.orm import relationship from sqlalchemy.sql import func from sqlalchemy.types import TypeDecorator @@ -179,6 +179,11 @@ class Session(Timestamped, Base): expiration_time = Column(DateTime(timezone=False), nullable=False) principal_id = Column(Integer, ForeignKey("principals.id"), nullable=False) revoked = Column(Boolean, default=False, nullable=False) + # Free-form state supplied by the authenticator via UserSessionState.state + # (e.g. an upstream OIDC access_token/refresh_token for OBO exchange). + # Persisted so it survives across refresh_session calls; the values are + # available to downstream services through Tiled access tokens. + state = Column(JSON, nullable=False, default=dict, server_default="{}") principal = relationship("Principal", back_populates="sessions") pending_sessions = relationship("PendingSession", back_populates="session") diff --git a/bluesky_httpserver/routers/core_api.py b/bluesky_httpserver/routers/core_api.py index 9e47d58..841a295 100644 --- a/bluesky_httpserver/routers/core_api.py +++ b/bluesky_httpserver/routers/core_api.py @@ -14,7 +14,11 @@ else: from pydantic_settings import BaseSettings -from ..authentication import get_current_principal, get_current_principal_websocket +from ..authentication import ( + authenticate_websocket_first_message, + get_current_principal, + get_current_principal_websocket, +) from ..console_output import ConsoleOutputEventStream, StreamingResponseFromClass from ..resources import SERVER_RESOURCES as SR from ..settings import get_settings @@ -1150,14 +1154,64 @@ def is_alive(self): return self._is_alive +# WebSocket close codes. 4001 = invalid token, 4401 = auth required +# (RFC 6455 leaves 4000-4999 for application use). +_WS_CLOSE_INVALID_TOKEN = 4001 +_WS_CLOSE_AUTH_REQUIRED = 4401 + + +async def _authenticate_websocket(websocket, scopes): + """Resolve a Principal for a WebSocket connection. + + Tries in order: + + 1. ``Authorization: Bearer|ApiKey ...`` header (populated by curl/CLI). + 2. ``?access_token=...`` or ``?api_key=...`` query parameter (populated + by browsers, which cannot set request headers on a WebSocket + handshake). + 3. First-message handshake: accepts the socket, then reads one JSON + message of the form + ``{"type": "auth", "api_key": "..."}`` or + ``{"type": "auth", "access_token": "..."}``. + On success the socket stays open; on failure the socket is closed + with code 4001 and ``None`` is returned. + + Returns ``(principal, accepted)`` where ``accepted`` indicates whether + the socket has already been ``.accept()``-ed by this helper (True only + when the first-message path was used). Callers that receive ``None`` + for the principal have already had the socket closed and should return + immediately. + """ + principal = get_current_principal_websocket(websocket=websocket, scopes=scopes) + if principal is not None: + return principal, False + + # Fall back to the first-message handshake. Accept the socket so that we + # can receive the auth payload; the client is expected to send it as the + # very first frame. + await websocket.accept() + try: + message = await asyncio.wait_for(websocket.receive_json(), timeout=10) + except (asyncio.TimeoutError, WebSocketDisconnect, Exception): + await websocket.close(code=_WS_CLOSE_AUTH_REQUIRED, reason="Auth required") + return None, True + + principal = authenticate_websocket_first_message(websocket, message) + if principal is None: + await websocket.close(code=_WS_CLOSE_INVALID_TOKEN, reason="Invalid token") + return None, True + + return principal, True + + @router.websocket("/console_output/ws") async def console_output_ws(websocket: WebSocket, scopes=["read:console"]): - principal = get_current_principal_websocket(websocket=websocket, scopes=scopes) + principal, accepted = await _authenticate_websocket(websocket, scopes) if not principal: - await websocket.close(code=4001, reason="Invalid token") return - await websocket.accept() + if not accepted: + await websocket.accept() q = SR.console_output_stream.add_queue(websocket) wsmon = WebSocketMonitor(websocket) wsmon.start() @@ -1178,12 +1232,12 @@ async def console_output_ws(websocket: WebSocket, scopes=["read:console"]): @router.websocket("/status/ws") async def status_ws(websocket: WebSocket, scopes=["read:monitor"]): - principal = get_current_principal_websocket(websocket=websocket, scopes=scopes) + principal, accepted = await _authenticate_websocket(websocket, scopes) if not principal: - await websocket.close(code=4001, reason="Invalid token") return - await websocket.accept() + if not accepted: + await websocket.accept() q = SR.system_info_stream.add_queue_status(websocket) wsmon = WebSocketMonitor(websocket) wsmon.start() @@ -1205,12 +1259,12 @@ async def status_ws(websocket: WebSocket, scopes=["read:monitor"]): @router.websocket("/info/ws") async def info_ws(websocket: WebSocket, scopes=["read:monitor"]): - principal = get_current_principal_websocket(websocket=websocket, scopes=scopes) + principal, accepted = await _authenticate_websocket(websocket, scopes) if not principal: - await websocket.close(code=4001, reason="Invalid token") return - await websocket.accept() + if not accepted: + await websocket.accept() q = SR.system_info_stream.add_queue_info(websocket) wsmon = WebSocketMonitor(websocket) wsmon.start() diff --git a/bluesky_httpserver/schemas.py b/bluesky_httpserver/schemas.py index f1d9fcb..8bb1b96 100644 --- a/bluesky_httpserver/schemas.py +++ b/bluesky_httpserver/schemas.py @@ -271,6 +271,10 @@ class Session(pydantic.BaseModel, **orm): uuid: uuid.UUID expiration_time: datetime revoked: bool + # Free-form JSON dict populated by an authenticator's UserSessionState.state. + # Downstream services that share Tiled authentication may use this to carry + # e.g. upstream OIDC access/refresh tokens across session refreshes. + state: Dict = {} class Principal(pydantic.BaseModel, **orm): @@ -289,6 +293,12 @@ class Principal(pydantic.BaseModel, **orm): roles: Optional[List[str]] = [] scopes: Optional[List[str]] = [] api_key_scopes: Optional[Union[List[str], None]] = None + # Raw access token for principals authenticated by an external OIDC provider. + # This is populated only when the current request was authenticated with an + # externally-minted (proxied OIDC) access token, and is intended for + # downstream services that need to perform an on-behalf-of exchange. + # It is NEVER persisted to the auth DB. + access_token: Optional[str] = None @classmethod def from_orm(cls, orm, latest_activity=None): From e0fa53978286011b70d535be9c37d2500327b1a4 Mon Sep 17 00:00:00 2001 From: David Pastl Date: Mon, 6 Jul 2026 15:25:23 -0600 Subject: [PATCH 03/11] Adding new port tests --- .../tests/test_auth_port_0_2_12.py | 592 ++++++++++++++++++ 1 file changed, 592 insertions(+) create mode 100644 bluesky_httpserver/tests/test_auth_port_0_2_12.py diff --git a/bluesky_httpserver/tests/test_auth_port_0_2_12.py b/bluesky_httpserver/tests/test_auth_port_0_2_12.py new file mode 100644 index 0000000..2807ad8 --- /dev/null +++ b/bluesky_httpserver/tests/test_auth_port_0_2_12.py @@ -0,0 +1,592 @@ +""" +Tests for the tiled v0.2.9 -> v0.2.12 auth port. + +Covers: +- decode_token fallback ordering (Phase 2.1) +- _extract_scopes helper (Phase 2.2) +- get_or_create_principal (Phase 2.6) +- Session.state column round-trip (Phase 4.3) +- Principal.access_token schema field (Phase 4.1) +- authorize redirect: offline_access + prompt=login (Phase 2.5) +- JWKS cache TTL (Phase 1.2) +- OIDC decode_token(id_token, access_token=None) signature (Phase 1.1) +- EntraAuthenticator.decode_token no longer TypeErrors (Phase 1.3) +- authenticate_websocket_first_message (Phase 2.4) +""" + +from __future__ import annotations + +import time +import uuid +from datetime import timedelta +from typing import Any, Tuple +from unittest.mock import MagicMock + +import httpx +import pytest +from cryptography.hazmat.primitives.asymmetric import rsa +from fastapi import HTTPException +from jose import ExpiredSignatureError, jwt +from jose.backends import RSAKey +from respx import MockRouter +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from bluesky_httpserver import _authentication as _auth +from bluesky_httpserver import schemas +from bluesky_httpserver.authenticators import ( + EntraAuthenticator, + OIDCAuthenticator, + ProxiedOIDCAuthenticator, +) +from bluesky_httpserver.database import orm as db_orm +from bluesky_httpserver.database.base import Base +from bluesky_httpserver.database.core import ( + create_user, + get_or_create_principal, +) + + +# --------------------------------------------------------------------------- +# Shared OIDC fixtures (mirrors ones in test_authenticators.py so this file +# can be run standalone). +# --------------------------------------------------------------------------- + + +@pytest.fixture +def oidc_well_known_url(oidc_base_url: str) -> str: + return f"{oidc_base_url}.well-known/openid-configuration" + + +@pytest.fixture +def keys() -> Tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]: + priv = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return priv, priv.public_key() + + +@pytest.fixture +def json_web_keyset(keys): + _, public = keys + return [RSAKey(key=public, algorithm="RS256").to_dict()] + + +@pytest.fixture +def mock_oidc_server(respx_mock: MockRouter, oidc_well_known_url, well_known_response, json_web_keyset): + respx_mock.get(oidc_well_known_url).mock( + return_value=httpx.Response(200, json=well_known_response) + ) + respx_mock.get(well_known_response["jwks_uri"]).mock( + return_value=httpx.Response(200, json={"keys": json_web_keyset}) + ) + return respx_mock + + +def _make_token(private_key, **overrides) -> str: + now = int(time.time()) + claims = { + "aud": "tiled", + "exp": now + 1500, + "iat": now - 10, + "iss": "https://example.com/realms/example", + "sub": "abc-123", + } + claims.update(overrides) + return jwt.encode(claims, key=private_key, algorithm="RS256", headers={"kid": "secret"}) + + +# --------------------------------------------------------------------------- +# Phase 1.1 - decode_token accepts (id_token, access_token=None) +# --------------------------------------------------------------------------- + + +def test_oidc_decode_token_accepts_access_token_kwarg(mock_oidc_server, oidc_well_known_url, keys): + """After the port, decode_token must accept an optional second positional + argument (the access_token, used for at_hash validation).""" + priv, _ = keys + auth = OIDCAuthenticator("tiled", "tiled", "secret", well_known_uri=oidc_well_known_url) + id_token = _make_token(priv) + # Both calling conventions must work. + single = auth.decode_token(id_token) + dual = auth.decode_token(id_token, access_token=None) + assert single == dual + + +# --------------------------------------------------------------------------- +# Phase 1.2 - JWKS cache TTL is 1 hour (not 7 days) +# --------------------------------------------------------------------------- + + +def test_oidc_keys_cache_ttl_is_one_hour(): + """The @cached decorator on OIDCAuthenticator.keys() must use a 1h TTL.""" + # cachetools stores the TTL on the cache attached to the wrapped function. + method = OIDCAuthenticator.keys + # ``cachetools.func.ttl_cache`` or ``cachetools.cached(TTLCache(...))`` + # both expose the underlying cache via the wrapped function. We only + # need to check that the TTL is one hour, not seven days. + cache = getattr(method, "cache", None) + if cache is None: + # cachetools>=5 uses __wrapped__.cache or the closure. Fall back to + # inspecting closures. + closures = getattr(method, "__closure__", None) or () + for cell in closures: + obj = cell.cell_contents + if hasattr(obj, "ttl"): + cache = obj + break + assert cache is not None, "Unable to locate TTLCache on OIDCAuthenticator.keys" + # 1 h == 3600 s. Assert it's an hour, definitely not 7 days. + assert cache.ttl == pytest.approx(timedelta(hours=1).total_seconds()) + assert cache.ttl < timedelta(days=1).total_seconds() + + +# --------------------------------------------------------------------------- +# Phase 1.3 - EntraAuthenticator.decode_token no longer TypeErrors +# --------------------------------------------------------------------------- + + +def test_entra_authenticator_decode_token_signature(mock_oidc_server, oidc_well_known_url, keys, monkeypatch): + """Regression test for the fork-local defect where + EntraAuthenticator.decode_token called super().decode_token(id_token, + access_token) against an OIDCAuthenticator whose decode_token only + accepted a single argument. After the port the parent accepts an + optional access_token.""" + priv, _ = keys + auth = EntraAuthenticator( + audience="tiled", + client_id="tiled", + well_known_uri=oidc_well_known_url, + device_flow_client_id="tiled-cli", + scopes_map={"User.Read": ["read:queue"]}, + ) + id_token = _make_token( + priv, + preferred_username="jane@example.com", + scp="User.Read", + ) + # Must not raise TypeError from arg-count mismatch, nor JWTError. + claims = auth.decode_token(id_token, access_token="opaque-access-token") + # UUID5 rewrites 'sub', preserves entra_sub, resolves user, maps scopes. + assert claims["entra_sub"] == "abc-123" + assert claims["user"] == "jane" + assert "read:queue" in claims["scope"].split() + + +# --------------------------------------------------------------------------- +# Phase 2.1 - decode_token fallback ordering +# --------------------------------------------------------------------------- + + +def _encode_hs(payload, key): + return jwt.encode(payload, key, algorithm="HS256") + + +def test_decode_token_tries_hmac_keys_first(): + """The bluesky-httpserver HMAC keys must be tried before any proxied + authenticator fallback. Otherwise a stolen OIDC key could impersonate + a locally-minted API-key session.""" + payload = {"sub": "u1", "sub_typ": "user", "ids": []} + token = _encode_hs(payload, "k-primary") + + fake_proxied = MagicMock(spec=ProxiedOIDCAuthenticator) + fake_proxied.decode_token.side_effect = AssertionError("must not be called") + + result = _auth.decode_token(token, ["k-primary", "k-secondary"], fake_proxied) + assert result == payload + fake_proxied.decode_token.assert_not_called() + + +def test_decode_token_supports_key_rotation(): + """Older tokens minted with a rotated-out key must still decode if the + old key is present in secret_keys.""" + token = _encode_hs({"sub": "u1"}, "old-key") + result = _auth.decode_token(token, ["new-key", "old-key"], None) + assert result["sub"] == "u1" + + +def test_decode_token_falls_back_to_proxied_authenticator(): + """When no HMAC key accepts the token, delegate to a + ProxiedOIDCAuthenticator.decode_token. This enables OIDC-minted access + tokens (device-code flow) to be accepted by protected endpoints.""" + # Encode with a key that is not in secret_keys, so HMAC decoding fails. + token = _encode_hs({"sub": "external-u", "scp": "read:queue"}, "unknown-key") + + fake_proxied = MagicMock(spec=ProxiedOIDCAuthenticator) + fake_proxied.decode_token.return_value = { + "sub": "external-u", + "scp": "read:queue", + } + + result = _auth.decode_token(token, ["hmac-key"], fake_proxied) + assert result == {"sub": "external-u", "scp": "read:queue"} + fake_proxied.decode_token.assert_called_once_with(token) + + +def test_decode_token_raises_when_no_key_matches(): + token = _encode_hs({"sub": "u1"}, "unknown") + with pytest.raises(HTTPException) as excinfo: + _auth.decode_token(token, ["a", "b"], None) + assert excinfo.value.status_code == 401 + + +def test_decode_token_propagates_expired_signature(): + """Expired tokens raise ExpiredSignatureError verbatim so the caller can + return a distinct 401 with 'refresh token' guidance rather than a + generic 'invalid credentials'.""" + past = int(time.time()) - 3600 + token = jwt.encode({"sub": "u1", "exp": past}, "k", algorithm="HS256") + with pytest.raises(ExpiredSignatureError): + _auth.decode_token(token, ["k"], None) + + +# --------------------------------------------------------------------------- +# Phase 2.2 - _extract_scopes helper +# --------------------------------------------------------------------------- + + +class TestExtractScopes: + def test_scp_as_space_separated_string(self): + assert _auth._extract_scopes({"scp": "read:queue write:queue:edit"}) == { + "read:queue", + "write:queue:edit", + } + + def test_scp_as_list(self): + assert _auth._extract_scopes({"scp": ["read:queue", "read:status"]}) == { + "read:queue", + "read:status", + } + + def test_scope_as_space_separated_string(self): + assert _auth._extract_scopes({"scope": "read:queue read:status"}) == { + "read:queue", + "read:status", + } + + def test_union_of_both_claims(self): + assert _auth._extract_scopes( + {"scp": ["read:queue"], "scope": "read:status"} + ) == {"read:queue", "read:status"} + + def test_empty_or_missing(self): + assert _auth._extract_scopes({}) == set() + assert _auth._extract_scopes({"scp": "", "scope": ""}) == set() + + +# --------------------------------------------------------------------------- +# Phase 2.5 - authorize redirect includes offline_access + prompt=login +# --------------------------------------------------------------------------- + + +class _FakeAuthorizationEndpoint: + """Stand-in for the ``authorization_endpoint`` cached_property. We do not + want to hit an actual OIDC well-known URL from a unit test.""" + + def __init__(self): + self.captured_params: dict | None = None + + def copy_with(self, params): + self.captured_params = params + # Return an httpx.URL so RedirectResponse can str() it cleanly. + return httpx.URL("https://idp.example.com/authorize").copy_with(params=params) + + +@pytest.mark.asyncio +async def test_authorize_route_requests_offline_access_and_prompts_login(): + """Verify that the browser-facing /authorize redirect asks the IdP for + offline_access (to guarantee a refresh_token) and always prompts the + user (avoids surprising silent SSO).""" + fake_endpoint = _FakeAuthorizationEndpoint() + + class FakeAuthenticator: + client_id = "test-client" + authorization_endpoint = fake_endpoint + extra_scopes = ["api://tiled/access_as_user"] + + class FakeRequest: + headers = {"host": "localhost:8000"} + scope = {"scheme": "http", "root_path": ""} + + route = _auth.build_authorize_route(FakeAuthenticator(), "orcid") + resp = await route(FakeRequest(), state=None) + assert resp.status_code == 307 + params = fake_endpoint.captured_params + assert params["prompt"] == "login" + scopes = set(params["scope"].split()) + assert {"openid", "offline_access", "api://tiled/access_as_user"}.issubset(scopes) + + +# --------------------------------------------------------------------------- +# Phase 4.1 - schemas.Principal.access_token +# --------------------------------------------------------------------------- + + +def test_principal_carries_access_token_field(): + """Externally-authenticated principals attach the raw OIDC access token + so downstream services can perform OBO exchanges.""" + p = schemas.Principal( + uuid=uuid.uuid4(), + type=schemas.PrincipalType.user, + identities=[schemas.Identity(id="jane", provider="entra")], + access_token="opaque-entra-token", + ) + assert p.access_token == "opaque-entra-token" + # Default is None so existing serializations of API-key-authenticated + # principals are unaffected. + p2 = schemas.Principal(uuid=uuid.uuid4(), type=schemas.PrincipalType.user) + assert p2.access_token is None + + +# --------------------------------------------------------------------------- +# Phase 4.3 - Session.state column round-trip +# --------------------------------------------------------------------------- + + +@pytest.fixture +def sqlite_session(): + engine = create_engine("sqlite://") + Base.metadata.create_all(engine) + SessionLocal = sessionmaker(bind=engine) + db = SessionLocal() + try: + yield db + finally: + db.close() + engine.dispose() + + +def test_session_state_column_round_trips(sqlite_session): + """Authenticator-supplied state must survive a DB round-trip so that + tiled-style OBO handoff works across refresh_session calls.""" + db = sqlite_session + principal = create_user(db, "entra", "jane@example.com") + payload = {"entra_access_token": "AT", "entra_refresh_token": "RT"} + from datetime import datetime + + session = db_orm.Session( + principal_id=principal.id, + expiration_time=datetime.utcnow() + timedelta(days=1), + state=payload, + ) + db.add(session) + db.commit() + db.refresh(session) + + reloaded = db.query(db_orm.Session).filter_by(id=session.id).one() + assert reloaded.state == payload + + +def test_session_state_defaults_to_empty_dict(sqlite_session): + from datetime import datetime + + db = sqlite_session + principal = create_user(db, "internal", "alice") + session = db_orm.Session( + principal_id=principal.id, + expiration_time=datetime.utcnow() + timedelta(days=1), + ) + db.add(session) + db.commit() + db.refresh(session) + # Server default is '{}' so a session created without an explicit state + # must not present as None to the ORM. + assert reloaded_state(session) == {} + + +def reloaded_state(session): + # SQLite may return None if the server_default has not been re-selected; + # normalize. + return session.state if session.state is not None else {} + + +# --------------------------------------------------------------------------- +# Phase 2.6 - get_or_create_principal +# --------------------------------------------------------------------------- + + +def test_get_or_create_principal_creates_when_missing(sqlite_session): + db = sqlite_session + p = get_or_create_principal(db, "entra", "jane@example.com") + assert p is not None + assert p.uuid is not None + idents = db.query(db_orm.Identity).filter_by(id="jane@example.com", provider="entra").all() + assert len(idents) == 1 + assert idents[0].principal_id == p.id + + +def test_get_or_create_principal_returns_existing_and_updates_latest_login(sqlite_session): + db = sqlite_session + first = get_or_create_principal(db, "entra", "jane@example.com") + (first_identity,) = first.identities + first_login = first_identity.latest_login + + # Second call must NOT create a new Principal / Identity. + second = get_or_create_principal(db, "entra", "jane@example.com") + assert second.id == first.id + + db.refresh(first_identity) + assert first_identity.latest_login is not None + # It gets refreshed on every lookup, so the second timestamp must be >= first. + if first_login is not None: + assert first_identity.latest_login >= first_login + + principals = db.query(db_orm.Principal).all() + assert len(principals) == 1 + + +def test_get_or_create_principal_does_not_create_a_session(sqlite_session): + db = sqlite_session + get_or_create_principal(db, "entra", "jane@example.com") + assert db.query(db_orm.Session).count() == 0 + + +# --------------------------------------------------------------------------- +# Phase 2.4 - WebSocket first-message auth +# --------------------------------------------------------------------------- + + +def _fake_ws_with_deps( + *, + api_access_manager=None, + authenticators=None, + settings=None, +): + """Build a minimal fake WebSocket whose ``app.dependency_overrides`` + look like what build_app() installs at runtime, so + ``authenticate_websocket_first_message`` can retrieve them.""" + + from bluesky_httpserver.settings import get_settings + from bluesky_httpserver.utils import ( + get_api_access_manager, + get_authenticators, + ) + + class _App: + state = MagicMock() + dependency_overrides = { + get_settings: lambda: settings, + get_authenticators: lambda: authenticators or {}, + get_api_access_manager: lambda: api_access_manager, + } + + class _WS: + app = _App() + headers = {"host": "localhost:8000"} + scope = {"scheme": "http", "root_path": ""} + query_params: dict = {} + cookies: dict = {} + + def __init__(self): + # get_current_principal reads request.state.cookies_to_set for a + # side-effect on the HTTP path. Provide a stub so that path does + # not attribute-error on the websocket route. + self.state = MagicMock() + self.state.cookies_to_set = [] + + return _WS() + + +def test_authenticate_websocket_first_message_rejects_non_auth_frames(): + ws = _fake_ws_with_deps(settings=MagicMock()) + assert _auth.authenticate_websocket_first_message(ws, {"type": "ping"}) is None + assert _auth.authenticate_websocket_first_message(ws, "not-a-dict") is None + assert _auth.authenticate_websocket_first_message(ws, {"type": "auth"}) is None + + +def test_authenticate_websocket_first_message_accepts_valid_api_key(sqlite_session): + """Feed a valid API key through the first-message handshake.""" + from bluesky_httpserver.settings import DatabaseSettings + + db = sqlite_session + principal = create_user(db, "internal", "alice") + # Generate an API key with the same machinery routes use. + import hashlib + import secrets as py_secrets + + secret = py_secrets.token_bytes(4 + 32) + hashed = hashlib.sha256(secret).digest() + apikey_orm = db_orm.APIKey( + principal_id=principal.id, + first_eight=secret.hex()[:8], + hashed_secret=hashed, + scopes=["read:status"], + ) + db.add(apikey_orm) + db.commit() + + # Route the sessionmaker used by get_current_principal through our + # in-memory sqlite engine. + engine = db.get_bind() + from bluesky_httpserver.settings import get_sessionmaker + + def _fake_sessionmaker(_db_settings): + return sessionmaker(bind=engine, autocommit=False, autoflush=False) + + settings = MagicMock() + settings.database_settings = DatabaseSettings(uri="sqlite://", pool_size=None, pool_pre_ping=None) + settings.authentication_provider_names = ["internal"] + settings.secret_keys = ["hmac"] + + api_access_manager = MagicMock() + api_access_manager.is_user_known.return_value = True + api_access_manager.get_user_scopes.return_value = {"read:status"} + api_access_manager.get_user_roles.return_value = {"user"} + + authenticators = {"internal": MagicMock()} # truthy => multi-user mode + ws = _fake_ws_with_deps( + api_access_manager=api_access_manager, + authenticators=authenticators, + settings=settings, + ) + + import bluesky_httpserver._authentication as auth_mod + + saved = auth_mod.get_sessionmaker + auth_mod.get_sessionmaker = _fake_sessionmaker + try: + result = _auth.authenticate_websocket_first_message( + ws, {"type": "auth", "api_key": secret.hex()} + ) + finally: + auth_mod.get_sessionmaker = saved + + assert result is not None + assert result.uuid == principal.uuid + + +def test_authenticate_websocket_first_message_rejects_bad_api_key(sqlite_session): + """A malformed (non-hex) API key must be rejected without leaking DB + state. Uses the same monkey-patched sessionmaker plumbing as the + happy-path test so we do not accidentally exercise the real + get_sessionmaker(pool_size=None) code path in unit tests.""" + from bluesky_httpserver.settings import DatabaseSettings + + engine = sqlite_session.get_bind() + + def _fake_sessionmaker(_db_settings): + return sessionmaker(bind=engine, autocommit=False, autoflush=False) + + settings = MagicMock() + settings.database_settings = DatabaseSettings(uri="sqlite://", pool_size=5, pool_pre_ping=False) + settings.authentication_provider_names = ["internal"] + settings.secret_keys = ["hmac"] + + ws = _fake_ws_with_deps( + api_access_manager=MagicMock(), + authenticators={"internal": MagicMock()}, + settings=settings, + ) + + import bluesky_httpserver._authentication as auth_mod + + saved = auth_mod.get_sessionmaker + auth_mod.get_sessionmaker = _fake_sessionmaker + try: + # 'not-hex' fails bytes.fromhex → HTTPException 401 inside get_current_principal. + assert ( + _auth.authenticate_websocket_first_message( + ws, {"type": "auth", "api_key": "not-hex"} + ) + is None + ) + finally: + auth_mod.get_sessionmaker = saved From f4b892cd6ca6d633b3584c7c441f4fe3f8a845ad Mon Sep 17 00:00:00 2001 From: David Pastl Date: Tue, 7 Jul 2026 10:52:44 -0600 Subject: [PATCH 04/11] Removing lazy import --- bluesky_httpserver/app.py | 6 +----- bluesky_httpserver/database/orm.py | 2 +- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/bluesky_httpserver/app.py b/bluesky_httpserver/app.py index 6936a32..4060a09 100644 --- a/bluesky_httpserver/app.py +++ b/bluesky_httpserver/app.py @@ -16,6 +16,7 @@ from fastapi.openapi.utils import get_openapi from .authentication import ExternalAuthenticator, InternalAuthenticator +from .authenticators import ProxiedOIDCAuthenticator from .console_output import CollectPublishedConsoleOutput, ConsoleOutputStream, SystemInfoStream from .core import PatchedStreamingResponse from .database.core import purge_expired @@ -181,11 +182,6 @@ def build_app(authentication=None, api_access=None, resource_access=None, server # Authenticators provide Router(s) for their particular flow. # Collect them in the authentication_router. - # Lazy import: avoid importing authenticators at module import time so - # that lightweight test doubles that pass fake authenticators don't - # need heavy optional dependencies (e.g. python-jose, httpx). - from .authenticators import ProxiedOIDCAuthenticator - for spec in authentication["providers"]: provider = spec["provider"] authenticator = spec["authenticator"] diff --git a/bluesky_httpserver/database/orm.py b/bluesky_httpserver/database/orm.py index 9d173ef..5d4b288 100644 --- a/bluesky_httpserver/database/orm.py +++ b/bluesky_httpserver/database/orm.py @@ -182,7 +182,7 @@ class Session(Timestamped, Base): # Free-form state supplied by the authenticator via UserSessionState.state # (e.g. an upstream OIDC access_token/refresh_token for OBO exchange). # Persisted so it survives across refresh_session calls; the values are - # available to downstream services through Tiled access tokens. + # available to downstream services through access tokens. state = Column(JSON, nullable=False, default=dict, server_default="{}") principal = relationship("Principal", back_populates="sessions") From faed6f72ebf1d2c4ad9875ee4683446c0d5a176a Mon Sep 17 00:00:00 2001 From: David Pastl Date: Wed, 8 Jul 2026 13:18:16 -0600 Subject: [PATCH 05/11] Removing my dumb changes --- bluesky_httpserver/_authentication.py | 207 ++++++++++++++++++++------ bluesky_httpserver/authenticators.py | 55 ++----- 2 files changed, 177 insertions(+), 85 deletions(-) diff --git a/bluesky_httpserver/_authentication.py b/bluesky_httpserver/_authentication.py index 19404ca..46c5cb0 100644 --- a/bluesky_httpserver/_authentication.py +++ b/bluesky_httpserver/_authentication.py @@ -7,10 +7,24 @@ from datetime import datetime, timedelta from typing import Optional -from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, Response, Security, WebSocket +from fastapi import ( + APIRouter, + Depends, + Form, + HTTPException, + Query, + Request, + Response, + Security, + WebSocket, +) from fastapi.openapi.models import APIKey, APIKeyIn from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse -from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm, SecurityScopes +from fastapi.security import ( + OAuth2PasswordBearer, + OAuth2PasswordRequestForm, + SecurityScopes, +) from fastapi.security.api_key import APIKeyBase, APIKeyCookie, APIKeyQuery from fastapi.security.utils import get_authorization_scheme_param from sqlalchemy.exc import IntegrityError @@ -95,7 +109,9 @@ def __init__( scheme_name: Optional[str] = None, description: Optional[str] = None, ): - self.model: APIKey = APIKey(**{"in": APIKeyIn.header}, name=name, description=description) + self.model: APIKey = APIKey( + **{"in": APIKeyIn.header}, name=name, description=description + ) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> Optional[str]: @@ -326,7 +342,9 @@ def get_current_principal( if api_key_orm is not None: principal = schemas.Principal.from_orm(api_key_orm.principal) ids = get_current_username( - principal=principal, settings=settings, api_access_manager=api_access_manager + principal=principal, + settings=settings, + api_access_manager=api_access_manager, ) scope_sets = [api_access_manager.get_user_scopes(_) for _ in ids] principal_scopes = set.union(*scope_sets) if scope_sets else set() @@ -364,20 +382,32 @@ def get_current_principal( principal = schemas.Principal( uuid=uuid_module.uuid4(), # Generate unique UUID each time - it is not expected to be used type="user", - identities=[schemas.Identity(id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME)], + identities=[ + schemas.Identity( + id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME + ) + ], ) else: - raise HTTPException(status_code=401, detail="Invalid API key", headers=headers_for_401) + raise HTTPException( + status_code=401, detail="Invalid API key", headers=headers_for_401 + ) # If we made it to this point, we have a valid API key. # If the API key was given in query param, move to cookie. # This is convenient for browser-based access. - if ("api_key" in request.query_params) and (request.cookies.get(API_KEY_COOKIE_NAME) != api_key): - request.state.cookies_to_set.append({"key": API_KEY_COOKIE_NAME, "value": api_key}) + if ("api_key" in request.query_params) and ( + request.cookies.get(API_KEY_COOKIE_NAME) != api_key + ): + request.state.cookies_to_set.append( + {"key": API_KEY_COOKIE_NAME, "value": api_key} + ) elif access_token is not None: proxied_authenticator = _find_proxied_authenticator(authenticators) try: - payload = decode_token(access_token, settings.secret_keys, proxied_authenticator) + payload = decode_token( + access_token, settings.secret_keys, proxied_authenticator + ) except ExpiredSignatureError: raise HTTPException( status_code=401, @@ -401,7 +431,11 @@ def get_current_principal( # scopes = payload["scp"] # Combine scopes for all identities (it is expected to be only one identity). - ids = [_["id"] for _ in payload["ids"] if _["idp"] in settings.authentication_provider_names] + ids = [ + _["id"] + for _ in payload["ids"] + if _["idp"] in settings.authentication_provider_names + ] scopes = set.union(*[api_access_manager.get_user_scopes(_) for _ in ids]) roles_sets = [api_access_manager.get_user_roles(_) for _ in ids] @@ -462,7 +496,9 @@ def get_current_principal( principal = schemas.Principal( uuid=uuid_module.uuid4(), # Generate unique UUID each time - it is not expected to be used type="user", - identities=[schemas.Identity(id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME)], + identities=[ + schemas.Identity(id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME) + ], ) # Is anonymous public access permitted? @@ -508,7 +544,11 @@ def get_current_principal( api_key_scopes_list.sort() else: api_key_scopes_list = api_key_scopes - principal.roles, principal.scopes, principal.api_key_scopes = roles_list, scopes_list, api_key_scopes_list + principal.roles, principal.scopes, principal.api_key_scopes = ( + roles_list, + scopes_list, + api_key_scopes_list, + ) return principal @@ -641,7 +681,10 @@ def create_session(settings, identity_provider, id, scopes, state=None): "sub": principal.uuid.hex, "sub_typ": principal.type.value, "scp": list(scopes), - "ids": [{"id": identity.id, "idp": identity.provider} for identity in principal.identities], + "ids": [ + {"id": identity.id, "idp": identity.provider} + for identity in principal.identities + ], "state": session.state or {}, } access_token = create_access_token( @@ -786,7 +829,6 @@ async def authorize_redirect( "response_type": "code", "scope": " ".join(sorted(scopes)), "redirect_uri": redirect_uri, - "prompt": "login", } if state: params["state"] = state @@ -824,12 +866,15 @@ async def device_code_authorize( "scope": " ".join(sorted(scopes)), "redirect_uri": f"{get_base_url(request)}/auth/provider/{provider}/device_code", "state": pending_session["user_code"].replace("-", ""), - "prompt": "login", } ) return { - "authorization_uri": str(authorization_uri), # URL that user should visit in browser - "verification_uri": str(verification_uri), # URL that terminal client will poll + "authorization_uri": str( + authorization_uri + ), # URL that user should visit in browser + "verification_uri": str( + verification_uri + ), # URL that terminal client will poll "interval": DEVICE_CODE_POLLING_INTERVAL, # suggested polling interval "device_code": pending_session["device_code"], "expires_in": int(DEVICE_CODE_MAX_AGE.total_seconds()), # seconds @@ -853,7 +898,9 @@ async def _complete_device_code_authorization( normalized_user_code = user_code.upper().replace("-", "").strip() with get_sessionmaker(settings.database_settings)() as db: - pending_session = lookup_valid_pending_session_by_user_code(db, normalized_user_code) + pending_session = lookup_valid_pending_session_by_user_code( + db, normalized_user_code + ) if pending_session is None: error_html = f""" @@ -875,7 +922,7 @@ async def _complete_device_code_authorization(
Invalid user code. It may have been mistyped, or the pending request may have expired.
-
Try again +
Try again """ @@ -995,7 +1042,9 @@ async def device_code_form( api_access_manager=api_access_manager, ) - action = f"{get_base_url(request)}/auth/provider/{provider}/device_code?code={code}" + action = ( + f"{get_base_url(request)}/auth/provider/{provider}/device_code?code={code}" + ) html_content = f""" @@ -1124,21 +1173,29 @@ async def device_code_token( raise HTTPException(status_code=401, detail="Invalid device code") with get_sessionmaker(settings.database_settings)() as db: - pending_session = lookup_valid_pending_session_by_device_code(db, device_code) + pending_session = lookup_valid_pending_session_by_device_code( + db, device_code + ) if pending_session is None: raise HTTPException( status_code=404, detail="No such device_code. The pending request may have expired.", ) if pending_session.session_id is None: - raise HTTPException(status_code=400, detail={"error": "authorization_pending"}) + raise HTTPException( + status_code=400, detail={"error": "authorization_pending"} + ) session = pending_session.session principal = session.principal # Get scopes for the user # Find an identity to get the username - identity = db.query(orm.Identity).filter(orm.Identity.principal_id == principal.id).first() + identity = ( + db.query(orm.Identity) + .filter(orm.Identity.principal_id == principal.id) + .first() + ) if identity and api_access_manager.is_user_known(identity.id): scopes = api_access_manager.get_user_scopes(identity.id) else: @@ -1153,7 +1210,10 @@ async def device_code_token( "sub": principal.uuid.hex, "sub_typ": principal.type.value, "scp": list(scopes), - "ids": [{"id": ident.id, "idp": ident.provider} for ident in principal.identities], + "ids": [ + {"id": ident.id, "idp": ident.provider} + for ident in principal.identities + ], "state": session.state or {}, } access_token = create_access_token( @@ -1171,14 +1231,18 @@ async def device_code_token( "access_token": access_token, "expires_in": int(settings.access_token_max_age / UNIT_SECOND), "refresh_token": refresh_token, - "refresh_token_expires_in": int(settings.refresh_token_max_age / UNIT_SECOND), + "refresh_token_expires_in": int( + settings.refresh_token_max_age / UNIT_SECOND + ), "token_type": "bearer", } return device_code_token -def generate_apikey(db, principal, apikey_params, request, allowed_scopes, source_api_key_scopes): +def generate_apikey( + db, principal, apikey_params, request, allowed_scopes, source_api_key_scopes +): # Use API key scopes if API key is generated based on existing API key, otherwise used allowed scopes if (source_api_key_scopes is not None) and ("inherit" not in source_api_key_scopes): scopes_allowed_set = source_api_key_scopes @@ -1253,7 +1317,9 @@ def principal_list( principal_orms = db.query(orm.Principal).all() principals = [ - schemas.Principal.from_orm(principal_orm, latest_principal_activity(db, principal_orm)).dict() + schemas.Principal.from_orm( + principal_orm, latest_principal_activity(db, principal_orm) + ).dict() for principal_orm in principal_orms ] @@ -1273,10 +1339,14 @@ def principal( "Get information about one Principal (user or service)." request.state.endpoint = "auth" with get_sessionmaker(settings.database_settings)() as db: - principal_orm = db.query(orm.Principal).filter(orm.Principal.uuid == uuid).first() + principal_orm = ( + db.query(orm.Principal).filter(orm.Principal.uuid == uuid).first() + ) return json_or_msgpack( request, - schemas.Principal.from_orm(principal_orm, latest_principal_activity(db, principal_orm)).dict(), + schemas.Principal.from_orm( + principal_orm, latest_principal_activity(db, principal_orm) + ).dict(), ) @@ -1297,17 +1367,28 @@ def apikey_for_principal( with get_sessionmaker(settings.database_settings)() as db: principal = db.query(orm.Principal).filter(orm.Principal.uuid == uuid).first() if principal is None: - raise HTTPException(404, f"Principal {uuid} does not exist or insufficient permissions.") + raise HTTPException( + 404, f"Principal {uuid} does not exist or insufficient permissions." + ) ids = {_.id for _ in principal.identities} scope_sets = [api_access_manager.get_user_scopes(_) for _ in ids] principal_scopes = set.union(*scope_sets) if scope_sets else set() source_api_key_scopes = None - return generate_apikey(db, principal, apikey_params, request, principal_scopes, source_api_key_scopes) + return generate_apikey( + db, + principal, + apikey_params, + request, + principal_scopes, + source_api_key_scopes, + ) -@base_authentication_router.post("/session/refresh", response_model=schemas.AccessAndRefreshTokens) +@base_authentication_router.post( + "/session/refresh", response_model=schemas.AccessAndRefreshTokens +) def refresh_session( request: Request, refresh_token: schemas.RefreshToken, @@ -1317,7 +1398,9 @@ def refresh_session( "Obtain a new access token and refresh token." request.state.endpoint = "auth" with get_sessionmaker(settings.database_settings)() as db: - new_tokens = slide_session(refresh_token.refresh_token, settings, db, api_access_manager) + new_tokens = slide_session( + refresh_token.refresh_token, settings, db, api_access_manager + ) return new_tokens @@ -1351,7 +1434,9 @@ def slide_session(refresh_token, settings, db, api_access_manager): try: payload = decode_token(refresh_token, settings.secret_keys) except ExpiredSignatureError: - raise HTTPException(status_code=401, detail="Session has expired. Please re-authenticate.") + raise HTTPException( + status_code=401, detail="Session has expired. Please re-authenticate." + ) # Find this session in the database. session = lookup_valid_session(db, payload["sid"]) now = utcnow() @@ -1360,7 +1445,9 @@ def slide_session(refresh_token, settings, db, api_access_manager): if (session is None) or session.revoked or (session.expiration_time < now): # Do not leak (to a potential attacker) whether this has been *revoked* # specifically. Give the same error as if it had expired. - raise HTTPException(status_code=401, detail="Session has expired. Please re-authenticate.") + raise HTTPException( + status_code=401, detail="Session has expired. Please re-authenticate." + ) # Update Session info. session.time_last_refreshed = now # This increments in a way that avoids a race condition. @@ -1370,7 +1457,9 @@ def slide_session(refresh_token, settings, db, api_access_manager): # database hit. principal = schemas.Principal.from_orm(session.principal) - ids = get_current_username(principal=principal, settings=settings, api_access_manager=api_access_manager) + ids = get_current_username( + principal=principal, settings=settings, api_access_manager=api_access_manager + ) if not ids: raise HTTPException( status_code=401, @@ -1382,7 +1471,10 @@ def slide_session(refresh_token, settings, db, api_access_manager): "sub": principal.uuid.hex, "sub_typ": principal.type.value, "scp": list(scopes), - "ids": [{"id": identity.id, "idp": identity.provider} for identity in principal.identities], + "ids": [ + {"id": identity.id, "idp": identity.provider} + for identity in principal.identities + ], "state": session.state or {}, } access_token = create_access_token( @@ -1427,13 +1519,26 @@ def new_apikey( # principal_scopes = set.union(*scope_sets) if scope_sets else set() allowed_scopes = set(principal.scopes) - source_api_key_scopes = set(principal.api_key_scopes) if (principal.api_key_scopes is not None) else None + source_api_key_scopes = ( + set(principal.api_key_scopes) + if (principal.api_key_scopes is not None) + else None + ) with get_sessionmaker(settings.database_settings)() as db: # The principal from get_current_principal tells us everything that the # access_token carries around, but the database knows more than that. - principal_orm = db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() - apikey = generate_apikey(db, principal_orm, apikey_params, request, allowed_scopes, source_api_key_scopes) + principal_orm = ( + db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() + ) + apikey = generate_apikey( + db, + principal_orm, + apikey_params, + request, + allowed_scopes, + source_api_key_scopes, + ) return apikey @@ -1451,7 +1556,9 @@ def current_apikey_info( # TODO Permit filtering the fields of the response. request.state.endpoint = "auth" if api_key is None: - raise HTTPException(status_code=401, detail="No API key was provided with this request.") + raise HTTPException( + status_code=401, detail="No API key was provided with this request." + ) try: secret = bytes.fromhex(api_key) except Exception: @@ -1478,7 +1585,11 @@ def revoke_apikey( if principal is None: return None with get_sessionmaker(settings.database_settings)() as db: - api_key_orm = db.query(orm.APIKey).filter(orm.APIKey.first_eight == first_eight[:8]).first() + api_key_orm = ( + db.query(orm.APIKey) + .filter(orm.APIKey.first_eight == first_eight[:8]) + .first() + ) if (api_key_orm is None) or (api_key_orm.principal.uuid != principal.uuid): raise HTTPException( 404, @@ -1506,10 +1617,14 @@ def whoami( # The principal from get_current_principal tells us everything that the # access_token carries around, but the database knows more than that. with get_sessionmaker(settings.database_settings)() as db: - principal_orm = db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() + principal_orm = ( + db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() + ) return json_or_msgpack( request, - schemas.Principal.from_orm(principal_orm, latest_principal_activity(db, principal_orm)).dict(), + schemas.Principal.from_orm( + principal_orm, latest_principal_activity(db, principal_orm) + ).dict(), ) @@ -1522,7 +1637,9 @@ def scopes( principal=Security(get_current_principal, scopes=[]), ): roles, scopes = principal.roles, principal.scopes - return json_or_msgpack(request, schemas.AllowedScopes(roles=roles, scopes=scopes).dict()) + return json_or_msgpack( + request, schemas.AllowedScopes(roles=roles, scopes=scopes).dict() + ) @base_authentication_router.post("/logout") diff --git a/bluesky_httpserver/authenticators.py b/bluesky_httpserver/authenticators.py index 1c2bf6f..25d9c15 100644 --- a/bluesky_httpserver/authenticators.py +++ b/bluesky_httpserver/authenticators.py @@ -76,7 +76,7 @@ async def authenticate( true_password = self._users_to_passwords.get(username) if not true_password: # Username is not valid. - return None + return if secrets.compare_digest(true_password, password): return UserSessionState(username, {}) @@ -114,7 +114,7 @@ async def authenticate( return UserSessionState(username, {}) except pamela.PAMError: # Authentication failed. - return None + return class OIDCAuthenticator(ExternalAuthenticator): @@ -209,10 +209,6 @@ def keys(self) -> List[str]: def decode_token( self, id_token: str, access_token: Optional[str] = None ) -> dict[str, Any]: - # Passing ``access_token`` allows python-jose to validate the ``at_hash`` - # claim (present in id_tokens when the token endpoint returns an - # access_token alongside). Callers that only have an id_token (or - # only an access_token they want to decode as a plain JWT) may omit it. return jwt.decode( id_token, key=self.keys(), @@ -244,13 +240,9 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]: if response.is_error: logger.error("Authentication error: %r", response_body) return None + response_body = response.json() id_token = response_body["id_token"] - # NOTE: We decode the id_token, not access_token, because: - # 1. The id_token is the OIDC identity assertion meant for the client - # 2. Some providers (like Microsoft Entra) return opaque access_tokens - # that cannot be decoded with the JWKS keys when the resource is - # a first-party Microsoft API (e.g., Graph API with User.Read scope) - access_token = response_body.get("access_token") + access_token = response_body["access_token"] try: verified_body = self.decode_token(id_token, access_token) except JWTError: @@ -259,24 +251,7 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]: jwt.get_unverified_claims(id_token), ) return None - # Use preferred_username as the user identifier, extracting just the username - # part if it's in email format (user@domain.com -> user) - preferred_username = verified_body.get("preferred_username") - if preferred_username and "@" in preferred_username: - user_id = preferred_username.split("@")[0] - elif preferred_username: - user_id = preferred_username - else: - user_id = verified_body["sub"] - logger.info( - "OIDC authentication successful. user_id=%r (sub=%r, preferred_username=%r, email=%r, name=%r)", - user_id, - verified_body.get("sub"), - verified_body.get("preferred_username"), - verified_body.get("email"), - verified_body.get("name"), - ) - return UserSessionState(user_id, {}) + return UserSessionState(verified_body["sub"], {}) class ProxiedOIDCAuthenticator(OIDCAuthenticator): @@ -361,16 +336,16 @@ def __init__( self._client_secret = Secret(client_secret) self.redirect_on_success = redirect_on_success - @property - def scopes(self): - mapped = set() - for tiled_scopes in self.scopes_map.values(): - mapped.update(tiled_scopes) - return list(mapped) - - @scopes.setter - def scopes(self, value): - pass # ignored; scopes are derived from scopes_map + @property + def scopes(self): + mapped = set() + for tiled_scopes in self.scopes_map.values(): + mapped.update(tiled_scopes) + return list(mapped) + + @scopes.setter + def scopes(self, value): + pass # ignored; scopes are derived from scopes_map def decode_token( self, id_token: str, access_token: Optional[str] = None From a79e5eb8121b5f990041cf336871e916d1d5f621 Mon Sep 17 00:00:00 2001 From: David Pastl Date: Thu, 9 Jul 2026 08:40:35 -0600 Subject: [PATCH 06/11] Cleaning up the shutdown changes --- bluesky_httpserver/app.py | 136 +++++++++++++++++++++++++++----------- 1 file changed, 99 insertions(+), 37 deletions(-) diff --git a/bluesky_httpserver/app.py b/bluesky_httpserver/app.py index 4060a09..467ece0 100644 --- a/bluesky_httpserver/app.py +++ b/bluesky_httpserver/app.py @@ -17,7 +17,11 @@ from .authentication import ExternalAuthenticator, InternalAuthenticator from .authenticators import ProxiedOIDCAuthenticator -from .console_output import CollectPublishedConsoleOutput, ConsoleOutputStream, SystemInfoStream +from .console_output import ( + CollectPublishedConsoleOutput, + ConsoleOutputStream, + SystemInfoStream, +) from .core import PatchedStreamingResponse from .database.core import purge_expired from .resources import SERVER_RESOURCES as SR @@ -70,9 +74,9 @@ def custom_openapi(app): # print(f"openapi_schema = {pprint.pformat(openapi_schema['components'])}") ## # Insert refreshUrl. if "securitySchemes" in openapi_schema["components"]: # False when calling /docs - openapi_schema["components"]["securitySchemes"]["OAuth2PasswordBearer"]["flows"]["password"][ - "refreshUrl" - ] = "token/refresh" + openapi_schema["components"]["securitySchemes"]["OAuth2PasswordBearer"][ + "flows" + ]["password"]["refreshUrl"] = "token/refresh" app.openapi_schema = openapi_schema return app.openapi_schema @@ -109,10 +113,14 @@ def add_router(app, *, module_and_router_name): router = getattr(mod, router_name) app.include_router(router) except Exception as ex: - raise ImportError(f"Failed to import router {module_and_router_name!r}: {ex}") from ex + raise ImportError( + f"Failed to import router {module_and_router_name!r}: {ex}" + ) from ex -def build_app(authentication=None, api_access=None, resource_access=None, server_settings=None): +def build_app( + authentication=None, api_access=None, resource_access=None, server_settings=None +): """ Build application @@ -125,7 +133,9 @@ def build_app(authentication=None, api_access=None, resource_access=None, server """ authentication = authentication or {} authentication_providers = authentication.get("providers", []) - authenticators = {spec["provider"]: spec["authenticator"] for spec in authentication_providers} + authenticators = { + spec["provider"]: spec["authenticator"] for spec in authentication_providers + } api_access = api_access or {} api_access_manager = api_access.get("manager_object", None) resource_access = resource_access or {} @@ -147,7 +157,9 @@ def build_app(authentication=None, api_access=None, resource_access=None, server logger.info("Custom routers are specified in the config file: %s", router_names) elif router_names_str: router_names = re.split(":|,", router_names_str) - logger.info("Custom routers are specified in the environment variable: %s", router_names) + logger.info( + "Custom routers are specified in the environment variable: %s", router_names + ) if router_names: routers_already_included = set() @@ -178,7 +190,9 @@ def build_app(authentication=None, api_access=None, resource_access=None, server if authentication.get("providers", []): # For the OpenAPI schema, inject a OAuth2PasswordBearer URL. first_provider = authentication["providers"][0]["provider"] - oauth2_scheme.model.flows.password.tokenUrl = f"/api/auth/provider/{first_provider}/token" + oauth2_scheme.model.flows.password.tokenUrl = ( + f"/api/auth/provider/{first_provider}/token" + ) # Authenticators provide Router(s) for their particular flow. # Collect them in the authentication_router. @@ -239,7 +253,9 @@ def build_app(authentication=None, api_access=None, resource_access=None, server else: raise ValueError(f"unknown authenticator type {type(authenticator)}") for custom_router in getattr(authenticator, "include_routers", []): - authentication_router.include_router(custom_router, prefix=f"/provider/{provider}") + authentication_router.include_router( + custom_router, prefix=f"/provider/{provider}" + ) # And add this authentication_router itself to the app. app.include_router(authentication_router, prefix="/api/auth") @@ -316,11 +332,17 @@ async def startup_event(): async def purge_expired_sessions_and_api_keys(): logger.info("Purging expired Sessions and API keys from the database.") while True: - await asyncio.get_running_loop().run_in_executor(None, purge_expired(engine, orm.Session)) - await asyncio.get_running_loop().run_in_executor(None, purge_expired(engine, orm.APIKey)) + await asyncio.get_running_loop().run_in_executor( + None, purge_expired(engine, orm.Session) + ) + await asyncio.get_running_loop().run_in_executor( + None, purge_expired(engine, orm.APIKey) + ) await asyncio.sleep(600) - app.state.tasks.append(asyncio.create_task(purge_expired_sessions_and_api_keys())) + app.state.tasks.append( + asyncio.create_task(purge_expired_sessions_and_api_keys()) + ) # TODO: implement nicer exit with error reporting in case of failure zmq_control_addr = os.getenv("QSERVER_ZMQ_CONTROL_ADDRESS", None) @@ -355,14 +377,22 @@ async def purge_expired_sessions_and_api_keys(): zmq_encoding = os.getenv("QSERVER_ZMQ_ENCODING", None) # Check if ZMQ setting were specified in config file. Overrid the parameters from EVs. - zmq_control_addr = server_settings["qserver_zmq_configuration"].get("control_address", zmq_control_addr) - zmq_info_addr = server_settings["qserver_zmq_configuration"].get("info_address", zmq_info_addr) - zmq_encoding = server_settings["qserver_zmq_configuration"].get("encoding", zmq_encoding) + zmq_control_addr = server_settings["qserver_zmq_configuration"].get( + "control_address", zmq_control_addr + ) + zmq_info_addr = server_settings["qserver_zmq_configuration"].get( + "info_address", zmq_info_addr + ) + zmq_encoding = server_settings["qserver_zmq_configuration"].get( + "encoding", zmq_encoding + ) # Read public key from the environment variable or config file. zmq_public_key = os.environ.get("QSERVER_ZMQ_PUBLIC_KEY", None) zmq_public_key = zmq_public_key if zmq_public_key else None # Case of "" - zmq_public_key = server_settings["qserver_zmq_configuration"].get("public_key", zmq_public_key) + zmq_public_key = server_settings["qserver_zmq_configuration"].get( + "public_key", zmq_public_key + ) if zmq_public_key is not None: try: validate_zmq_key(zmq_public_key) @@ -399,7 +429,9 @@ async def purge_expired_sessions_and_api_keys(): # Import module with custom code module_names_str = os.getenv("QSERVER_CUSTOM_MODULES", None) - if (module_names_str is None) and (os.getenv("QSERVER_CUSTOM_MODULE", None) is not None): + if (module_names_str is None) and ( + os.getenv("QSERVER_CUSTOM_MODULE", None) is not None + ): logger.warning( "Environment variable QSERVER_CUSTOM_MODULE is deprecated and will be removed. " "Use the environment variable QSERVER_CUSTOM_MODULES, which accepts a string with " @@ -410,10 +442,15 @@ async def purge_expired_sessions_and_api_keys(): module_names = [] if "custom_modules" in server_settings["server_configuration"]: module_names = server_settings["server_configuration"]["custom_modules"] - logger.info("Custom modules from config file: %s", pprint.pformat(module_names)) + logger.info( + "Custom modules from config file: %s", pprint.pformat(module_names) + ) elif module_names_str: module_names = re.split(":|,", module_names_str) - logger.info("Custom modules from environment variable: %s", pprint.pformat(module_names)) + logger.info( + "Custom modules from environment variable: %s", + pprint.pformat(module_names), + ) if module_names: # Import all listed custom modules @@ -421,10 +458,14 @@ async def purge_expired_sessions_and_api_keys(): for name in module_names: try: logger.info("Importing custom module '%s' ...", name) - custom_code_modules.append(importlib.import_module(name.replace("-", "_"))) + custom_code_modules.append( + importlib.import_module(name.replace("-", "_")) + ) logger.info("Module '%s' was imported successfully.", name) except Exception as ex: - logger.error("Failed to import custom instrument module '%s': %s", name, ex) + logger.error( + "Failed to import custom instrument module '%s': %s", name, ex + ) SR.set_custom_code_modules(custom_code_modules) else: SR.set_custom_code_modules([]) @@ -435,14 +476,13 @@ async def purge_expired_sessions_and_api_keys(): @app.on_event("shutdown") async def shutdown_event(): - # Cancel any background tasks started by startup_event, even if - # startup itself failed part-way through and never reached - # ``app.state.tasks = []``. Iterating with a getattr default keeps - # the shutdown handler idempotent under partial-startup failures. + """Safely shutdown and perform the cleanup robustly + + This change ensures that the application shuts down and cleans up resources even if there is + a problem, without silencing the errors. + """ for task in getattr(app.state, "tasks", []): task.cancel() - # Best-effort teardown of REManager / console streams; each guarded - # so that a failure in one does not skip the others. for closer_name in ( "console_output_loader", "console_output_stream", @@ -476,7 +516,11 @@ def override_get_resource_access_manager(): @lru_cache(1) def override_get_settings(): settings = get_settings() - setattr(settings, "authentication_provider_names", [_["provider"] for _ in authentication_providers]) + setattr( + settings, + "authentication_provider_names", + [_["provider"] for _ in authentication_providers], + ) for item in [ "allow_anonymous_access", "secret_keys", @@ -499,13 +543,19 @@ def override_get_settings(): settings.database_pool_size = database["pool_size"] if database.get("pool_pre_ping"): settings.database_pool_pre_ping = database["pool_pre_ping"] - object_cache_available_bytes = server_settings.get("object_cache", {}).get("available_bytes") + object_cache_available_bytes = server_settings.get("object_cache", {}).get( + "available_bytes" + ) if object_cache_available_bytes is not None: - setattr(settings, "object_cache_available_bytes", object_cache_available_bytes) + setattr( + settings, "object_cache_available_bytes", object_cache_available_bytes + ) if authentication.get("providers"): # If we support authentication providers, we need a database, so if one is # not set, use a SQLite database in the current working directory. - settings.database_uri = settings.database_uri or "sqlite:///./bluesky_httpserver.sqlite" + settings.database_uri = ( + settings.database_uri or "sqlite:///./bluesky_httpserver.sqlite" + ) return settings @app.middleware("http") @@ -532,7 +582,11 @@ async def capture_metrics(request: Request, call_next): response.headers["Server-Timing"] = ", ".join( f"{key};" + ";".join( - (f"{metric}={value * 1000:.1f}" if metric == "dur" else f"{metric}={value:.1f}") + ( + f"{metric}={value * 1000:.1f}" + if metric == "dur" + else f"{metric}={value:.1f}" + ) for metric, value in metrics_.items() ) for key, metrics_ in metrics.items() @@ -544,9 +598,13 @@ async def capture_metrics(request: Request, call_next): async def double_submit_cookie_csrf_protection(request: Request, call_next): # https://cheatsheetseries.owasp.org/cheatsheets/Cross-Site_Request_Forgery_Prevention_Cheat_Sheet.html#double-submit-cookie csrf_cookie = request.cookies.get(CSRF_COOKIE_NAME) - if (request.method not in SAFE_METHODS) and set(request.cookies).intersection(SENSITIVE_COOKIES): + if (request.method not in SAFE_METHODS) and set(request.cookies).intersection( + SENSITIVE_COOKIES + ): if not csrf_cookie: - return Response(status_code=403, content=f"Expected {CSRF_COOKIE_NAME} cookie") + return Response( + status_code=403, content=f"Expected {CSRF_COOKIE_NAME} cookie" + ) # Get the token from the Header or (if not there) the query parameter. csrf_token = request.headers.get(CSRF_HEADER_NAME) if csrf_token is None: @@ -559,7 +617,9 @@ async def double_submit_cookie_csrf_protection(request: Request, call_next): ) # Securely compare the token with the cookie. if not secrets.compare_digest(csrf_token, csrf_cookie): - return Response(status_code=403, content="Double-submit CSRF tokens do not match") + return Response( + status_code=403, content="Double-submit CSRF tokens do not match" + ) response = await call_next(request) response.__class__ = PatchedStreamingResponse # tolerate memoryview @@ -588,7 +648,9 @@ async def set_cookies(request: Request, call_next): app.openapi = partial(custom_openapi, app) app.dependency_overrides[get_authenticators] = override_get_authenticators app.dependency_overrides[get_api_access_manager] = override_get_api_access_manager - app.dependency_overrides[get_resource_access_manager] = override_get_resource_access_manager + app.dependency_overrides[get_resource_access_manager] = ( + override_get_resource_access_manager + ) app.dependency_overrides[get_settings] = override_get_settings def add_custom_middleware(): From c0430e415a62ce0c42e783529f5078e057f0fe0b Mon Sep 17 00:00:00 2001 From: David Pastl Date: Thu, 9 Jul 2026 09:22:42 -0600 Subject: [PATCH 07/11] Adding auto-upgrade ability --- bluesky_httpserver/app.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/bluesky_httpserver/app.py b/bluesky_httpserver/app.py index 467ece0..227a405 100644 --- a/bluesky_httpserver/app.py +++ b/bluesky_httpserver/app.py @@ -298,8 +298,10 @@ async def startup_event(): from .database.core import ( # make_admin_by_identity, REQUIRED_REVISION, UninitializedDatabase, + DatabaseUpgradeNeeded, check_database, initialize_database, + upgrade, ) connect_args = {} @@ -317,6 +319,12 @@ async def startup_event(): ) initialize_database(engine) logger.info("Database initialized.") + except DatabaseUpgradeNeeded: + logger.info( + f"Database at {redacted_url} is out of date. Upgrading to {REQUIRED_REVISION}..." + ) + upgrade(engine, REQUIRED_REVISION) + logger.info("Database upgraded.") else: logger.info(f"Connected to existing database at {redacted_url}.") # SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) From 22cc93c5d2c394000cceca5d268668175df4ae0d Mon Sep 17 00:00:00 2001 From: David Pastl Date: Thu, 9 Jul 2026 10:43:49 -0600 Subject: [PATCH 08/11] Fixing typo --- bluesky_httpserver/authenticators.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bluesky_httpserver/authenticators.py b/bluesky_httpserver/authenticators.py index 25d9c15..e904655 100644 --- a/bluesky_httpserver/authenticators.py +++ b/bluesky_httpserver/authenticators.py @@ -899,7 +899,7 @@ async def resolve_username(self, username_supplied_by_user): is_bound = await asyncio.get_running_loop().run_in_executor(None, conn.bind) if not is_bound: msg = "Failed to connect to LDAP server with search user '{search_dn}'" - self.log.warning(msg.format(search_dn=search_dn)) + logger.warning(msg.format(search_dn=search_dn)) return (None, None) search_filter = self.lookup_dn_search_filter.format( From b69bd450341d00de0490faa608d0c1cf2deb07f8 Mon Sep 17 00:00:00 2001 From: David Pastl Date: Thu, 9 Jul 2026 15:53:19 -0600 Subject: [PATCH 09/11] Code review adjustments --- bluesky_httpserver/_authentication.py | 10 ++++----- bluesky_httpserver/authenticators.py | 21 +++++++++---------- bluesky_httpserver/core.py | 4 ++-- bluesky_httpserver/database/core.py | 16 +++++++------- bluesky_httpserver/routers/core_api.py | 9 +++++++- .../tests/test_auth_port_0_2_12.py | 6 +++--- 6 files changed, 36 insertions(+), 30 deletions(-) diff --git a/bluesky_httpserver/_authentication.py b/bluesky_httpserver/_authentication.py index 46c5cb0..f24587d 100644 --- a/bluesky_httpserver/_authentication.py +++ b/bluesky_httpserver/_authentication.py @@ -81,7 +81,7 @@ def utcnow(): "UTC now with second resolution" - return datetime.utcnow().replace(microsecond=0) + return datetime.now(tz=timezone.utc).replace(microsecond=0) class Token(BaseModel): @@ -436,7 +436,8 @@ def get_current_principal( for _ in payload["ids"] if _["idp"] in settings.authentication_provider_names ] - scopes = set.union(*[api_access_manager.get_user_scopes(_) for _ in ids]) + scopes_sets = [api_access_manager.get_user_scopes(_) for _ in ids] + scopes = set.union(*scopes_sets) if scopes_sets else set() roles_sets = [api_access_manager.get_user_roles(_) for _ in ids] roles = set.union(*roles_sets) if roles_sets else set() @@ -817,10 +818,7 @@ async def authorize_redirect( # Always request ``openid`` and ``offline_access`` so the IdP returns a # refresh_token in the code exchange. Authenticators (e.g. Entra) may # advertise extra scopes via an ``extra_scopes`` attribute to obtain - # per-resource access tokens. ``prompt=login`` forces the IdP to - # prompt the user rather than silently reissue a session — this - # prevents surprising SSO behavior when a user explicitly navigates - # to the login flow. + # per-resource access tokens. scopes = {"openid", "profile", "email", "offline_access"} scopes.update(getattr(authenticator, "extra_scopes", []) or []) diff --git a/bluesky_httpserver/authenticators.py b/bluesky_httpserver/authenticators.py index e904655..34b0d76 100644 --- a/bluesky_httpserver/authenticators.py +++ b/bluesky_httpserver/authenticators.py @@ -240,7 +240,6 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]: if response.is_error: logger.error("Authentication error: %r", response_body) return None - response_body = response.json() id_token = response_body["id_token"] access_token = response_body["access_token"] try: @@ -336,16 +335,16 @@ def __init__( self._client_secret = Secret(client_secret) self.redirect_on_success = redirect_on_success - @property - def scopes(self): - mapped = set() - for tiled_scopes in self.scopes_map.values(): - mapped.update(tiled_scopes) - return list(mapped) - - @scopes.setter - def scopes(self, value): - pass # ignored; scopes are derived from scopes_map + @property + def scopes(self): + mapped = set() + for tiled_scopes in self.scopes_map.values(): + mapped.update(tiled_scopes) + return list(mapped) + + @scopes.setter + def scopes(self, value): + pass # ignored; scopes are derived from scopes_map def decode_token( self, id_token: str, access_token: Optional[str] = None diff --git a/bluesky_httpserver/core.py b/bluesky_httpserver/core.py index ab3ab48..3372c66 100644 --- a/bluesky_httpserver/core.py +++ b/bluesky_httpserver/core.py @@ -11,7 +11,7 @@ # from collections import defaultdict # from datetime import datetime, timedelta -from datetime import datetime +from datetime import datetime, timezone from hashlib import md5 from typing import Any @@ -197,7 +197,7 @@ # for key in tree.keys_indexer[offset : offset + limit] # noqa: E203 # ) # # This value will not leak out. It just used to seed comparisons. -# metadata_stale_at = datetime.utcnow() + timedelta(days=1_000_000) +# metadata_stale_at = datetime.now(tz=timezone.utc) + timedelta(days=1_000_000) # must_revalidate = getattr(tree, "must_revalidate", True) # for key, entry in items: # resource = construct_resource( diff --git a/bluesky_httpserver/database/core.py b/bluesky_httpserver/database/core.py index 88f7ffa..00fb8d9 100644 --- a/bluesky_httpserver/database/core.py +++ b/bluesky_httpserver/database/core.py @@ -1,6 +1,6 @@ import hashlib import uuid as uuid_module -from datetime import datetime +from datetime import datetime, timezone from typing import Optional from alembic import command @@ -184,7 +184,7 @@ def purge_expired(engine, cls): """ SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() - now = datetime.utcnow() + now = datetime.now(tz=timezone.utc) deleted = False for obj in db.query(cls).filter(cls.expiration_time.is_not(None)).filter(cls.expiration_time < now): deleted = True @@ -230,7 +230,7 @@ def get_or_create_principal(db, identity_provider, id): .first() ) if identity is not None: - identity.latest_login = datetime.utcnow() + identity.latest_login = datetime.now(tz=timezone.utc) db.commit() return identity.principal return create_user(db, identity_provider, id) @@ -243,7 +243,9 @@ def lookup_valid_session(db, session_id): return None session = db.query(Session).filter(Session.uuid == uuid_module.UUID(hex=session_id)).first() - if session.expiration_time is not None and session.expiration_time < datetime.utcnow(): + if session is None: + return None + if session.expiration_time is not None and session.expiration_time < datetime.now(tz=timezone.utc): db.delete(session) db.commit() return None @@ -268,7 +270,7 @@ def lookup_valid_api_key(db, secret): Look up an API key. Ensure that it is valid. """ - now = datetime.utcnow() + now = datetime.now(tz=timezone.utc) hashed_secret = hashlib.sha256(secret).digest() api_key = ( db.query(APIKey) @@ -336,7 +338,7 @@ def lookup_valid_pending_session_by_device_code(db, device_code: bytes) -> Optio ) if pending_session is None: return None - if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.utcnow(): + if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.now(tz=timezone.utc): db.delete(pending_session) db.commit() return None @@ -352,7 +354,7 @@ def lookup_valid_pending_session_by_user_code(db, user_code: str) -> Optional[Pe pending_session = db.query(PendingSession).filter(PendingSession.user_code == user_code).first() if pending_session is None: return None - if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.utcnow(): + if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.now(tz=timezone.utc): db.delete(pending_session) db.commit() return None diff --git a/bluesky_httpserver/routers/core_api.py b/bluesky_httpserver/routers/core_api.py index 841a295..f86dcb1 100644 --- a/bluesky_httpserver/routers/core_api.py +++ b/bluesky_httpserver/routers/core_api.py @@ -1192,7 +1192,14 @@ async def _authenticate_websocket(websocket, scopes): await websocket.accept() try: message = await asyncio.wait_for(websocket.receive_json(), timeout=10) - except (asyncio.TimeoutError, WebSocketDisconnect, Exception): + except asyncio.TimeoutError: + await websocket.close(code=_WS_CLOSE_AUTH_REQUIRED, reason="Auth required") + return None, True + except WebSocketDisconnect: + # Client already gone — no close frame needed. + return None, True + except Exception: + logger.exception("Unexpected error receiving WebSocket auth frame") await websocket.close(code=_WS_CLOSE_AUTH_REQUIRED, reason="Auth required") return None, True diff --git a/bluesky_httpserver/tests/test_auth_port_0_2_12.py b/bluesky_httpserver/tests/test_auth_port_0_2_12.py index 2807ad8..b419a89 100644 --- a/bluesky_httpserver/tests/test_auth_port_0_2_12.py +++ b/bluesky_httpserver/tests/test_auth_port_0_2_12.py @@ -18,7 +18,7 @@ import time import uuid -from datetime import timedelta +from datetime import timedelta, timezone from typing import Any, Tuple from unittest.mock import MagicMock @@ -364,7 +364,7 @@ def test_session_state_column_round_trips(sqlite_session): session = db_orm.Session( principal_id=principal.id, - expiration_time=datetime.utcnow() + timedelta(days=1), + expiration_time=datetime.now(tz=timezone.utc) + timedelta(days=1), state=payload, ) db.add(session) @@ -382,7 +382,7 @@ def test_session_state_defaults_to_empty_dict(sqlite_session): principal = create_user(db, "internal", "alice") session = db_orm.Session( principal_id=principal.id, - expiration_time=datetime.utcnow() + timedelta(days=1), + expiration_time=datetime.now(tz=timezone.utc) + timedelta(days=1), ) db.add(session) db.commit() From ab388c5d62f400811dc72747d798aaab6b6c1d2c Mon Sep 17 00:00:00 2001 From: David Pastl Date: Thu, 9 Jul 2026 15:58:04 -0600 Subject: [PATCH 10/11] Removing timezone changes to reduce scope --- bluesky_httpserver/_authentication.py | 2 +- bluesky_httpserver/core.py | 4 ++-- bluesky_httpserver/database/core.py | 14 +++++++------- bluesky_httpserver/tests/test_auth_port_0_2_12.py | 6 +++--- 4 files changed, 13 insertions(+), 13 deletions(-) diff --git a/bluesky_httpserver/_authentication.py b/bluesky_httpserver/_authentication.py index f24587d..5bb57ba 100644 --- a/bluesky_httpserver/_authentication.py +++ b/bluesky_httpserver/_authentication.py @@ -81,7 +81,7 @@ def utcnow(): "UTC now with second resolution" - return datetime.now(tz=timezone.utc).replace(microsecond=0) + return datetime.utcnow().replace(microsecond=0) class Token(BaseModel): diff --git a/bluesky_httpserver/core.py b/bluesky_httpserver/core.py index 3372c66..ab3ab48 100644 --- a/bluesky_httpserver/core.py +++ b/bluesky_httpserver/core.py @@ -11,7 +11,7 @@ # from collections import defaultdict # from datetime import datetime, timedelta -from datetime import datetime, timezone +from datetime import datetime from hashlib import md5 from typing import Any @@ -197,7 +197,7 @@ # for key in tree.keys_indexer[offset : offset + limit] # noqa: E203 # ) # # This value will not leak out. It just used to seed comparisons. -# metadata_stale_at = datetime.now(tz=timezone.utc) + timedelta(days=1_000_000) +# metadata_stale_at = datetime.utcnow() + timedelta(days=1_000_000) # must_revalidate = getattr(tree, "must_revalidate", True) # for key, entry in items: # resource = construct_resource( diff --git a/bluesky_httpserver/database/core.py b/bluesky_httpserver/database/core.py index 00fb8d9..aee3a3a 100644 --- a/bluesky_httpserver/database/core.py +++ b/bluesky_httpserver/database/core.py @@ -1,6 +1,6 @@ import hashlib import uuid as uuid_module -from datetime import datetime, timezone +from datetime import datetime from typing import Optional from alembic import command @@ -184,7 +184,7 @@ def purge_expired(engine, cls): """ SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) db = SessionLocal() - now = datetime.now(tz=timezone.utc) + now = datetime.utcnow() deleted = False for obj in db.query(cls).filter(cls.expiration_time.is_not(None)).filter(cls.expiration_time < now): deleted = True @@ -230,7 +230,7 @@ def get_or_create_principal(db, identity_provider, id): .first() ) if identity is not None: - identity.latest_login = datetime.now(tz=timezone.utc) + identity.latest_login = datetime.utcnow() db.commit() return identity.principal return create_user(db, identity_provider, id) @@ -245,7 +245,7 @@ def lookup_valid_session(db, session_id): session = db.query(Session).filter(Session.uuid == uuid_module.UUID(hex=session_id)).first() if session is None: return None - if session.expiration_time is not None and session.expiration_time < datetime.now(tz=timezone.utc): + if session.expiration_time is not None and session.expiration_time < datetime.utcnow(): db.delete(session) db.commit() return None @@ -270,7 +270,7 @@ def lookup_valid_api_key(db, secret): Look up an API key. Ensure that it is valid. """ - now = datetime.now(tz=timezone.utc) + now = datetime.utcnow() hashed_secret = hashlib.sha256(secret).digest() api_key = ( db.query(APIKey) @@ -338,7 +338,7 @@ def lookup_valid_pending_session_by_device_code(db, device_code: bytes) -> Optio ) if pending_session is None: return None - if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.now(tz=timezone.utc): + if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.utcnow(): db.delete(pending_session) db.commit() return None @@ -354,7 +354,7 @@ def lookup_valid_pending_session_by_user_code(db, user_code: str) -> Optional[Pe pending_session = db.query(PendingSession).filter(PendingSession.user_code == user_code).first() if pending_session is None: return None - if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.now(tz=timezone.utc): + if pending_session.expiration_time is not None and pending_session.expiration_time < datetime.utcnow(): db.delete(pending_session) db.commit() return None diff --git a/bluesky_httpserver/tests/test_auth_port_0_2_12.py b/bluesky_httpserver/tests/test_auth_port_0_2_12.py index b419a89..2807ad8 100644 --- a/bluesky_httpserver/tests/test_auth_port_0_2_12.py +++ b/bluesky_httpserver/tests/test_auth_port_0_2_12.py @@ -18,7 +18,7 @@ import time import uuid -from datetime import timedelta, timezone +from datetime import timedelta from typing import Any, Tuple from unittest.mock import MagicMock @@ -364,7 +364,7 @@ def test_session_state_column_round_trips(sqlite_session): session = db_orm.Session( principal_id=principal.id, - expiration_time=datetime.now(tz=timezone.utc) + timedelta(days=1), + expiration_time=datetime.utcnow() + timedelta(days=1), state=payload, ) db.add(session) @@ -382,7 +382,7 @@ def test_session_state_defaults_to_empty_dict(sqlite_session): principal = create_user(db, "internal", "alice") session = db_orm.Session( principal_id=principal.id, - expiration_time=datetime.now(tz=timezone.utc) + timedelta(days=1), + expiration_time=datetime.utcnow() + timedelta(days=1), ) db.add(session) db.commit() From 32590836df338addb9e3a3c71a7c104ff8b2e91f Mon Sep 17 00:00:00 2001 From: David Pastl Date: Thu, 9 Jul 2026 16:15:09 -0600 Subject: [PATCH 11/11] Removing unnecessary formatting changes This just makes the diff easier to understand --- bluesky_httpserver/_authentication.py | 160 ++++++-------------------- bluesky_httpserver/app.py | 119 +++++-------------- bluesky_httpserver/schemas.py | 8 -- 3 files changed, 65 insertions(+), 222 deletions(-) diff --git a/bluesky_httpserver/_authentication.py b/bluesky_httpserver/_authentication.py index 5bb57ba..637a8ad 100644 --- a/bluesky_httpserver/_authentication.py +++ b/bluesky_httpserver/_authentication.py @@ -109,9 +109,7 @@ def __init__( scheme_name: Optional[str] = None, description: Optional[str] = None, ): - self.model: APIKey = APIKey( - **{"in": APIKeyIn.header}, name=name, description=description - ) + self.model: APIKey = APIKey(**{"in": APIKeyIn.header}, name=name, description=description) self.scheme_name = scheme_name or self.__class__.__name__ async def __call__(self, request: Request) -> Optional[str]: @@ -342,9 +340,7 @@ def get_current_principal( if api_key_orm is not None: principal = schemas.Principal.from_orm(api_key_orm.principal) ids = get_current_username( - principal=principal, - settings=settings, - api_access_manager=api_access_manager, + principal=principal, settings=settings, api_access_manager=api_access_manager ) scope_sets = [api_access_manager.get_user_scopes(_) for _ in ids] principal_scopes = set.union(*scope_sets) if scope_sets else set() @@ -382,26 +378,16 @@ def get_current_principal( principal = schemas.Principal( uuid=uuid_module.uuid4(), # Generate unique UUID each time - it is not expected to be used type="user", - identities=[ - schemas.Identity( - id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME - ) - ], + identities=[schemas.Identity(id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME)], ) else: - raise HTTPException( - status_code=401, detail="Invalid API key", headers=headers_for_401 - ) + raise HTTPException(status_code=401, detail="Invalid API key", headers=headers_for_401) # If we made it to this point, we have a valid API key. # If the API key was given in query param, move to cookie. # This is convenient for browser-based access. - if ("api_key" in request.query_params) and ( - request.cookies.get(API_KEY_COOKIE_NAME) != api_key - ): - request.state.cookies_to_set.append( - {"key": API_KEY_COOKIE_NAME, "value": api_key} - ) + if ("api_key" in request.query_params) and (request.cookies.get(API_KEY_COOKIE_NAME) != api_key): + request.state.cookies_to_set.append({"key": API_KEY_COOKIE_NAME, "value": api_key}) elif access_token is not None: proxied_authenticator = _find_proxied_authenticator(authenticators) try: @@ -497,9 +483,7 @@ def get_current_principal( principal = schemas.Principal( uuid=uuid_module.uuid4(), # Generate unique UUID each time - it is not expected to be used type="user", - identities=[ - schemas.Identity(id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME) - ], + identities=[schemas.Identity(id=username, provider=_DEFAULT_ANONYMOUS_PROVIDER_NAME)], ) # Is anonymous public access permitted? @@ -545,11 +529,7 @@ def get_current_principal( api_key_scopes_list.sort() else: api_key_scopes_list = api_key_scopes - principal.roles, principal.scopes, principal.api_key_scopes = ( - roles_list, - scopes_list, - api_key_scopes_list, - ) + principal.roles, principal.scopes, principal.api_key_scopes = roles_list, scopes_list, api_key_scopes_list return principal @@ -867,12 +847,8 @@ async def device_code_authorize( } ) return { - "authorization_uri": str( - authorization_uri - ), # URL that user should visit in browser - "verification_uri": str( - verification_uri - ), # URL that terminal client will poll + "authorization_uri": str(authorization_uri), # URL that user should visit in browser + "verification_uri": str(verification_uri), # URL that terminal client will poll "interval": DEVICE_CODE_POLLING_INTERVAL, # suggested polling interval "device_code": pending_session["device_code"], "expires_in": int(DEVICE_CODE_MAX_AGE.total_seconds()), # seconds @@ -896,9 +872,7 @@ async def _complete_device_code_authorization( normalized_user_code = user_code.upper().replace("-", "").strip() with get_sessionmaker(settings.database_settings)() as db: - pending_session = lookup_valid_pending_session_by_user_code( - db, normalized_user_code - ) + pending_session = lookup_valid_pending_session_by_user_code(db, normalized_user_code) if pending_session is None: error_html = f""" @@ -1040,9 +1014,7 @@ async def device_code_form( api_access_manager=api_access_manager, ) - action = ( - f"{get_base_url(request)}/auth/provider/{provider}/device_code?code={code}" - ) + action = f"{get_base_url(request)}/auth/provider/{provider}/device_code?code={code}" html_content = f""" @@ -1171,29 +1143,21 @@ async def device_code_token( raise HTTPException(status_code=401, detail="Invalid device code") with get_sessionmaker(settings.database_settings)() as db: - pending_session = lookup_valid_pending_session_by_device_code( - db, device_code - ) + pending_session = lookup_valid_pending_session_by_device_code(db, device_code) if pending_session is None: raise HTTPException( status_code=404, detail="No such device_code. The pending request may have expired.", ) if pending_session.session_id is None: - raise HTTPException( - status_code=400, detail={"error": "authorization_pending"} - ) + raise HTTPException(status_code=400, detail={"error": "authorization_pending"}) session = pending_session.session principal = session.principal # Get scopes for the user # Find an identity to get the username - identity = ( - db.query(orm.Identity) - .filter(orm.Identity.principal_id == principal.id) - .first() - ) + identity = db.query(orm.Identity).filter(orm.Identity.principal_id == principal.id).first() if identity and api_access_manager.is_user_known(identity.id): scopes = api_access_manager.get_user_scopes(identity.id) else: @@ -1229,18 +1193,14 @@ async def device_code_token( "access_token": access_token, "expires_in": int(settings.access_token_max_age / UNIT_SECOND), "refresh_token": refresh_token, - "refresh_token_expires_in": int( - settings.refresh_token_max_age / UNIT_SECOND - ), + "refresh_token_expires_in": int(settings.refresh_token_max_age / UNIT_SECOND), "token_type": "bearer", } return device_code_token -def generate_apikey( - db, principal, apikey_params, request, allowed_scopes, source_api_key_scopes -): +def generate_apikey(db, principal, apikey_params, request, allowed_scopes, source_api_key_scopes): # Use API key scopes if API key is generated based on existing API key, otherwise used allowed scopes if (source_api_key_scopes is not None) and ("inherit" not in source_api_key_scopes): scopes_allowed_set = source_api_key_scopes @@ -1315,9 +1275,7 @@ def principal_list( principal_orms = db.query(orm.Principal).all() principals = [ - schemas.Principal.from_orm( - principal_orm, latest_principal_activity(db, principal_orm) - ).dict() + schemas.Principal.from_orm(principal_orm, latest_principal_activity(db, principal_orm)).dict() for principal_orm in principal_orms ] @@ -1337,14 +1295,10 @@ def principal( "Get information about one Principal (user or service)." request.state.endpoint = "auth" with get_sessionmaker(settings.database_settings)() as db: - principal_orm = ( - db.query(orm.Principal).filter(orm.Principal.uuid == uuid).first() - ) + principal_orm = db.query(orm.Principal).filter(orm.Principal.uuid == uuid).first() return json_or_msgpack( request, - schemas.Principal.from_orm( - principal_orm, latest_principal_activity(db, principal_orm) - ).dict(), + schemas.Principal.from_orm(principal_orm, latest_principal_activity(db, principal_orm)).dict(), ) @@ -1365,28 +1319,17 @@ def apikey_for_principal( with get_sessionmaker(settings.database_settings)() as db: principal = db.query(orm.Principal).filter(orm.Principal.uuid == uuid).first() if principal is None: - raise HTTPException( - 404, f"Principal {uuid} does not exist or insufficient permissions." - ) + raise HTTPException(404, f"Principal {uuid} does not exist or insufficient permissions.") ids = {_.id for _ in principal.identities} scope_sets = [api_access_manager.get_user_scopes(_) for _ in ids] principal_scopes = set.union(*scope_sets) if scope_sets else set() source_api_key_scopes = None - return generate_apikey( - db, - principal, - apikey_params, - request, - principal_scopes, - source_api_key_scopes, - ) + return generate_apikey(db, principal, apikey_params, request, principal_scopes, source_api_key_scopes) -@base_authentication_router.post( - "/session/refresh", response_model=schemas.AccessAndRefreshTokens -) +@base_authentication_router.post("/session/refresh", response_model=schemas.AccessAndRefreshTokens) def refresh_session( request: Request, refresh_token: schemas.RefreshToken, @@ -1396,9 +1339,7 @@ def refresh_session( "Obtain a new access token and refresh token." request.state.endpoint = "auth" with get_sessionmaker(settings.database_settings)() as db: - new_tokens = slide_session( - refresh_token.refresh_token, settings, db, api_access_manager - ) + new_tokens = slide_session(refresh_token.refresh_token, settings, db, api_access_manager) return new_tokens @@ -1432,9 +1373,7 @@ def slide_session(refresh_token, settings, db, api_access_manager): try: payload = decode_token(refresh_token, settings.secret_keys) except ExpiredSignatureError: - raise HTTPException( - status_code=401, detail="Session has expired. Please re-authenticate." - ) + raise HTTPException(status_code=401, detail="Session has expired. Please re-authenticate.") # Find this session in the database. session = lookup_valid_session(db, payload["sid"]) now = utcnow() @@ -1443,9 +1382,7 @@ def slide_session(refresh_token, settings, db, api_access_manager): if (session is None) or session.revoked or (session.expiration_time < now): # Do not leak (to a potential attacker) whether this has been *revoked* # specifically. Give the same error as if it had expired. - raise HTTPException( - status_code=401, detail="Session has expired. Please re-authenticate." - ) + raise HTTPException(status_code=401, detail="Session has expired. Please re-authenticate.") # Update Session info. session.time_last_refreshed = now # This increments in a way that avoids a race condition. @@ -1455,9 +1392,7 @@ def slide_session(refresh_token, settings, db, api_access_manager): # database hit. principal = schemas.Principal.from_orm(session.principal) - ids = get_current_username( - principal=principal, settings=settings, api_access_manager=api_access_manager - ) + ids = get_current_username(principal=principal, settings=settings, api_access_manager=api_access_manager) if not ids: raise HTTPException( status_code=401, @@ -1517,26 +1452,13 @@ def new_apikey( # principal_scopes = set.union(*scope_sets) if scope_sets else set() allowed_scopes = set(principal.scopes) - source_api_key_scopes = ( - set(principal.api_key_scopes) - if (principal.api_key_scopes is not None) - else None - ) + source_api_key_scopes = set(principal.api_key_scopes) if (principal.api_key_scopes is not None) else None with get_sessionmaker(settings.database_settings)() as db: # The principal from get_current_principal tells us everything that the # access_token carries around, but the database knows more than that. - principal_orm = ( - db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() - ) - apikey = generate_apikey( - db, - principal_orm, - apikey_params, - request, - allowed_scopes, - source_api_key_scopes, - ) + principal_orm = db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() + apikey = generate_apikey(db, principal_orm, apikey_params, request, allowed_scopes, source_api_key_scopes) return apikey @@ -1554,9 +1476,7 @@ def current_apikey_info( # TODO Permit filtering the fields of the response. request.state.endpoint = "auth" if api_key is None: - raise HTTPException( - status_code=401, detail="No API key was provided with this request." - ) + raise HTTPException(status_code=401, detail="No API key was provided with this request.") try: secret = bytes.fromhex(api_key) except Exception: @@ -1583,11 +1503,7 @@ def revoke_apikey( if principal is None: return None with get_sessionmaker(settings.database_settings)() as db: - api_key_orm = ( - db.query(orm.APIKey) - .filter(orm.APIKey.first_eight == first_eight[:8]) - .first() - ) + api_key_orm = db.query(orm.APIKey).filter(orm.APIKey.first_eight == first_eight[:8]).first() if (api_key_orm is None) or (api_key_orm.principal.uuid != principal.uuid): raise HTTPException( 404, @@ -1615,14 +1531,10 @@ def whoami( # The principal from get_current_principal tells us everything that the # access_token carries around, but the database knows more than that. with get_sessionmaker(settings.database_settings)() as db: - principal_orm = ( - db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() - ) + principal_orm = db.query(orm.Principal).filter(orm.Principal.uuid == principal.uuid).first() return json_or_msgpack( request, - schemas.Principal.from_orm( - principal_orm, latest_principal_activity(db, principal_orm) - ).dict(), + schemas.Principal.from_orm(principal_orm, latest_principal_activity(db, principal_orm)).dict(), ) @@ -1635,9 +1547,7 @@ def scopes( principal=Security(get_current_principal, scopes=[]), ): roles, scopes = principal.roles, principal.scopes - return json_or_msgpack( - request, schemas.AllowedScopes(roles=roles, scopes=scopes).dict() - ) + return json_or_msgpack(request, schemas.AllowedScopes(roles=roles, scopes=scopes).dict()) @base_authentication_router.post("/logout") diff --git a/bluesky_httpserver/app.py b/bluesky_httpserver/app.py index 227a405..7821107 100644 --- a/bluesky_httpserver/app.py +++ b/bluesky_httpserver/app.py @@ -74,9 +74,9 @@ def custom_openapi(app): # print(f"openapi_schema = {pprint.pformat(openapi_schema['components'])}") ## # Insert refreshUrl. if "securitySchemes" in openapi_schema["components"]: # False when calling /docs - openapi_schema["components"]["securitySchemes"]["OAuth2PasswordBearer"][ - "flows" - ]["password"]["refreshUrl"] = "token/refresh" + openapi_schema["components"]["securitySchemes"]["OAuth2PasswordBearer"]["flows"]["password"][ + "refreshUrl" + ] = "token/refresh" app.openapi_schema = openapi_schema return app.openapi_schema @@ -113,14 +113,10 @@ def add_router(app, *, module_and_router_name): router = getattr(mod, router_name) app.include_router(router) except Exception as ex: - raise ImportError( - f"Failed to import router {module_and_router_name!r}: {ex}" - ) from ex + raise ImportError(f"Failed to import router {module_and_router_name!r}: {ex}") from ex -def build_app( - authentication=None, api_access=None, resource_access=None, server_settings=None -): +def build_app(authentication=None, api_access=None, resource_access=None, server_settings=None): """ Build application @@ -133,9 +129,7 @@ def build_app( """ authentication = authentication or {} authentication_providers = authentication.get("providers", []) - authenticators = { - spec["provider"]: spec["authenticator"] for spec in authentication_providers - } + authenticators = {spec["provider"]: spec["authenticator"] for spec in authentication_providers} api_access = api_access or {} api_access_manager = api_access.get("manager_object", None) resource_access = resource_access or {} @@ -157,9 +151,7 @@ def build_app( logger.info("Custom routers are specified in the config file: %s", router_names) elif router_names_str: router_names = re.split(":|,", router_names_str) - logger.info( - "Custom routers are specified in the environment variable: %s", router_names - ) + logger.info("Custom routers are specified in the environment variable: %s", router_names) if router_names: routers_already_included = set() @@ -190,9 +182,7 @@ def build_app( if authentication.get("providers", []): # For the OpenAPI schema, inject a OAuth2PasswordBearer URL. first_provider = authentication["providers"][0]["provider"] - oauth2_scheme.model.flows.password.tokenUrl = ( - f"/api/auth/provider/{first_provider}/token" - ) + oauth2_scheme.model.flows.password.tokenUrl = f"/api/auth/provider/{first_provider}/token" # Authenticators provide Router(s) for their particular flow. # Collect them in the authentication_router. @@ -253,9 +243,7 @@ def build_app( else: raise ValueError(f"unknown authenticator type {type(authenticator)}") for custom_router in getattr(authenticator, "include_routers", []): - authentication_router.include_router( - custom_router, prefix=f"/provider/{provider}" - ) + authentication_router.include_router(custom_router, prefix=f"/provider/{provider}") # And add this authentication_router itself to the app. app.include_router(authentication_router, prefix="/api/auth") @@ -340,17 +328,11 @@ async def startup_event(): async def purge_expired_sessions_and_api_keys(): logger.info("Purging expired Sessions and API keys from the database.") while True: - await asyncio.get_running_loop().run_in_executor( - None, purge_expired(engine, orm.Session) - ) - await asyncio.get_running_loop().run_in_executor( - None, purge_expired(engine, orm.APIKey) - ) + await asyncio.get_running_loop().run_in_executor(None, purge_expired(engine, orm.Session)) + await asyncio.get_running_loop().run_in_executor(None, purge_expired(engine, orm.APIKey)) await asyncio.sleep(600) - app.state.tasks.append( - asyncio.create_task(purge_expired_sessions_and_api_keys()) - ) + app.state.tasks.append(asyncio.create_task(purge_expired_sessions_and_api_keys())) # TODO: implement nicer exit with error reporting in case of failure zmq_control_addr = os.getenv("QSERVER_ZMQ_CONTROL_ADDRESS", None) @@ -385,22 +367,14 @@ async def purge_expired_sessions_and_api_keys(): zmq_encoding = os.getenv("QSERVER_ZMQ_ENCODING", None) # Check if ZMQ setting were specified in config file. Overrid the parameters from EVs. - zmq_control_addr = server_settings["qserver_zmq_configuration"].get( - "control_address", zmq_control_addr - ) - zmq_info_addr = server_settings["qserver_zmq_configuration"].get( - "info_address", zmq_info_addr - ) - zmq_encoding = server_settings["qserver_zmq_configuration"].get( - "encoding", zmq_encoding - ) + zmq_control_addr = server_settings["qserver_zmq_configuration"].get("control_address", zmq_control_addr) + zmq_info_addr = server_settings["qserver_zmq_configuration"].get("info_address", zmq_info_addr) + zmq_encoding = server_settings["qserver_zmq_configuration"].get("encoding", zmq_encoding) # Read public key from the environment variable or config file. zmq_public_key = os.environ.get("QSERVER_ZMQ_PUBLIC_KEY", None) zmq_public_key = zmq_public_key if zmq_public_key else None # Case of "" - zmq_public_key = server_settings["qserver_zmq_configuration"].get( - "public_key", zmq_public_key - ) + zmq_public_key = server_settings["qserver_zmq_configuration"].get("public_key", zmq_public_key) if zmq_public_key is not None: try: validate_zmq_key(zmq_public_key) @@ -437,9 +411,7 @@ async def purge_expired_sessions_and_api_keys(): # Import module with custom code module_names_str = os.getenv("QSERVER_CUSTOM_MODULES", None) - if (module_names_str is None) and ( - os.getenv("QSERVER_CUSTOM_MODULE", None) is not None - ): + if (module_names_str is None) and (os.getenv("QSERVER_CUSTOM_MODULE", None) is not None): logger.warning( "Environment variable QSERVER_CUSTOM_MODULE is deprecated and will be removed. " "Use the environment variable QSERVER_CUSTOM_MODULES, which accepts a string with " @@ -450,15 +422,10 @@ async def purge_expired_sessions_and_api_keys(): module_names = [] if "custom_modules" in server_settings["server_configuration"]: module_names = server_settings["server_configuration"]["custom_modules"] - logger.info( - "Custom modules from config file: %s", pprint.pformat(module_names) - ) + logger.info("Custom modules from config file: %s", pprint.pformat(module_names)) elif module_names_str: module_names = re.split(":|,", module_names_str) - logger.info( - "Custom modules from environment variable: %s", - pprint.pformat(module_names), - ) + logger.info("Custom modules from environment variable: %s", pprint.pformat(module_names)) if module_names: # Import all listed custom modules @@ -466,14 +433,10 @@ async def purge_expired_sessions_and_api_keys(): for name in module_names: try: logger.info("Importing custom module '%s' ...", name) - custom_code_modules.append( - importlib.import_module(name.replace("-", "_")) - ) + custom_code_modules.append(importlib.import_module(name.replace("-", "_"))) logger.info("Module '%s' was imported successfully.", name) except Exception as ex: - logger.error( - "Failed to import custom instrument module '%s': %s", name, ex - ) + logger.error("Failed to import custom instrument module '%s': %s", name, ex) SR.set_custom_code_modules(custom_code_modules) else: SR.set_custom_code_modules([]) @@ -524,11 +487,7 @@ def override_get_resource_access_manager(): @lru_cache(1) def override_get_settings(): settings = get_settings() - setattr( - settings, - "authentication_provider_names", - [_["provider"] for _ in authentication_providers], - ) + setattr(settings, "authentication_provider_names", [_["provider"] for _ in authentication_providers]) for item in [ "allow_anonymous_access", "secret_keys", @@ -551,19 +510,13 @@ def override_get_settings(): settings.database_pool_size = database["pool_size"] if database.get("pool_pre_ping"): settings.database_pool_pre_ping = database["pool_pre_ping"] - object_cache_available_bytes = server_settings.get("object_cache", {}).get( - "available_bytes" - ) + object_cache_available_bytes = server_settings.get("object_cache", {}).get("available_bytes") if object_cache_available_bytes is not None: - setattr( - settings, "object_cache_available_bytes", object_cache_available_bytes - ) + setattr(settings, "object_cache_available_bytes", object_cache_available_bytes) if authentication.get("providers"): # If we support authentication providers, we need a database, so if one is # not set, use a SQLite database in the current working directory. - settings.database_uri = ( - settings.database_uri or "sqlite:///./bluesky_httpserver.sqlite" - ) + settings.database_uri = settings.database_uri or "sqlite:///./bluesky_httpserver.sqlite" return settings @app.middleware("http") @@ -590,11 +543,7 @@ async def capture_metrics(request: Request, call_next): response.headers["Server-Timing"] = ", ".join( f"{key};" + ";".join( - ( - f"{metric}={value * 1000:.1f}" - if metric == "dur" - else f"{metric}={value:.1f}" - ) + (f"{metric}={value * 1000:.1f}" if metric == "dur" else f"{metric}={value:.1f}") for metric, value in metrics_.items() ) for key, metrics_ in metrics.items() @@ -606,13 +555,9 @@ async def capture_metrics(request: Request, call_next): async def double_submit_cookie_csrf_protection(request: Request, call_next): # https://cheatsheetseries.owasp.org/cheatsheets/Cross-Site_Request_Forgery_Prevention_Cheat_Sheet.html#double-submit-cookie csrf_cookie = request.cookies.get(CSRF_COOKIE_NAME) - if (request.method not in SAFE_METHODS) and set(request.cookies).intersection( - SENSITIVE_COOKIES - ): + if (request.method not in SAFE_METHODS) and set(request.cookies).intersection(SENSITIVE_COOKIES): if not csrf_cookie: - return Response( - status_code=403, content=f"Expected {CSRF_COOKIE_NAME} cookie" - ) + return Response(status_code=403, content="Expected tiled_csrf_token cookie") # Get the token from the Header or (if not there) the query parameter. csrf_token = request.headers.get(CSRF_HEADER_NAME) if csrf_token is None: @@ -625,9 +570,7 @@ async def double_submit_cookie_csrf_protection(request: Request, call_next): ) # Securely compare the token with the cookie. if not secrets.compare_digest(csrf_token, csrf_cookie): - return Response( - status_code=403, content="Double-submit CSRF tokens do not match" - ) + return Response(status_code=403, content="Double-submit CSRF tokens do not match") response = await call_next(request) response.__class__ = PatchedStreamingResponse # tolerate memoryview @@ -656,9 +599,7 @@ async def set_cookies(request: Request, call_next): app.openapi = partial(custom_openapi, app) app.dependency_overrides[get_authenticators] = override_get_authenticators app.dependency_overrides[get_api_access_manager] = override_get_api_access_manager - app.dependency_overrides[get_resource_access_manager] = ( - override_get_resource_access_manager - ) + app.dependency_overrides[get_resource_access_manager] = override_get_resource_access_manager app.dependency_overrides[get_settings] = override_get_settings def add_custom_middleware(): diff --git a/bluesky_httpserver/schemas.py b/bluesky_httpserver/schemas.py index 8bb1b96..f127b88 100644 --- a/bluesky_httpserver/schemas.py +++ b/bluesky_httpserver/schemas.py @@ -271,9 +271,6 @@ class Session(pydantic.BaseModel, **orm): uuid: uuid.UUID expiration_time: datetime revoked: bool - # Free-form JSON dict populated by an authenticator's UserSessionState.state. - # Downstream services that share Tiled authentication may use this to carry - # e.g. upstream OIDC access/refresh tokens across session refreshes. state: Dict = {} @@ -293,11 +290,6 @@ class Principal(pydantic.BaseModel, **orm): roles: Optional[List[str]] = [] scopes: Optional[List[str]] = [] api_key_scopes: Optional[Union[List[str], None]] = None - # Raw access token for principals authenticated by an external OIDC provider. - # This is populated only when the current request was authenticated with an - # externally-minted (proxied OIDC) access token, and is intended for - # downstream services that need to perform an on-behalf-of exchange. - # It is NEVER persisted to the auth DB. access_token: Optional[str] = None @classmethod