feat: implement dynamic server URL handling for JWT issuer

fastid/auth/config.py

@@ -29,24 +29,10 @@ class AuthSettings(BaseSettings):
     app_expires_in_seconds: int = 60
 
     @property
-    def issuer(self) -> str:
-        return core_settings.frontend_url
-
-    @property
-    def authorization_endpoint(self) -> str:
-        return f"{core_settings.frontend_url}/authorize"
-
-    @property
-    def token_endpoint(self) -> str:
-        return f"{core_settings.api_url}/token"
-
-    @property
-    def userinfo_endpoint(self) -> str:
-        return f"{core_settings.api_url}/userinfo"
-
-    @property
-    def jwks_uri(self) -> str:
-        return f"{core_settings.frontend_url}/.well-known/jwks.json"
+    def server_url(self) -> str | None:
+        if core_settings.public_url is None:
+            return None
+        return core_settings.public_url.rstrip("/")
 
 
 auth_settings = AuthSettings()

fastid/auth/grants.py

@@ -30,6 +30,7 @@
     TokenResponse,
     UserDTO,
 )
+from fastid.auth.server import ServerURLDep
 from fastid.cache.dependencies import CacheDep
 from fastid.cache.exceptions import KeyNotFoundError
 from fastid.core.base import UseCase
@@ -43,9 +44,10 @@
 
 
 class Grant(UseCase):
-    def __init__(self, uow: UOWDep, cache: CacheDep) -> None:
+    def __init__(self, uow: UOWDep, cache: CacheDep, server_url: ServerURLDep) -> None:
         self.uow = uow
         self.cache = cache
+        self.server_url = server_url
         self.token_backend = jwt_backend
 
     @abstractmethod
@@ -125,14 +127,14 @@ def _get_token_response(self, scope: str, tokens: dict[str, Any]) -> TokenRespon
 
     def _issue_at(self, user: User, scope: str, tokens: dict[str, dict[str, Any]]) -> None:
         schema = JWTPayload(sub=str(user.id), scope=scope)
-        token, payload = self.token_backend.create("access", schema)
+        token, payload = self.token_backend.create("access", schema, issuer=self.server_url)
         tokens["access"]["is_issued"] = True
         tokens["access"]["token"] = token
         tokens["access"]["payload"] = payload
 
     def _issue_rt(self, user: User, scope: str, tokens: dict[str, dict[str, Any]]) -> None:
         schema = JWTPayload(sub=str(user.id), scope=scope)
-        token, payload = self.token_backend.create("refresh", schema)
+        token, payload = self.token_backend.create("refresh", schema, issuer=self.server_url)
         tokens["refresh"]["is_issued"] = True
         tokens["refresh"]["token"] = token
         tokens["refresh"]["payload"] = payload
@@ -146,7 +148,7 @@ def _issue_it(self, user: User, tokens: dict[str, dict[str, Any]]) -> None:
             email=user.email,
             email_verified=user.is_verified,
         )
-        token, payload = self.token_backend.create("id", schema)
+        token, payload = self.token_backend.create("id", schema, issuer=self.server_url)
         tokens["id"]["is_issued"] = True
         tokens["id"]["token"] = token
         tokens["id"]["payload"] = payload
@@ -203,7 +205,7 @@ async def authorize(
     ) -> AuthorizationResponse:
         await self.authenticate_client(form.client_id, form.client_secret)
         try:
-            content = self.token_backend.validate("refresh", form.refresh_token)
+            content = self.token_backend.validate("refresh", form.refresh_token, issuer=self.server_url)
         except JWTError as e:
             raise InvalidTokenError from e
         assert content.scope is not None
