"""User endpoints — /v1/users/*""" 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 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 pk_user_ed25519: str # base64 raw 32B pk_user_x25519: str # base64 raw 32B keypair_bundle: str | None = None # AES-GCM encrypted bundle (web 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("password") @classmethod def password_strength(cls, v: str) -> str: if len(v) < 8: raise ValueError("password must be at least 8 characters") return v class LoginRequest(BaseModel): username: str password: str 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") pw_hash, pw_salt = hash_password(body.password) 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=current_pw_version(), pk_ed25519=body.pk_user_ed25519, pk_x25519=body.pk_user_x25519, hub_id=hub_id, keypair_bundle=body.keypair_bundle, ) 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 user or 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") if user.status != "active": raise HTTPException(status_code=403, detail=f"Account {user.status}") if pw_needs_rehash(user.pw_version): new_hash, new_salt = hash_password(body.password) 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() resp = { "access_token": access_token, "refresh_token": raw_rt, "token_type": "bearer", "expires_in": _ttl(), } if user.keypair_bundle: resp["keypair_bundle"] = user.keypair_bundle # encrypted, for web clients return resp @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, } @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") return { "user_id": target.id, "username": target.username, "pk_ed25519": target.pk_ed25519, "pk_x25519": target.pk_x25519, } 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"