"""Sign-in, sign-out, lock, unlock and the Me payload.""" from __future__ import annotations from typing import Any from fastapi import APIRouter, Depends, Request, Response from pydantic import BaseModel, Field from .. import audit, events from ..auth import devices, signin, stepup from ..auth import directory as userdir from ..auth import pin as pins from .deps import ( Db, Principal, can_register_device, clear_session_cookie, current_device, current_principal, current_principal_any, directory_of, get_db, ip_of, resolve, session_key, set_session_cookie, settings_of, user_agent_of, ) from .errors import ApiError router = APIRouter(prefix="/api/auth", tags=["auth"]) class PasswordSignin(BaseModel): login: str = Field(max_length=256) password: str = Field(max_length=1024) class PinSignin(BaseModel): emp_no: str = Field(max_length=64) pin: str = Field(max_length=64) class Visibility(BaseModel): hidden: bool class Unlock(BaseModel): factor: str = Field(max_length=16) secret: str = Field(max_length=1024) def me_payload(conn: Any, p: Principal) -> dict[str, Any]: s = p.session return { "user": userdir.public_user(p.user), "roles": p.roles, "screens": p.screens, "groups": p.groups, "session": { "id": str(s["id"]), "kind": p.kind, "created_at": s["created_at"], "expires_at": s["expires_at"], "locked": s.get("locked_since") is not None, "locked_since": s.get("locked_since"), "device": devices.public(p.device), }, "pin": pins.state(conn, p.user_id).public(), "can_register_device": can_register_device(p), } def _me_after_signin(request: Request, response: Response, d: Db, result: signin.SigninResult) -> dict[str, Any]: set_session_cookie(request, response, result.token) # Resolve the new session the normal way so the payload is exactly what GET /me returns p = resolve(request, d, allow_locked=True, token=result.token) return me_payload(d.conn, p) @router.get("/context") def context(request: Request, d: Db = Depends(get_db)) -> dict[str, Any]: # noqa: B008 s = settings_of(request) device = None if request.cookies.get("ps_device"): device = devices.public(current_device(request, d.conn)) return { "device": device, "stack": s.stack, "insecure_ldap": s.ldap_allow_insecure, "pin_digits": {"min": s.pin_min_digits, "max": s.pin_max_digits}, } @router.post("/signin/password") def signin_password( body: PasswordSignin, request: Request, response: Response, d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: key = session_key(request) result = signin.password_signin( d.conn, directory_of(request), settings_of(request), key, login=body.login, password=body.password, ip=ip_of(request), user_agent=user_agent_of(request), ) return _me_after_signin(request, response, d, result) @router.post("/signin/pin") def signin_pin( body: PinSignin, request: Request, response: Response, d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: key = session_key(request) device = current_device(request, d.conn) if device is None: raise ApiError(403, "device_required", "PIN sign-in works only on a registered device.") result = signin.pin_signin( d.conn, directory_of(request), settings_of(request), key, device=device, emp_no=body.emp_no, pin=body.pin, ip=ip_of(request), user_agent=user_agent_of(request), ) return _me_after_signin(request, response, d, result) @router.get("/me") def me(p: Principal = Depends(current_principal_any), d: Db = Depends(get_db)) -> dict[str, Any]: # noqa: B008 return me_payload(d.conn, p) @router.post("/signout", status_code=204) def signout( request: Request, p: Principal = Depends(current_principal_any), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> Response: from ..auth import sessions conn = d.conn with conn.transaction(): if sessions.revoke(conn, p.session_id, "signout", notify=False): audit.append( conn, action="auth.signout", user_id=p.user_id, session_id=p.session_id, role_snapshot=p.role_snapshot, ip=p.ip, ) response = Response(status_code=204) clear_session_cookie(request, response) return response def _lock(d: Db, p: Principal, source: str) -> None: conn = d.conn with conn.transaction(): row = conn.execute( "UPDATE sessions SET locked_since = now() WHERE id = %s AND locked_since IS NULL RETURNING id", (p.session_id,), ).fetchone() if row is not None: audit.append( conn, action="auth.lock", user_id=p.user_id, session_id=p.session_id, role_snapshot=p.role_snapshot, detail={"source": source}, ip=p.ip, ) events.publish(conn, {"event": "session", "session_id": str(p.session_id), "action": "locked"}) @router.post("/lock", status_code=204) def lock(p: Principal = Depends(current_principal), d: Db = Depends(get_db)) -> Response: # noqa: B008 _lock(d, p, "manual") return Response(status_code=204) @router.post("/visibility", status_code=204) def visibility( body: Visibility, p: Principal = Depends(current_principal_any), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> Response: # Hidden is only recorded. The lock comes from `resolve` once the report is older than # SESSION_HIDDEN_LOCK_SECONDS (the old build's "tab hidden over 1 min"), so a reload or an # in-app navigation, which also fire a hidden report, never lock the screen. if body.hidden: d.conn.execute( "UPDATE sessions SET hidden_since = coalesce(hidden_since, now()) WHERE id = %s", (p.session_id,) ) return Response(status_code=204) @router.post("/unlock") def unlock( body: Unlock, request: Request, p: Principal = Depends(current_principal_any), # noqa: B008 d: Db = Depends(get_db), # noqa: B008 ) -> dict[str, Any]: conn = d.conn if p.locked: stepup.verify( conn, directory_of(request), settings_of(request), p.actor, factor=body.factor, secret=body.secret, kind="unlock", ip=p.ip, ) with conn.transaction(): conn.execute("UPDATE sessions SET locked_since = NULL, last_seen_at = now() WHERE id = %s", (p.session_id,)) audit.append( conn, action="auth.unlock", user_id=p.user_id, session_id=p.session_id, role_snapshot=p.role_snapshot, detail={"factor": body.factor}, ip=p.ip, ) events.publish(conn, {"event": "session", "session_id": str(p.session_id), "action": "unlocked"}) p.session["locked_since"] = None return me_payload(conn, p)