Vineetiitg
feat(backend): integrate Redis workers, persistent model caching, Cohere V2 fallback, and 5x TTFT fast-path RAG
d816f3a
Raw
History Blame Contribute Delete
1.94 kB
import jwt
from datetime import datetime, timedelta, timezone
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from passlib.context import CryptContext
from app.auth.models import UserContext
from app.core.config import settings
from app.core.errors import CopilotError
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login", auto_error=False)
# Dummy local DB for auth
USERS = {
"admin": {"password_hash": pwd_context.hash("admin123"), "role": "admin"},
"user": {"password_hash": pwd_context.hash("user123"), "role": "user"}
}
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")
def resolve_user(token: str | None = Depends(oauth2_scheme)) -> UserContext:
if not settings.AUTH_ENABLED:
return UserContext(role="admin", user_id="local-dev")
if not token:
return UserContext(role="guest", user_id="guest")
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
username: str | None = payload.get("sub")
role: str | None = payload.get("role")
if username is None or role is None:
return UserContext(role="guest", user_id="guest")
return UserContext(role=role, user_id=username)
except jwt.PyJWTError:
return UserContext(role="guest", user_id="guest")
def require_admin(user: UserContext = Depends(resolve_user)) -> None:
if user.role != "admin":
raise CopilotError("Admin role required.", status_code=403)