tfaour
acdb929a99
Build and push server image / build-and-push (push) Successful in 50s
Calendar rule/grid lines were light gray, which dithers away to near-invisible on the 6-color e-ink palette -- now black. CalDAV accounts (Nextcloud, Fastmail, iCloud, ...) can now be linked alongside the existing single ICS subscription, since one account can expose several calendars. A frame's Calendar tab now lists calendars per person rather than one opt-in per person: your own row shows every calendar you have available with a full add/remove toggle, while other linked users' rows show only calendars they've included, toggleable off (mute) but not on -- only a calendar's owner can add it to a shared frame. FrameCalendar replaces the old single-boolean UserFrame.calendar_included; existing opt-ins are migrated forward.
725 lines
28 KiB
Python
725 lines
28 KiB
Python
"""HTML page routes: first-run setup, login/logout, user settings, and
|
|
the admin panel. The per-frame pages (Photos/Configuration/Stats) live in
|
|
routers/frame_pages.py.
|
|
|
|
All POSTs here are plain HTML forms, so CSRF rides a hidden form field
|
|
(checked explicitly) rather than the X-CSRF-Token header the JSON API
|
|
uses."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hmac
|
|
import logging
|
|
import time
|
|
|
|
from fastapi import APIRouter, Depends, Form, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import caldav_client, mail
|
|
from ..auth import (
|
|
SESSION_COOKIE,
|
|
SESSION_LIFETIME_S,
|
|
consume_password_reset_token,
|
|
create_password_reset_token,
|
|
create_session,
|
|
current_session,
|
|
current_user,
|
|
destroy_session,
|
|
get_server_settings,
|
|
hash_password,
|
|
require_user_api,
|
|
users_exist,
|
|
verify_password,
|
|
)
|
|
from ..db import get_db
|
|
from ..models import Frame, PasswordResetToken, PendingClaim, User, UserFrame
|
|
from .common import valid_http_url
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
USERNAME_MAX_LEN = 64
|
|
PASSWORD_MIN_LEN = 8
|
|
PENDING_CLAIM_TTL_S = 24 * 3600
|
|
|
|
|
|
def _set_session_cookie(response, cookie_value: str) -> None:
|
|
# No Secure flag: the server itself is plain HTTP by design (TLS is a
|
|
# reverse proxy's job, see README) and a LAN deployment without HTTPS
|
|
# must still be able to log in.
|
|
response.set_cookie(
|
|
SESSION_COOKIE,
|
|
cookie_value,
|
|
max_age=SESSION_LIFETIME_S,
|
|
httponly=True,
|
|
samesite="lax",
|
|
)
|
|
|
|
|
|
def _check_form_csrf(request: Request, db: Session, csrf_token: str) -> None:
|
|
session = current_session(request, db)
|
|
if session is None or not hmac.compare_digest(csrf_token, session.csrf_token):
|
|
raise HTTPException(403, "Missing or invalid CSRF token")
|
|
|
|
|
|
def _normalize_username(username: str) -> str:
|
|
return username.strip().lower()
|
|
|
|
|
|
def _validate_credentials(username: str, password: str) -> str:
|
|
username = _normalize_username(username)
|
|
if not username or len(username) > USERNAME_MAX_LEN:
|
|
raise HTTPException(400, "Invalid username")
|
|
if len(password) < PASSWORD_MIN_LEN:
|
|
raise HTTPException(400, f"Password must be at least {PASSWORD_MIN_LEN} characters")
|
|
return username
|
|
|
|
|
|
@router.get("/setup", response_class=HTMLResponse)
|
|
def setup_page(request: Request, db: Session = Depends(get_db)):
|
|
if users_exist(db):
|
|
return RedirectResponse("/login", status_code=303)
|
|
return templates.TemplateResponse("setup.html", {"request": request, "error": None})
|
|
|
|
|
|
@router.post("/setup")
|
|
def setup_submit(
|
|
request: Request,
|
|
username: str = Form(...),
|
|
display_name: str = Form(""),
|
|
password: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Creates admin #1 -- only ever available while no users exist, so it
|
|
needs no CSRF/session (there is nothing to ride). Links every
|
|
existing frame (i.e. the migrated frame #1) to the new admin, makes
|
|
them its owner + controller, and inherits the migrated Immich creds
|
|
onto their account (that's how env/config.json creds become per-user
|
|
state)."""
|
|
if users_exist(db):
|
|
raise HTTPException(403, "Setup has already been completed")
|
|
username = _validate_credentials(username, password)
|
|
|
|
admin = User(
|
|
username=username,
|
|
display_name=display_name.strip() or username,
|
|
password_hash=hash_password(password),
|
|
is_admin=True,
|
|
created_at=time.time(),
|
|
)
|
|
db.add(admin)
|
|
db.flush()
|
|
|
|
for frame in db.scalars(select(Frame)):
|
|
db.add(UserFrame(user_id=admin.id, frame_id=frame.id))
|
|
if frame.owner_user_id is None:
|
|
frame.owner_user_id = admin.id
|
|
frame.claimed_at = time.time()
|
|
if frame.controlled_by_user_id is None:
|
|
frame.controlled_by_user_id = admin.id
|
|
if not admin.immich_url and frame.immich_url and frame.immich_api_key:
|
|
admin.immich_url = frame.immich_url
|
|
admin.immich_api_key = frame.immich_api_key
|
|
|
|
db.commit()
|
|
logger.info("First-run setup: created admin '%s' and linked %s", username,
|
|
", ".join(f"frame #{f.id}" for f in db.scalars(select(Frame))) or "no frames")
|
|
|
|
cookie_value, _ = create_session(db, admin)
|
|
response = RedirectResponse("/", status_code=303)
|
|
_set_session_cookie(response, cookie_value)
|
|
return response
|
|
|
|
|
|
def _safe_next(next_url: str) -> str:
|
|
"""Same-site relative paths only -- a login redirect target from a
|
|
query param must never become an open redirect."""
|
|
if next_url.startswith("/") and not next_url.startswith("//"):
|
|
return next_url
|
|
return "/"
|
|
|
|
|
|
@router.get("/login", response_class=HTMLResponse)
|
|
def login_page(request: Request, next: str = "", db: Session = Depends(get_db)):
|
|
if not users_exist(db):
|
|
return RedirectResponse("/setup", status_code=303)
|
|
if current_user(request, db) is not None:
|
|
return RedirectResponse(_safe_next(next), status_code=303)
|
|
return templates.TemplateResponse(
|
|
"login.html", {"request": request, "error": None, "next": next}
|
|
)
|
|
|
|
|
|
@router.post("/login")
|
|
def login_submit(
|
|
request: Request,
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
next: str = Form(""),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = db.scalars(
|
|
select(User).where(User.username == _normalize_username(username))
|
|
).first()
|
|
if user is None or not user.password_hash or not verify_password(password, user.password_hash):
|
|
return templates.TemplateResponse(
|
|
"login.html",
|
|
{"request": request, "error": "Wrong username or password.", "next": next},
|
|
status_code=401,
|
|
)
|
|
cookie_value, _ = create_session(db, user)
|
|
response = RedirectResponse(_safe_next(next), status_code=303)
|
|
_set_session_cookie(response, cookie_value)
|
|
return response
|
|
|
|
|
|
@router.post("/logout")
|
|
def logout(request: Request, csrf_token: str = Form(""), db: Session = Depends(get_db)):
|
|
_check_form_csrf(request, db, csrf_token)
|
|
destroy_session(db, request)
|
|
response = RedirectResponse("/login", status_code=303)
|
|
response.delete_cookie(SESSION_COOKIE)
|
|
return response
|
|
|
|
|
|
@router.get("/forgot-password", response_class=HTMLResponse)
|
|
def forgot_password_page(request: Request):
|
|
return templates.TemplateResponse(
|
|
"forgot_password.html", {"request": request, "sent": False, "error": None}
|
|
)
|
|
|
|
|
|
@router.post("/forgot-password", response_class=HTMLResponse)
|
|
def forgot_password_submit(
|
|
request: Request, email: str = Form(...), db: Session = Depends(get_db)
|
|
):
|
|
"""Always shows the same "check your email" result regardless of
|
|
whether the address matches an account -- otherwise this endpoint
|
|
would let anyone enumerate registered emails. Silently no-ops (same
|
|
response) if SMTP isn't configured or the user has no email set."""
|
|
email = email.strip().lower()
|
|
user = db.scalars(select(User).where(User.email != "").where(User.email == email)).first()
|
|
if user is not None:
|
|
token = create_password_reset_token(db, user)
|
|
reset_url = str(request.base_url).rstrip("/") + f"/reset-password/{token}"
|
|
settings = get_server_settings(db)
|
|
mail.send_email(
|
|
settings, user.email, "Reset your ESPresso Frame password",
|
|
f"Someone (hopefully you) asked to reset the password for '{user.username}'.\n\n"
|
|
f"Reset it here (valid for 1 hour): {reset_url}\n\n"
|
|
"If you didn't request this, ignore this email.",
|
|
)
|
|
return templates.TemplateResponse(
|
|
"forgot_password.html", {"request": request, "sent": True, "error": None}
|
|
)
|
|
|
|
|
|
@router.get("/reset-password/{token}", response_class=HTMLResponse)
|
|
def reset_password_page(token: str, request: Request, db: Session = Depends(get_db)):
|
|
row = db.get(PasswordResetToken, token)
|
|
valid = row is not None and row.expires_at > time.time()
|
|
return templates.TemplateResponse(
|
|
"reset_password.html", {"request": request, "token": token, "valid": valid, "error": None}
|
|
)
|
|
|
|
|
|
@router.post("/reset-password/{token}", response_class=HTMLResponse)
|
|
def reset_password_submit(
|
|
token: str, request: Request, password: str = Form(...), db: Session = Depends(get_db)
|
|
):
|
|
if len(password) < PASSWORD_MIN_LEN:
|
|
return templates.TemplateResponse(
|
|
"reset_password.html",
|
|
{"request": request, "token": token, "valid": True,
|
|
"error": f"Password must be at least {PASSWORD_MIN_LEN} characters."},
|
|
)
|
|
user = consume_password_reset_token(db, token)
|
|
if user is None:
|
|
return templates.TemplateResponse(
|
|
"reset_password.html",
|
|
{"request": request, "token": token, "valid": False, "error": None},
|
|
)
|
|
user.password_hash = hash_password(password)
|
|
db.commit()
|
|
logger.info("Password reset via email link for user '%s'", user.username)
|
|
cookie_value, _ = create_session(db, user)
|
|
response = RedirectResponse("/", status_code=303)
|
|
_set_session_cookie(response, cookie_value)
|
|
return response
|
|
|
|
|
|
def _normalize_device_id(device_id: str) -> str:
|
|
device_id = device_id.strip().lower()
|
|
if len(device_id) != 12 or any(c not in "0123456789abcdef" for c in device_id):
|
|
raise HTTPException(400, "Invalid device id")
|
|
return device_id
|
|
|
|
|
|
def _attempt_claim(db: Session, user: User, device_id: str) -> str:
|
|
"""Claims the frame for `user` if it has registered, else records a
|
|
pending claim the frame's first check-in will attach (see
|
|
auth._register_frame). Returns "claimed" or "pending"."""
|
|
frame = db.scalars(select(Frame).where(Frame.device_id == device_id)).first()
|
|
now = time.time()
|
|
if frame is not None:
|
|
if frame.owner_user_id is not None:
|
|
raise HTTPException(409, "That frame is already claimed")
|
|
frame.owner_user_id = user.id
|
|
frame.claimed_at = now
|
|
if frame.controlled_by_user_id is None:
|
|
frame.controlled_by_user_id = user.id
|
|
if db.get(UserFrame, (user.id, frame.id)) is None:
|
|
db.add(UserFrame(user_id=user.id, frame_id=frame.id))
|
|
pending = db.get(PendingClaim, device_id)
|
|
if pending is not None:
|
|
db.delete(pending)
|
|
db.commit()
|
|
logger.info("User '%s' claimed frame #%d (%s)", user.username, frame.id, device_id)
|
|
return "claimed"
|
|
|
|
pending = db.get(PendingClaim, device_id)
|
|
if pending is None:
|
|
pending = PendingClaim(device_id=device_id, user_id=user.id, created_at=now,
|
|
expires_at=now + PENDING_CLAIM_TTL_S)
|
|
db.add(pending)
|
|
else:
|
|
pending.user_id = user.id
|
|
pending.expires_at = now + PENDING_CLAIM_TTL_S
|
|
db.commit()
|
|
logger.info("User '%s' filed a pending claim for device %s", user.username, device_id)
|
|
return "pending"
|
|
|
|
|
|
def _render_claim(request: Request, db: Session, device_id: str, error: str | None = None):
|
|
user = current_user(request, db)
|
|
session = current_session(request, db) if user else None
|
|
frame = db.scalars(select(Frame).where(Frame.device_id == device_id)).first()
|
|
if frame is None:
|
|
status = "unregistered"
|
|
pending = db.get(PendingClaim, device_id)
|
|
pending_yours = bool(pending and user and pending.user_id == user.id
|
|
and pending.expires_at > time.time())
|
|
elif frame.owner_user_id is None:
|
|
status, pending_yours = "unclaimed", False
|
|
elif user is not None and (
|
|
frame.owner_user_id == user.id or db.get(UserFrame, (user.id, frame.id)) is not None
|
|
):
|
|
status, pending_yours = "claimed_yours", False
|
|
else:
|
|
status, pending_yours = "claimed", False
|
|
return templates.TemplateResponse(
|
|
"claim.html",
|
|
{
|
|
"request": request,
|
|
"device_id": device_id,
|
|
"status": status,
|
|
"pending_yours": pending_yours,
|
|
"user": user,
|
|
"csrf_token": session.csrf_token if session else None,
|
|
"error": error,
|
|
},
|
|
)
|
|
|
|
|
|
@router.get("/claim", response_class=HTMLResponse)
|
|
def claim_page(request: Request, device_id: str = "", db: Session = Depends(get_db)):
|
|
"""Where the captive portal's post-provisioning redirect lands. Also
|
|
the enrollment gate: a valid device id is what entitles a stranger to
|
|
create an account (signup form on this page); everyone else gets
|
|
enrolled by the admin."""
|
|
device_id = _normalize_device_id(device_id)
|
|
return _render_claim(request, db, device_id)
|
|
|
|
|
|
@router.post("/claim")
|
|
def claim_submit(
|
|
request: Request,
|
|
device_id: str = Form(...),
|
|
csrf_token: str = Form(""),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
device_id = _normalize_device_id(device_id)
|
|
user = current_user(request, db)
|
|
if user is None:
|
|
return RedirectResponse(f"/login?next=/claim%3Fdevice_id%3D{device_id}", status_code=303)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
try:
|
|
_attempt_claim(db, user, device_id)
|
|
except HTTPException as e:
|
|
if e.status_code == 409:
|
|
return _render_claim(request, db, device_id, error=e.detail)
|
|
raise
|
|
return RedirectResponse(f"/claim?device_id={device_id}", status_code=303)
|
|
|
|
|
|
@router.post("/claim/signup")
|
|
def claim_signup(
|
|
request: Request,
|
|
device_id: str = Form(...),
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Account creation, gated on a plausible frame claim: the device id
|
|
must belong to a frame that is unclaimed (or not yet registered --
|
|
the user beat the device here after provisioning). A fabricated id
|
|
can create an orphan account whose pending claim expires in 24h --
|
|
accepted at household scale, and visible in /admin."""
|
|
device_id = _normalize_device_id(device_id)
|
|
frame = db.scalars(select(Frame).where(Frame.device_id == device_id)).first()
|
|
if frame is not None and frame.owner_user_id is not None:
|
|
return _render_claim(request, db, device_id,
|
|
error="That frame is already claimed -- log in instead.")
|
|
username = _validate_credentials(username, password)
|
|
if db.scalars(select(User).where(User.username == username)).first() is not None:
|
|
return _render_claim(request, db, device_id,
|
|
error=f"Username '{username}' is taken -- log in instead?")
|
|
|
|
user = User(
|
|
username=username,
|
|
display_name=username,
|
|
password_hash=hash_password(password),
|
|
is_admin=False,
|
|
created_at=time.time(),
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
logger.info("User '%s' signed up via claim gate for device %s", username, device_id)
|
|
|
|
_attempt_claim(db, user, device_id)
|
|
cookie_value, _ = create_session(db, user)
|
|
response = RedirectResponse(f"/claim?device_id={device_id}", status_code=303)
|
|
_set_session_cookie(response, cookie_value)
|
|
return response
|
|
|
|
|
|
def _settings_context(request: Request, db: Session, user, saved: bool, error: str | None) -> dict:
|
|
from .common import shell_context
|
|
|
|
ctx = shell_context(request, db, user, active_nav="settings")
|
|
ctx.update({"saved": saved, "error": error})
|
|
return ctx
|
|
|
|
|
|
@router.get("/settings", response_class=HTMLResponse)
|
|
def settings_page(request: Request, db: Session = Depends(get_db)):
|
|
user = current_user(request, db)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
return templates.TemplateResponse(
|
|
"settings.html", _settings_context(request, db, user, saved=False, error=None)
|
|
)
|
|
|
|
|
|
@router.post("/settings", response_class=HTMLResponse)
|
|
def settings_submit(
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
display_name: str = Form(""),
|
|
email: str = Form(""),
|
|
immich_url: str = Form(""),
|
|
immich_api_key: str = Form(""),
|
|
calendar_ics_url: str = Form(""),
|
|
calendar_caldav_url: str = Form(""),
|
|
calendar_caldav_username: str = Form(""),
|
|
calendar_caldav_password: str = Form(""),
|
|
current_password: str = Form(""),
|
|
new_password: str = Form(""),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
user = current_user(request, db)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
|
|
error = None
|
|
user.display_name = display_name.strip() or user.username
|
|
user.email = email.strip().lower()
|
|
user.immich_url = immich_url.strip()
|
|
# Blank API key field = keep the existing one (it's never echoed back
|
|
# into the form -- a secret that round-trips through HTML is a secret
|
|
# in every browser's autofill store).
|
|
if immich_api_key.strip():
|
|
user.immich_api_key = immich_api_key.strip()
|
|
|
|
# Unlike the API key, this isn't a secret -- it round-trips visibly in
|
|
# the form, so blank means an explicit clear (there needs to be some
|
|
# way to actually remove a linked calendar), not "keep existing".
|
|
stripped_ics = calendar_ics_url.strip()
|
|
if stripped_ics and not valid_http_url(stripped_ics):
|
|
error = "Calendar URL must be a plain http:// or https:// URL."
|
|
else:
|
|
user.calendar_ics_url = stripped_ics
|
|
|
|
stripped_caldav_url = calendar_caldav_url.strip()
|
|
if stripped_caldav_url and not valid_http_url(stripped_caldav_url):
|
|
error = "CalDAV URL must be a plain http:// or https:// URL."
|
|
else:
|
|
if stripped_caldav_url != user.calendar_caldav_url:
|
|
# Server (and likely account) changed -- last discovery no
|
|
# longer describes what's actually there.
|
|
user.calendar_caldav_calendars = None
|
|
user.calendar_caldav_checked_at = 0.0
|
|
user.calendar_caldav_url = stripped_caldav_url
|
|
user.calendar_caldav_username = calendar_caldav_username.strip()
|
|
# Blank password field = keep the existing one, same idiom as the
|
|
# Immich API key -- a secret that round-trips through HTML is a
|
|
# secret in every browser's autofill store.
|
|
if calendar_caldav_password.strip():
|
|
user.calendar_caldav_password = calendar_caldav_password.strip()
|
|
|
|
if new_password:
|
|
if not user.password_hash or not verify_password(current_password, user.password_hash):
|
|
error = "Current password is wrong -- password not changed."
|
|
elif len(new_password) < PASSWORD_MIN_LEN:
|
|
error = f"New password must be at least {PASSWORD_MIN_LEN} characters."
|
|
else:
|
|
user.password_hash = hash_password(new_password)
|
|
|
|
db.commit()
|
|
return templates.TemplateResponse(
|
|
"settings.html", _settings_context(request, db, user, saved=error is None, error=error)
|
|
)
|
|
|
|
|
|
@router.post("/api/settings/caldav-discover")
|
|
def api_caldav_discover(request: Request, db: Session = Depends(get_db)):
|
|
"""Lists the calendars in the CalDAV account already saved on this
|
|
user's Settings (not whatever's currently typed in the form but not
|
|
yet saved -- same idiom as /api/frames/{id}/albums using the frame's
|
|
already-saved Immich creds). Caches the result on the user row so
|
|
every frame's Calendar tab can offer it without a live round-trip."""
|
|
user = require_user_api(request, db)
|
|
if not user.calendar_caldav_url or not user.calendar_caldav_username:
|
|
raise HTTPException(400, "Save a CalDAV URL and username first")
|
|
try:
|
|
calendars = caldav_client.discover_calendars(
|
|
user.calendar_caldav_url, user.calendar_caldav_username, user.calendar_caldav_password
|
|
)
|
|
except caldav_client.CalDavError as e:
|
|
raise HTTPException(502, f"Could not discover calendars: {e}") from e
|
|
user.calendar_caldav_calendars = calendars
|
|
user.calendar_caldav_checked_at = time.time()
|
|
db.commit()
|
|
return calendars
|
|
|
|
|
|
def _require_admin_page(request: Request, db: Session) -> User:
|
|
user = current_user(request, db)
|
|
if user is None or not user.is_admin:
|
|
raise HTTPException(403, "Admin only")
|
|
return user
|
|
|
|
|
|
def _render_admin(request: Request, db: Session, admin: User, notice: str | None = None,
|
|
error: str | None = None) -> HTMLResponse:
|
|
from .common import shell_context
|
|
|
|
users = list(db.scalars(select(User).order_by(User.id)))
|
|
frames = list(db.scalars(select(Frame).order_by(Frame.id)))
|
|
links = list(db.scalars(select(UserFrame)))
|
|
links_by_frame: dict[int, list[User]] = {}
|
|
users_by_id = {u.id: u for u in users}
|
|
for link in links:
|
|
links_by_frame.setdefault(link.frame_id, []).append(users_by_id[link.user_id])
|
|
ctx = shell_context(request, db, admin, active_nav="admin")
|
|
ctx.update({
|
|
"users": users,
|
|
"frames": frames,
|
|
"links_by_frame": links_by_frame,
|
|
"smtp": get_server_settings(db),
|
|
"notice": notice,
|
|
"error": error,
|
|
})
|
|
return templates.TemplateResponse("admin.html", ctx)
|
|
|
|
|
|
@router.get("/admin", response_class=HTMLResponse)
|
|
def admin_page(request: Request, db: Session = Depends(get_db)):
|
|
user = current_user(request, db)
|
|
if user is None:
|
|
return RedirectResponse("/login", status_code=303)
|
|
if not user.is_admin:
|
|
raise HTTPException(403, "Admin only")
|
|
return _render_admin(request, db, user)
|
|
|
|
|
|
@router.post("/admin/users", response_class=HTMLResponse)
|
|
def admin_create_user(
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
username: str = Form(...),
|
|
password: str = Form(...),
|
|
is_admin: bool = Form(False),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
username = _validate_credentials(username, password)
|
|
if db.scalars(select(User).where(User.username == username)).first() is not None:
|
|
return _render_admin(request, db, admin, error=f"Username '{username}' already exists.")
|
|
db.add(User(
|
|
username=username,
|
|
display_name=username,
|
|
password_hash=hash_password(password),
|
|
is_admin=is_admin,
|
|
created_at=time.time(),
|
|
))
|
|
db.commit()
|
|
return _render_admin(request, db, admin, notice=f"User '{username}' created.")
|
|
|
|
|
|
@router.post("/admin/users/{user_id}/reset-password", response_class=HTMLResponse)
|
|
def admin_reset_password(
|
|
user_id: int,
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
password: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
target = db.get(User, user_id)
|
|
if target is None:
|
|
return _render_admin(request, db, admin, error="No such user.")
|
|
if len(password) < PASSWORD_MIN_LEN:
|
|
return _render_admin(request, db, admin,
|
|
error=f"Password must be at least {PASSWORD_MIN_LEN} characters.")
|
|
target.password_hash = hash_password(password)
|
|
db.commit()
|
|
return _render_admin(request, db, admin, notice=f"Password reset for '{target.username}'.")
|
|
|
|
|
|
@router.post("/admin/users/{user_id}/delete", response_class=HTMLResponse)
|
|
def admin_delete_user(
|
|
user_id: int,
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
if user_id == admin.id:
|
|
return _render_admin(request, db, admin, error="You can't delete your own account.")
|
|
target = db.get(User, user_id)
|
|
if target is None:
|
|
return _render_admin(request, db, admin, error="No such user.")
|
|
name = target.username
|
|
db.delete(target) # sessions/links cascade; frames.owner goes NULL
|
|
db.commit()
|
|
return _render_admin(request, db, admin, notice=f"User '{name}' deleted.")
|
|
|
|
|
|
@router.post("/admin/frames/{frame_id}/link-user", response_class=HTMLResponse)
|
|
def admin_link_user(
|
|
frame_id: int,
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
username: str = Form(...),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
frame = db.get(Frame, frame_id)
|
|
target = db.scalars(select(User).where(User.username == _normalize_username(username))).first()
|
|
if frame is None or target is None:
|
|
return _render_admin(request, db, admin, error="No such frame or user.")
|
|
if db.get(UserFrame, (target.id, frame_id)) is not None:
|
|
return _render_admin(request, db, admin, error=f"'{target.username}' is already linked.")
|
|
db.add(UserFrame(user_id=target.id, frame_id=frame_id))
|
|
if frame.owner_user_id is None:
|
|
# Linking to an unclaimed frame claims it -- the admin flow for
|
|
# adopting a frame that self-registered without a pending claim.
|
|
frame.owner_user_id = target.id
|
|
frame.claimed_at = time.time()
|
|
db.commit()
|
|
return _render_admin(request, db, admin,
|
|
notice=f"Linked '{target.username}' to frame #{frame_id}.")
|
|
|
|
|
|
@router.post("/admin/frames/{frame_id}/end-legacy", response_class=HTMLResponse)
|
|
def admin_end_legacy(
|
|
frame_id: int,
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Closes the legacy-token migration window once the device is
|
|
confirmed on per-frame auth (device_token_ack + recent last_seen in
|
|
the frames table below)."""
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
frame = db.get(Frame, frame_id)
|
|
if frame is None:
|
|
return _render_admin(request, db, admin, error="No such frame.")
|
|
frame.legacy_token_enabled = False
|
|
db.commit()
|
|
return _render_admin(request, db, admin, notice=f"Legacy token disabled for frame #{frame_id}.")
|
|
|
|
|
|
@router.post("/admin/smtp", response_class=HTMLResponse)
|
|
def admin_smtp_save(
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
smtp_host: str = Form(""),
|
|
smtp_port: int = Form(587),
|
|
smtp_username: str = Form(""),
|
|
smtp_password: str = Form(""),
|
|
smtp_from_address: str = Form(""),
|
|
smtp_encryption: str = Form("starttls"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
"""Saves the SMTP config used for password-reset emails and battery-
|
|
threshold alerts. Blank password = keep the existing one, same
|
|
round-trip-avoidance as the Immich API key field in /settings."""
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
settings = get_server_settings(db)
|
|
settings.smtp_host = smtp_host.strip()
|
|
settings.smtp_port = max(1, min(65535, smtp_port))
|
|
settings.smtp_username = smtp_username.strip()
|
|
if smtp_password.strip():
|
|
settings.smtp_password = smtp_password.strip()
|
|
settings.smtp_from_address = smtp_from_address.strip()
|
|
settings.smtp_encryption = smtp_encryption if smtp_encryption in ("none", "starttls", "ssl") else "starttls"
|
|
db.commit()
|
|
return _render_admin(request, db, admin, notice="SMTP settings saved.")
|
|
|
|
|
|
@router.post("/admin/smtp/test", response_class=HTMLResponse)
|
|
def admin_smtp_test(request: Request, csrf_token: str = Form(""), db: Session = Depends(get_db)):
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
if not admin.email:
|
|
return _render_admin(request, db, admin, error="Set an email on your own account (Settings) to test SMTP.")
|
|
settings = get_server_settings(db)
|
|
ok = mail.send_email(
|
|
settings, admin.email, "ESPresso Frame test email",
|
|
"If you're reading this, SMTP is configured correctly.",
|
|
)
|
|
if ok:
|
|
return _render_admin(request, db, admin, notice=f"Test email sent to {admin.email}.")
|
|
return _render_admin(request, db, admin, error="Failed to send -- check the SMTP settings and server logs.")
|
|
|
|
|
|
@router.post("/admin/frames/{frame_id}/delete", response_class=HTMLResponse)
|
|
def admin_delete_frame(
|
|
frame_id: int,
|
|
request: Request,
|
|
csrf_token: str = Form(""),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
admin = _require_admin_page(request, db)
|
|
_check_form_csrf(request, db, csrf_token)
|
|
frame = db.get(Frame, frame_id)
|
|
if frame is None:
|
|
return _render_admin(request, db, admin, error="No such frame.")
|
|
db.delete(frame) # links/battery log cascade
|
|
db.commit()
|
|
return _render_admin(request, db, admin, notice=f"Frame #{frame_id} deleted.")
|