CLI main and RuntimeConfig cleanup. Added a session_ctx wrapper function for easier access and avoiding hostutil import in db.

This commit is contained in:
Leo Vasanko
2026-02-19 16:28:58 +00:00
parent 4caef13f60
commit f746085ead
14 changed files with 166 additions and 178 deletions
+43 -84
View File
@@ -1,16 +1,16 @@
import argparse import argparse
import json
import logging import logging
import os import os
from urllib.parse import urlparse from urllib.parse import urlparse
import msgspec
from fastapi_vue import server from fastapi_vue import server
from fastapi_vue.hostutil import parse_endpoints from fastapi_vue.hostutil import parse_endpoints
from paskia.config import PaskiaConfig
from paskia.db.jsonl import load_readonly from paskia.db.jsonl import load_readonly
from paskia.util import startupbox from paskia.util import startupbox
from paskia.util.hostutil import normalize_origin from paskia.util.hostutil import normalize_origin
from paskia.util.runtime import RuntimeConfig
DEFAULT_PORT = 4401 DEFAULT_PORT = 4401
DEVMODE = os.getenv("PASKIA_DEV") == "1" DEVMODE = os.getenv("PASKIA_DEV") == "1"
@@ -51,7 +51,6 @@ def add_common_options(p: argparse.ArgumentParser) -> None:
"--origin", "--origin",
action="append", action="append",
dest="origins", dest="origins",
default=[],
metavar="URL", metavar="URL",
help="Allowed origin URL(s). May be specified multiple times. If any are specified, only those origins are permitted for WebSocket authentication.", help="Allowed origin URL(s). May be specified multiple times. If any are specified, only those origins are permitted for WebSocket authentication.",
) )
@@ -91,106 +90,66 @@ def main():
args = parser.parse_args() args = parser.parse_args()
# Handle clearing options # Load stored config (read-only, no writes, no global state)
if getattr(args, "auth_host", None) == "":
args.auth_host = None
if getattr(args, "rp_name", None) == "":
args.rp_name = None
if getattr(args, "listen", None) == "":
args.listen = None
# Read-only load to get stored config (no writes, no global state)
db_path = os.environ.get("PASKIA_DB", f"{args.rp_id}.paskiadb") db_path = os.environ.get("PASKIA_DB", f"{args.rp_id}.paskiadb")
stored_db = load_readonly(db_path, rp_id=args.rp_id) config = load_readonly(db_path, rp_id=args.rp_id).config
stored_config = stored_db.config
# Apply defaults from stored config # Override stored config with CLI args, or clear with empty string
if args.rp_name is None and stored_config.rp_name is not None: if args.rp_name is not None:
args.rp_name = stored_config.rp_name config.rp_name = args.rp_name or None
if args.origins is None and stored_config.origins is not None: if args.auth_host is not None:
args.origins = stored_config.origins config.auth_host = args.auth_host or None
if args.auth_host is None and stored_config.auth_host is not None: if args.origins is not None:
args.auth_host = stored_config.auth_host config.origins = None if args.origins == [""] else args.origins
if args.listen is None and stored_config.listen is not None: if args.listen is not None:
args.listen = stored_config.listen config.listen = None if args.listen == [""] else args.listen
# Parse first endpoint for config display and site_url
ep = next(iter(parse_endpoints(args.listen, DEFAULT_PORT)), {})
host, port, uds = ep.get("host"), ep.get("port"), ep.get("uds")
# Process and normalize auth_host # Process and normalize auth_host
if args.auth_host: if config.auth_host:
if "://" not in args.auth_host: if "://" not in config.auth_host:
args.auth_host = f"https://{args.auth_host}" config.auth_host = f"https://{config.auth_host}"
args.auth_host = args.auth_host.rstrip("/") config.auth_host = config.auth_host.rstrip("/")
validate_auth_host(args.auth_host, args.rp_id) validate_auth_host(config.auth_host, config.rp_id)
if args.origins: if config.origins:
args.origins.insert(0, args.auth_host) # Ensure first in origins config.origins.insert(0, config.auth_host) # Ensure first in origins
# Normalize, strip trailing slashes, and deduplicate while preserving order # Normalize and deduplicate while preserving order
origins = list({normalize_origin(o).rstrip("/"): ... for o in (args.origins)}) if config.origins:
config.origins = list({normalize_origin(o): ... for o in config.origins})
# Compute site_url and site_path for reset links # Parse first endpoint for site_url fallback
# Priority: auth_host > first configured origin > PASKIA_VITE_URL (devserver) > http://localhost:port > https://rp_id ep = next(iter(parse_endpoints(config.listen, DEFAULT_PORT)), {})
port = ep.get("port")
# Compute site_url and site_path
# Priority: auth_host > origins[0] > PASKIA_VITE_URL > http://localhost:port > https://rp_id
site_path = "/auth/" site_path = "/auth/"
if args.auth_host: if config.auth_host:
site_url = args.auth_host site_url, site_path = config.auth_host, "/"
site_path = "/" elif config.origins:
elif origins: site_url = config.origins[0]
# Find localhost origin if rp_id is localhost, else use first origin
localhost_origin = (
next((o for o in origins if "://localhost" in o), None)
if args.rp_id == "localhost"
else None
)
site_url = localhost_origin or origins[0]
elif vite_url := os.environ.get("PASKIA_VITE_URL"): elif vite_url := os.environ.get("PASKIA_VITE_URL"):
site_url = vite_url.rstrip("/") # Devserver site_url = vite_url.rstrip("/") # Devserver
elif args.rp_id == "localhost" and port: elif config.rp_id == "localhost" and port:
site_url = f"http://localhost:{port}" # Backend directly if we can site_url = f"http://localhost:{port}" # Backend directly if we can
else: else:
site_url = f"https://{args.rp_id}" # Assume external reverse proxy site_url = f"https://{config.rp_id}" # Assume external reverse proxy
# Build runtime configuration # Build runtime configuration for the server
config = PaskiaConfig( runtime = RuntimeConfig(
rp_id=args.rp_id, config=config,
rp_name=args.rp_name or None,
origins=origins or None,
auth_host=args.auth_host or None,
site_url=site_url, site_url=site_url,
site_path=site_path, site_path=site_path,
host=host, save=args.save,
port=port,
uds=uds,
) )
startupbox.print_startup_config(runtime)
os.environ["PASKIA_CONFIG"] = msgspec.json.encode(runtime).decode()
# Export configuration via single JSON env variable for worker processes # Run the server (spawns processes in dev mode)
# Include cli_config and save flag so lifespan can handle bootstrap/persistence
cli_config = {
"rp_id": args.rp_id,
"rp_name": args.rp_name,
"origins": args.origins,
"auth_host": args.auth_host,
"listen": args.listen,
}
config_json = {
"rp_id": config.rp_id,
"rp_name": config.rp_name,
"origins": config.origins,
"auth_host": config.auth_host,
"site_url": config.site_url,
"site_path": config.site_path,
"save": args.save,
"cli_config": cli_config,
}
os.environ["PASKIA_CONFIG"] = json.dumps(config_json)
startupbox.print_startup_config(config)
dev = {"reload": True, "reload_dirs": ["paskia"]} if DEVMODE else {} dev = {"reload": True, "reload_dirs": ["paskia"]} if DEVMODE else {}
server.run( server.run(
"paskia.fastapi.mainapp:app", "paskia.fastapi.mainapp:app",
listen=args.listen, listen=config.listen,
default_port=DEFAULT_PORT, default_port=DEFAULT_PORT,
log_level="warning", log_level="warning",
access_log=False, access_log=False,
+6 -1
View File
@@ -23,6 +23,11 @@ if TYPE_CHECKING:
EXPIRES = SESSION_LIFETIME EXPIRES = SESSION_LIFETIME
def session_ctx(auth: str, host: str | None = None):
"""Get session context with normalized host."""
return db.data().session_ctx(auth, hostutil.normalize_host(host))
def expires() -> datetime: def expires() -> datetime:
return datetime.now(UTC) + EXPIRES return datetime.now(UTC) + EXPIRES
@@ -42,7 +47,7 @@ def get_reset(token: str) -> "ResetToken":
def delete_credential(credential_uuid: UUID, auth: str, host: str | None = None): def delete_credential(credential_uuid: UUID, auth: str, host: str | None = None):
"""Delete a specific credential for the current user.""" """Delete a specific credential for the current user."""
ctx = db.data().session_ctx(auth, hostutil.normalize_host(host)) ctx = session_ctx(auth, host)
if not ctx: if not ctx:
raise ValueError("Session expired") raise ValueError("Session expired")
db.delete_credential(credential_uuid, ctx.user.uuid) db.delete_credential(credential_uuid, ctx.user.uuid)
-17
View File
@@ -1,4 +1,3 @@
from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
# Shared configuration constants for session management. # Shared configuration constants for session management.
@@ -6,19 +5,3 @@ SESSION_LIFETIME = timedelta(hours=24)
# Lifetime for reset links created by admins # Lifetime for reset links created by admins
RESET_LIFETIME = timedelta(days=14) RESET_LIFETIME = timedelta(days=14)
@dataclass
class PaskiaConfig:
"""Runtime configuration for the Paskia authentication server."""
rp_id: str
rp_name: str | None
origins: list[str] | None
auth_host: str | None
site_url: str # Base URL without trailing path (e.g. https://example.com)
site_path: str # Path to auth UI: "/" if auth_host, else "/auth/"
# Listen address (one of host:port or uds)
host: str | None = None
port: int | None = None
uds: str | None = None
+4 -7
View File
@@ -9,7 +9,6 @@ import msgspec
import uuid7 import uuid7
from paskia import db from paskia import db
from paskia.util import hostutil
from paskia.util import passphrase as passphrase_util from paskia.util import passphrase as passphrase_util
from paskia.util.crypto import hash_secret from paskia.util.crypto import hash_secret
@@ -601,14 +600,14 @@ class OIDC(msgspec.Struct, dict=True):
key: bytes | None = None key: bytes | None = None
class Config(msgspec.Struct, frozen=True, dict=True, omit_defaults=True): class Config(msgspec.Struct, omit_defaults=True):
"""Stored configuration for the instance.""" """Stored configuration for the instance."""
rp_id: str rp_id: str
rp_name: str | None = None rp_name: str | None = None
origins: list[str] | None = None
auth_host: str | None = None auth_host: str | None = None
listen: str | None = None origins: list[str] | None = None
listen: list[str] | None = None
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
@@ -679,10 +678,8 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
if s.client_uuid is not None: if s.client_uuid is not None:
return None return None
# Normalize host for comparison (stored hosts are already normalized)
normalized_input = hostutil.normalize_host(host)
# Validate host matches (sessions are always created with a host) # Validate host matches (sessions are always created with a host)
normalized_input = host
if s.host != normalized_input: if s.host != normalized_input:
# Session bound to different host # Session bound to different host
return None return None
+4 -4
View File
@@ -15,7 +15,7 @@ from fastapi.security import HTTPBearer
from paskia import authcode, db from paskia import authcode, db
from paskia._version import __version__ from paskia._version import __version__
from paskia.authsession import EXPIRES, get_reset from paskia.authsession import EXPIRES, get_reset, session_ctx
from paskia.fastapi import authz, session, user from paskia.fastapi import authz, session, user
from paskia.fastapi.response import MsgspecResponse from paskia.fastapi.response import MsgspecResponse
from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME, get_client_ip from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME, get_client_ip
@@ -202,7 +202,7 @@ async def api_user_info(
detail="Authentication required", detail="Authentication required",
mode="login", mode="login",
) )
ctx = db.data().session_ctx(auth, request.headers.get("host")) ctx = session_ctx(auth, request.headers.get("host"))
if not ctx: if not ctx:
raise authz.AuthException( raise authz.AuthException(
status_code=401, status_code=401,
@@ -249,7 +249,7 @@ async def api_logout(request: Request, response: Response, auth=AUTH_COOKIE):
if not auth: if not auth:
return {"message": "Already logged out"} return {"message": "Already logged out"}
host = request.headers.get("host") host = request.headers.get("host")
ctx = db.data().session_ctx(auth, host) ctx = session_ctx(auth, host)
if not ctx: if not ctx:
return {"message": "Already logged out"} return {"message": "Already logged out"}
with suppress(Exception): with suppress(Exception):
@@ -282,7 +282,7 @@ async def api_set_session(
secret = a.session_key secret = a.session_key
# Verify the session exists # Verify the session exists
ctx = db.data().session_ctx(secret, host) ctx = session_ctx(secret, host)
if not ctx: if not ctx:
raise HTTPException(401, f"Session not found on {host}") raise HTTPException(401, f"Session not found on {host}")
+10 -13
View File
@@ -1,9 +1,9 @@
import json
import logging import logging
import os import os
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from pathlib import Path from pathlib import Path
import msgspec
from fastapi import FastAPI, HTTPException, Request, Response from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.responses import FileResponse, RedirectResponse from fastapi.responses import FileResponse, RedirectResponse
@@ -13,7 +13,6 @@ from paskia.bootstrap import bootstrap_if_needed
from paskia.db import start_background, stop_background from paskia.db import start_background, stop_background
from paskia.db.background import flush from paskia.db.background import flush
from paskia.db.logging import configure_db_logging from paskia.db.logging import configure_db_logging
from paskia.db.structs import Config
from paskia.fastapi import admin, api, auth_host, oid, ws from paskia.fastapi import admin, api, auth_host, oid, ws
# Import frontend instance # Import frontend instance
@@ -21,6 +20,7 @@ from paskia.fastapi.front import frontend
from paskia.fastapi.logging import AccessLogMiddleware, configure_access_logging from paskia.fastapi.logging import AccessLogMiddleware, configure_access_logging
from paskia.fastapi.session import AUTH_COOKIE from paskia.fastapi.session import AUTH_COOKIE
from paskia.util import hostutil, passphrase, vitedev from paskia.util import hostutil, passphrase, vitedev
from paskia.util.runtime import RuntimeConfig
# Configure custom logging # Configure custom logging
configure_access_logging() configure_access_logging()
@@ -40,13 +40,13 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path
so that uvicorn reload / multiprocess workers inherit the settings. so that uvicorn reload / multiprocess workers inherit the settings.
All keys are guaranteed to exist; values are already normalized by __main__.py. All keys are guaranteed to exist; values are already normalized by __main__.py.
""" """
config = json.loads(os.environ["PASKIA_CONFIG"]) runtime = msgspec.json.decode(os.environ["PASKIA_CONFIG"], type=RuntimeConfig)
try: try:
await globals.init( await globals.init(
rp_id=config["rp_id"], rp_id=runtime.config.rp_id,
rp_name=config["rp_name"], rp_name=runtime.config.rp_name,
origins=config["origins"], origins=runtime.config.origins,
bootstrap=False, bootstrap=False,
) )
except ValueError as e: except ValueError as e:
@@ -55,13 +55,10 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path
raise raise
# Bootstrap and persist config now that the full DB is loaded # Bootstrap and persist config now that the full DB is loaded
cli_config_data = config.get("cli_config") await bootstrap_if_needed(config=runtime.config)
if cli_config_data: if runtime.save:
cli_config = Config(**cli_config_data) await db.update_config(runtime.config)
await bootstrap_if_needed(config=cli_config) await flush()
if config.get("save"):
await db.update_config(cli_config)
await flush()
# Restore uvicorn info logging (suppressed during startup in dev mode) # Restore uvicorn info logging (suppressed during startup in dev mode)
# Keep uvicorn.error at WARNING to suppress WebSocket "connection open/closed" messages # Keep uvicorn.error at WARNING to suppress WebSocket "connection open/closed" messages
+6 -5
View File
@@ -13,6 +13,7 @@ from paskia import db
from paskia.authsession import ( from paskia.authsession import (
delete_credential, delete_credential,
expires, expires,
session_ctx,
) )
from paskia.fastapi import authz, session from paskia.fastapi import authz, session
from paskia.fastapi.response import MsgspecResponse from paskia.fastapi.response import MsgspecResponse
@@ -45,7 +46,7 @@ async def user_update_display_name(
status_code=401, detail="Authentication Required", mode="login" status_code=401, detail="Authentication Required", mode="login"
) )
host = request.headers.get("host") host = request.headers.get("host")
ctx = db.data().session_ctx(auth, host) ctx = session_ctx(auth, host)
if not ctx: if not ctx:
raise authz.AuthException( raise authz.AuthException(
status_code=401, detail="Session expired", mode="login" status_code=401, detail="Session expired", mode="login"
@@ -74,7 +75,7 @@ async def user_update_info(
raise authz.AuthException( raise authz.AuthException(
status_code=401, detail="Authentication Required", mode="login" status_code=401, detail="Authentication Required", mode="login"
) )
ctx = db.data().session_ctx(auth, request.headers.get("host")) ctx = session_ctx(auth, request.headers.get("host"))
if not ctx: if not ctx:
raise authz.AuthException( raise authz.AuthException(
status_code=401, detail="Session expired", mode="login" status_code=401, detail="Session expired", mode="login"
@@ -112,7 +113,7 @@ async def user_update_theme(
raise authz.AuthException( raise authz.AuthException(
status_code=401, detail="Authentication Required", mode="login" status_code=401, detail="Authentication Required", mode="login"
) )
ctx = db.data().session_ctx(auth, request.headers.get("host")) ctx = session_ctx(auth, request.headers.get("host"))
if not ctx: if not ctx:
raise authz.AuthException( raise authz.AuthException(
status_code=401, detail="Session expired", mode="login" status_code=401, detail="Session expired", mode="login"
@@ -129,7 +130,7 @@ async def api_logout_all(request: Request, response: Response, auth=AUTH_COOKIE)
if not auth: if not auth:
return {"message": "Already logged out"} return {"message": "Already logged out"}
host = request.headers.get("host") host = request.headers.get("host")
ctx = db.data().session_ctx(auth, host) ctx = session_ctx(auth, host)
if not ctx: if not ctx:
raise authz.AuthException( raise authz.AuthException(
status_code=401, detail="Session expired", mode="login" status_code=401, detail="Session expired", mode="login"
@@ -151,7 +152,7 @@ async def api_delete_session(
status_code=401, detail="Authentication Required", mode="login" status_code=401, detail="Authentication Required", mode="login"
) )
host = request.headers.get("host") host = request.headers.get("host")
ctx = db.data().session_ctx(auth, host) ctx = session_ctx(auth, host)
if not ctx: if not ctx:
raise authz.AuthException( raise authz.AuthException(
status_code=401, detail="Session expired", mode="login" status_code=401, detail="Session expired", mode="login"
+2 -2
View File
@@ -7,7 +7,7 @@ from fastapi import FastAPI, WebSocket
from paskia import authcode, db from paskia import authcode, db
from paskia.authcode import CookieCode, OIDCCode from paskia.authcode import CookieCode, OIDCCode
from paskia.authsession import get_reset from paskia.authsession import get_reset, session_ctx
from paskia.db.structs import Session from paskia.db.structs import Session
from paskia.fastapi import authz, remote from paskia.fastapi import authz, remote
from paskia.fastapi.session import AUTH_COOKIE, infodict from paskia.fastapi.session import AUTH_COOKIE, infodict
@@ -195,7 +195,7 @@ async def websocket_authenticate(
# If there's an existing session, restrict to that user's credentials (reauth) # If there's an existing session, restrict to that user's credentials (reauth)
session_user_uuid = None session_user_uuid = None
if auth: if auth:
existing_ctx = db.data().session_ctx(auth, host) existing_ctx = session_ctx(auth, host)
if existing_ctx: if existing_ctx:
session_user_uuid = existing_ctx.user.uuid session_user_uuid = existing_ctx.user.uuid
+3 -2
View File
@@ -7,6 +7,7 @@ from uuid import UUID
from fastapi import WebSocket from fastapi import WebSocket
from paskia import db from paskia import db
from paskia.authsession import session_ctx
from paskia.db import Credential, SessionContext from paskia.db import Credential, SessionContext
from paskia.fastapi.session import infodict from paskia.fastapi.session import infodict
from paskia.fastapi.wsutil import validate_origin from paskia.fastapi.wsutil import validate_origin
@@ -90,7 +91,7 @@ async def authenticate_and_login(
# Get credential IDs if restricting to a user's credentials # Get credential IDs if restricting to a user's credentials
credential_ids = None credential_ids = None
if auth: if auth:
existing_ctx = db.data().session_ctx(auth, host) existing_ctx = session_ctx(auth, host)
if existing_ctx: if existing_ctx:
credential_ids = existing_ctx.user.credential_ids or None credential_ids = existing_ctx.user.credential_ids or None
@@ -107,7 +108,7 @@ async def authenticate_and_login(
) )
# Fetch and return the full session context # Fetch and return the full session context
ctx = db.data().session_ctx(secret, normalized_host) ctx = session_ctx(secret, host)
if not ctx: if not ctx:
raise ValueError("Failed to create session context") raise ValueError("Failed to create session context")
return ctx, secret return ctx, secret
+1 -1
View File
@@ -13,7 +13,7 @@ import httpx
from paskia import db from paskia import db
from paskia.util import oidjwt from paskia.util import oidjwt
from paskia.util.hostutil import _load_config from paskia.util.runtime import _load_config
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
+26 -16
View File
@@ -1,27 +1,23 @@
"""Utilities for determining the auth UI host and base URLs.""" """Utilities for determining the auth UI host and base URLs."""
import json
import os
from functools import lru_cache
from urllib.parse import urlparse, urlsplit from urllib.parse import urlparse, urlsplit
from paskia.util.runtime import _load_config
@lru_cache(maxsize=1)
def _load_config() -> dict: def _cfg():
"""Load PASKIA_CONFIG JSON.""" return _load_config()
config_json = os.getenv("PASKIA_CONFIG")
if not config_json:
return {}
return json.loads(config_json)
def is_root_mode() -> bool: def is_root_mode() -> bool:
return _load_config().get("auth_host") is not None cfg = _cfg()
return cfg is not None and cfg.config.auth_host is not None
def dedicated_auth_host() -> str | None: def dedicated_auth_host() -> str | None:
"""Return configured auth_host netloc, or None.""" """Return configured auth_host netloc, or None."""
auth_host = _load_config().get("auth_host") cfg = _cfg()
auth_host = cfg.config.auth_host if cfg else None
if not auth_host: if not auth_host:
return None return None
@@ -35,8 +31,10 @@ def ui_base_path() -> str:
def auth_site_url() -> str: def auth_site_url() -> str:
"""Return the base URL for the auth site UI (computed at startup).""" """Return the base URL for the auth site UI (computed at startup)."""
cfg = _load_config() cfg = _cfg()
return cfg.get("site_url", "https://localhost") + cfg.get("site_path", "/auth/") if cfg:
return cfg.site_url + cfg.site_path
return "https://localhost/auth/"
def reset_link_url(token: str) -> str: def reset_link_url(token: str) -> str:
@@ -45,10 +43,10 @@ def reset_link_url(token: str) -> str:
def normalize_origin(origin: str) -> str: def normalize_origin(origin: str) -> str:
"""Normalize an origin URL by adding https:// if no scheme is present.""" """Normalize an origin URL by adding https:// if no scheme is present, removing trailing slashes."""
if "://" not in origin: if "://" not in origin:
return f"https://{origin}" return f"https://{origin}"
return origin return origin.rstrip("/")
def reload_config() -> None: def reload_config() -> None:
@@ -74,3 +72,15 @@ def normalize_host(raw_host: str | None) -> str | None:
# Strip port from host:port # Strip port from host:port
netloc = netloc.rsplit(":", 1)[0] netloc = netloc.rsplit(":", 1)[0]
return netloc.lower() or None return netloc.lower() or None
def format_endpoint(ep: dict) -> str:
"""Format an endpoint dict to a listen string (e.g. 'unix:/path' or 'host:port')."""
if uds := ep.get("uds"):
return f"unix:{uds}"
host = ep["host"]
port = ep["port"]
# Bracket IPv6 addresses
if ":" in host:
host = f"[{host}]"
return f"{host}:{port}"
+2 -2
View File
@@ -3,7 +3,7 @@
from collections.abc import Sequence from collections.abc import Sequence
from fnmatch import fnmatchcase from fnmatch import fnmatchcase
from paskia import db from paskia.authsession import session_ctx
from paskia.util.hostutil import normalize_host from paskia.util.hostutil import normalize_host
__all__ = ["has_any", "has_all", "session_context"] __all__ = ["has_any", "has_all", "session_context"]
@@ -40,4 +40,4 @@ async def session_context(auth: str | None, host: str | None = None):
if not auth: if not auth:
return None return None
normalized_host = normalize_host(host) if host else None normalized_host = normalize_host(host) if host else None
return db.data().session_ctx(auth, normalized_host) return session_ctx(auth, normalized_host)
+31
View File
@@ -0,0 +1,31 @@
"""Runtime configuration utilities."""
import os
from functools import lru_cache
import msgspec
from paskia.db.structs import Config
class RuntimeConfig(msgspec.Struct):
"""Runtime configuration for the Paskia authentication server.
Wraps the db Config (CLI/stored settings) with computed runtime fields.
Serialized to PASKIA_CONFIG env var as JSON via msgspec.
"""
config: Config # CLI/stored configuration to persist
site_url: str # Base URL without trailing path (e.g. https://example.com)
site_path: str # Path to auth UI: "/" if auth_host, else "/auth/"
save: bool = False # Whether to persist config to database
@lru_cache(maxsize=1)
def _load_config() -> "RuntimeConfig | None":
"""Load RuntimeConfig from PASKIA_CONFIG env var."""
config_json = os.getenv("PASKIA_CONFIG")
if not config_json:
return None
return msgspec.json.decode(config_json.encode(), type=RuntimeConfig)
+28 -24
View File
@@ -1,14 +1,19 @@
"""Startup configuration box formatting utilities.""" """Startup configuration box formatting utilities."""
from __future__ import annotations
import os import os
import re import re
from sys import stderr from sys import stderr
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from fastapi_vue.hostutil import parse_endpoints
from paskia._version import __version__ from paskia._version import __version__
from paskia.util.hostutil import format_endpoint
if TYPE_CHECKING: if TYPE_CHECKING:
from paskia.config import PaskiaConfig from paskia.util.runtime import RuntimeConfig
BOX_WIDTH = 60 # Inner width (excluding box chars) BOX_WIDTH = 60 # Inner width (excluding box chars)
@@ -42,7 +47,7 @@ def bottom() -> str:
return "" + "" * (BOX_WIDTH + 2) + "\n" return "" + "" * (BOX_WIDTH + 2) + "\n"
def print_startup_config(config: "PaskiaConfig") -> None: def print_startup_config(runtime: RuntimeConfig) -> None:
"""Print server configuration on startup.""" """Print server configuration on startup."""
# Key graphic with yellow shading (bright for highlights, dark for body) # Key graphic with yellow shading (bright for highlights, dark for body)
y = YELLOW # Dark yellow for main body y = YELLOW # Dark yellow for main body
@@ -57,41 +62,40 @@ def print_startup_config(config: "PaskiaConfig") -> None:
lines.append( lines.append(
line( line(
f"{b}{y} {b}{y}▀▀▀▀{b}{y}▀▀{b}{y}▀▀{b}{r} {w}" f"{b}{y} {b}{y}▀▀▀▀{b}{y}▀▀{b}{y}▀▀{b}{r} {w}"
+ config.site_url + runtime.site_url
+ config.site_path + runtime.site_path
+ r + r
) )
) )
lines.append(line(f" {y}▀▀▀▀▀{r}")) lines.append(line(f" {y}▀▀▀▀▀{r}"))
# Format auth host section # Format auth host section
if config.auth_host: if runtime.config.auth_host:
lines.append(line(f"Auth Host: {config.auth_host}")) lines.append(line(f"Auth Host: {runtime.config.auth_host}"))
from paskia.__main__ import DEFAULT_PORT as P # noqa: PLC0415 - circular
from paskia.__main__ import DEVMODE # noqa: PLC0415 - circular
# Show frontend URL if in dev mode # Show frontend URL if in dev mode
devmode = os.environ.get("PASKIA_VITE_URL") if DEVMODE:
if devmode: lines.append(line(f"Dev Frontend: {os.environ.get('PASKIA_VITE_URL')}"))
lines.append(line(f"Dev Frontend: {devmode}"))
# Format listen address with scheme # Format listen endpoints (dev mode only uses the first endpoint)
if config.uds:
listen = f"unix:{config.uds}" endpoints = list(parse_endpoints(runtime.config.listen, P))
elif config.host: if DEVMODE:
listen = f"http://{config.host}:{config.port}" endpoints = endpoints[:1] # server.run reload=True uses only one
else: parts = [format_endpoint(ep) for ep in endpoints]
listen = f"http://0.0.0.0:{config.port} + [::]:{config.port}" lines.append(line(f"Backend: {' '.join(parts)}"))
lines.append(line(f"Backend: {listen}"))
# Relying Party line (omit name if same as id) # Relying Party line (omit name if same as id)
rp_id = config.rp_id rp_id = runtime.config.rp_id
rp_name = config.rp_name rp_name = runtime.config.rp_name
if rp_name and rp_name != rp_id: suffix = f" ({rp_name})" if rp_name and rp_name != rp_id else ""
lines.append(line(f"Relying Party: {rp_id} ({rp_name})")) lines.append(line(f"Relying Party: {rp_id}{suffix}"))
else:
lines.append(line(f"Relying Party: {rp_id}"))
# Format origins section # Format origins section
allowed = config.origins allowed = runtime.config.origins
if allowed: if allowed:
lines.append(line("Permitted Origins:")) lines.append(line("Permitted Origins:"))
for origin in sorted(allowed): for origin in sorted(allowed):