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

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

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

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
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,
    decrypt_email,
    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,
    UserDevice, UserPreference,
)
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(),
    }


# ── Device authentication ────────────────────────────────────────────────────
#
# A device signs in with an Ed25519 key instead of re-deriving one from the
# passphrase every time. The passphrase remains the account's credential and its
# only recovery path; this is the day-to-day path once a device is registered.
#
# This is **not** the key directory that was H3, and the difference matters:
# nothing reads these but the hub, no group key is ever wrapped for one, and it
# is a different key from the per-node identity keys, which never leave the
# device-node relationship. What it does cost is metadata — the hub now knows
# how many devices an account has and when each last signed in.

DEVICE_AUTH_TIMESTAMP_WINDOW = 60  # seconds, as for node auth


class DeviceRegisterRequest(BaseModel):
    pk_auth_ed25519: str          # base64 raw 32 bytes
    label: str = ""


class DeviceAuthRequest(BaseModel):
    username:  str
    timestamp: int                # unix epoch seconds
    signature: str                # base64 Ed25519 over the message below


@router.post("/devices", status_code=201)
async def register_device(
    body: DeviceRegisterRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """
    Register a device's hub authentication key.

    Requires an existing session, which in practice means the passphrase was
    entered on this device a moment ago. A device cannot enrol itself.
    """
    try:
        raw = base64.b64decode(body.pk_auth_ed25519)
        Ed25519PublicKey.from_public_bytes(raw)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid Ed25519 public key")

    existing = await db.execute(
        select(UserDevice).where(
            UserDevice.pk_auth_ed25519 == body.pk_auth_ed25519))
    found = existing.scalar_one_or_none()
    if found:
        if found.user_id != current_user.id:
            # One key, one account. Sharing it would make "who signed in" a
            # question with two answers.
            raise HTTPException(status_code=409,
                                detail="That key belongs to another account")
        return {"id": found.id, "label": found.label, "existing": True}

    count = await db.execute(
        select(UserDevice).where(UserDevice.user_id == current_user.id))
    if len(count.scalars().all()) >= 10:
        raise HTTPException(status_code=409,
                            detail="Too many devices — remove one first")

    device = UserDevice(user_id=current_user.id,
                        pk_auth_ed25519=body.pk_auth_ed25519,
                        label=body.label[:64])
    db.add(device)
    await db.commit()
    await db.refresh(device)
    return {"id": device.id, "label": device.label, "existing": False}


@router.get("/devices")
async def list_devices(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(UserDevice).where(UserDevice.user_id == current_user.id)
        .order_by(UserDevice.created_at))
    return {"devices": [
        {"id": d.id, "label": d.label,
         "created_at": d.created_at.isoformat() if d.created_at else None,
         "last_seen": d.last_seen.isoformat() if d.last_seen else None}
        for d in result.scalars().all()
    ]}


@router.delete("/devices/{device_id}")
async def delete_device(
    device_id: str,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """Retire a device's hub key. Its per-node identities are separate and are
    revoked on each node, which the hub cannot do and should not be able to."""
    result = await db.execute(
        select(UserDevice).where(UserDevice.id == device_id,
                                 UserDevice.user_id == current_user.id))
    device = result.scalar_one_or_none()
    if not device:
        raise HTTPException(status_code=404, detail="No such device")
    await db.delete(device)
    await db.commit()
    return {"status": "deleted", "id": device_id}


@router.post("/auth")
@limiter.limit("10/minute")
async def device_auth(
    body: DeviceAuthRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    """
    Sign in with a registered device key. Same shape as `/v1/nodes/auth`.

    The timestamp window is what stops a captured signature being replayed
    later; the signature covers the username as well, so one collected for a
    different account is not usable here.
    """
    now = int(time.time())
    if abs(now - body.timestamp) > DEVICE_AUTH_TIMESTAMP_WINDOW:
        raise HTTPException(status_code=401,
                            detail="Timestamp too old or too far in the future")

    result = await db.execute(select(User).where(User.username == body.username))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    if user.status != "active":
        raise HTTPException(status_code=403, detail=f"Account {user.status}")

    devices = await db.execute(
        select(UserDevice).where(UserDevice.user_id == user.id))
    message = f"meshbay:user_auth:{body.username}:{body.timestamp}".encode()
    try:
        sig = base64.b64decode(body.signature)
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid signature encoding")

    matched = None
    for device in devices.scalars().all():
        try:
            pk = Ed25519PublicKey.from_public_bytes(
                base64.b64decode(device.pk_auth_ed25519))
            pk.verify(sig, message)
        except Exception:
            continue
        matched = device
        break

    if matched is None:
        db.add(IPLog(user_id=user.id, event="device_auth_fail",
                     ip_address=client_ip(request), detail=body.username))
        await db.commit()
        raise HTTPException(status_code=401, detail="Invalid signature")

    matched.last_seen = datetime.now(timezone.utc)

    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()
    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
    db.add(RefreshToken(user_id=user.id, token_hash=rt_hash,
                        family_id=str(uuid.uuid4()), expires_at=expires_at))
    db.add(IPLog(user_id=user.id, event="device_auth",
                 ip_address=client_ip(request)))
    await db.commit()

    return {
        "access_token":  access_token,
        "refresh_token": raw_rt,
        "token_type":    "bearer",
        "expires_in":    _ttl(),
        "device_id":     matched.id,
    }


@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),
):
    email = ""
    try:
        email = decrypt_email(current_user.email) if current_user.email else ""
    except Exception:
        pass
    return {
        "user_id":  current_user.id,
        "username": current_user.username,
        "email":    email,
        "role":     current_user.role,
        "status":   current_user.status,
    }


class UpdateProfileRequest(BaseModel):
    email: str | None = None

    @field_validator("email")
    @classmethod
    def email_valid(cls, v: str | None) -> str | None:
        if v is None:
            return v
        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


@router.patch("/me")
async def update_profile(
    body: UpdateProfileRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    if body.email is not None:
        current_user.email = encrypt_email(body.email)
    await db.commit()
    email = ""
    try:
        email = decrypt_email(current_user.email) if current_user.email else ""
    except Exception:
        pass
    return {
        "user_id":  current_user.id,
        "username": current_user.username,
        "email":    email,
        "role":     current_user.role,
        "status":   current_user.status,
    }


# ── User preferences ────────────────────────────────────────────────────────

ALLOWED_PREF_KEYS = frozenset([
    "notifications_disabled",
    "default_tab",
    "music_keep_screen_on",
])

def _valid_pref_key(key: str) -> bool:
    if key in ALLOWED_PREF_KEYS:
        return True
    if key.startswith("default_tab:"):
        return True
    return False


@router.get("/me/preferences")
async def get_preferences(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(UserPreference).where(UserPreference.user_id == current_user.id))
    prefs = {p.key: p.value for p in result.scalars().all()}
    return prefs


class PrefValue(BaseModel):
    value: str


@router.put("/me/preferences/{key:path}")
async def set_preference(
    key: str,
    body: PrefValue,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    if not _valid_pref_key(key):
        raise HTTPException(status_code=400, detail=f"Unknown preference key: {key}")
    result = await db.execute(
        select(UserPreference).where(
            UserPreference.user_id == current_user.id,
            UserPreference.key == key))
    pref = result.scalar_one_or_none()
    if pref:
        pref.value = body.value
    else:
        db.add(UserPreference(
            user_id=current_user.id, key=key, value=body.value))
    await db.commit()
    return {"key": key, "value": body.value}


@router.delete("/me/preferences/{key:path}")
async def delete_preference(
    key: str,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(UserPreference).where(
            UserPreference.user_id == current_user.id,
            UserPreference.key == key))
    pref = result.scalar_one_or_none()
    if pref:
        await db.delete(pref)
        await db.commit()
    return {"status": "deleted", "key": key}


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}


@router.delete("/me/node_key")
async def unlink_node_key(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Remove the linked node key from the operator's account."""
    current_user.pk_node_ed25519 = None
    await db.commit()
    return {"status": "unlinked"}


# 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(UserPreference).where(UserPreference.user_id == user.id))
    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