More robust server startup, startup logo and info screen, renewed devmode script.

This commit is contained in:
Leo Vasanko
2025-12-05 19:06:42 +00:00
parent 8b98cb6325
commit da503a3081
10 changed files with 287 additions and 99 deletions
+15 -4
View File
@@ -30,11 +30,22 @@ Install [UV](https://docs.astral.sh/uv/getting-started/installation/) and run:
uvx paskia serve --rp-id example.com uvx paskia serve --rp-id example.com
``` ```
On the first run it downloads the software and prints a registration link for the Admin. Consider `uv tool install paskia` for a permanent install of `paskia` CLI. On the first run it downloads the software and prints a registration link for the Admin. If you are going to be connecting `localhost` directly, for testing, leave out the rp-id.
The server will start up on [localhost:4401](http://localhost:4401) "for authentication required", serving for `*.example.com`. The server will start up on [localhost:4401](http://localhost:4401) "for authentication required", serving for `*.example.com`.
If you are going to be connecting `localhost` directly, for testing, leave out the rp-id. Otherwise you will need a web server such as [Caddy](https://caddyserver.com/) to serve HTTPS on your actual domain names and proxy requests to Paskia and your backend apps. Otherwise you will need a web server such as [Caddy](https://caddyserver.com/) to serve HTTPS on your actual domain names and proxy requests to Paskia and your backend apps.
A quick example without any config file:
```fish
sudo caddy reverse-proxy --from example.com --to :4401
```
For a permanent install of `paskia` CLI command, not needing `uvx`:
```fish
uv tool install paskia
```
## Configuration ## Configuration
@@ -44,12 +55,12 @@ There is no config file. Pass only the options on CLI:
paskia serve [options] paskia serve [options]
``` ```
Options (all optional): Optional options:
- Listen address (one of): - Listen address (one of):
* `[host]:port`: Address and port (default: `localhost:4401`) * `[host]:port`: Address and port (default: `localhost:4401`)
* `unix:/path.sock`: Unix socket * `unix:/path.sock`: Unix socket
- `--rp-id <domain>`: Domain name for authentication (required for production) - `--rp-id <domain>`: Main domain (required for production)
- `--rp-name "<text>"`: Name of your company or site (default: same as rp-id) - `--rp-name "<text>"`: Name of your company or site (default: same as rp-id)
- `--origin <url>`: Explicit single site (default: `https://<rp-id>`) - `--origin <url>`: Explicit single site (default: `https://<rp-id>`)
- `--auth-host <domain>`: Dedicated authentication site (e.g., `auth.example.com`) - `--auth-host <domain>`: Dedicated authentication site (e.g., `auth.example.com`)
+1 -1
View File
@@ -55,7 +55,7 @@
<div class="section"> <div class="section">
<h2>Browser Mode (full page)</h2> <h2>Browser Mode (full page)</h2>
<p>Block access to otherwise open site - intended for forward-auth mechanism (Caddy, Nxinx):</p> <p>Block access to otherwise open site - intended for forward-auth mechanism (Caddy, Nginx):</p>
<button onclick="browserNav('/auth/api/forward')">🔐 Basic Auth</button> <button onclick="browserNav('/auth/api/forward')">🔐 Basic Auth</button>
<button onclick="browserNav('/auth/api/forward?max_age=10s')">🔄 Reauth (max_age=10s)</button> <button onclick="browserNav('/auth/api/forward?max_age=10s')">🔄 Reauth (max_age=10s)</button>
<button onclick="browserNav('/auth/api/forward?perm=auth:admin')">🛡️ Admin Only</button> <button onclick="browserNav('/auth/api/forward?perm=auth:admin')">🛡️ Admin Only</button>
+44 -15
View File
@@ -111,7 +111,13 @@ def add_common_options(p: argparse.ArgumentParser) -> None:
"--rp-id", default="localhost", help="Relying Party ID (default: localhost)" "--rp-id", default="localhost", help="Relying Party ID (default: localhost)"
) )
p.add_argument("--rp-name", help="Relying Party name (default: same as rp-id)") p.add_argument("--rp-name", help="Relying Party name (default: same as rp-id)")
p.add_argument("--origin", help="Origin URL (default: https://<rp-id>)") p.add_argument(
"--origin",
action="append",
dest="origins",
metavar="URL",
help="Allowed origin URL(s). May be specified multiple times. If any are specified, only those origins are permitted for WebSocket authentication.",
)
p.add_argument( p.add_argument(
"--auth-host", "--auth-host",
help=( help=(
@@ -166,39 +172,62 @@ def main():
else: else:
host = port = uds = all_ifaces = None # type: ignore host = port = uds = all_ifaces = None # type: ignore
# Export configuration via environment for lifespan initialization in each process # Collect origins and handle auth_host
os.environ.setdefault("PASKIA_RP_ID", args.rp_id) origins = getattr(args, "origins", None) or []
if args.rp_name: if not getattr(args, "auth_host", None):
os.environ["PASKIA_RP_NAME"] = args.rp_name
if args.origin:
os.environ["PASKIA_ORIGIN"] = args.origin
if getattr(args, "auth_host", None):
os.environ["PASKIA_AUTH_HOST"] = args.auth_host
else:
# Preserve pre-set env variable if CLI option omitted # Preserve pre-set env variable if CLI option omitted
args.auth_host = os.environ.get("PASKIA_AUTH_HOST") args.auth_host = os.environ.get("PASKIA_AUTH_HOST")
if args.auth_host: if args.auth_host:
# Normalize auth_host with scheme
if "://" not in args.auth_host:
args.auth_host = f"https://{args.auth_host}"
validate_auth_host(args.auth_host, args.rp_id) validate_auth_host(args.auth_host, args.rp_id)
from paskia.util import hostutil as _hostutil # local import from paskia.util import hostutil as _hostutil # local import
_hostutil.reload_config() _hostutil.reload_config()
# If origins are configured, ensure auth_host is included at top
if origins:
# Insert auth_host at the beginning (Passkey.__init__ will normalize/dedupe)
origins.insert(0, args.auth_host)
# Export configuration via single JSON env variable for worker processes
# (PASKIA_DEVMODE is kept separate as it's externally defined)
# All keys are always present; None is used where no value is configured
import json
config = {
"rp_id": args.rp_id,
"rp_name": args.rp_name or None,
"origins": origins or None,
"auth_host": args.auth_host or None,
"default_admin": os.getenv("PASKIA_DEFAULT_ADMIN") or None,
"default_org": os.getenv("PASKIA_DEFAULT_ORG") or None,
}
os.environ["PASKIA_CONFIG"] = json.dumps(config)
# One-time initialization + bootstrap before starting any server processes. # One-time initialization + bootstrap before starting any server processes.
# Lifespan in worker processes will call globals.init with bootstrap disabled. # Lifespan in worker processes will call globals.init with bootstrap disabled.
from paskia import globals as _globals # local import from paskia import globals as _globals # local import
asyncio.run( asyncio.run(
_globals.init( _globals.init(
rp_id=args.rp_id, rp_id=config["rp_id"],
rp_name=args.rp_name, rp_name=config["rp_name"],
origin=args.origin, origins=config["origins"],
default_admin=os.getenv("PASKIA_DEFAULT_ADMIN") or None, default_admin=config["default_admin"],
default_org=os.getenv("PASKIA_DEFAULT_ORG") or None, default_org=config["default_org"],
bootstrap=True, bootstrap=True,
) )
) )
# Print startup configuration
from paskia.util import startupbox
startupbox.print_startup_config(_globals.passkey.instance, args, host, port, uds)
# Handle recover-admin command (no server start) # Handle recover-admin command (no server start)
if args.command == "reset": if args.command == "reset":
from paskia.fastapi import reset as reset_cmd # local import from paskia.fastapi import reset as reset_cmd # local import
+11 -13
View File
@@ -15,26 +15,24 @@ from paskia.util import frontend, hostutil, passphrase
async def lifespan(app: FastAPI): # pragma: no cover - startup path async def lifespan(app: FastAPI): # pragma: no cover - startup path
"""Application lifespan to ensure globals (DB, passkey) are initialized in each process. """Application lifespan to ensure globals (DB, passkey) are initialized in each process.
We populate configuration from environment variables (set by the CLI entrypoint) Configuration is passed via PASKIA_CONFIG JSON env variable (set by the CLI entrypoint)
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.
""" """
import json
from paskia import globals from paskia import globals
rp_id = os.getenv("PASKIA_RP_ID", "localhost") config = json.loads(os.environ["PASKIA_CONFIG"])
rp_name = os.getenv("PASKIA_RP_NAME") or None
origin = os.getenv("PASKIA_ORIGIN") or None
default_admin = (
os.getenv("PASKIA_DEFAULT_ADMIN") or None
) # still passed for context
default_org = os.getenv("PASKIA_DEFAULT_ORG") or None
try: try:
# CLI (__main__) performs bootstrap once; here we skip to avoid duplicate work # CLI (__main__) performs bootstrap once; here we skip to avoid duplicate work
await globals.init( await globals.init(
rp_id=rp_id, rp_id=config["rp_id"],
rp_name=rp_name, rp_name=config["rp_name"],
origin=origin, origins=config["origins"],
default_admin=default_admin, default_admin=config["default_admin"],
default_org=default_org, default_org=config["default_org"],
bootstrap=False, bootstrap=False,
) )
except ValueError as e: except ValueError as e:
+17 -6
View File
@@ -42,19 +42,30 @@ def websocket_error_handler(func):
app = FastAPI() app = FastAPI()
def _validate_origin(ws: WebSocket) -> str:
"""Extract and validate origin from WebSocket request headers.
Raises:
ValueError: If origin header is missing or not in allowed list
"""
origin = ws.headers.get("origin")
if not origin:
raise ValueError("Origin header is required for WebSocket connections")
return passkey.instance.validate_origin(origin)
async def register_chat( async def register_chat(
ws: WebSocket, ws: WebSocket,
user_uuid: UUID, user_uuid: UUID,
user_name: str, user_name: str,
origin: str,
credential_ids: list[bytes] | None = None, credential_ids: list[bytes] | None = None,
origin: str | None = None,
): ):
"""Generate registration options and send them to the client.""" """Generate registration options and send them to the client."""
options, challenge = passkey.instance.reg_generate_options( options, challenge = passkey.instance.reg_generate_options(
user_id=user_uuid, user_id=user_uuid,
user_name=user_name, user_name=user_name,
credential_ids=credential_ids, credential_ids=credential_ids,
origin=origin,
) )
await ws.send_json({"optionsJSON": options}) await ws.send_json({"optionsJSON": options})
response = await ws.receive_json() response = await ws.receive_json()
@@ -75,7 +86,7 @@ async def websocket_register_add(
- Normal session via auth cookie (requires recent authentication) - Normal session via auth cookie (requires recent authentication)
- Reset token supplied as ?reset=... (auth cookie ignored) - Reset token supplied as ?reset=... (auth cookie ignored)
""" """
origin = ws.headers["origin"] origin = _validate_origin(ws)
host = origin.split("://", 1)[1] host = origin.split("://", 1)[1]
if reset is not None: if reset is not None:
if not passphrase.is_well_formed(reset): if not passphrase.is_well_formed(reset):
@@ -100,7 +111,7 @@ async def websocket_register_add(
challenge_ids = await db.instance.get_credentials_by_user_uuid(user_uuid) challenge_ids = await db.instance.get_credentials_by_user_uuid(user_uuid)
# WebAuthn registration # WebAuthn registration
credential = await register_chat(ws, user_uuid, user_name, challenge_ids, origin) credential = await register_chat(ws, user_uuid, user_name, origin, challenge_ids)
# Create a new session and store everything in database # Create a new session and store everything in database
token = create_token() token = create_token()
@@ -131,7 +142,7 @@ async def websocket_register_add(
@app.websocket("/authenticate") @app.websocket("/authenticate")
@websocket_error_handler @websocket_error_handler
async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE): async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
origin = ws.headers["origin"] origin = _validate_origin(ws)
host = origin.split("://", 1)[1] host = origin.split("://", 1)[1]
# 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)
@@ -166,7 +177,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
raise ValueError("This passkey belongs to a different account") raise ValueError("This passkey belongs to a different account")
# Verify the credential matches the stored data # Verify the credential matches the stored data
passkey.instance.auth_verify(credential, challenge, stored_cred, origin=origin) passkey.instance.auth_verify(credential, challenge, stored_cred, origin)
# Update both credential and user's last_seen timestamp # Update both credential and user's last_seen timestamp
await db.instance.login(stored_cred.user_uuid, stored_cred) await db.instance.login(stored_cred.user_uuid, stored_cred)
+2 -2
View File
@@ -29,7 +29,7 @@ class Manager(Generic[T]):
async def init( async def init(
rp_id: str = "localhost", rp_id: str = "localhost",
rp_name: str | None = None, rp_name: str | None = None,
origin: str | None = None, origins: list[str] | None = None,
default_admin: str | None = None, default_admin: str | None = None,
default_org: str | None = None, default_org: str | None = None,
*, *,
@@ -45,7 +45,7 @@ async def init(
passkey.instance = Passkey( passkey.instance = Passkey(
rp_id=rp_id, rp_id=rp_id,
rp_name=rp_name or rp_id, rp_name=rp_name or rp_id,
origin=origin, origins=origins,
) )
# Test if we have a database already initialized, otherwise use SQL # Test if we have a database already initialized, otherwise use SQL
+39 -16
View File
@@ -47,7 +47,7 @@ class Passkey:
self, self,
rp_id: str, rp_id: str,
rp_name: str | None = None, rp_name: str | None = None,
origin: str | None = None, origins: list[str] | None = None,
supported_pub_key_algs: list[COSEAlgorithmIdentifier] | None = None, supported_pub_key_algs: list[COSEAlgorithmIdentifier] | None = None,
): ):
""" """
@@ -56,27 +56,30 @@ class Passkey:
Args: Args:
rp_id: Your security domain (e.g. "example.com") rp_id: Your security domain (e.g. "example.com")
rp_name: The relying party display name (e.g. "Example App"). May be shown in authenticators. rp_name: The relying party display name (e.g. "Example App"). May be shown in authenticators.
origin: The origin URL of the application (e.g. "https://app.example.com"). origins: List of allowed origin URLs (e.g. ["https://app.example.com", "https://auth.example.com"]).
If no scheme is provided, "https://" will be prepended. Each must be a subdomain or same as rp_id. If not provided, any subdomain of rp_id is allowed.
Must be a subdomain or same as rp_id, with port and scheme but no path included.
supported_pub_key_algs: List of supported COSE algorithms (default is EDDSA, ECDSA_SHA_256, RSASSA_PKCS1_v1_5_SHA_256). supported_pub_key_algs: List of supported COSE algorithms (default is EDDSA, ECDSA_SHA_256, RSASSA_PKCS1_v1_5_SHA_256).
Raises: Raises:
ValueError: If the origin domain doesn't match or isn't a subdomain of rp_id. ValueError: If any origin domain doesn't match or isn't a subdomain of rp_id.
""" """
self.rp_id = rp_id self.rp_id = rp_id
self.rp_name = rp_name or rp_id self.rp_name = rp_name or rp_id
self.origin = self._normalize_and_validate_origin(origin, rp_id) self.allowed_origins: set[str] | None = None
if origins:
# Normalize and deduplicate origins into a set for O(1) lookups
self.allowed_origins = {
self._normalize_and_validate_origin(o, rp_id) for o in origins
}
self.supported_pub_key_algs = supported_pub_key_algs or [ self.supported_pub_key_algs = supported_pub_key_algs or [
COSEAlgorithmIdentifier.EDDSA, COSEAlgorithmIdentifier.EDDSA,
COSEAlgorithmIdentifier.ECDSA_SHA_256, COSEAlgorithmIdentifier.ECDSA_SHA_256,
COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_256, COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_256,
] ]
def _normalize_and_validate_origin(self, origin: str | None, rp_id: str) -> str: def _normalize_and_validate_origin(self, origin: str, rp_id: str) -> str:
if origin is None: """Normalize and validate an origin URL against the rp_id."""
origin = f"https://{rp_id}" if "://" not in origin:
elif "://" not in origin:
origin = f"https://{origin}" origin = f"https://{origin}"
hostname = urlparse(origin).hostname hostname = urlparse(origin).hostname
@@ -90,6 +93,24 @@ class Passkey:
f"Origin domain '{hostname}' must be the same as or a subdomain of rp_id '{rp_id}'" f"Origin domain '{hostname}' must be the same as or a subdomain of rp_id '{rp_id}'"
) )
def validate_origin(self, origin: str) -> str:
"""Validate that origin is allowed and return the normalized form.
Args:
origin: The origin URL to validate (from WebSocket request header)
Returns:
The validated origin URL
Raises:
ValueError: If origin is not in the allowed list (when origins are configured)
or if origin is not a valid subdomain of rp_id
"""
normalized = self._normalize_and_validate_origin(origin, self.rp_id)
if self.allowed_origins is not None and normalized not in self.allowed_origins:
raise ValueError(f"Origin '{origin}' is not in the allowed origins list")
return normalized
### Registration Methods ### ### Registration Methods ###
def reg_generate_options( def reg_generate_options(
@@ -137,14 +158,16 @@ class Passkey:
response_json: dict | str, response_json: dict | str,
expected_challenge: bytes, expected_challenge: bytes,
user_uuid: UUID, user_uuid: UUID,
origin: str | None = None, origin: str,
) -> Credential: ) -> Credential:
""" """
Verify registration response. Verify registration response.
Args: Args:
credential: The credential response from the client response_json: The credential response from the client
expected_challenge: The expected challenge bytes expected_challenge: The expected challenge bytes
user_uuid: The user's UUID
origin: The origin URL (required, must be pre-validated)
Returns: Returns:
Registration verification result Registration verification result
@@ -153,7 +176,7 @@ class Passkey:
registration = verify_registration_response( registration = verify_registration_response(
credential=credential, credential=credential,
expected_challenge=expected_challenge, expected_challenge=expected_challenge,
expected_origin=origin or self.origin, expected_origin=origin,
expected_rp_id=self.rp_id, expected_rp_id=self.rp_id,
) )
return Credential( return Credential(
@@ -206,7 +229,7 @@ class Passkey:
credential: AuthenticationCredential, credential: AuthenticationCredential,
expected_challenge: bytes, expected_challenge: bytes,
stored_cred: Credential, stored_cred: Credential,
origin: str | None = None, origin: str,
) -> VerifiedAuthentication: ) -> VerifiedAuthentication:
""" """
Verify authentication response against locally stored credential data. Verify authentication response against locally stored credential data.
@@ -215,13 +238,13 @@ class Passkey:
credential: The authentication credential response from the client credential: The authentication credential response from the client
expected_challenge: The earlier generated challenge bytes expected_challenge: The earlier generated challenge bytes
stored_cred: The server stored credential record (modified by this function) stored_cred: The server stored credential record (modified by this function)
origin: The origin URL (required, must be pre-validated)
""" """
expected_origin = origin or self.origin
# Verify the authentication response # Verify the authentication response
verification = verify_authentication_response( verification = verify_authentication_response(
credential=credential, credential=credential,
expected_challenge=expected_challenge, expected_challenge=expected_challenge,
expected_origin=expected_origin, expected_origin=origin,
expected_rp_id=self.rp_id, expected_rp_id=self.rp_id,
credential_public_key=stored_cred.public_key, credential_public_key=stored_cred.public_key,
credential_current_sign_count=stored_cred.sign_count, credential_current_sign_count=stored_cred.sign_count,
+27 -11
View File
@@ -1,28 +1,38 @@
"""Utilities for determining the auth UI host and base URLs.""" """Utilities for determining the auth UI host and base URLs."""
import json
import os import os
from functools import lru_cache from functools import lru_cache
from urllib.parse import urlparse, urlsplit from urllib.parse import urlparse, urlsplit
from paskia.globals import passkey as global_passkey from paskia.globals import passkey as global_passkey
_AUTH_HOST_ENV = "PASKIA_AUTH_HOST"
def _default_origin_scheme() -> str: def _default_origin_scheme() -> str:
origin_url = urlparse(global_passkey.instance.origin) """Get the default scheme from configured origins, or fallback to https."""
return origin_url.scheme or "https" allowed = global_passkey.instance.allowed_origins
if allowed:
# Pick any origin from the set
origin_url = urlparse(next(iter(allowed)))
return origin_url.scheme or "https"
return "https"
@lru_cache(maxsize=1) @lru_cache(maxsize=1)
def _load_config() -> tuple[str | None, str] | None: def _load_config() -> tuple[str | None, str] | None:
raw = os.getenv(_AUTH_HOST_ENV) """Load auth_host from PASKIA_CONFIG JSON, falling back to PASKIA_AUTH_HOST."""
# Try PASKIA_CONFIG first (set by CLI)
config_json = os.getenv("PASKIA_CONFIG")
if config_json:
config = json.loads(config_json)
raw = config["auth_host"] # Always present, may be None
else:
# Fallback for external usage (e.g., PASKIA_AUTH_HOST set directly)
raw = os.getenv("PASKIA_AUTH_HOST")
if not raw: if not raw:
return None return None
candidate = raw.strip() parsed = urlparse(raw if "://" in raw else f"//{raw}")
if not candidate:
return None
parsed = urlparse(candidate if "://" in candidate else f"//{candidate}")
netloc = parsed.netloc or parsed.path netloc = parsed.netloc or parsed.path
if not netloc: if not netloc:
return None return None
@@ -53,8 +63,14 @@ def auth_site_base_url(scheme: str | None = None, host: str | None = None) -> st
scheme_to_use = scheme or _default_origin_scheme() scheme_to_use = scheme or _default_origin_scheme()
netloc = host.strip("/") netloc = host.strip("/")
else: else:
origin = global_passkey.instance.origin.rstrip("/") # Use the first allowed origin, or fallback to rp_id
return f"{origin}{ui_base_path()}" allowed = global_passkey.instance.allowed_origins
if allowed:
origin = allowed[0].rstrip("/")
return f"{origin}{ui_base_path()}"
# Fallback: construct from rp_id
rp_id = global_passkey.instance.rp_id
return f"https://{rp_id}{ui_base_path()}"
base = f"{scheme_to_use}://{netloc}".rstrip("/") base = f"{scheme_to_use}://{netloc}".rstrip("/")
path = ui_base_path().lstrip("/") path = ui_base_path().lstrip("/")
+73
View File
@@ -0,0 +1,73 @@
"""Startup configuration box formatting utilities."""
import os
BOX_WIDTH = 60 # Inner width (excluding box chars)
def line(text: str = "") -> str:
"""Format a line inside the box with proper padding, truncating if needed."""
if len(text) > BOX_WIDTH:
text = text[: BOX_WIDTH - 1] + ""
return f"{text:<{BOX_WIDTH}}\n"
def top() -> str:
return "" + "" * (BOX_WIDTH + 2) + "\n"
def bottom() -> str:
return "" + "" * (BOX_WIDTH + 2) + "\n"
def print_startup_config(passkey_instance, args, host, port, uds) -> None:
"""Print server configuration on startup."""
from sys import stderr
from paskia._version import __version__
lines = [top()]
lines.append(line(" ▄▄▄▄▄"))
lines.append(line("█ █ Paskia " + __version__))
lines.append(line("█ █▄▄▄▄▄▄▄▄▄▄▄▄"))
lines.append(line("█ █▀▀▀▀█▀▀█▀▀█"))
lines.append(line(" ▀▀▀▀▀"))
# Format auth host section
auth_host = getattr(args, "auth_host", None)
if auth_host:
lines.append(line(f"Auth Host: {auth_host}"))
# Show frontend URL if in dev mode
devmode = os.environ.get("PASKIA_DEVMODE")
if devmode:
lines.append(line(f"Dev Frontend: {devmode}"))
# Format listen address with scheme
if uds:
listen = f"unix:{uds}"
elif host:
listen = f"http://{host}:{port}"
else:
listen = f"http://0.0.0.0:{port} + [::]:{port}"
lines.append(line(f"Backend: {listen}"))
# Relying Party line (omit name if same as id)
rp_id = passkey_instance.rp_id
rp_name = passkey_instance.rp_name
if rp_name and rp_name != rp_id:
lines.append(line(f"Relying Party: {rp_id} ({rp_name})"))
else:
lines.append(line(f"Relying Party: {rp_id}"))
# Format origins section
allowed = passkey_instance.allowed_origins
if allowed:
lines.append(line("Permitted Origins:"))
for origin in sorted(allowed):
lines.append(line(f" - {origin}"))
else:
lines.append(line(f"Origin: {rp_id} and all subdomains allowed"))
lines.append(bottom())
stderr.write("".join(lines))
Regular → Executable
+58 -31
View File
@@ -1,16 +1,19 @@
#!/usr/bin/env python3 #!/usr/bin/env -S uv run
"""Development server script for Paskia. """Run Vite development server for frontend and FastAPI backend with auto-reload.
This script is only available when running from the git repository source, This script is only available when running from the git repository source,
not from the installed package. It starts both the Vite frontend dev server not from the installed package. It starts both the Vite frontend dev server
and the FastAPI backend with auto-reload enabled. and the FastAPI backend with auto-reload enabled.
Usage: Usage:
python scripts/dev.py [options...] uv run scripts/dev.py [host:port] [options...]
All options are forwarded to `paskia serve`. The optional host:port argument sets where the Vite frontend listens.
All other options are forwarded to `paskia serve`.
Backend always listens on localhost:4402.
""" """
import argparse
import atexit import atexit
import os import os
import shutil import shutil
@@ -21,14 +24,10 @@ from pathlib import Path
from sys import stderr from sys import stderr
from threading import Thread from threading import Thread
# Set dev mode environment variable BEFORE importing anything from paskia from paskia.fastapi.__main__ import parse_endpoint
os.environ["PASKIA_DEVMODE"] = "1"
# Ensure the package is importable when running from repo root DEFAULT_VITE_PORT = 4403 # overrides by CLI option
sys.path.insert(0, str(Path(__file__).parent.parent)) BACKEND_PORT = 4402 # hardcoded, also in vite.config.ts
DEFAULT_DEV_PORT = 4402
DEV_SERVER = "http://localhost:4403"
NO_FRONTEND_TOOL = """\ NO_FRONTEND_TOOL = """\
┃ ⚠️ deno, npm or bunx needed to run the frontend server. ┃ ⚠️ deno, npm or bunx needed to run the frontend server.
@@ -48,17 +47,19 @@ BUN_BUG = """\
NO_FRONTEND = """\ NO_FRONTEND = """\
Note: only static build of the frontend is served at localhost:4402. The backend will still try reaching Vite at {vite_url}
The page will not update with frontend code changes. for various frontend assets, so make sure to start it manually.
""" """
def run_vite(): def run_vite(vite_url: str, vite_host: str | None, vite_port: int):
"""Spawn the frontend dev server (deno, npm, or bunx) as a background process.""" """Spawn the frontend dev server (deno, npm, or bunx) as a background process."""
devpath = Path(__file__).parent.parent / "frontend" devpath = Path(__file__).parent.parent / "frontend"
if not (devpath / "package.json").exists(): if not (devpath / "package.json").exists():
stderr.write(f"┃ ⚠️ Frontend source not found at {devpath}\n") stderr.write(
stderr.write(NO_FRONTEND) f"┃ ⚠️ Frontend source not found at {devpath}\n"
+ NO_FRONTEND.format(vite_url=vite_url)
)
return return
options = [ options = [
@@ -74,24 +75,31 @@ def run_vite():
tool_name = option[0] tool_name = option[0]
break break
# Add Vite CLI args for host/port
vite_args = ["--port", str(vite_port)]
if vite_host:
vite_args.extend(["--host", vite_host])
vite_process = None vite_process = None
def start_vite(): def start_vite():
nonlocal vite_process nonlocal vite_process
if cmd is None: if cmd is None:
stderr.write(NO_FRONTEND_TOOL) stderr.write(NO_FRONTEND_TOOL + NO_FRONTEND.format(vite_url=vite_url))
stderr.write(NO_FRONTEND)
return return
assert tool_name is not None assert tool_name is not None
try: try:
if tool_name == "bunx": if tool_name == "bunx":
stderr.write(BUN_BUG) stderr.write(BUN_BUG)
stderr.write(f">>> {' '.join([tool_name, *cmd[1:]])}\n") full_cmd = cmd + vite_args
vite_process = subprocess.Popen(cmd, cwd=str(devpath), shell=False) stderr.write(f">>> {' '.join([tool_name, *full_cmd[1:]])}\n")
vite_process = subprocess.Popen(full_cmd, cwd=str(devpath), shell=False)
except Exception as e: except Exception as e:
stderr.write(f"┃ ⚠️ Vite couldn't start: {e}\n") stderr.write(
stderr.write(NO_FRONTEND) f"┃ ⚠️ Vite couldn't start: {e}\n"
+ NO_FRONTEND.format(vite_url=vite_url)
)
def cleanup(): def cleanup():
if vite_process: if vite_process:
@@ -108,20 +116,39 @@ def run_vite():
def main(): def main():
# Start Vite dev server first # Parse optional hostport argument for Vite frontend
run_vite() parser = argparse.ArgumentParser(add_help=False)
parser.add_argument("hostport", nargs="?", default=None)
args, remaining = parser.parse_known_args()
# Set default origin for Vite if not specified # Parse Vite endpoint
if "--origin" not in sys.argv: vite_host, vite_port, vite_uds, all_ifaces = parse_endpoint(
os.environ.setdefault("PASKIA_ORIGIN", DEV_SERVER) args.hostport, DEFAULT_VITE_PORT
)
# Build argv for the main CLI if vite_uds:
# Dev mode always listens on localhost:4402 (security: prevents public exposure) raise SystemExit("┃ ⚠️ Unix sockets are not supported for Vite frontend")
# User args come after, allowing overrides of other options
sys.argv = ["paskia", "serve", f"localhost:{DEFAULT_DEV_PORT}"] + sys.argv[1:]
# Handle all-interfaces case (:port syntax)
# Vite uses 0.0.0.0 to listen on all interfaces (IPv4 only, sufficient for dev)
if all_ifaces:
vite_host = "0.0.0.0"
# Build Vite URL for PASKIA_DEVMODE (always use localhost for URL)
vite_url = f"http://localhost:{vite_port}"
# Start Vite dev server
run_vite(vite_url, vite_host, vite_port)
# Set dev mode with Vite URL
os.environ["PASKIA_DEVMODE"] = vite_url
# Import CLI after environment is set up
from paskia.fastapi.__main__ import main as cli_main from paskia.fastapi.__main__ import main as cli_main
# Build argv for the main CLI in Dev mode
# Backend always listens on localhost only (Vite proxies API requests)
sys.argv = ["paskia", "serve", f"localhost:{BACKEND_PORT}"] + remaining
cli_main() cli_main()