inkwell/src/thoughtsync/auth.py

from __future__ import annotations

import functools
import logging
import uuid
from datetime import datetime, timezone

from quart import Blueprint, g, jsonify, request, session
from sqlalchemy import func, select

from .common import iso
from .db import session_scope
from .models.device_token import DeviceToken
from .models.user import User
from .proxy import client_address
from .ratelimit import (
    register_by_address,
    sign_in_by_account,
    sign_in_by_address,
)
from .security import dummy_verify, generate_token, hash_password, hash_token, verify_password
from .settings import get_setting, set_settings

bp = Blueprint("auth", __name__, url_prefix="/api/auth")

# Every credential event goes to the app log — there is no audit TABLE yet (see task
# 2939), and until there is, `docker compose logs` is the only way to know whether
# anyone is knocking. That matters most in exactly the window this was written for: a
# freshly-exposed instance.
#
# The attempted email is included deliberately. It is the operator's own server, and
# "somebody failed a login" without saying against WHICH account tells you nothing you
# can act on. Passwords, obviously, never appear.
logger = logging.getLogger(__name__)

SESSION_KEY = "user_id"
MIN_PASSWORD_LEN = 8
DEVICE_NAME_CAP = 100


def _serialize_user(user: User) -> dict:
    return {
        "id": str(user.id),
        "email": user.email,
        "display_name": user.display_name,
        "email_verified": user.email_verified,
        "is_admin": user.is_admin,
    }


def _session_user_id() -> uuid.UUID | None:
    raw = session.get(SESSION_KEY)
    if not raw:
        return None
    try:
        return uuid.UUID(raw)
    except (ValueError, TypeError):
        session.pop(SESSION_KEY, None)
        return None


def _bearer_token() -> str | None:
    """Extract a `Authorization: Bearer <token>` device token, if present."""
    header = request.headers.get("Authorization", "")
    if header.startswith("Bearer "):
        return header[7:].strip() or None
    return None


async def _user_id_from_bearer() -> uuid.UUID | None:
    """Resolve a device bearer token to its owner, refreshing last_used_at. Native
    clients (Tauri/Android) authenticate sync this way instead of a session cookie."""
    token = _bearer_token()
    if not token:
        return None
    async with session_scope() as db:
        row = await db.scalar(select(DeviceToken).where(DeviceToken.token_hash == hash_token(token)))
        if row is None:
            return None
        # Cheap liveness stamp; sync calls are user-initiated/periodic, not per-keystroke.
        row.last_used_at = datetime.now(timezone.utc)
        await db.commit()
        return row.user_id


def login_required(fn):
    """Guard: 401 unless authenticated. Accepts a web session cookie OR a device
    bearer token (native clients). Sets g.user_id for the view. The session path
    stays DB-free (fast); only bearer auth does a token lookup."""

    @functools.wraps(fn)
    async def wrapper(*args, **kwargs):
        uid = _session_user_id()
        if uid is None:
            uid = await _user_id_from_bearer()
        if uid is None:
            return jsonify({"error": "authentication required"}), 401
        g.user_id = uid
        return await fn(*args, **kwargs)

    return wrapper


def require_admin(fn):
    """Guard: 401 unauthenticated, 403 non-admin. Checks is_admin live from the DB
    so a demoted admin loses access immediately."""

    @functools.wraps(fn)
    async def wrapper(*args, **kwargs):
        uid = _session_user_id()
        if uid is None:
            return jsonify({"error": "authentication required"}), 401
        async with session_scope() as db:
            user = await db.get(User, uid)
            if user is None:
                session.pop(SESSION_KEY, None)
                return jsonify({"error": "authentication required"}), 401
            if not user.is_admin:
                return jsonify({"error": "admin access required"}), 403
        g.user_id = uid
        return await fn(*args, **kwargs)

    return wrapper


def _throttled(retry_after: int):
    """The 429 every throttled credential route returns.

    Deliberately says nothing about WHICH limit was hit or how many attempts are
    left — that would tell someone probing whether the email they guessed exists.
    `Retry-After` is standard and is the one thing a legitimate client (or person)
    genuinely needs.
    """
    logger.warning("throttled credential attempt from=%s retry_after=%ss", client_address(), retry_after)
    return (
        jsonify({"error": "too many attempts — try again shortly"}),
        429,
        {"Retry-After": str(retry_after)},
    )


def _sign_in_block(email: str) -> int | None:
    """Seconds to wait before this sign-in may be attempted, or None to proceed.

    Checked BEFORE the password is verified, so a throttled attempt costs no bcrypt
    — which is the other half of what this protects: hashing is deliberately slow,
    and an unauthenticated caller who can trigger it without limit has a CPU
    exhaustion primitive, not just a guessing one.
    """
    address = client_address()
    waits = [
        sign_in_by_address.retry_after(address),
        sign_in_by_account.retry_after(email) if email else None,
    ]
    live = [w for w in waits if w is not None]
    return max(live) if live else None


