Initial commit: StackPilot Phase 1 (Core)
Self-hosted Docker Compose manager. - Backend: FastAPI + docker-py + SQLite (JWT auth, file-first stacks, lifecycle, live status, WebSocket logs, docker-run converter, audit log) - Frontend: React + Vite + Tailwind (login/setup, dashboard, stacks, stack detail, Monaco editor, dark/light theme) - Deployment: docker-compose.yml, Dockerfiles, nginx reverse proxy Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
+132
@@ -0,0 +1,132 @@
|
||||
"""JWT auth + user management."""
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from jose import JWTError, jwt
|
||||
from passlib.context import CryptContext
|
||||
from sqlmodel import Session, select
|
||||
|
||||
from config import settings
|
||||
from database import get_session
|
||||
from models.user import User
|
||||
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login", auto_error=False)
|
||||
|
||||
|
||||
# --- password helpers ---
|
||||
|
||||
|
||||
def hash_password(password: str) -> str:
|
||||
return pwd_context.hash(password)
|
||||
|
||||
|
||||
def verify_password(plain: str, hashed: str) -> bool:
|
||||
return pwd_context.verify(plain, hashed)
|
||||
|
||||
|
||||
# --- token helpers ---
|
||||
|
||||
|
||||
def _create_token(sub: str, role: str, token_type: str, expires: timedelta) -> str:
|
||||
now = datetime.now(timezone.utc)
|
||||
payload = {
|
||||
"sub": sub,
|
||||
"role": role,
|
||||
"type": token_type,
|
||||
"iat": now,
|
||||
"exp": now + expires,
|
||||
}
|
||||
return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
|
||||
|
||||
|
||||
def create_access_token(user: User) -> str:
|
||||
return _create_token(
|
||||
user.username,
|
||||
user.role,
|
||||
"access",
|
||||
timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
|
||||
)
|
||||
|
||||
|
||||
def create_refresh_token(user: User) -> str:
|
||||
return _create_token(
|
||||
user.username,
|
||||
user.role,
|
||||
"refresh",
|
||||
timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS),
|
||||
)
|
||||
|
||||
|
||||
def decode_token(token: str, expected_type: str = "access") -> dict:
|
||||
try:
|
||||
payload = jwt.decode(
|
||||
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
|
||||
)
|
||||
except JWTError as exc:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Invalid token",
|
||||
) from exc
|
||||
if payload.get("type") != expected_type:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Wrong token type",
|
||||
)
|
||||
return payload
|
||||
|
||||
|
||||
# --- user lookups ---
|
||||
|
||||
|
||||
def get_user(session: Session, username: str) -> Optional[User]:
|
||||
return session.exec(select(User).where(User.username == username)).first()
|
||||
|
||||
|
||||
def authenticate(session: Session, username: str, password: str) -> Optional[User]:
|
||||
user = get_user(session, username)
|
||||
if not user or not user.is_active:
|
||||
return None
|
||||
if not verify_password(password, user.hashed_password):
|
||||
return None
|
||||
return user
|
||||
|
||||
|
||||
def users_exist(session: Session) -> bool:
|
||||
return session.exec(select(User)).first() is not None
|
||||
|
||||
|
||||
# --- FastAPI dependencies ---
|
||||
|
||||
|
||||
def get_current_user(
|
||||
token: Optional[str] = Depends(oauth2_scheme),
|
||||
session: Session = Depends(get_session),
|
||||
) -> User:
|
||||
if not token:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Not authenticated",
|
||||
headers={"WWW-Authenticate": "Bearer"},
|
||||
)
|
||||
payload = decode_token(token, "access")
|
||||
user = get_user(session, payload.get("sub", ""))
|
||||
if not user or not user.is_active:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="User not found or inactive",
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
def require_admin(user: User = Depends(get_current_user)) -> User:
|
||||
if user.role != "admin":
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Admin privileges required",
|
||||
)
|
||||
return user
|
||||
Reference in New Issue
Block a user