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

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

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

from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, EmailStr, field_validator
from sqlalchemy import select
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.config import HubConfig
from meshbay_hub.db.engine import get_db
from meshbay_hub.db.models import GroupMember, IPLog, RefreshToken, User
from meshbay_hub.api.deps import get_current_user, require_user_scope

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
    pk_user_ed25519: str        # base64 raw 32B
    pk_user_x25519:  str        # base64 raw 32B

    @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


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,
        pk_ed25519=body.pk_user_ed25519,
        pk_x25519=body.pk_user_x25519,
        hub_id=hub_id,
    )
    db.add(user)
    db.add(IPLog(
        event="account_create",
        ip_address=_client_ip(request),
        detail=body.username,
    ))
    await db.commit()
    await db.refresh(user)

    # Set user_id in IPLog after commit
    await db.execute(
        IPLog.__table__.update()
        .where(IPLog.user_id == None)  # noqa: E711
        .values(user_id=user.id))
    await db.commit()

    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, user.pk_ed25519, 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, user.pk_ed25519, 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}


class RotateKeysRequest(BaseModel):
    pk_user_ed25519: str  # base64 raw 32B
    pk_user_x25519: str   # base64 raw 32B


@router.put("/me/keys")
async def rotate_browser_keys(
    body: RotateKeysRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    for field, label in [
        (body.pk_user_ed25519, "Ed25519"),
        (body.pk_user_x25519, "X25519"),
    ]:
        try:
            raw = base64.b64decode(field)
            if len(raw) != 32:
                raise ValueError
        except Exception:
            raise HTTPException(
                status_code=400,
                detail=f"Invalid {label} public key (need 32 bytes base64)",
            )

    current_user.pk_ed25519 = body.pk_user_ed25519
    current_user.pk_x25519 = body.pk_user_x25519
    await db.commit()
    return {
        "status": "updated",
        "pk_ed25519": body.pk_user_ed25519,
        "pk_x25519": body.pk_user_x25519,
    }


@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")
    resp = {
        "user_id":    target.id,
        "username":   target.username,
        "pk_ed25519": target.pk_ed25519,
        "pk_x25519":  target.pk_x25519,
    }
    if target.pk_node_ed25519:
        resp["pk_node_ed25519"] = target.pk_node_ed25519
    return resp


def _client_ip(request: Request) -> str:
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.client.host if request.client else "unknown"