path: root/packages/meshbay-hub/src/meshbay_hub/api/users.py

"""User endpoints — /v1/users/*"""

import base64
import logging
import uuid
from datetime import datetime, timezone, timedelta

from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, field_validator
from sqlalchemy import delete, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from meshbay_hub.auth import (
    current_pw_version,
    decode_access_token,
    encrypt_email,
    generate_refresh_token,
    hash_password,
    hash_refresh_token,
    hub_public_key_pem,
    issue_access_token,
    pw_needs_rehash,
    verify_password,
)
from meshbay_hub.api.middleware import limiter
from meshbay_hub.api.netutil import client_ip
from meshbay_hub.config import HubConfig
from meshbay_hub.db.engine import get_db
from meshbay_hub.db.models import (
    Group, GroupMember, IPLog, Node, Notification, RefreshToken, User,
)
from meshbay_hub.api.deps import get_current_user, require_user_scope

log = logging.getLogger(__name__)

router = APIRouter(prefix="/v1/users", tags=["users"])

_cfg: HubConfig | None = None

def set_config(cfg: HubConfig) -> None:
    global _cfg
    _cfg = cfg

def _ttl() -> int:
    return _cfg.jwt.access_token_ttl if _cfg else 3600

def _refresh_ttl() -> int:
    return _cfg.jwt.refresh_token_ttl if _cfg else 86400 * 30


# ── Models ────────────────────────────────────────────────────────────────────

class RegisterRequest(BaseModel):
    username:        str
    email:           str
    password:        str | None = None    # deprecated — legacy native clients
    auth_key:        str | None = None    # PBKDF2-derived, new clients

    @field_validator("username")
    @classmethod
    def username_valid(cls, v: str) -> str:
        v = v.strip()
        if len(v) < 3 or len(v) > 64:
            raise ValueError("username must be 3-64 chars")
        if not v.replace("_", "").replace("-", "").replace(".", "").isalnum():
            raise ValueError("username: only letters, digits, -, _, .")
        return v

    @field_validator("email")
    @classmethod
    def email_valid(cls, v: str) -> str:
        """
        Sanity-check the address (L6): the field was plain `str`, so any junk was
        accepted and stored encrypted forever. Deliberately not RFC 5322 — full
        validation would pull in the email-validator dependency for little gain,
        and the address is only ever used for recovery and legal contact.
        """
        v = v.strip()
        local, sep, domain = v.partition("@")
        if (not sep or not local or not domain
                or "." not in domain
                or len(v) > 254
                or any(c.isspace() or ord(c) < 32 for c in v)):
            raise ValueError("invalid email address")
        return v


class LoginRequest(BaseModel):
    username: str
    password: str | None = None    # legacy (raw password) for migration
    auth_key: str | None = None    # PBKDF2-derived auth key (new scheme)


class RefreshRequest(BaseModel):
    refresh_token: str


# ── Endpoints ─────────────────────────────────────────────────────────────────

@router.post("/register", status_code=201)
@limiter.limit("5/minute")
async def register(
    body: RegisterRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    existing = await db.execute(
        select(User).where(User.username == body.username))
    if existing.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="Username already taken")

    credential = body.auth_key or body.password
    if not credential:
        raise HTTPException(status_code=400, detail="auth_key or password required")

    pw_hash, pw_salt = hash_password(credential)
    # auth_key → pw_version 3 (password split); raw password → pw_version 2 (legacy)
    pw_ver = current_pw_version() if body.auth_key else 2
    hub_id = _cfg.identity.id if _cfg else "meshbay.org"
    user = User(
        username=body.username,
        email=encrypt_email(body.email),
        pw_hash=pw_hash,
        pw_salt=pw_salt,
        pw_version=pw_ver,
        hub_id=hub_id,
    )
    db.add(user)
    # flush assigns user.id so the log row can be attributed directly.
    #
    # Finding M6: this used to insert the row with a NULL user_id and then run
    #   UPDATE ip_logs SET user_id = <new user> WHERE user_id IS NULL
    # which claimed *every* unattributed row in the table — failed logins for other
    # usernames, other registrations racing this one — and stamped them with the
    # account just created. For logs retained a year to answer legal requests, that
    # attributed other people's connections to the wrong person.
    await db.flush()
    db.add(IPLog(
        user_id=user.id,
        event="account_create",
        ip_address=client_ip(request),
        detail=body.username,
    ))
    await db.commit()
    await db.refresh(user)

    return {"user_id": user.id}


@router.post("/login")
@limiter.limit("10/minute")
async def login(
    body: LoginRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(User).where(User.username == body.username))
    user = result.scalar_one_or_none()

    ip = client_ip(request)

    if not body.auth_key and not body.password:
        raise HTTPException(status_code=401, detail="No credentials provided")

    if not user:
        db.add(IPLog(event="login_fail", ip_address=ip, detail=body.username))
        await db.commit()
        raise HTTPException(status_code=401, detail="Invalid credentials")

    if user.pw_version >= 3:
        # New scheme: verify auth_key
        if not body.auth_key or not verify_password(
            body.auth_key, user.pw_hash, user.pw_salt, version=user.pw_version
        ):
            db.add(IPLog(event="login_fail", ip_address=ip, detail=body.username))
            await db.commit()
            raise HTTPException(status_code=401, detail="Invalid credentials")
    else:
        # Legacy scheme: need raw password
        if not body.password:
            raise HTTPException(status_code=401, detail="auth_upgrade_required")
        if not verify_password(
            body.password, user.pw_hash, user.pw_salt, version=user.pw_version
        ):
            db.add(IPLog(event="login_fail", ip_address=ip, detail=body.username))
            await db.commit()
            raise HTTPException(status_code=401, detail="Invalid credentials")
        # Migrate to new scheme if auth_key provided alongside password
        if body.auth_key:
            new_hash, new_salt = hash_password(body.auth_key)
            user.pw_hash = new_hash
            user.pw_salt = new_salt
            user.pw_version = current_pw_version()
        elif user.pw_version < 2:
            # Legacy rehash: upgrade Argon2 params within the password scheme (v1 -> v2)
            new_hash, new_salt = hash_password(body.password)
            user.pw_hash = new_hash
            user.pw_salt = new_salt
            user.pw_version = 2

    if user.status != "active":
        raise HTTPException(status_code=403, detail=f"Account {user.status}")

    # Rehash within the auth_key scheme if Argon2 params upgraded beyond v3
    if user.pw_version >= 3 and pw_needs_rehash(user.pw_version):
        new_hash, new_salt = hash_password(body.auth_key)
        user.pw_hash = new_hash
        user.pw_salt = new_salt
        user.pw_version = current_pw_version()

    memberships = await db.execute(
        select(GroupMember.group_id).where(GroupMember.user_id == user.id))
    group_ids = [gid for (gid,) in memberships.all()]
    access_token = issue_access_token(user.id, ttl=_ttl(), groups=group_ids)
    raw_rt, rt_hash = generate_refresh_token()
    family_id = str(uuid.uuid4())

    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
    db.add(RefreshToken(
        user_id=user.id, token_hash=rt_hash,
        family_id=family_id, expires_at=expires_at,
    ))
    db.add(IPLog(user_id=user.id, event="login", ip_address=ip))
    await db.commit()

    return {
        "access_token":  access_token,
        "refresh_token": raw_rt,
        "token_type":    "bearer",
        "expires_in":    _ttl(),
    }


@router.post("/token/refresh")
@limiter.limit("20/minute")
async def token_refresh(
    body: RefreshRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    rt_hash = hash_refresh_token(body.refresh_token)
    result = await db.execute(
        select(RefreshToken).where(RefreshToken.token_hash == rt_hash))
    rt = result.scalar_one_or_none()

    if not rt:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    if rt.revoked:
        # Reuse detected — revoke entire token family
        await db.execute(
            RefreshToken.__table__.update()
            .where(RefreshToken.family_id == rt.family_id)
            .values(revoked=True))
        await db.commit()
        raise HTTPException(status_code=401, detail="Token reuse detected — family revoked")

    if rt.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc):
        raise HTTPException(status_code=401, detail="Expired refresh token")

    user = await db.get(User, rt.user_id)
    if not user or user.status != "active":
        raise HTTPException(status_code=401, detail="User not found or suspended")

    # Revoke old token
    rt.revoked = True

    # Issue new refresh token in the same family
    new_raw_rt, new_rt_hash = generate_refresh_token()
    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
    db.add(RefreshToken(
        user_id=user.id, token_hash=new_rt_hash,
        family_id=rt.family_id, expires_at=expires_at,
    ))

    memberships = await db.execute(
        select(GroupMember.group_id).where(GroupMember.user_id == user.id))
    group_ids = [gid for (gid,) in memberships.all()]
    new_access = issue_access_token(user.id, ttl=_ttl(), groups=group_ids)
    await db.commit()

    return {
        "access_token": new_access,
        "refresh_token": new_raw_rt,
        "token_type": "bearer",
        "expires_in": _ttl(),
    }


@router.get("/me")
async def get_current_user_info(
    current_user: User = Depends(get_current_user),
):
    return {
        "user_id":  current_user.id,
        "username": current_user.username,
        "role":     current_user.role,
        "status":   current_user.status,
    }


class NodeKeyRequest(BaseModel):
    pk_node_ed25519: str   # base64 raw 32B Ed25519 public key


@router.put("/me/node_key")
async def register_node_key(
    body: NodeKeyRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """Link a node daemon's Ed25519 public key to the operator's account."""
    try:
        raw = base64.b64decode(body.pk_node_ed25519)
        if len(raw) != 32:
            raise ValueError
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid Ed25519 public key (need 32 bytes base64)")

    current_user.pk_node_ed25519 = body.pk_node_ed25519
    await db.commit()
    return {"status": "stored", "pk_node_ed25519": body.pk_node_ed25519}


# Key rotation used to live here (`PUT /me/keys`). Identity keys are per node
# now, so rotating means `meshbay-node member unpin <user>` and pairing again with
# a fresh code — an operator decision on the machine that pinned it, not a hub
# call that silently changes what every node believes about someone.


# ── Account deletion ─────────────────────────────────────────────────────────

async def erase_account(db: AsyncSession, user: User) -> dict:
    """
    Erase an account, keeping only what the law asked us to keep.

    Gone: credentials, email, node key, group memberships, notifications, refresh
    tokens, node registrations. The username is released.

    Kept: the row itself, emptied, and the IP log that points at it. Those logs
    exist for one year to answer legal requests, and a log that cannot say whose
    connection it recorded does not do that — detaching them would keep the data
    and lose the only thing it is for. So the account becomes a tombstone rather
    than a hole in the table.

    Not touched, because the hub cannot: files this person uploaded to nodes, and
    the identity keys nodes pinned for them. Those live on machines the hub does
    not command, and only their operators can remove them.
    """
    owned = (await db.execute(
        select(Group).where(Group.admin_id == user.id))).scalars().all()
    if owned:
        raise HTTPException(
            status_code=409,
            detail=("This account still owns groups: "
                    + ", ".join(g.name for g in owned)
                    + ". Delete them or hand them over first — deleting the "
                      "account would strand their members."),
        )

    await db.execute(delete(GroupMember).where(GroupMember.user_id == user.id))
    await db.execute(delete(Notification).where(Notification.user_id == user.id))
    await db.execute(delete(RefreshToken).where(RefreshToken.user_id == user.id))
    await db.execute(delete(Node).where(Node.user_id == user.id))

    username = user.username
    # Before the name is released: the connection log is kept for its legal
    # retention period and has to stay readable, which means saying who this was
    # and not "deleted-3f9a1c". Nothing else keeps it.
    await db.execute(
        update(IPLog).where(IPLog.user_id == user.id).values(username=username))
    user.username = f"deleted-{user.id[:8]}"
    user.email = ""
    user.pw_hash = b""
    user.pw_salt = b""
    user.pk_node_ed25519 = None
    user.status = "deleted"
    user.role = "user"
    await db.commit()
    log.info("Account erased: %s (%s)", username, user.id[:8])
    return {"status": "deleted", "username": username}


class DeleteAccountRequest(BaseModel):
    auth_key: str


@router.delete("/me")
async def delete_own_account(
    body: DeleteAccountRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """
    Erase your own account. The passphrase is re-checked here.

    A live access token is not enough for something irreversible: it may be a
    borrowed laptop or a session left open. Same value as at sign-in, so the hub
    still never sees the passphrase itself.
    """
    if not verify_password(body.auth_key, current_user.pw_hash, current_user.pw_salt,
                           current_user.pw_version):
        raise HTTPException(status_code=403, detail="Passphrase does not match")
    return await erase_account(db, current_user)


@router.get("/{username}/pubkeys")
async def get_user_pubkeys(
    username: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).where(User.username == username))
    target = result.scalar_one_or_none()
    if not target:
        raise HTTPException(status_code=404, detail="User not found")
    # Account lookup, not a key directory. `user_id` is how a username is resolved
    # for an invitation, and `pk_node_ed25519` is a node's own linking key. The
    # user identity keys this used to return were H3: whoever asked wrapped the
    # group key for whatever came back.
    resp = {
        "user_id":  target.id,
        "username": target.username,
    }
    if target.pk_node_ed25519:
        resp["pk_node_ed25519"] = target.pk_node_ed25519
    return resp