devplacepy/devplacepy/routers/devrant/auth.py

# retoor <retoor@molodetz.nl>

import logging
import re

from fastapi import APIRouter, Request
from fastapi.responses import Response

from devplacepy.cache import TTLCache
from devplacepy.config import SECONDS_PER_DAY
from devplacepy.database import get_table, get_setting, is_account_active
from devplacepy.utils import verify_password_async, register_account_async
from devplacepy.services.audit import record as audit
from devplacepy.services.devrant.params import merge_params
from devplacepy.services.devrant.tokens import issue_token, resolve_user, revoke_all
from devplacepy.services.devrant.profile import build_profile
from devplacepy.services.devrant.ids import user_by_id
from devplacepy.services.devrant.avatar import render_png
from devplacepy.routers.devrant._shared import dr_ok, dr_error, resolve_actor, unauthorized

logger = logging.getLogger(__name__)
router = APIRouter()

USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9_-]{3,32}$")
PASSWORD_MIN = 6

_avatar_cache = TTLCache(ttl=SECONDS_PER_DAY, max_size=4096)


@router.post("/users/auth-token")
async def auth_token(request: Request):
    params = await merge_params(request)
    identifier = (params.get("username") or "").strip()
    password = params.get("password") or ""
    invalid = "Invalid login credentials entered. Please try again."
    if not identifier or not password:
        return dr_error(invalid, 400)
    users = get_table("users")
    user = users.find_one(username=identifier) or users.find_one(
        email=identifier.lower()
    )
    if (
        not user
        or not is_account_active(user)
        or not await verify_password_async(password, user["password_hash"])
    ):
        audit.record(
            request,
            "auth.login.failure",
            user=None,
            actor_kind="guest",
            result="failure",
            origin="devrant",
            metadata={"identifier": identifier},
            summary=f"failed devrant login for {identifier}",
        )
        return dr_error(invalid, 400)
    token = issue_token(user)
    logger.info("devrant login for %s", user["username"])
    audit.record(
        request,
        "auth.login.success",
        user=user,
        target_type="user",
        target_uid=user["uid"],
        target_label=user["username"],
        origin="devrant",
        summary=f"user {user['username']} logged in via devrant",
        links=[audit.target("user", user["uid"], user["username"])],
    )
    return dr_ok(auth_token=token)


@router.post("/users")
async def register(request: Request):
    params = await merge_params(request)
    username = (params.get("username") or "").strip()
    email = (params.get("email") or "").strip().lower()
    password = params.get("password") or ""
    if get_setting("registration_open", "1") != "1":
        return dr_error("Registration is closed.", error_field="username")
    if not USERNAME_PATTERN.match(username):
        return dr_error(
            "Username must be 3-32 letters, numbers, hyphens or underscores.",
            error_field="username",
        )
    if "@" not in email:
        return dr_error("Please enter a valid email address.", error_field="email")
    if len(password) < PASSWORD_MIN:
        return dr_error(
            "Password must be at least 6 characters.", error_field="password"
        )
    users = get_table("users")
    if users.find_one(username=username):
        return dr_error("That username is already taken.", error_field="username")
    if users.find_one(email=email):
        return dr_error("That email is already registered.", error_field="email")
    uid, role, is_first = await register_account_async(username, email, password)
    logger.info("devrant registration for %s", username)
    audit.record(
        request,
        "auth.signup",
        user={"uid": uid, "username": username, "role": role},
        target_type="user",
        target_uid=uid,
        target_label=username,
        new_value=role,
        origin="devrant",
        metadata={"first_user": is_first},
        summary=f"user {username} registered via devrant",
        links=[audit.target("user", uid, username)],
    )
    user = users.find_one(uid=uid)
    return dr_ok(auth_token=issue_token(user))


@router.get("/get-user-id")
async def get_user_id(request: Request):
    params = await merge_params(request)
    username = (params.get("username") or "").strip()
    user = get_table("users").find_one(username=username) if username else None
    if not user:
        return dr_error("User not found.")
    return dr_ok(user_id=int(user["id"]))


@router.get("/users/{user_id}")
async def profile(request: Request, user_id: str):
    params = await merge_params(request)
    user = user_by_id(user_id)
    if not user:
        return dr_error("User not found.")
    viewer = resolve_actor(request, params)
    return dr_ok(profile=build_profile(user, viewer))


@router.post("/users/me/edit-profile")
async def edit_profile(request: Request):
    params = await merge_params(request)
    user = resolve_actor(request, params)
    if not user:
        return unauthorized()
    updates = {"uid": user["uid"]}
    field_map = {
        "profile_about": "bio",
        "profile_location": "location",
        "profile_github": "git_link",
        "profile_website": "website",
    }
    for source, column in field_map.items():
        if source in params:
            updates[column] = (params.get(source) or "")[:500]
    if len(updates) > 1:
        get_table("users").update(updates, ["uid"])
        audit.record(
            request,
            "profile.update",
            user=user,
            target_type="user",
            target_uid=user["uid"],
            target_label=user["username"],
            origin="devrant",
            summary=f"{user['username']} edited profile via devrant",
            links=[audit.target("user", user["uid"], user["username"])],
        )
    return dr_ok()


@router.post("/users/forgot-password")
async def forgot_password(request: Request):
    await merge_params(request)
    return dr_ok()


@router.post("/users/me/mark-news-read")
async def mark_news_read(request: Request):
    await merge_params(request)
    return dr_ok()


@router.post("/users/me/resend-confirm")
async def resend_confirm(request: Request):
    await merge_params(request)
    return dr_ok()


@router.delete("/users/me")
async def delete_account(request: Request):
    params = await merge_params(request)
    user = resolve_user(params)
    if not user:
        return unauthorized()
    from devplacepy.services.moderation import deletion

    username = user["username"]
    revoke_all(user["uid"])
    result = deletion.delete_account(user)
    if result is None:
        return dr_error("This account is already being deleted.")
    logger.info("devrant account deleted for %s", username)
    audit.record(
        request,
        "account.delete.request",
        user=user,
        target_type="user",
        target_uid=user["uid"],
        target_label=username,
        origin="devrant",
        metadata={
            "stamp": result["stamp"],
            "rows": result["rows"],
            "grace_hours": result["grace_hours"],
        },
        summary=f"{username} deleted account via devrant",
        links=[audit.target("user", user["uid"], username)],
    )
    return dr_ok()


@router.get("/avatars/u/{seed}.png")
async def avatar_image(request: Request, seed: str, size: int = 128):
    size = max(16, min(512, size))
    cache_key = f"{seed}:{size}"
    png = _avatar_cache.get(cache_key)
    if png is None:
        png = render_png(seed, size)
        _avatar_cache.set(cache_key, png)
    headers = {"Cache-Control": f"public, max-age={SECONDS_PER_DAY}, immutable"}
    return Response(content=png, media_type="image/png", headers=headers)