@@ -228,7 +230,7 @@ async def _issue_app_tokens(self, app: App, scope: str) -> dict[str, dict[str, A
             for token_type in ["access", "refresh", "id"]
         }
         schema = JWTPayload(sub=str(app.id), scope=scope)
-        token, payload = await self.token_backend.create_async("access", schema)
+        token, payload = await self.token_backend.create_async("access", schema, issuer=self.server_url)
         tokens["access"]["is_issued"] = True
         tokens["access"]["token"] = token
         tokens["access"]["payload"] = payload

fastid/auth/server.py

@@ -0,0 +1,28 @@
+from typing import Annotated
+
+from fastapi import Depends, Request
+
+from fastid.auth.config import auth_settings
+from fastid.core.config import CoreSettings
+from fastid.core.dependencies import get_core_settings
+
+
+def get_server_url(
+    request: Request,
+    settings: CoreSettings = Depends(get_core_settings),
+) -> str:
+    if auth_settings.server_url is not None:
+        return auth_settings.server_url.rstrip("/")
+    if settings.behind_proxy:
+        forwarded_host = request.headers.get("X-Forwarded-Host")
+        forwarded_proto = request.headers.get("X-Forwarded-Proto", "http")
+        if forwarded_host:
+            return f"{forwarded_proto}://{forwarded_host}"
+    base_url = str(request.base_url).rstrip("/")
+    root_path = str(request.scope.get("root_path", "")).rstrip("/")
+    if root_path and base_url.endswith(root_path):
+        return base_url[: -len(root_path)]
+    return base_url
+
+
+ServerURLDep = Annotated[str, Depends(get_server_url)]

fastid/auth/use_cases.py

@@ -2,6 +2,7 @@
 from fastid.auth.models import User
 from fastid.auth.repositories import EmailUserSpecification
 from fastid.auth.schemas import UserCreate
+from fastid.auth.server import ServerURLDep
 from fastid.core.base import UseCase
 from fastid.database.dependencies import UOWDep
 from fastid.database.exceptions import NoResultFoundError
@@ -11,8 +12,9 @@
 
 
 class AuthUseCases(UseCase):
-    def __init__(self, uow: UOWDep) -> None:
+    def __init__(self, uow: UOWDep, server_url: ServerURLDep) -> None:
         self.uow = uow
+        self.server_url = server_url
 
     async def register(self, dto: UserCreate) -> User:
         try:
@@ -28,7 +30,7 @@ async def register(self, dto: UserCreate) -> User:
 
     async def get_userinfo(self, token: str, *, token_type: str = "access") -> User:  # noqa: S107
         try:
-            payload = jwt_backend.validate(token_type, token)
+            payload = jwt_backend.validate(token_type, token, issuer=self.server_url)
         except JWTError as e:
             raise InvalidTokenError from e
         try:

fastid/core/config.py

@@ -1,9 +1,11 @@
 from enum import auto
+from urllib.parse import urlsplit
 
-from pydantic import Field
+from pydantic import Field, field_validator
 from pydantic_settings import SettingsConfigDict
 
 from fastid.core.schemas import ENV_PREFIX, BaseEnum, BaseSettings
+from fastid.core.urls import join_url_path
 
 
 class Environment(BaseEnum):
@@ -14,23 +16,28 @@ class Environment(BaseEnum):
 
 
 class CoreSettings(BaseSettings):
-    env: Environment = Environment.local
+    env: Environment = Environment.prod
     debug: bool = False
+    behind_proxy: bool = True
+    public_url: str | None = None
 
-    base_url: str = "http://localhost:8012"
     api_path: str = "/api/v1"
     admin_path: str = "/admin"
     frontend_path: str = ""
 
-    @property
-    def api_url(self) -> str:
-        return f"{self.base_url}/{self.api_path[1:]}"
-
-    @property
-    def frontend_url(self) -> str:
-        if self.frontend_path == "":
-            return self.base_url
-        return f"{self.base_url}/{self.frontend_path[1:]}"
+    @field_validator("public_url")
+    @classmethod
+    def validate_public_url(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        parsed = urlsplit(value)
+        if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+            msg = "public_url must be an absolute HTTP(S) URL"
+            raise ValueError(msg)
+        if parsed.query or parsed.fragment:
+            msg = "public_url must not contain a query or fragment"
+            raise ValueError(msg)
+        return value.rstrip("/")
 
 
 core_settings = CoreSettings()
@@ -42,8 +49,8 @@ class BrandingSettings(BaseSettings):
     api_swagger_title: str = Field(default_factory=lambda data: f"{data['title']} API")
     frontend_swagger_title: str = Field(default_factory=lambda data: f"{data['title']} Frontend")
     admin_swagger_title: str = Field(default_factory=lambda data: f"{data['title']} Admin")
-    admin_favicon_url: str = f"{core_settings.frontend_url}/static/assets/favicon.png"
-    admin_logo_url: str = f"{core_settings.frontend_url}/static/assets/logo_text.png"
+    admin_favicon_url: str = join_url_path(core_settings.frontend_path, "static/assets/favicon.png")
+    admin_logo_url: str = join_url_path(core_settings.frontend_path, "static/assets/logo_text.png")
     notify_from: str = Field(default_factory=lambda data: data["title"])
 
     model_config = SettingsConfigDict(env_prefix=f"{ENV_PREFIX}branding_")

fastid/core/dependencies.py

@@ -1,6 +1,10 @@
-from fastid.core.config import branding_settings
+from fastid.core.config import CoreSettings, branding_settings, core_settings
 from fastid.core.logging.config import config_dict
 from fastid.core.logging.provider import LogProvider
 
 log_provider = LogProvider(config_dict)
 log = log_provider.logger(branding_settings.service_name)
+
+
+def get_core_settings() -> CoreSettings:
+    return core_settings

fastid/core/urls.py

@@ -0,0 +1,4 @@
+def join_url_path(*parts: str) -> str:
+    """Join application URL path components into one root-relative path."""
+    segments = [segment.strip("/") for part in parts if part for segment in part.split("/") if segment]
+    return f"/{'/'.join(segments)}"

fastid/frontend/dependencies.py

@@ -8,6 +8,7 @@
 from fastid.auth.grants import AuthorizationCodeGrant
 from fastid.auth.models import User
 from fastid.auth.schemas import OAuth2ConsentRequest
+from fastid.auth.server import ServerURLDep
 from fastid.security.exceptions import JWTError
 from fastid.security.jwt import (
     jwt_backend,
@@ -42,12 +43,13 @@ async def get_user(
 
 def is_action_verified(
     request: Request,
+    server_url: ServerURLDep,
 ) -> bool:
     token = vt_transport.get_token(request)
     if token is None:
         return False
     try:
-        jwt_backend.validate("verify", token)
+        jwt_backend.validate("verify", token, issuer=server_url)
     except JWTError:
         return False
     return True

fastid/frontend/openid.py

@@ -5,21 +5,25 @@
 
 from fastid.auth.config import auth_settings
 from fastid.auth.schemas import JWK, JWKS, DiscoveryDocument
+from fastid.core.config import core_settings
+from fastid.core.urls import join_url_path
 from fastid.security.schemas import JWTPayload
 
-discovery_document = DiscoveryDocument(
-    issuer=auth_settings.issuer,
-    authorization_endpoint=auth_settings.authorization_endpoint,
-    token_endpoint=auth_settings.token_endpoint,
-    userinfo_endpoint=auth_settings.userinfo_endpoint,
-    jwks_uri=auth_settings.jwks_uri,
-    scopes_supported=["openid", "email", "name", "offline_access"],
-    response_types_supported=["code"],
-    grant_types_supported=["authorization_code", "refresh_token"],
-    subject_types_supported=["public"],
-    id_token_signing_alg_values_supported=["RS256"],
-    claims_supported=list(JWTPayload.model_fields),
-)
+
+def get_discovery_document(server_url: str) -> DiscoveryDocument:
+    return DiscoveryDocument(
+        issuer=server_url,
+        authorization_endpoint=f"{server_url}{join_url_path(core_settings.frontend_path, 'authorize')}",
+        token_endpoint=f"{server_url}{join_url_path(core_settings.api_path, 'token')}",
+        userinfo_endpoint=f"{server_url}{join_url_path(core_settings.api_path, 'userinfo')}",
+        jwks_uri=f"{server_url}{join_url_path(core_settings.frontend_path, '.well-known/jwks.json')}",
+        scopes_supported=["openid", "email", "name", "offline_access"],
+        response_types_supported=["code"],
+        grant_types_supported=["authorization_code", "refresh_token"],
+        subject_types_supported=["public"],
+        id_token_signing_alg_values_supported=["RS256"],
+        claims_supported=list(JWTPayload.model_fields),
+    )
 
 
 def get_jwks() -> JWKS:

fastid/frontend/router.py

@@ -7,13 +7,14 @@
 from fastid.auth.grants import AuthorizationCodeGrant
 from fastid.auth.models import User
 from fastid.auth.schemas import JWKS, DiscoveryDocument, OAuth2ConsentRequest
+from fastid.auth.server import ServerURLDep
 from fastid.frontend.dependencies import (
     get_user,
     get_user_or_none,
     is_action_verified,
     valid_consent,
 )
-from fastid.frontend.openid import discovery_document, jwks
+from fastid.frontend.openid import get_discovery_document, jwks
 from fastid.frontend.templating import templates
 from fastid.notify.schemas import UserAction
 from fastid.oauth.dependencies import OAuthAccountsDep
@@ -155,10 +156,10 @@ def logout() -> Any:
 
 
 @router.get("/.well-known/openid-configuration")
-def openid_configuration() -> DiscoveryDocument:
-    return discovery_document
+def openid_configuration(server_url: ServerURLDep) -> DiscoveryDocument:
+    return get_discovery_document(server_url)
 
 
-@router.get("/.well-known/jwks.json")
+@router.get("/.well-known/jwks.json", name="openid_jwks")
 def get_jwks() -> JWKS:
     return jwks

fastid/integrations/config.py

@@ -1,4 +1,3 @@
-from fastid.core.config import core_settings
 from fastid.core.schemas import BaseSettings
 
 
@@ -19,17 +18,5 @@ class IntegrationSettings(BaseSettings):
     telegram_notification_enabled: bool = False
     telegram_bot_token: str = ""
 
-    @property
-    def base_authorization_url(self) -> str:
-        return f"{core_settings.api_url}/oauth/login"
-
-    @property
-    def base_redirect_url(self) -> str:
-        return f"{core_settings.api_url}/oauth/callback"
-
-    @property
-    def base_revoke_url(self) -> str:
-        return f"{core_settings.api_url}/oauth/revoke"
-
 
 integration_settings = IntegrationSettings()