Changed origin config to take multiple origins and if any are configured, restrict access to these. Removed bootstrap name options of created org and user (both can be easily renamed from web ui). Cleanup.

This commit is contained in:
2025-12-06 00:51:18 +00:00
parent 127e06179b
commit 8937905c9c
10 changed files with 40 additions and 106 deletions
+6 -36
View File
@@ -53,16 +53,10 @@ async def _create_and_log_admin_reset_link(user_uuid, message, session_type) ->
return reset_link
async def bootstrap_system(
user_name: str | None = None, org_name: str | None = None
) -> dict:
async def bootstrap_system() -> dict:
"""
Bootstrap the entire system with default data.
Args:
user_name: Display name for the admin user (default: "Admin")
org_name: Display name for the organization (default: "Organization")
Returns:
dict: Contains information about created entities and reset link
"""
@@ -70,7 +64,7 @@ async def bootstrap_system(
perm0 = Permission(id="auth:admin", display_name="Master Admin")
await globals.db.instance.create_permission(perm0)
org = Org(uuid7.create(), org_name or "Organization")
org = Org(uuid7.create(), "Organization")
await globals.db.instance.create_organization(org)
# After creation, org.permissions now includes the auto-created org admin permission
@@ -89,7 +83,7 @@ async def bootstrap_system(
user = User(
uuid=uuid7.create(),
display_name=user_name or "Admin",
display_name="Admin",
role_uuid=role.uuid,
created_at=datetime.now(timezone.utc),
visits=0,
@@ -159,16 +153,10 @@ async def check_admin_credentials() -> bool:
return False
async def bootstrap_if_needed(
default_admin: str | None = None, default_org: str | None = None
) -> bool:
async def bootstrap_if_needed() -> bool:
"""
Check if system needs bootstrapping and perform it if necessary.
Args:
default_admin: Display name for the admin user
default_org: Display name for the organization
Returns:
bool: True if bootstrapping was performed, False if system was already set up
"""
@@ -185,35 +173,17 @@ async def bootstrap_if_needed(
# No admin permission found, need to bootstrap
# Bootstrap creates the admin user AND the reset link, so no need to check credentials after
await bootstrap_system(default_admin, default_org)
await bootstrap_system()
return True
# CLI interface
async def main():
"""Main CLI entry point for bootstrapping."""
import argparse
# Configure logging for CLI usage
logging.basicConfig(level=logging.INFO, format="%(message)s", force=True)
parser = argparse.ArgumentParser(
description="Bootstrap passkey authentication system"
)
parser.add_argument(
"--user-name",
default=None,
help="Name for the admin user (default: Admin)",
)
parser.add_argument(
"--org-name",
default=None,
help="Name for the organization (default: Organization)",
)
args = parser.parse_args()
await globals.init(default_admin=args.user_name, default_org=args.org_name)
await globals.init()
if __name__ == "__main__":
+5 -3
View File
@@ -5,6 +5,7 @@ This module provides an async database layer using SQLAlchemy async mode
for managing users and credentials in a WebAuthn authentication system.
"""
import os
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from uuid import UUID
@@ -40,7 +41,7 @@ from paskia.db import (
)
from paskia.globals import db
DB_PATH = "sqlite+aiosqlite:///paskia.sqlite"
DB_PATH_DEFAULT = "sqlite+aiosqlite:///paskia.sqlite"
def _normalize_dt(value: datetime | None) -> datetime | None:
@@ -52,7 +53,8 @@ def _normalize_dt(value: datetime | None) -> datetime | None:
async def init(*args, **kwargs):
db.instance = DB()
db_path = os.environ.get("PASKIA_DB", DB_PATH_DEFAULT)
db.instance = DB(db_path)
await db.instance.init_db()
@@ -289,7 +291,7 @@ class RolePermission(Base):
class DB(DatabaseInterface):
"""Database class that handles its own connections."""
def __init__(self, db_path: str = DB_PATH):
def __init__(self, db_path: str = DB_PATH_DEFAULT):
"""Initialize with database path."""
self.engine = create_async_engine(db_path, echo=False)
# Ensure SQLite foreign key enforcement is ON for every new connection
-8
View File
@@ -174,10 +174,6 @@ def main():
# Collect origins and handle auth_host
origins = getattr(args, "origins", None) or []
if not getattr(args, "auth_host", None):
# Preserve pre-set env variable if CLI option omitted
args.auth_host = os.environ.get("PASKIA_AUTH_HOST")
if args.auth_host:
# Normalize auth_host with scheme
if "://" not in args.auth_host:
@@ -203,8 +199,6 @@ def main():
"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)
@@ -217,8 +211,6 @@ def main():
rp_id=config["rp_id"],
rp_name=config["rp_name"],
origins=config["origins"],
default_admin=config["default_admin"],
default_org=config["default_org"],
bootstrap=True,
)
)
+1 -3
View File
@@ -452,9 +452,7 @@ async def admin_create_user_registration_link(
expiry=expiry,
token_type=token_type,
)
url = hostutil.reset_link_url(
token, request.url.scheme, request.headers.get("host")
)
url = hostutil.reset_link_url(token)
return {
"url": url,
"expires": (
+2 -1
View File
@@ -21,7 +21,7 @@ from paskia.authsession import (
session_expiry,
)
from paskia.fastapi import authz, session, user
from paskia.fastapi.session import AUTH_COOKIE
from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME
from paskia.globals import db
from paskia.globals import passkey as global_passkey
from paskia.util import frontend, hostutil, htmlutil, passphrase, userinfo
@@ -199,6 +199,7 @@ async def get_settings():
"rp_name": pk.rp_name,
"ui_base_path": base_path,
"auth_host": hostutil.configured_auth_host(),
"session_cookie": AUTH_COOKIE_NAME,
}
-2
View File
@@ -31,8 +31,6 @@ async def lifespan(app: FastAPI): # pragma: no cover - startup path
rp_id=config["rp_id"],
rp_name=config["rp_name"],
origins=config["origins"],
default_admin=config["default_admin"],
default_org=config["default_org"],
bootstrap=False,
)
except ValueError as e:
+1 -1
View File
@@ -12,7 +12,7 @@ from fastapi import Cookie, Request, Response, WebSocket
from paskia.authsession import EXPIRES
AUTH_COOKIE_NAME = "__Host-auth"
AUTH_COOKIE_NAME = "__Host-paskia"
AUTH_COOKIE = Cookie(None, alias=AUTH_COOKIE_NAME)
+1 -3
View File
@@ -150,9 +150,7 @@ async def api_create_link(
expiry=expiry,
token_type="device addition",
)
url = hostutil.reset_link_url(
token, request.url.scheme, request.headers.get("host")
)
url = hostutil.reset_link_url(token)
return {
"message": "Registration link generated successfully",
"url": url,
+1 -3
View File
@@ -30,8 +30,6 @@ async def init(
rp_id: str = "localhost",
rp_name: str | None = None,
origins: list[str] | None = None,
default_admin: str | None = None,
default_org: str | None = None,
*,
bootstrap: bool = True,
) -> None:
@@ -60,7 +58,7 @@ async def init(
# Bootstrap system if needed
from .bootstrap import bootstrap_if_needed
await bootstrap_if_needed(default_admin, default_org)
await bootstrap_if_needed()
# Global instances
+21 -44
View File
@@ -8,35 +8,24 @@ from urllib.parse import urlparse, urlsplit
from paskia.globals import passkey as global_passkey
def _default_origin_scheme() -> str:
"""Get the default scheme from configured origins, or fallback to 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)
def _load_config() -> tuple[str | None, str] | None:
"""Load auth_host from PASKIA_CONFIG JSON, falling back to PASKIA_AUTH_HOST."""
# Try PASKIA_CONFIG first (set by CLI)
def _load_config() -> tuple[str, str] | None:
"""Load auth_host from PASKIA_CONFIG JSON.
Returns (scheme, netloc) tuple if configured, None otherwise.
"""
config_json = os.getenv("PASKIA_CONFIG")
if config_json:
if not config_json:
return None
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:
return None
parsed = urlparse(raw if "://" in raw else f"//{raw}")
netloc = parsed.netloc or parsed.path
if not netloc:
return None
return (parsed.scheme or None, netloc.strip("/"))
return (parsed.scheme or "https", netloc.strip("/"))
def configured_auth_host() -> str | None:
@@ -52,36 +41,24 @@ def ui_base_path() -> str:
return "/" if is_root_mode() else "/auth/"
def auth_site_base_url(scheme: str | None = None, host: str | None = None) -> str:
def auth_site_base_url() -> str:
"""Return the base URL for the auth site UI.
If auth_host is configured (root mode), returns its URL.
Otherwise, constructs URL from rp_id with /auth/ path.
"""
cfg = _load_config()
if cfg:
cfg_scheme, cfg_host = cfg
scheme_to_use = cfg_scheme or scheme or _default_origin_scheme()
netloc = cfg_host
else:
if host:
scheme_to_use = scheme or _default_origin_scheme()
netloc = host.strip("/")
else:
# Use the first allowed origin, or fallback to rp_id
allowed = global_passkey.instance.allowed_origins
if allowed:
origin = allowed[0].rstrip("/")
return f"{origin}{ui_base_path()}"
# Fallback: construct from rp_id
scheme, netloc = cfg
return f"{scheme}://{netloc}/"
# Not in root mode: use rp_id with /auth/ path
rp_id = global_passkey.instance.rp_id
return f"https://{rp_id}{ui_base_path()}"
base = f"{scheme_to_use}://{netloc}".rstrip("/")
path = ui_base_path().lstrip("/")
return f"{base}/{path}" if path else f"{base}/"
return f"https://{rp_id}/auth/"
def reset_link_url(
token: str, scheme: str | None = None, host: str | None = None
) -> str:
base = auth_site_base_url(scheme, host)
return f"{base}{token}"
def reset_link_url(token: str) -> str:
return f"{auth_site_base_url()}{token}"
def reload_config() -> None: