Create registration links on the same host (subdomain) that is being used by the one who creates it.

This commit is contained in:
Leo Vasanko
2025-10-02 12:30:50 -06:00
parent eb38995cca
commit fbfd0bbb47
3 changed files with 34 additions and 6 deletions
+6 -4
View File
@@ -1,13 +1,13 @@
import logging import logging
from uuid import UUID, uuid4 from uuid import UUID, uuid4
from fastapi import Body, Cookie, FastAPI, HTTPException from fastapi import Body, Cookie, FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, JSONResponse from fastapi.responses import FileResponse, JSONResponse
from ..authsession import expires from ..authsession import expires
from ..globals import db from ..globals import db
from ..globals import passkey as global_passkey from ..globals import passkey as global_passkey
from ..util import frontend, passphrase, permutil, querysafe, tokens from ..util import frontend, hostutil, passphrase, permutil, querysafe, tokens
from . import authz from . import authz
app = FastAPI() app = FastAPI()
@@ -335,7 +335,7 @@ async def admin_update_user_role(
@app.post("/orgs/{org_uuid}/users/{user_uuid}/create-link") @app.post("/orgs/{org_uuid}/users/{user_uuid}/create-link")
async def admin_create_user_registration_link( async def admin_create_user_registration_link(
org_uuid: UUID, user_uuid: UUID, auth=Cookie(None) org_uuid: UUID, user_uuid: UUID, request: Request, auth=Cookie(None)
): ):
try: try:
user_org, _role_name = await db.instance.get_user_organization(user_uuid) user_org, _role_name = await db.instance.get_user_organization(user_uuid)
@@ -358,7 +358,9 @@ async def admin_create_user_registration_link(
expires=expires(), expires=expires(),
info={"type": "device addition", "created_by_admin": True}, info={"type": "device addition", "created_by_admin": True},
) )
origin = global_passkey.instance.origin origin = hostutil.effective_origin(
request.url.scheme, request.headers.get("host"), global_passkey.instance.rp_id
)
url = f"{origin}/auth/{token}" url = f"{origin}/auth/{token}"
return {"url": url, "expires": expires().isoformat()} return {"url": url, "expires": expires().isoformat()}
+4 -2
View File
@@ -29,7 +29,7 @@ from ..authsession import (
) )
from ..globals import db from ..globals import db
from ..globals import passkey as global_passkey from ..globals import passkey as global_passkey
from ..util import passphrase, permutil, tokens from ..util import hostutil, passphrase, permutil, tokens
from ..util.tokens import session_key from ..util.tokens import session_key
from . import authz, session from . import authz, session
@@ -267,7 +267,9 @@ async def api_create_link(request: Request, auth=Cookie(None)):
expires=expires(), expires=expires(),
info=session.infodict(request, "device addition"), info=session.infodict(request, "device addition"),
) )
origin = global_passkey.instance.origin.rstrip("/") origin = hostutil.effective_origin(
request.url.scheme, request.headers.get("host"), global_passkey.instance.rp_id
)
url = f"{origin}/auth/{token}" url = f"{origin}/auth/{token}"
return { return {
"message": "Registration link generated successfully", "message": "Registration link generated successfully",
+24
View File
@@ -0,0 +1,24 @@
"""Utilities for host validation and origin determination."""
from ..globals import passkey as global_passkey
def effective_origin(scheme: str, host: str | None, rp_id: str) -> str:
"""Determine the effective origin for a request.
Uses the provided host if it's compatible with the relying party ID,
otherwise falls back to the configured origin.
Args:
scheme: The URL scheme (e.g. "https")
host: The host header value (e.g. "example.com" or "sub.example.com:8080")
rp_id: The relying party ID (e.g. "example.com")
Returns:
The effective origin URL to use
"""
if host:
hostname = host.split(":")[0] # Remove port if present
if hostname == rp_id or hostname.endswith(f".{rp_id}"):
return f"{scheme}://{host}"
return global_passkey.instance.origin