"""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"