def _sign_in_failed(email: str) -> None:
    address = client_address()
    sign_in_by_address.record(address)
    if email:
        sign_in_by_account.record(email)


def _sign_in_succeeded(email: str) -> None:
    """Clear the account's history on success. The address keeps its count: one
    correct password does not vouch for the other attempts from there."""
    if email:
        sign_in_by_account.forget(email)


@bp.post("/register")
async def register():
    data = await request.get_json(silent=True) or {}
    email = (data.get("email") or "").strip().lower()
    password = data.get("password") or ""
    display_name = (data.get("display_name") or "").strip()

    if not email or "@" not in email:
        return jsonify({"error": "a valid email is required"}), 400
    if len(password) < MIN_PASSWORD_LEN:
        return jsonify({"error": f"password must be at least {MIN_PASSWORD_LEN} characters"}), 400
    if not display_name:
        display_name = email.split("@", 1)[0]

    # Counted by attempt rather than by failure: a rejected registration still cost a
    # round trip and a uniqueness check, and on an instance with signups open the
    # thing worth bounding is how fast accounts can appear at all.
    wait = register_by_address.retry_after(client_address())
    if wait is not None:
        return _throttled(wait)
    register_by_address.record(client_address())

    async with session_scope() as db:
        user_count = await db.scalar(select(func.count()).select_from(User)) or 0
        is_first = user_count == 0
        # The first account bootstraps the admin and is always allowed, even when
        # registration is otherwise closed.
        if not is_first and not await get_setting(db, "allow_registration"):
            logger.warning("registration refused (closed) email=%s from=%s", email, client_address())
            return jsonify({"error": "registration is closed"}), 403
        existing = await db.scalar(select(User).where(User.email == email))
        if existing is not None:
            return jsonify({"error": "an account with that email already exists"}), 409
        user = User(
            email=email,
            password_hash=hash_password(password),
            display_name=display_name,
            is_admin=is_first,
        )
        db.add(user)
        if is_first:
            # Registration CLOSES the moment the instance has an owner.
            #
            # Not "defaults closed" — that would still need the first person to get in
            # somehow. Closed as a CONSEQUENCE of the admin account existing, which is
            # the only formulation with no open window in it. Leaving the setting on
            # meant the gap between "my account exists" and "I remembered to turn it
            # off in Settings" was wide open, and on a public host that gap is the
            # entire exposure — it starts the moment DNS resolves.
            #
            # An admin who wants a second person turns it back on in Settings → Access,
            # adds them, and turns it off. Crude until invites exist, but it is a
            # deliberate act rather than a default nobody chose.
            await set_settings(db, {"allow_registration": False})
        await db.commit()
        await db.refresh(user)
        session[SESSION_KEY] = str(user.id)
        session.permanent = True
        logger.info(
            "account created email=%s admin=%s from=%s", email, is_first, client_address()
        )
        return jsonify(_serialize_user(user)), 201


@bp.post("/login")
async def login():
    data = await request.get_json(silent=True) or {}
    email = (data.get("email") or "").strip().lower()
    password = data.get("password") or ""

    wait = _sign_in_block(email)
    if wait is not None:
        return _throttled(wait)

    async with session_scope() as db:
        user = await db.scalar(select(User).where(User.email == email))
        if user is None or not user.password_hash:
            # Hash anyway. Without this, "no such account" returns in microseconds
            # while a wrong password takes bcrypt's deliberate ~100ms, and the
            # difference is a reliable oracle for which emails have accounts here.
            dummy_verify(password)
            _sign_in_failed(email)
            logger.warning("sign-in failed (no such account) email=%s from=%s", email, client_address())
            return jsonify({"error": "invalid email or password"}), 401
        if not verify_password(password, user.password_hash):
            _sign_in_failed(email)
            logger.warning("sign-in failed (bad password) email=%s from=%s", email, client_address())
            return jsonify({"error": "invalid email or password"}), 401
        _sign_in_succeeded(email)
        session[SESSION_KEY] = str(user.id)
        session.permanent = True
        logger.info("sign-in ok email=%s from=%s", email, client_address())
        return jsonify(_serialize_user(user))


@bp.post("/logout")
async def logout():
    session.pop(SESSION_KEY, None)
    return jsonify({"ok": True})


@bp.get("/me")
@login_required
async def me():
    async with session_scope() as db:
        user = await db.get(User, g.user_id)
        if user is None:
            session.pop(SESSION_KEY, None)
            return jsonify({"error": "authentication required"}), 401
        return jsonify(_serialize_user(user))


# --- Device (bearer) tokens for native clients — M8 sync hub ---


