Admin Server Options panel added for configuring rp-name, auth-host and origins.
This commit is contained in:
+14
-34
@@ -1,7 +1,6 @@
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import msgspec
|
||||
from fastapi_vue import server
|
||||
@@ -9,7 +8,11 @@ from fastapi_vue.hostutil import parse_endpoints
|
||||
|
||||
from paskia.db.jsonl import load_readonly
|
||||
from paskia.util import startupbox
|
||||
from paskia.util.hostutil import normalize_origin
|
||||
from paskia.util.hostutil import (
|
||||
normalize_auth_host_and_origins,
|
||||
normalize_origin,
|
||||
validate_auth_host,
|
||||
)
|
||||
from paskia.util.runtime import RuntimeConfig
|
||||
|
||||
DEFAULT_PORT = 4401
|
||||
@@ -21,27 +24,6 @@ Example:
|
||||
"""
|
||||
|
||||
|
||||
def is_subdomain(sub: str, domain: str) -> bool:
|
||||
"""Check if sub is a subdomain of domain (or equal)."""
|
||||
sub_parts = sub.lower().split(".")
|
||||
domain_parts = domain.lower().split(".")
|
||||
if len(sub_parts) < len(domain_parts):
|
||||
return False
|
||||
return sub_parts[-len(domain_parts) :] == domain_parts
|
||||
|
||||
|
||||
def validate_auth_host(auth_host: str, rp_id: str) -> None:
|
||||
"""Validate that auth_host is a subdomain of rp_id."""
|
||||
parsed = urlparse(auth_host if "://" in auth_host else f"//{auth_host}")
|
||||
host = parsed.hostname or parsed.path
|
||||
if not host:
|
||||
raise SystemExit(f"Invalid auth-host: '{auth_host}'")
|
||||
if not is_subdomain(host, rp_id):
|
||||
raise SystemExit(
|
||||
f"auth-host '{auth_host}' is not a subdomain of rp-id '{rp_id}'"
|
||||
)
|
||||
|
||||
|
||||
def add_common_options(p: argparse.ArgumentParser) -> None:
|
||||
p.add_argument(
|
||||
"--rp-id", default="localhost", help="Relying Party ID (default: localhost)"
|
||||
@@ -104,18 +86,16 @@ def main():
|
||||
if args.listen is not None:
|
||||
config.listen = None if args.listen == [""] else args.listen
|
||||
|
||||
# Process and normalize auth_host
|
||||
if config.auth_host:
|
||||
if "://" not in config.auth_host:
|
||||
config.auth_host = f"https://{config.auth_host}"
|
||||
config.auth_host = config.auth_host.rstrip("/")
|
||||
validate_auth_host(config.auth_host, config.rp_id)
|
||||
if config.origins:
|
||||
config.origins.insert(0, config.auth_host) # Ensure first in origins
|
||||
|
||||
# Normalize and deduplicate while preserving order
|
||||
# Process and normalize auth_host and origins
|
||||
try:
|
||||
validate_auth_host(config.auth_host, config.rp_id) if config.auth_host else None
|
||||
except ValueError as e:
|
||||
raise SystemExit(str(e))
|
||||
if config.origins:
|
||||
config.origins = list({normalize_origin(o): ... for o in config.origins})
|
||||
config.origins = [normalize_origin(o) for o in config.origins]
|
||||
config.auth_host, config.origins = normalize_auth_host_and_origins(
|
||||
config.auth_host, config.origins
|
||||
)
|
||||
|
||||
# Parse first endpoint for site_url fallback
|
||||
ep = next(iter(parse_endpoints(config.listen, DEFAULT_PORT)), {})
|
||||
|
||||
+78
-1
@@ -12,12 +12,13 @@ from paskia.db import Permission as PermDC
|
||||
from paskia.db import Role as RoleDC
|
||||
from paskia.db import User as UserDC
|
||||
from paskia.db.operations import _UNSET
|
||||
from paskia.db.structs import Client
|
||||
from paskia.db.structs import Client, Config
|
||||
from paskia.fastapi import authz
|
||||
from paskia.fastapi.front import frontend
|
||||
from paskia.fastapi.response import MsgspecResponse
|
||||
from paskia.fastapi.session import AUTH_COOKIE
|
||||
from paskia.globals import passkey
|
||||
from paskia.sansio import Passkey
|
||||
from paskia.util import (
|
||||
hostutil,
|
||||
permutil,
|
||||
@@ -1176,3 +1177,79 @@ async def admin_delete_oidc_client(
|
||||
raise HTTPException(status_code=404, detail=str(e))
|
||||
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
# -------------------- Server Configuration --------------------
|
||||
|
||||
|
||||
@app.get("/server-config")
|
||||
async def admin_get_server_config(
|
||||
request: Request,
|
||||
auth=AUTH_COOKIE,
|
||||
):
|
||||
"""Get current server configuration (master admin only)."""
|
||||
await authz.verify(auth, ["auth:admin"], host=request.headers.get("host"))
|
||||
pk = passkey.instance
|
||||
config = db.data().config
|
||||
return {
|
||||
"rp_name": pk.rp_name,
|
||||
"auth_host": config.auth_host or "",
|
||||
"origins": list(pk.allowed_origins) if pk.allowed_origins else [],
|
||||
}
|
||||
|
||||
|
||||
@app.patch("/server-config")
|
||||
async def admin_update_server_config(
|
||||
request: Request,
|
||||
payload: dict = Body(...),
|
||||
auth=AUTH_COOKIE,
|
||||
):
|
||||
"""Update server configuration (master admin only).
|
||||
|
||||
Updates rp_name, auth_host, and origins in both the runtime Passkey
|
||||
instance and the persisted database config.
|
||||
"""
|
||||
await authz.verify(
|
||||
auth, ["auth:admin"], host=request.headers.get("host"), max_age="5m"
|
||||
)
|
||||
config = db.data().config
|
||||
pk = passkey.instance
|
||||
|
||||
rp_name = payload.get("rp_name", "").strip() or None
|
||||
auth_host = payload.get("auth_host", "").strip() or None
|
||||
raw_origins = payload.get("origins", [])
|
||||
origins = [
|
||||
hostutil.normalize_origin(o.strip()) for o in raw_origins if o.strip()
|
||||
] or None
|
||||
|
||||
# Normalize auth_host and origins (matching CLI startup behavior)
|
||||
if auth_host:
|
||||
try:
|
||||
hostutil.validate_auth_host(auth_host, config.rp_id)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
auth_host, origins = hostutil.normalize_auth_host_and_origins(auth_host, origins)
|
||||
|
||||
# Validate origins against the current rp_id
|
||||
if origins:
|
||||
for o in origins:
|
||||
Passkey(rp_id=config.rp_id, origins=[o]) # validates or raises
|
||||
|
||||
# Update runtime Passkey instance
|
||||
pk.rp_name = rp_name or config.rp_id
|
||||
pk.allowed_origins = set(origins) if origins else None
|
||||
|
||||
# Persist to database
|
||||
new_config = Config(
|
||||
rp_id=config.rp_id,
|
||||
rp_name=rp_name,
|
||||
auth_host=auth_host,
|
||||
origins=origins,
|
||||
listen=config.listen,
|
||||
)
|
||||
await db.update_config(new_config)
|
||||
|
||||
# Reload hostutil cached config so auth_host changes take effect
|
||||
hostutil.reload_config()
|
||||
|
||||
return {"status": "ok"}
|
||||
|
||||
@@ -185,7 +185,8 @@ async def get_settings():
|
||||
auth_site_url=hostutil.auth_site_url(),
|
||||
session_cookie=AUTH_COOKIE_NAME,
|
||||
version=__version__,
|
||||
)
|
||||
),
|
||||
headers={"Access-Control-Allow-Origin": "*", "Vary": "Origin"},
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -49,6 +49,51 @@ def normalize_origin(origin: str) -> str:
|
||||
return origin.rstrip("/")
|
||||
|
||||
|
||||
def is_subdomain(sub: str, domain: str) -> bool:
|
||||
"""Check if sub is a subdomain of domain (or equal)."""
|
||||
sub_parts = sub.lower().split(".")
|
||||
domain_parts = domain.lower().split(".")
|
||||
if len(sub_parts) < len(domain_parts):
|
||||
return False
|
||||
return sub_parts[-len(domain_parts) :] == domain_parts
|
||||
|
||||
|
||||
def validate_auth_host(auth_host: str, rp_id: str) -> None:
|
||||
"""Validate that auth_host is a subdomain of rp_id.
|
||||
|
||||
Raises ValueError on invalid auth_host.
|
||||
"""
|
||||
parsed = urlparse(auth_host if "://" in auth_host else f"//{auth_host}")
|
||||
host = parsed.hostname or parsed.path
|
||||
if not host:
|
||||
raise ValueError(f"Invalid auth-host: '{auth_host}'")
|
||||
if not is_subdomain(host, rp_id):
|
||||
raise ValueError(
|
||||
f"auth-host '{auth_host}' is not a subdomain of rp-id '{rp_id}'"
|
||||
)
|
||||
|
||||
|
||||
def normalize_auth_host_and_origins(
|
||||
auth_host: str | None, origins: list[str] | None
|
||||
) -> tuple[str | None, list[str] | None]:
|
||||
"""Normalize auth_host and origins, matching CLI startup behavior.
|
||||
|
||||
- Adds https:// to auth_host if no scheme present, strips trailing slashes
|
||||
- Validates auth_host is a well-formed subdomain (caller provides rp_id via validate_auth_host)
|
||||
- Inserts auth_host as first origin if both are specified and not already present
|
||||
- Deduplicates origins while preserving order
|
||||
"""
|
||||
if auth_host:
|
||||
if "://" not in auth_host:
|
||||
auth_host = f"https://{auth_host}"
|
||||
auth_host = auth_host.rstrip("/")
|
||||
if origins is not None and auth_host not in origins:
|
||||
origins.insert(0, auth_host)
|
||||
if origins:
|
||||
origins = list(dict.fromkeys(origins))
|
||||
return auth_host, origins
|
||||
|
||||
|
||||
def reload_config() -> None:
|
||||
_load_config.cache_clear()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user