import argparse import asyncio import ipaddress import logging import os from urllib.parse import urlparse import uvicorn from passkey.util import frontend DEFAULT_HOST = "localhost" DEFAULT_SERVE_PORT = 4401 DEFAULT_DEV_PORT = 4402 def parse_endpoint( value: str | None, default_port: int ) -> tuple[str | None, int | None, str | None, bool]: """Parse an endpoint using stdlib (urllib.parse, ipaddress). Returns (host, port, uds_path). If uds_path is not None, host/port are None. Supported forms: - host[:port] - :port (uses default host) - [ipv6][:port] (bracketed for port usage) - ipv6 (unbracketed, no port allowed -> default port) - unix:/path/to/socket.sock - None -> defaults (localhost:4401) Notes: - For IPv6 with an explicit port you MUST use brackets (e.g. [::1]:8080) - Unbracketed IPv6 like ::1 implies the default port. """ if not value: return DEFAULT_HOST, default_port, None, False # Port only (numeric) -> localhost:port if value.isdigit(): try: port_only = int(value) except ValueError: # pragma: no cover (isdigit guards) raise SystemExit(f"Invalid port '{value}'") return DEFAULT_HOST, port_only, None, False # Leading colon :port -> bind all interfaces (0.0.0.0 + ::) if value.startswith(":") and value != ":": port_part = value[1:] if not port_part.isdigit(): raise SystemExit(f"Invalid port in '{value}'") return None, int(port_part), None, True # UNIX domain socket if value.startswith("unix:"): uds_path = value[5:] or None if uds_path is None: raise SystemExit("unix: path must not be empty") return None, None, uds_path, False # Unbracketed IPv6 (cannot safely contain a port) -> detect by multiple colons if value.count(":") > 1 and not value.startswith("["): try: ipaddress.IPv6Address(value) except ValueError as e: # pragma: no cover raise SystemExit(f"Invalid IPv6 address '{value}': {e}") return value, default_port, None, False # Use urllib.parse for everything else (host[:port], :port, [ipv6][:port]) parsed = urlparse(f"//{value}") # // prefix lets urlparse treat it as netloc host = parsed.hostname port = parsed.port # Host may be None if empty (e.g. ':5500') if not host: host = DEFAULT_HOST if port is None: port = default_port # Validate IP literals (optional; hostname passes through) try: # Strip brackets if somehow present (urlparse removes them already) ipaddress.ip_address(host) except ValueError: # Not an IP address -> treat as hostname; no action pass return host, port, None, False def add_common_options(p: argparse.ArgumentParser) -> None: p.add_argument( "--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("--origin", help="Origin URL (default: https://)") def main(): # Configure logging to remove the "ERROR:root:" prefix logging.basicConfig(level=logging.INFO, format="%(message)s", force=True) parser = argparse.ArgumentParser( prog="passkey-auth", description="Passkey authentication server" ) sub = parser.add_subparsers(dest="command", required=True) # serve subcommand serve = sub.add_parser( "serve", help="Run the server (production style, no auto-reload)" ) serve.add_argument( "hostport", nargs="?", help=( "Endpoint (default: localhost:4401). Forms: host[:port] | :port | " "[ipv6][:port] | ipv6 | unix:/path.sock" ), ) add_common_options(serve) # dev subcommand dev = sub.add_parser("dev", help="Run the server in development (auto-reload)") dev.add_argument( "hostport", nargs="?", help=( "Endpoint (default: localhost:4402). Forms: host[:port] | :port | " "[ipv6][:port] | ipv6 | unix:/path.sock" ), ) add_common_options(dev) args = parser.parse_args() default_port = DEFAULT_DEV_PORT if args.command == "dev" else DEFAULT_SERVE_PORT host, port, uds, all_ifaces = parse_endpoint(args.hostport, default_port) devmode = args.command == "dev" # Determine origin (dev mode default override) origin = args.origin if devmode and not args.origin and not args.rp_id: # Dev mode: Vite runs on another port, override: origin = "http://localhost:4403" # Export configuration via environment for lifespan initialization in each process os.environ.setdefault("PASSKEY_RP_ID", args.rp_id) if args.rp_name: os.environ["PASSKEY_RP_NAME"] = args.rp_name if origin: os.environ["PASSKEY_ORIGIN"] = origin # One-time initialization + bootstrap before starting any server processes. # Lifespan in worker processes will call globals.init with bootstrap disabled. from passkey import globals as _globals # local import asyncio.run( _globals.init( rp_id=args.rp_id, rp_name=args.rp_name, origin=origin, default_admin=os.getenv("PASSKEY_DEFAULT_ADMIN") or None, default_org=os.getenv("PASSKEY_DEFAULT_ORG") or None, bootstrap=True, ) ) run_kwargs: dict = { "reload": devmode, "log_level": "info", } if uds: run_kwargs["uds"] = uds else: # For :port form (all interfaces) we will handle separately if not all_ifaces: run_kwargs["host"] = host run_kwargs["port"] = port if devmode: # Spawn frontend dev server (bun or npm) only once in parent process if os.environ.get("PASSKEY_BUN_PARENT") != "1": os.environ["PASSKEY_BUN_PARENT"] = "1" frontend.run_dev() if all_ifaces and not uds: # If reload enabled, fallback to single dual-stack attempt (::) to keep reload simple if devmode: run_kwargs["host"] = "::" run_kwargs["port"] = port uvicorn.run("passkey.fastapi:app", **run_kwargs) else: # Start two servers concurrently: IPv4 and IPv6 from uvicorn import Config, Server # noqa: E402 local import from passkey.fastapi import app as fastapi_app # noqa: E402 local import async def serve_both(): servers = [] assert port is not None for h in ("0.0.0.0", "::"): try: cfg = Config( app=fastapi_app, host=h, port=port, log_level="info", ) servers.append(Server(cfg)) except Exception as e: # pragma: no cover logging.warning(f"Failed to configure server for {h}: {e}") tasks = [asyncio.create_task(s.serve()) for s in servers] await asyncio.gather(*tasks) asyncio.run(serve_both()) else: uvicorn.run("passkey.fastapi:app", **run_kwargs) if __name__ == "__main__": main()