def _serialize_device(d: DeviceToken) -> dict:
    return {
        "id": str(d.id),
        "name": d.name,
        "created_at": iso(d.created_at),
        "last_used_at": iso(d.last_used_at),
    }


async def _issue_device_token(db, user_id: uuid.UUID, name: str) -> tuple[DeviceToken, str]:
    """Create a device token; return the row plus the ONE-TIME plaintext token."""
    token = generate_token()
    row = DeviceToken(
        user_id=user_id,
        token_hash=hash_token(token),
        name=(name or "").strip()[:DEVICE_NAME_CAP] or "Device",
    )
    db.add(row)
    await db.flush()
    return row, token


@bp.post("/device-login")
async def device_login():
    """Native first-link: exchange email+password for a device bearer token. Public
    (no existing session) — this is how a fresh native install authenticates."""
    data = await request.get_json(silent=True) or {}
    email = (data.get("email") or "").strip().lower()
    password = data.get("password") or ""
    if not email or not password:
        return jsonify({"error": "email and password are required"}), 400

    # Same budget as the web sign-in, and the SAME counters — this route hands out a
    # long-lived bearer token, so leaving it unthrottled would just move the guessing
    # here from /login.
    wait = _sign_in_block(email)
    if wait is not None:
        return _throttled(wait)

    async with session_scope() as db:
        user = await db.scalar(select(User).where(User.email == email))
        if user is None or not user.password_hash:
            dummy_verify(password)
            _sign_in_failed(email)
            logger.warning("device-login failed (no such account) email=%s from=%s", email, client_address())
            return jsonify({"error": "invalid email or password"}), 401
        if not verify_password(password, user.password_hash):
            _sign_in_failed(email)
            logger.warning("device-login failed (bad password) email=%s from=%s", email, client_address())
            return jsonify({"error": "invalid email or password"}), 401
        _sign_in_succeeded(email)
        row, token = await _issue_device_token(db, user.id, data.get("name") or "")
        # A device token outlives the session that made it, so its creation is the
        # most consequential thing on this blueprint.
        logger.info(
            "device token issued email=%s device=%s from=%s", email, row.name, client_address()
        )
        await db.commit()
        return jsonify({"token": token, "device": _serialize_device(row), "user": _serialize_user(user)}), 201


@bp.post("/devices")
@login_required
async def create_device():
    """Issue a device token for the already-authenticated user (web 'Link a device')."""
    data = await request.get_json(silent=True) or {}
    async with session_scope() as db:
        row, token = await _issue_device_token(db, g.user_id, data.get("name") or "")
        await db.commit()
        return jsonify({"token": token, "device": _serialize_device(row)}), 201


@bp.get("/devices")
@login_required
async def list_devices():
    async with session_scope() as db:
        rows = (
            await db.scalars(
                select(DeviceToken).where(DeviceToken.user_id == g.user_id).order_by(DeviceToken.created_at.desc())
            )
        ).all()
        return jsonify({"devices": [_serialize_device(d) for d in rows]})


@bp.delete("/devices/self")
@login_required
async def revoke_own_device():
    """Revoke the device token presented on THIS request.

    What makes "unlink" on a native client actually stop its access. The client
    can't use the id-keyed route below, because it doesn't reliably know its own
    device id: a token pasted from the web app arrives without one, and `/me`
    describes the user, not the device row. Identifying the row by the presented
    token needs nothing the caller doesn't already hold, so it works for both ways
    a client can be linked.

    Declared above the `<device_id>` rule for reading order only — Werkzeug ranks a
    static rule ahead of a converter regardless of registration order.
    """
    token = _bearer_token()
    if token is None:
        # A session-cookie caller holds no device token, so "revoke the one I'm
        # using" is meaningless rather than merely unauthorized. The web app
        # revokes by id.
        return jsonify({"error": "no device token was presented"}), 400
    async with session_scope() as db:
        row = await db.scalar(
            select(DeviceToken).where(
                DeviceToken.token_hash == hash_token(token),
                # Owner-scoped like every other device route. The hash already pins
                # a single row; the guarantee shouldn't rest on one column.
                DeviceToken.user_id == g.user_id,
            )
        )
        if row is None:
            return jsonify({"error": "not found"}), 404
        await db.delete(row)
        await db.commit()
        return jsonify({"ok": True})


@bp.delete("/devices/<device_id>")
@login_required
async def revoke_device(device_id: str):
    try:
        did = uuid.UUID(device_id)
    except (ValueError, TypeError):
        return jsonify({"error": "not found"}), 404
    async with session_scope() as db:
        row = await db.scalar(
            select(DeviceToken).where(DeviceToken.id == did, DeviceToken.user_id == g.user_id)
        )
        if row is None:
            return jsonify({"error": "not found"}), 404
        await db.delete(row)
        await db.commit()
        return jsonify({"ok": True})