diff --git a/bluesky_httpserver/app.py b/bluesky_httpserver/app.py index 0d96667..946673e 100644 --- a/bluesky_httpserver/app.py +++ b/bluesky_httpserver/app.py @@ -15,10 +15,15 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.openapi.utils import get_openapi -from .authentication import ExternalAuthenticator, InternalAuthenticator -from .console_output import CollectPublishedConsoleOutput, ConsoleOutputStream, SystemInfoStream +from .authenticators import ProxiedOIDCAuthenticator +from .console_output import ( + CollectPublishedConsoleOutput, + ConsoleOutputStream, + SystemInfoStream, +) from .core import PatchedStreamingResponse from .database.core import purge_expired +from .protocols import ExternalAuthenticator, InternalAuthenticator from .resources import SERVER_RESOURCES as SR from .routers import core_api from .settings import get_settings @@ -158,14 +163,9 @@ def build_app(authentication=None, api_access=None, resource_access=None, server logger.info("All custom routers are included successfully.") from .authentication import ( + add_external_routes, + add_internal_routes, base_authentication_router, - build_auth_code_route, - build_authorize_route, - build_device_code_authorize_route, - build_device_code_form_route, - build_device_code_submit_route, - build_device_code_token_route, - build_handle_credentials_route, oauth2_scheme, ) @@ -185,38 +185,11 @@ def build_app(authentication=None, api_access=None, resource_access=None, server provider = spec["provider"] authenticator = spec["authenticator"] if isinstance(authenticator, InternalAuthenticator): - authentication_router.post(f"/provider/{provider}/token")( - build_handle_credentials_route(authenticator, provider) - ) + add_internal_routes(authentication_router, provider, authenticator) elif isinstance(authenticator, ExternalAuthenticator): - # Standard OAuth callback route (authorization code flow) - authentication_router.get(f"/provider/{provider}/code")( - build_auth_code_route(authenticator, provider) - ) - authentication_router.post(f"/provider/{provider}/code")( - build_auth_code_route(authenticator, provider) - ) - # Device code flow routes for CLI/headless clients - # GET /authorize - redirects browser to OIDC provider - authentication_router.get(f"/provider/{provider}/authorize")( - build_authorize_route(authenticator, provider) - ) - # POST /authorize - initiates device code flow (returns device_code, user_code, etc.) - authentication_router.post(f"/provider/{provider}/authorize")( - build_device_code_authorize_route(authenticator, provider) - ) - # GET /device_code - shows user code entry form - authentication_router.get(f"/provider/{provider}/device_code")( - build_device_code_form_route(authenticator, provider) - ) - # POST /device_code - handles user code submission after browser auth - authentication_router.post(f"/provider/{provider}/device_code")( - build_device_code_submit_route(authenticator, provider) - ) - # POST /token - CLI client polls this for tokens - authentication_router.post(f"/provider/{provider}/token")( - build_device_code_token_route(authenticator, provider) - ) + add_external_routes(authentication_router, provider, authenticator) + 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", []): @@ -262,9 +235,11 @@ async def startup_event(): from .database import orm from .database.core import ( # make_admin_by_identity, REQUIRED_REVISION, + DatabaseUpgradeNeeded, UninitializedDatabase, check_database, initialize_database, + upgrade, ) connect_args = {} @@ -282,6 +257,10 @@ 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) @@ -416,10 +395,30 @@ 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() + """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() + 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(): diff --git a/bluesky_httpserver/_authentication.py b/bluesky_httpserver/authentication.py similarity index 72% rename from bluesky_httpserver/_authentication.py rename to bluesky_httpserver/authentication.py index c1144f5..655c3b4 100644 --- a/bluesky_httpserver/_authentication.py +++ b/bluesky_httpserver/authentication.py @@ -1,19 +1,36 @@ import asyncio import hashlib +import logging import secrets import uuid as uuid_module import warnings from datetime import datetime, timedelta -from typing import Optional - -from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, Response, Security, WebSocket +from typing import Any, List, Optional + +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 +from .authenticators import ProxiedOIDCAuthenticator + # To hide third-party warning # .../jose/backends/cryptography_backend.py:18: CryptographyDeprecationWarning: # int_from_bytes is deprecated, use int.from_bytes instead @@ -36,12 +53,14 @@ 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, lookup_valid_pending_session_by_user_code, lookup_valid_session, ) +from .protocols import InternalAuthenticator from .settings import get_sessionmaker, get_settings from .utils import ( API_KEY_COOKIE_NAME, @@ -60,6 +79,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 +161,9 @@ def create_refresh_token(session_id, secret_key, expires_delta): return encoded_jwt -def decode_token(token, secret_keys): +def decode_token( + token: str, secret_keys: List[str], proxied_authenticator: Optional[ProxiedOIDCAuthenticator] = None +) -> dict[str, Any]: credentials_exception = HTTPException( status_code=401, detail="Could not validate credentials", @@ -152,16 +175,33 @@ def decode_token(token, secret_keys): for secret_key in secret_keys: try: payload = jwt.decode(token, secret_key, algorithms=[ALGORITHM]) - break + return payload except ExpiredSignatureError: - # Do not let this be caught below with the other JWTError types. raise except JWTError: - # Try the next key in the key rotation. continue - else: - raise credentials_exception - return payload + # If none of the keys worked, try the proxied authenticator + # (e.g. tokens issued directly by an OIDC provider in the device code flow). + if proxied_authenticator: + return proxied_authenticator.decode_token(token) + raise credentials_exception + + +def _extract_scopes( + decoded_access_token: dict[str, Any], +) -> set[str]: + """Extract scopes from a decoded access token. + + Tiled-minted tokens (auth code flow) store scopes as a list under "scp". + OIDC-provider tokens (device code flow) store them as a space-separated + string under "scope". Handle both. + """ + if "scp" in decoded_access_token: + scp = decoded_access_token["scp"] + return set(scp) if isinstance(scp, list) else set(scp.split(" ")) + if "scope" in decoded_access_token: + return set(decoded_access_token["scope"].split(" ")) + return set() async def get_api_key( @@ -175,10 +215,56 @@ async def get_api_key( return None +def headers_for_401(request: Request, security_scopes: SecurityScopes): + # call directly from methods, rather than as a dependency, to avoid calling + # when not needed. + if security_scopes.scopes: + authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' + else: + authenticate_value = "Bearer" + headers_for_401 = { + "WWW-Authenticate": authenticate_value, + "X-Tiled-Root": get_base_url(request), + } + return headers_for_401 + + +async def get_decoded_access_token( + request: Request, + security_scopes: SecurityScopes, + access_token: str = Depends(oauth2_scheme), + settings: BaseSettings = Depends(get_settings), +): + if not access_token: + return None + try: + payload = decode_token(access_token, settings.secret_keys, settings.authenticator) + except ExpiredSignatureError: + raise HTTPException( + status_code=401, + detail="Access token has expired. Refresh token.", + headers=headers_for_401(request, security_scopes), + ) + return payload + + +def move_api_key(request: Request, api_key: Optional[str] = Depends(get_api_key)): + """ + Move API key from query parameter to cookie. + + When a URL with an API key in the query parameter is opened in a browser, + the API key is set as a cookie so that subsequent requests from the browser + are authenticated. (This approach was inspired by Jupyter notebook.) + """ + 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}) + + def get_current_principal( request: Request, security_scopes: SecurityScopes, access_token: str = Depends(oauth2_scheme), + decoded_access_token: str = Depends(get_decoded_access_token), api_key: str = Depends(get_api_key), settings: BaseSettings = Depends(get_settings), authenticators=Depends(get_authenticators), @@ -195,160 +281,197 @@ def get_current_principal( If this server is configured with a "single-user API key", then the Principal will be SpecialUsers.admin always. """ - if security_scopes.scopes: - authenticate_value = f'Bearer scope="{security_scopes.scope_str}"' - else: - authenticate_value = "Bearer" - headers_for_401 = { - "WWW-Authenticate": authenticate_value, - "X-Tiled-Root": get_base_url(request), - } # 'api_key_scopes' is a set of allowed scopes for API key if authorized with API key. # otherwise it is None. The original set of API key scopes is used for generating new # API keys. - roles, scopes, api_key_scopes = {}, {}, None if api_key is not None: if authenticators: - # Tiled is in a multi-user configuration with authentication providers. with get_sessionmaker(settings.database_settings)() as db: - # We store the hashed value of the API key secret. - # By comparing hashes we protect against timing attacks. - # By storing only the hash of the (high-entropy) secret - # we reduce the value of that an attacker can extracted from a - # stolen database backup. - try: - secret = bytes.fromhex(api_key) - except Exception: - # Not valid hex, therefore not a valid API key - raise HTTPException( - status_code=401, - detail="Invalid API key", - headers=headers_for_401, - ) - api_key_orm = lookup_valid_api_key(db, secret) - 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 - ) - scope_sets = [api_access_manager.get_user_scopes(_) for _ in ids] - principal_scopes = set.union(*scope_sets) if scope_sets else set() - - roles_sets = [api_access_manager.get_user_roles(_) for _ in ids] - roles = set.union(*roles_sets) if roles_sets else set() - - # principal_scopes = set().union(*[role.scopes for role in principal.roles]) - - # This intersection addresses the case where the Principal has - # lost a scope that they had when this key was created. - api_key_scopes = set(api_key_orm.scopes) - scopes = api_key_scopes.intersection(principal_scopes | {"inherit"}) - if "inherit" in scopes: - # The scope "inherit" is a metascope that confers all the - # scopes for the Principal associated with this API, - # resolved at access time. - scopes.update(principal_scopes) - scopes.discard("inherit") - api_key_orm.latest_activity = utcnow() - db.commit() - else: - raise HTTPException( - status_code=401, - detail="Invalid API key", - headers=headers_for_401, - ) - else: - # HTTP Server is in a "single user" mode with only one API key. - if secrets.compare_digest(api_key, settings.single_user_api_key): - username = SpecialUsers.single_user.value - scopes = api_access_manager.get_user_scopes(username) - roles = api_access_manager.get_user_roles(username) - - 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)], + principal = get_current_principal_from_api_key( + api_key, authenticators, db, settings, api_access_manager ) - - else: - 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}) - elif access_token is not None: - try: - payload = decode_token(access_token, settings.secret_keys) - except ExpiredSignatureError: + else: + principal = get_current_principal_from_single_user_api_key(api_key, settings, api_access_manager) + if principal is None: raise HTTPException( status_code=401, - detail="Access token has expired. Refresh token.", - headers=headers_for_401, + detail="Invalid API key", + headers=headers_for_401(request, security_scopes), ) - 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"] - ], + move_api_key(request, api_key) + elif decoded_access_token is not None: + principal = get_current_principal_from_token( + authenticators, access_token, decoded_access_token, settings, api_access_manager, request ) + else: + principal = get_current_principal_public_access(settings, api_access_manager) - # scopes = payload["scp"] + check_scopes(request, security_scopes, principal) - # 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]) + return principal + + +def get_current_principal_from_api_key( + api_key: str, + authenticators, + db, + settings: BaseSettings, + api_access_manager, +) -> schemas.Principal or None: + """ + Tiled is in a multi-user configuration with authentication providers. + We store the hashed value of the API key secret. + By comparing hashes we protect against timing attacks. + By storing only the hash of the (high-entropy) secret + we reduce the value of that an attacker can extracted from a + stolen database backup. + """ + try: + secret = bytes.fromhex(api_key) + except Exception: + return None + + api_key_orm = lookup_valid_api_key(db, secret) + 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) + scope_sets = [api_access_manager.get_user_scopes(_) for _ in ids] + principal_scopes = set.union(*scope_sets) if scope_sets else set() roles_sets = [api_access_manager.get_user_roles(_) for _ in ids] roles = set.union(*roles_sets) if roles_sets else set() + # This intersection addresses the case where the Principal has + # lost a scope that they had when this key was created. + api_key_scopes = set(api_key_orm.scopes) + scopes = api_key_scopes.intersection(principal_scopes | {"inherit"}) + if "inherit" in scopes: + # The scope "inherit" is a metascope that confers all the + # scopes for the Principal associated with this API, + # resolved at access time. + scopes.update(principal_scopes) + scopes.discard("inherit") + api_key_orm.latest_activity = utcnow() + db.commit() + return cleanup_principal_scopes(roles, scopes, api_key_scopes, principal) else: - # No form of authentication is present. - username = SpecialUsers.public.value - # This is a 'dummy' principal used to pass data within the server. Not saved to the databased. + return None + + +def get_current_principal_from_single_user_api_key( + api_key: str, settings: BaseSettings, api_access_manager +) -> schemas.Principal or None: + """Validates single user api key and sets the scopes and roles""" + if secrets.compare_digest(api_key, settings.single_user_api_key): + username = SpecialUsers.single_user.value + scopes = api_access_manager.get_user_scopes(username) + roles = api_access_manager.get_user_roles(username) + 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)], ) + return cleanup_principal_scopes(roles, scopes, None, principal) + else: + return None - # Is anonymous public access permitted? - if settings.allow_anonymous_access: - # Any user who can see the server can make unauthenticated requests. - # This is a sentinel that has special meaning to the authorization - # code (the access control policies). - scopes = api_access_manager.get_user_scopes(username) - roles = api_access_manager.get_user_roles(username) +def get_current_principal_from_token( + authenticators, access_token, decoded_access_token, settings, api_access_manager, request +) -> schemas.Principal or None: + """Get a principal from the stored token and set the scopes appropriately""" + + if "sub_typ" in decoded_access_token: + principal = schemas.Principal( + uuid=uuid_module.UUID(hex=decoded_access_token["sub"]), + type=decoded_access_token["sub_typ"], + identities=[ + schemas.Identity(id=identity["id"], provider=identity["idp"]) + for identity in decoded_access_token["ids"] + ], + ) + + ids = [_["id"] for _ in decoded_access_token["ids"] if _["idp"] in settings.authentication_provider_names] + 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() + else: + + identity_id = decoded_access_token.get("user") or decoded_access_token.get("sub") + provider = request.app.state.provider + + 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(decoded_access_token) + 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: - # In this mode, there may still be entries that are visible to all, - # but users have to authenticate as *someone* to see anything. - # They can still access the / and /docs routes. - scopes = {} - roles = {} - - # Scope enforcement happens here. - # https://fastapi.tiangolo.com/advanced/security/oauth2-scopes/ - if not set(security_scopes.scopes).issubset(scopes): - # Include a link to the root page which provides a list of - # authenticators. The use case here is: - # 1. User is emailed a link like https://example.com/subpath/node/metadata/a/b/c - # 2. Tiled Client tries to connect to that and gets 401. - # 3. Client can use this header to find its way to - # https://examples.com/subpath/ and obtain a list of - # authentication providers and endpoints. + extra_scopes = set() + roles = set() + scopes = set(token_scopes) | set(extra_scopes) + return cleanup_principal_scopes(roles, scopes, None, principal) + + +def get_current_principal_public_access(settings: BaseSettings, api_access_manager): + """Check if public access is enabled and create a principal if it is""" + roles, scopes = {}, {} + # No form of authentication is present. + username = SpecialUsers.public.value + # This is a 'dummy' principal used to pass data within the server. Not saved to the databased. + 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)], + ) + + # Is anonymous public access permitted? + if settings.allow_anonymous_access: + # Any user who can see the server can make unauthenticated requests. + # This is a sentinel that has special meaning to the authorization + # code (the access control policies). + scopes = api_access_manager.get_user_scopes(username) + roles = api_access_manager.get_user_roles(username) + + else: + # In this mode, there may still be entries that are visible to all, + # but users have to authenticate as *someone* to see anything. + # They can still access the / and /docs routes. + scopes = {} + roles = {} + return cleanup_principal_scopes(roles, scopes, None, principal) + + +def check_scopes(request: Request, security_scopes: SecurityScopes, principal: schemas.Principal): + """Enforce scope limits""" + if not set(security_scopes.scopes).issubset(principal.scopes): raise HTTPException( status_code=401, detail=( "Not enough permissions. " f"Requires scopes {security_scopes.scopes}. " - f"Request had scopes {list(scopes)}" + f"Request had scopes {list(principal.scopes)}" ), - headers=headers_for_401, + headers=headers_for_401(request, security_scopes), ) + +def cleanup_principal_scopes(roles, scopes, api_key_scopes, principal): + """Sort the scopes and include them to the principals list of scopes""" roles_list, scopes_list = list(roles), list(scopes) roles_list.sort() scopes_list.sort() @@ -373,12 +496,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 +526,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 +601,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 +614,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 +646,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 +654,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. @@ -481,9 +665,10 @@ async def auth_code( return auth_code -def build_handle_credentials_route(authenticator, provider): +def add_internal_routes(router: APIRouter, provider: str, authenticator: InternalAuthenticator): "Register a handle_credentials route function for this Authenticator." + @router.post(f"/provider/{provider}/token") async def handle_credentials( request: Request, form_data: OAuth2PasswordRequestForm = Depends(), @@ -495,6 +680,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,12 +697,41 @@ 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 +def add_external_routes(router: APIRouter, provider: str, authenticator: InternalAuthenticator): + router.get(f"/provider/{provider}/code")(build_auth_code_route(authenticator, provider)) + router.post(f"/provider/{provider}/code")(build_auth_code_route(authenticator, provider)) + # Device code flow routes for CLI/headless clients + # GET /authorize - redirects browser to OIDC provider + router.get(f"/provider/{provider}/authorize")(build_authorize_route(authenticator, provider)) + # POST /authorize - initiates device code flow (returns device_code, user_code, etc.) + router.post(f"/provider/{provider}/authorize")(build_device_code_authorize_route(authenticator, provider)) + # GET /device_code - shows user code entry form + router.get(f"/provider/{provider}/device_code")(build_device_code_form_route(authenticator, provider)) + # POST /device_code - handles user code submission after browser auth + router.post(f"/provider/{provider}/device_code")(build_device_code_submit_route(authenticator, provider)) + # POST /token - CLI client polls this for tokens + 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, + ) + + def create_pending_session(db): """ Create a pending session for device code flow. @@ -558,11 +773,19 @@ 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. + 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 +814,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 { @@ -648,7 +874,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 """ @@ -684,6 +910,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 +937,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 +1064,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 +1091,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 +1153,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 +1382,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/authentication/__init__.py b/bluesky_httpserver/authentication/__init__.py deleted file mode 100644 index 85d835e..0000000 --- a/bluesky_httpserver/authentication/__init__.py +++ /dev/null @@ -1,35 +0,0 @@ -from .._authentication import ( - base_authentication_router, - build_auth_code_route, - build_authorize_route, - build_device_code_authorize_route, - build_device_code_form_route, - build_device_code_submit_route, - build_device_code_token_route, - build_handle_credentials_route, - get_current_principal, - get_current_principal_websocket, - oauth2_scheme, -) -from .authenticator_base import ( - ExternalAuthenticator, - InternalAuthenticator, - UserSessionState, -) - -__all__ = [ - "ExternalAuthenticator", - "InternalAuthenticator", - "UserSessionState", - "get_current_principal", - "get_current_principal_websocket", - "base_authentication_router", - "build_auth_code_route", - "build_authorize_route", - "build_device_code_authorize_route", - "build_device_code_form_route", - "build_device_code_submit_route", - "build_device_code_token_route", - "build_handle_credentials_route", - "oauth2_scheme", -] diff --git a/bluesky_httpserver/authenticators.py b/bluesky_httpserver/authenticators.py index a58fedf..fba259d 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 @@ -16,7 +17,7 @@ from pydantic import Secret from starlette.responses import RedirectResponse -from .authentication import ( +from .protocols import ( ExternalAuthenticator, InternalAuthenticator, UserSessionState, @@ -26,6 +27,10 @@ logger = logging.getLogger(__name__) +class AuthCodeExchangeException(Exception): + pass + + class DummyAuthenticator(InternalAuthenticator): """ For test and demo purposes only! @@ -71,7 +76,7 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi 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, {}) @@ -105,7 +110,7 @@ async def authenticate(self, username: str, password: str) -> Optional[UserSessi return UserSessionState(username, {}) except pamela.PAMError: # Authentication failed. - return None + return class OIDCAuthenticator(ExternalAuthenticator): @@ -189,17 +194,18 @@ def device_authorization_endpoint(self) -> str: 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", []) - def decode_token(self, token: str) -> dict[str, Any]: + def decode_token(self, id_token: str, access_token: Optional[str] = None) -> dict[str, Any]: 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]: @@ -223,37 +229,16 @@ async def authenticate(self, request: Request) -> Optional[UserSessionState]: logger.error("Authentication error: %r", response_body) return None 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["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", 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): @@ -310,18 +295,302 @@ 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, + graph_username_attribute: Optional[str] = None, + ): + """A MS Entra specific version of the OIDC authenticator + + It attempts to extract a username from the standard list of claims returned + from the token Entra provides. Alternatively if a graph_username_attribute + is used then a call is made to MSGraphAPI to get the provided user attribute + and use it as the username instead. + + The graph API call is the recommended way to authenticate with MS products, as all + claims in the token are inconsistent and not guaranteed. + + """ + 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 + self.graph_username_attribute = graph_username_attribute + + @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) + + user_claims_list = [f"{key}:{value}" for key, value in claims.items()] + logger.debug("Claims:\n%s", "\n".join(user_claims_list)) + # 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 graph_lookup(self, access_token, user_param): + """Uses the access token provided in the auth flow to lookup a user parameter""" + headers = {"Authorization": f"Bearer {access_token}"} + + async with httpx.AsyncClient() as client: + response = await client.get( + "https://graph.microsoft.com/v1.0/me", + params={"$select": user_param}, + headers=headers, + ) + + response.raise_for_status() + + return response.json() + + def log_token_claims(self, verified_body): + """log token claims + Includes logging of the token claims so misconfigurations are easier + to diagnose. Keep at debug level 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 preferred_username=%r", + verified_body.get("entra_username"), + verified_body.get("user"), + verified_body.get("entra_sub"), + verified_body.get("preferred_username"), + ) + + async def get_username_from_graph(self, access_token): + """Attempts to get the username from either claims or MSGraphAPI call + + If no username is found, there are errors in looking up the graphAPI + username, or whatever it returns None + """ + try: + profile = await self.graph_lookup(access_token, self.graph_username_attribute) + logger.debug("Graph Profile: %r", profile) + except (httpx.HTTPStatusError, httpx.RequestError, ValueError): + logger.warning("Graph lookup failed") + username = None + if profile: + username = profile.get(self.graph_username_attribute) + if not username: + logger.warning( + "Graph lookup succeeded but %s was empty", + self.graph_username_attribute, + ) + return username + + def create_usersession(self, access_token, refresh_token, username): + """Create usersession from tokens and final username + + 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 auth_code_exchange(self, request: Request): + """Perform the authorization code exchange""" + code = request.query_params.get("code") + if not code: + logger.warning("Authentication failed: No authorization code parameter provided.") + raise AuthCodeExchangeException + 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) + raise AuthCodeExchangeException + logger.debug("Response: %s", response_body) + return response_body + + 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. + + When an error occurs, the authenticate function will return None + instead of a UserSessionState + """ + try: + response_body = await self.auth_code_exchange(request) + except AuthCodeExchangeException: + 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 + self.log_token_claims(verified_body) + + if self.graph_username_attribute is not None: + username = await self.get_username_from_graph(access_token) + else: + username = verified_body.get("user") or verified_body["sub"] + + if username is not None: + return self.create_usersession(access_token, refresh_token, username) + else: + return None + + 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 +600,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 +608,6 @@ async def exchange_code( class SAMLAuthenticator(ExternalAuthenticator): - def __init__( self, saml_settings, # See EXAMPLE_SAML_SETTINGS below. @@ -420,7 +689,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. @@ -682,7 +950,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( diff --git a/bluesky_httpserver/database/core.py b/bluesky_httpserver/database/core.py index 52d102f..dbf3083 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,28 @@ 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. @@ -216,6 +238,8 @@ 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 is None: + return None if session.expiration_time is not None and session.expiration_time < datetime.utcnow(): db.delete(session) db.commit() 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..5d4b288 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 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/authentication/authenticator_base.py b/bluesky_httpserver/protocols.py similarity index 100% rename from bluesky_httpserver/authentication/authenticator_base.py rename to bluesky_httpserver/protocols.py diff --git a/bluesky_httpserver/routers/core_api.py b/bluesky_httpserver/routers/core_api.py index 9e47d58..f86dcb1 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,71 @@ 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: + 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 + + 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 +1239,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 +1266,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..f127b88 100644 --- a/bluesky_httpserver/schemas.py +++ b/bluesky_httpserver/schemas.py @@ -271,6 +271,7 @@ class Session(pydantic.BaseModel, **orm): uuid: uuid.UUID expiration_time: datetime revoked: bool + state: Dict = {} class Principal(pydantic.BaseModel, **orm): @@ -289,6 +290,7 @@ class Principal(pydantic.BaseModel, **orm): roles: Optional[List[str]] = [] scopes: Optional[List[str]] = [] api_key_scopes: Optional[Union[List[str], None]] = None + access_token: Optional[str] = None @classmethod def from_orm(cls, orm, latest_activity=None): diff --git a/bluesky_httpserver/tests/conftest.py b/bluesky_httpserver/tests/conftest.py index 8a81df9..dd2f288 100644 --- a/bluesky_httpserver/tests/conftest.py +++ b/bluesky_httpserver/tests/conftest.py @@ -1,14 +1,22 @@ import os import time as ttime +from typing import Any, Tuple +import httpx import pytest import requests from bluesky_queueserver.manager.comms import zmq_single_request from bluesky_queueserver.manager.tests.common import re_manager_cmd # noqa: F401 from bluesky_queueserver.manager.tests.common import set_qserver_zmq_encoding # noqa: F401 +from cryptography.hazmat.primitives.asymmetric import rsa +from jose.backends import RSAKey +from respx import MockRouter +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker from xprocess import ProcessStarter import bluesky_httpserver.server as bqss +from bluesky_httpserver.database.base import Base SERVER_ADDRESS = "localhost" SERVER_PORT = "60610" @@ -231,6 +239,56 @@ def wait_for_ip_kernel_idle(timeout, polling_period=0.2, api_key=API_KEY_FOR_TES return False +# ============================================================================ +# AUTH Test Fixtures +# ============================================================================ + + +@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]: + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + public_key = private_key.public_key() + return (private_key, public_key) + + +@pytest.fixture +def json_web_keyset(keys: Tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]) -> list[dict[str, Any]]: + _, public_key = keys + return [RSAKey(key=public_key, algorithm="RS256").to_dict()] + + +@pytest.fixture +def mock_oidc_server( + respx_mock: MockRouter, + oidc_well_known_url: str, + well_known_response: dict[str, Any], + json_web_keyset: list[dict[str, Any]], +) -> MockRouter: + respx_mock.get(oidc_well_known_url).mock(return_value=httpx.Response(httpx.codes.OK, json=well_known_response)) + respx_mock.get(well_known_response["jwks_uri"]).mock( + return_value=httpx.Response(httpx.codes.OK, json={"keys": json_web_keyset}) + ) + return respx_mock + + +@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() + + # ============================================================================ # OIDC Test Fixtures # ============================================================================ diff --git a/bluesky_httpserver/tests/test_auth_for_websockets.py b/bluesky_httpserver/tests/test_auth_for_websockets.py index 44bac42..0b53e9c 100644 --- a/bluesky_httpserver/tests/test_auth_for_websockets.py +++ b/bluesky_httpserver/tests/test_auth_for_websockets.py @@ -2,11 +2,17 @@ import pprint import threading import time as ttime +from unittest.mock import MagicMock import pytest from bluesky_queueserver.manager.tests.common import re_manager, re_manager_cmd, re_manager_factory # noqa F401 +from sqlalchemy.orm import sessionmaker from websockets.sync.client import connect +from bluesky_httpserver import authentication as _auth +from bluesky_httpserver.database import orm as db_orm +from bluesky_httpserver.database.core import create_user + from .conftest import fastapi_server_fs # noqa: F401 from .conftest import ( SERVER_ADDRESS, @@ -173,3 +179,143 @@ def test_websocket_auth_01( assert isinstance(msg["msg"], dict) else: assert False, f"Unknown authentication type: {ws_auth_type!r}" + + +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() + + 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 diff --git a/bluesky_httpserver/tests/test_authenticators.py b/bluesky_httpserver/tests/test_authenticators.py index 7b7dd4b..b3c91db 100644 --- a/bluesky_httpserver/tests/test_authenticators.py +++ b/bluesky_httpserver/tests/test_authenticators.py @@ -1,17 +1,30 @@ import asyncio +import logging import os import time +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 fastapi.security import SecurityScopes from jose import ExpiredSignatureError, jwt -from jose.backends import RSAKey from respx import MockRouter from starlette.datastructures import URL, QueryParams +from starlette.requests import Request -from ..authenticators import LDAPAuthenticator, OIDCAuthenticator, ProxiedOIDCAuthenticator, UserSessionState +from bluesky_httpserver import authentication as _auth + +from ..authenticators import ( + EntraAuthenticator, + LDAPAuthenticator, + OIDCAuthenticator, + ProxiedOIDCAuthenticator, + UserSessionState, +) LDAP_TEST_HOST = os.environ.get("QSERVER_TEST_LDAP_HOST", "localhost") LDAP_TEST_PORT = int(os.environ.get("QSERVER_TEST_LDAP_PORT", "1389")) @@ -60,38 +73,6 @@ async def testing(): asyncio.run(testing()) -@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]: - private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) - public_key = private_key.public_key() - return (private_key, public_key) - - -@pytest.fixture -def json_web_keyset(keys: Tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]) -> list[dict[str, Any]]: - _, public_key = keys - return [RSAKey(key=public_key, algorithm="RS256").to_dict()] - - -@pytest.fixture -def mock_oidc_server( - respx_mock: MockRouter, - oidc_well_known_url: str, - well_known_response: dict[str, Any], - json_web_keyset: list[dict[str, Any]], -) -> MockRouter: - respx_mock.get(oidc_well_known_url).mock(return_value=httpx.Response(httpx.codes.OK, json=well_known_response)) - respx_mock.get(well_known_response["jwks_uri"]).mock( - return_value=httpx.Response(httpx.codes.OK, json={"keys": json_web_keyset}) - ) - return respx_mock - - def token(issued: bool, expired: bool) -> dict[str, str]: now = time.time() return { @@ -168,6 +149,36 @@ def test_oidc_decoding( authenticator.decode_token(encrypted_access_token) +def test_entra_decoding_ignores_unmapped_scopes(caplog): + def mock_decode_token(self, id_token, access_token): + return { + "iss": "https://login.microsoftonline.com/example-tenant/v2.0", + "sub": "opaque-sub", + "preferred_username": "alice@example.org", + "scp": "known.scope unknown.scope", + } + + original_decode_token = OIDCAuthenticator.decode_token + OIDCAuthenticator.decode_token = mock_decode_token + try: + caplog.set_level(logging.WARNING) + + authenticator = object.__new__(EntraAuthenticator) + authenticator.scopes_map = {"known.scope": ["read:metadata"]} + claims = authenticator.decode_token("id-token", "access-token") + + assert claims["entra_sub"] == "opaque-sub" + assert claims["entra_username"] == "alice@example.org" + assert claims["user"] == "alice" + assert claims["scope"] == "read:metadata" + assert any( + "Unmapped Entra scope in 'scp': unknown.scope" in record.message + for record in caplog.records + ) + finally: + OIDCAuthenticator.decode_token = original_decode_token + + @pytest.mark.asyncio async def test_proxied_oidc_token_retrieval(oidc_well_known_url: str, mock_oidc_server: MockRouter): authenticator = ProxiedOIDCAuthenticator("tiled", "tiled", oidc_well_known_url, @@ -293,3 +304,342 @@ async def test_OIDCAuthenticator_token_exchange_failure( result = await authenticator.authenticate(mock_request) assert result is None + + +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) + + +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"}) + + +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 + + +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() + + +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() + + +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_empty_or_missing(self): + assert _auth._extract_scopes({}) == set() + assert _auth._extract_scopes({"scp": "", "scope": ""}) == {""} + + +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) + + +def _make_request(*, query_string: bytes = b"", path: str = "/api/status") -> Request: + return Request( + { + "type": "http", + "scheme": "http", + "server": ("localhost", 8000), + "path": path, + "query_string": query_string, + "root_path": "", + "headers": [(b"host", b"localhost:8000")], + } + ) + + +def test_headers_for_401_includes_scope_and_root(): + request = _make_request() + headers = _auth.headers_for_401(request, SecurityScopes(scopes=["read:status"])) + assert headers["WWW-Authenticate"] == 'Bearer scope="read:status"' + assert headers["X-Tiled-Root"] == "http://localhost:8000/api" + + +def test_check_scopes_raises_for_missing_scope(): + principal = _auth.schemas.Principal( + uuid="123e4567-e89b-12d3-a456-426614174000", + type="user", + scopes={"read:status"}, + ) + request = _make_request() + with pytest.raises(HTTPException) as excinfo: + _auth.check_scopes(request, SecurityScopes(scopes=["admin:read:principals"]), principal) + assert excinfo.value.status_code == 401 + assert "Not enough permissions" in excinfo.value.detail + + +def test_cleanup_principal_scopes_assigns_sorted_fields(): + principal = _auth.schemas.Principal( + uuid="123e4567-e89b-12d3-a456-426614174000", + type="user", + identities=[_auth.schemas.Identity(id="alice", provider="internal")], + ) + result = _auth.cleanup_principal_scopes( + roles={"expert", "admin"}, + scopes={"read:status", "read:queue"}, + api_key_scopes={"read:status"}, + principal=principal, + ) + assert result.roles == ["admin", "expert"] + assert result.scopes == ["read:queue", "read:status"] + assert result.api_key_scopes == ["read:status"] + + +def test_get_current_principal_rejects_invalid_single_user_api_key(monkeypatch): + class _DummySessionMaker: + def __call__(self): + class _DummyCtx: + def __enter__(self): + return MagicMock() + + def __exit__(self, exc_type, exc, tb): + return False + + return _DummyCtx() + + monkeypatch.setattr(_auth, "get_sessionmaker", lambda _db_settings: _DummySessionMaker()) + + settings = MagicMock() + settings.database_settings = MagicMock() + settings.single_user_api_key = "expected-key" + + api_access_manager = MagicMock() + api_access_manager.get_user_scopes.return_value = {"read:status"} + api_access_manager.get_user_roles.return_value = {"single_user"} + + request = _make_request() + with pytest.raises(HTTPException) as excinfo: + _auth.get_current_principal( + request=request, + security_scopes=SecurityScopes(scopes=[]), + access_token=None, + decoded_access_token=None, + api_key="wrong-key", + settings=settings, + authenticators={}, + api_access_manager=api_access_manager, + ) + assert excinfo.value.status_code == 401 + assert "Invalid API key" in excinfo.value.detail + + +def test_get_current_principal_preserves_api_key_scopes(sqlite_session, monkeypatch): + import hashlib + import secrets as py_secrets + + from sqlalchemy.orm import sessionmaker + + from bluesky_httpserver.database import orm as db_orm + from bluesky_httpserver.database.core import create_user + from bluesky_httpserver.settings import DatabaseSettings + + db = sqlite_session + principal = create_user(db, "internal", "alice") + secret = py_secrets.token_bytes(4 + 32) + apikey_orm = db_orm.APIKey( + principal_id=principal.id, + first_eight=secret.hex()[:8], + hashed_secret=hashlib.sha256(secret).digest(), + scopes=["read:status"], + ) + db.add(apikey_orm) + db.commit() + + engine = db.get_bind() + + def _fake_sessionmaker(_db_settings): + return sessionmaker(bind=engine, autocommit=False, autoflush=False) + + monkeypatch.setattr(_auth, "get_sessionmaker", _fake_sessionmaker) + + settings = MagicMock() + settings.database_settings = DatabaseSettings(uri="sqlite://", pool_size=None, pool_pre_ping=None) + settings.authentication_provider_names = ["internal"] + + api_access_manager = MagicMock() + api_access_manager.get_user_scopes.return_value = {"read:status", "write:queue"} + api_access_manager.get_user_roles.return_value = {"user"} + + request = _make_request() + resolved = _auth.get_current_principal( + request=request, + security_scopes=SecurityScopes(scopes=[]), + access_token=None, + decoded_access_token=None, + api_key=secret.hex(), + settings=settings, + authenticators={"internal": MagicMock()}, + api_access_manager=api_access_manager, + ) + assert resolved.api_key_scopes == ["read:status"] diff --git a/bluesky_httpserver/tests/test_console_output.py b/bluesky_httpserver/tests/test_console_output.py index 3907c58..25a150d 100644 --- a/bluesky_httpserver/tests/test_console_output.py +++ b/bluesky_httpserver/tests/test_console_output.py @@ -1,13 +1,14 @@ import json import pprint import re +import socket import threading import time as ttime from typing import Any import pytest import requests -from bluesky_queueserver.manager.tests.common import re_manager_cmd, re_manager_factory # noqa F401 +from bluesky_queueserver.manager.tests.common import re_manager_factory # noqa F401 from websockets.sync.client import connect from bluesky_httpserver.tests.conftest import ( # noqa F401 @@ -22,6 +23,12 @@ ) +def get_free_tcp_port(): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + class _ReceiveStreamedConsoleOutput(threading.Thread): """ Catch streaming console output and save messages to the buffer. The method is intended @@ -84,17 +91,18 @@ def __del__(self): self.stop() -@pytest.mark.parametrize("zmq_port", (None, 60619)) +@pytest.mark.parametrize("use_custom_port", (False, True)) def test_http_server_stream_console_output_1( monkeypatch, re_manager_cmd, fastapi_server_fs, - zmq_port, # noqa F811 + use_custom_port, ): """ Test for ``stream_console_output`` API """ # Start HTTP Server + zmq_port = get_free_tcp_port() if use_custom_port else None if zmq_port is not None: monkeypatch.setenv("QSERVER_ZMQ_INFO_ADDRESS", f"tcp://localhost:{zmq_port}") fastapi_server_fs() @@ -167,18 +175,19 @@ def test_http_server_stream_console_output_1( @pytest.mark.parametrize("zmq_encoding", (None, "json", "msgpack")) -@pytest.mark.parametrize("zmq_port", (None, 60619)) +@pytest.mark.parametrize("use_custom_port", (False, True)) def test_http_server_console_output_1( monkeypatch, re_manager_cmd, fastapi_server_fs, - zmq_port, zmq_encoding, # noqa F811 + use_custom_port, ): """ Test for ``console_output`` API (not a streaming version). """ # Start HTTP Server + zmq_port = get_free_tcp_port() if use_custom_port else None if zmq_port is not None: monkeypatch.setenv("QSERVER_ZMQ_INFO_ADDRESS", f"tcp://localhost:{zmq_port}") if zmq_encoding is not None: @@ -249,17 +258,18 @@ def test_http_server_console_output_1( assert re.search(expected_output, console_output) -@pytest.mark.parametrize("zmq_port", (None, 60619)) +@pytest.mark.parametrize("use_custom_port", (False, True)) def test_http_server_console_output_update_1( monkeypatch, re_manager_cmd, fastapi_server_fs, - zmq_port, # noqa F811 + use_custom_port, ): """ Test for ``console_output`` API (not a streaming version). """ # Start HTTP Server + zmq_port = get_free_tcp_port() if use_custom_port else None if zmq_port is not None: monkeypatch.setenv("QSERVER_ZMQ_INFO_ADDRESS", f"tcp://localhost:{zmq_port}") fastapi_server_fs() @@ -393,17 +403,18 @@ def __del__(self): self.stop() -@pytest.mark.parametrize("zmq_port", (None, 60619)) +@pytest.mark.parametrize("use_custom_port", (False, True)) def test_http_server_console_output_socket_1( monkeypatch, re_manager_cmd, fastapi_server_fs, - zmq_port, # noqa F811 + use_custom_port, ): """ Test for ``/console_output/ws`` websocket """ # Start HTTP Server + zmq_port = get_free_tcp_port() if use_custom_port else None if zmq_port is not None: monkeypatch.setenv("QSERVER_ZMQ_INFO_ADDRESS", f"tcp://localhost:{zmq_port}") fastapi_server_fs() diff --git a/bluesky_httpserver/tests/test_database.py b/bluesky_httpserver/tests/test_database.py new file mode 100644 index 0000000..a923619 --- /dev/null +++ b/bluesky_httpserver/tests/test_database.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import uuid +from datetime import timedelta + +from bluesky_httpserver import schemas +from bluesky_httpserver.database import orm as db_orm +from bluesky_httpserver.database.core import ( + create_user, + get_or_create_principal, +) + + +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 + + +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 {} + + +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 diff --git a/bluesky_httpserver/tests/test_oidc_authenticators.py b/bluesky_httpserver/tests/test_oidc_authenticators.py index f3249cd..e7e4333 100644 --- a/bluesky_httpserver/tests/test_oidc_authenticators.py +++ b/bluesky_httpserver/tests/test_oidc_authenticators.py @@ -3,49 +3,12 @@ import time from typing import Any, Tuple -import httpx import pytest from cryptography.hazmat.primitives.asymmetric import rsa from jose import ExpiredSignatureError, jwt -from jose.backends import RSAKey from respx import MockRouter -from bluesky_httpserver.authenticators import OIDCAuthenticator, ProxiedOIDCAuthenticator - - -@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]: - """Generate RSA key pair for testing.""" - private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) - public_key = private_key.public_key() - return (private_key, public_key) - - -@pytest.fixture -def json_web_keyset(keys: Tuple[rsa.RSAPrivateKey, rsa.RSAPublicKey]) -> list[dict[str, Any]]: - """Create a JSON Web Key Set from the test keys.""" - _, public_key = keys - return [RSAKey(key=public_key, algorithm="RS256").to_dict()] - - -@pytest.fixture -def mock_oidc_server( - respx_mock: MockRouter, - oidc_well_known_url: str, - well_known_response: dict[str, Any], - json_web_keyset: list[dict[str, Any]], -) -> MockRouter: - """Set up mock OIDC server endpoints.""" - respx_mock.get(oidc_well_known_url).mock(return_value=httpx.Response(httpx.codes.OK, json=well_known_response)) - respx_mock.get(well_known_response["jwks_uri"]).mock( - return_value=httpx.Response(httpx.codes.OK, json={"keys": json_web_keyset}) - ) - return respx_mock +from bluesky_httpserver.authenticators import OIDCAuthenticator def create_token(issued: bool, expired: bool) -> dict[str, Any]: @@ -170,50 +133,3 @@ def test_oidc_authenticator_properties( assert authenticator.confirmation_message == "Logged in as {id}" assert authenticator.redirect_on_success == "https://app.example.com/success" assert authenticator.redirect_on_failure == "https://app.example.com/failure" - - -@pytest.mark.filterwarnings("ignore::DeprecationWarning") -class TestProxiedOIDCAuthenticator: - """Tests for ProxiedOIDCAuthenticator class.""" - - @pytest.mark.asyncio - async def test_proxied_oidc_oauth2_schema( - self, - mock_oidc_server: MockRouter, - oidc_well_known_url: str, - ): - """Test that ProxiedOIDCAuthenticator extracts bearer token correctly.""" - authenticator = ProxiedOIDCAuthenticator( - audience="test_client", - client_id="test_client", - well_known_uri=oidc_well_known_url, - device_flow_client_id="test_cli_client", - ) - - # Create a mock request with Authorization header - test_request = httpx.Request( - "GET", - "http://example.com/api/test", - headers={"Authorization": "Bearer TEST_TOKEN"}, - ) - - # The oauth2_schema should extract the bearer token - token = await authenticator.oauth2_schema(test_request) - assert token == "TEST_TOKEN" - - def test_proxied_oidc_with_scopes( - self, - mock_oidc_server: MockRouter, - oidc_well_known_url: str, - ): - """Test ProxiedOIDCAuthenticator with custom scopes.""" - authenticator = ProxiedOIDCAuthenticator( - audience="test_client", - client_id="test_client", - well_known_uri=oidc_well_known_url, - device_flow_client_id="test_cli_client", - scopes=["openid", "profile", "email"], - ) - - assert authenticator.scopes == ["openid", "profile", "email"] - assert authenticator.device_flow_client_id == "test_cli_client" diff --git a/bluesky_httpserver/tests/test_oidc_proxied_authenticators.py b/bluesky_httpserver/tests/test_oidc_proxied_authenticators.py new file mode 100644 index 0000000..044ac76 --- /dev/null +++ b/bluesky_httpserver/tests/test_oidc_proxied_authenticators.py @@ -0,0 +1,54 @@ +"""Tests for OIDC Authenticator functionality.""" + +import httpx +import pytest +from respx import MockRouter + +from bluesky_httpserver.authenticators import ProxiedOIDCAuthenticator + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +class TestProxiedOIDCAuthenticator: + """Tests for ProxiedOIDCAuthenticator class.""" + + @pytest.mark.asyncio + async def test_proxied_oidc_oauth2_schema( + self, + mock_oidc_server: MockRouter, + oidc_well_known_url: str, + ): + """Test that ProxiedOIDCAuthenticator extracts bearer token correctly.""" + authenticator = ProxiedOIDCAuthenticator( + audience="test_client", + client_id="test_client", + well_known_uri=oidc_well_known_url, + device_flow_client_id="test_cli_client", + ) + + # Create a mock request with Authorization header + test_request = httpx.Request( + "GET", + "http://example.com/api/test", + headers={"Authorization": "Bearer TEST_TOKEN"}, + ) + + # The oauth2_schema should extract the bearer token + token = await authenticator.oauth2_schema(test_request) + assert token == "TEST_TOKEN" + + def test_proxied_oidc_with_scopes( + self, + mock_oidc_server: MockRouter, + oidc_well_known_url: str, + ): + """Test ProxiedOIDCAuthenticator with custom scopes.""" + authenticator = ProxiedOIDCAuthenticator( + audience="test_client", + client_id="test_client", + well_known_uri=oidc_well_known_url, + device_flow_client_id="test_cli_client", + scopes=["openid", "profile", "email"], + ) + + assert authenticator.scopes == ["openid", "profile", "email"] + assert authenticator.device_flow_client_id == "test_cli_client" diff --git a/bluesky_httpserver/tests/test_system_info_socket.py b/bluesky_httpserver/tests/test_system_info_socket.py index 6b5eb49..9c8f2f6 100644 --- a/bluesky_httpserver/tests/test_system_info_socket.py +++ b/bluesky_httpserver/tests/test_system_info_socket.py @@ -1,5 +1,6 @@ import json import pprint +import socket import threading import time as ttime @@ -20,6 +21,12 @@ ) +def get_free_tcp_port(): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + class _ReceiveSystemInfoSocket(threading.Thread): """ Catch streaming console output by connecting to /console_output/ws socket and @@ -62,15 +69,16 @@ def __del__(self): self.stop() -@pytest.mark.parametrize("zmq_port", (None, 60619)) +@pytest.mark.parametrize("use_custom_port", (False, True)) @pytest.mark.parametrize("endpoint", ["/info/ws", "/status/ws"]) def test_http_server_system_info_socket_1( - monkeypatch, re_manager_cmd, fastapi_server_fs, zmq_port, endpoint # noqa F811 + monkeypatch, re_manager_cmd, fastapi_server_fs, use_custom_port, endpoint # noqa F811 ): """ Test for ``/info/ws`` and ``/status/ws`` websockets """ # Start HTTP Server + zmq_port = get_free_tcp_port() if use_custom_port else None if zmq_port is not None: monkeypatch.setenv("QSERVER_ZMQ_INFO_ADDRESS", f"tcp://localhost:{zmq_port}") fastapi_server_fs()