LeoVasanko
10af29f92d
- Serve multiple domains (RP IDs) from one instance: host-based dispatch, per-domain credentials and sessions, domains managed at runtime in the admin UI — previously one RP per instance - Cross-domain sign-in via Related Origin Requests: per-domain related-origins list with a served .well-known/webauthn document - Explicit per-domain origin lists with shell-glob wildcards (**. for apex + any subdomain depth, *. for one level), editable in the admin UI with validation and self-lockout guards - Per-domain auth hosts: the account/admin UI can live on a different host per domain, no longer confined to subdomains of a single RP - CLI: 'paskia init <rp-id [rp-name]' initializes or adds a domain to an existing database; 'paskia migrate' converts legacy databases BREAKING CHANGES (v2.0): - Database schema: config is now per-domain and credentials/sessions carry an rp_id — existing databases must be converted with 'paskia migrate' - Origins are now explicit: main implicitly allowed every subdomain of the RP; configure '**.' origins to reproduce that behavior - CLI: the flat '--rp-id/--rp-name/--origin/--auth/--save' flags are replaced by the 'init' and 'migrate' subcommandsReviewed-on: #4
85 lines
2.5 KiB
Python
85 lines
2.5 KiB
Python
"""Avatar storage and URL helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import hashlib
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
from fastapi import HTTPException, UploadFile
|
|
|
|
from paskia.db.paths import users_root_path
|
|
from paskia.domains import current_domain
|
|
|
|
MAX_UPLOAD_BYTES = 10 * 1024 * 1024
|
|
|
|
|
|
def avatar_path(user_uuid: UUID) -> Path:
|
|
"""Return the avatar file path for a user."""
|
|
return users_root_path(create_root=True) / str(user_uuid) / "profile.webp"
|
|
|
|
|
|
def avatar_public_path(user_uuid: UUID) -> str:
|
|
"""Return the public relative path for a user's avatar."""
|
|
return f"/auth/api/user/{user_uuid}/profile.webp"
|
|
|
|
|
|
def avatar_browser_url(user_uuid: UUID) -> str | None:
|
|
"""Return the browser-facing avatar URL."""
|
|
if not avatar_path(user_uuid).is_file():
|
|
return None
|
|
return avatar_public_path(user_uuid)
|
|
|
|
|
|
def avatar_url(user_uuid: UUID) -> str | None:
|
|
"""Return the absolute public avatar URL for a user, or None."""
|
|
if not avatar_path(user_uuid).is_file():
|
|
return None
|
|
return current_domain().api_url(f"user/{user_uuid}/profile.webp")
|
|
|
|
|
|
def remove_avatar_file(user_uuid: UUID) -> None:
|
|
"""Delete a stored avatar file if it exists."""
|
|
with contextlib.suppress(FileNotFoundError):
|
|
avatar_path(user_uuid).unlink()
|
|
|
|
|
|
def read_avatar_bytes(user_uuid: UUID) -> bytes | None:
|
|
"""Read the stored avatar file for a user, if present."""
|
|
path = avatar_path(user_uuid)
|
|
if not path.is_file():
|
|
return None
|
|
return path.read_bytes()
|
|
|
|
|
|
def _is_webp(data: bytes) -> bool:
|
|
"""Return True when bytes look like a RIFF WebP file."""
|
|
return len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP"
|
|
|
|
|
|
async def read_upload(upload: UploadFile) -> bytes:
|
|
"""Read an uploaded avatar and require it to already be WebP."""
|
|
data = await upload.read(MAX_UPLOAD_BYTES + 1)
|
|
if not data:
|
|
raise HTTPException(status_code=400, detail="No avatar file uploaded")
|
|
if len(data) > MAX_UPLOAD_BYTES:
|
|
raise HTTPException(status_code=413, detail="Avatar upload too large")
|
|
|
|
if not _is_webp(data):
|
|
raise HTTPException(status_code=400, detail="Avatar upload must be WebP")
|
|
|
|
return data
|
|
|
|
|
|
def store_avatar(user_uuid: UUID, data: bytes) -> None:
|
|
"""Store avatar bytes."""
|
|
path = avatar_path(user_uuid)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
|
|
|
|
def avatar_etag(data: bytes) -> str:
|
|
"""Return a stable ETag value for avatar bytes."""
|
|
return hashlib.sha256(data).hexdigest()[:16]
|