Spaces:
Runtime error
Runtime error
File size: 2,354 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """Tenant registration and login."""
from __future__ import annotations
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field, model_validator
from backend.api.deps import get_current_tenant
from backend.api import security
router = APIRouter(prefix="/auth", tags=["auth"])
class Credentials(BaseModel):
tenant_id: str = Field(min_length=1, max_length=64)
password: str = Field(default="", max_length=256)
passphrase: str = Field(default="", max_length=256)
@model_validator(mode="after")
def _resolve_secret(self) -> Credentials:
secret = (self.password or self.passphrase or "").strip()
if len(secret) < 6:
raise ValueError("password or passphrase must be at least 6 characters")
object.__setattr__(self, "password", secret)
return self
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
tenant_id: str = ""
expires_at: int | None = None
class WhoAmIResponse(BaseModel):
tenant_id: str
def _token_response(tenant_id: str) -> TokenResponse:
import time
from backend.config import settings
expires_at = int(time.time()) + settings.jwt_expiry_minutes * 60
return TokenResponse(
access_token=security.create_token(tenant_id),
tenant_id=tenant_id,
expires_at=expires_at,
)
@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
def register(creds: Credentials) -> TokenResponse:
if not security.register_user(creds.tenant_id, creds.password):
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Tenant already exists")
from backend.core.tenant_provision import provision_tenant_from_default
provision_tenant_from_default(creds.tenant_id)
return _token_response(creds.tenant_id)
@router.post("/login", response_model=TokenResponse)
def login(creds: Credentials) -> TokenResponse:
if not security.authenticate(creds.tenant_id, creds.password):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
return _token_response(creds.tenant_id)
@router.get("/me", response_model=WhoAmIResponse)
def me(tenant_id: str = Depends(get_current_tenant)) -> WhoAmIResponse:
return WhoAmIResponse(tenant_id=tenant_id)
|