Hermes Agent
f22802658e
- Add hash_password/verify_password to app/auth.py using bcrypt - Fix /login to reject bootstrap creds (403) when admin users exist - Add normal password-based login path for existing admin users - Eagerly load User.roles to avoid lazy-load outside async session - Fix test_user_models.py VARCHAR assertion (SQLAlchemy 2.0 uses 'string') - Use shared-cache SQLite in conftest for endpoint-level tests - Add 23 tests covering all 3 required scenarios plus edge cases: 1. Bootstrap login with hardcoded creds when no admins → PASS (200) 2. Bootstrap login with hardcoded creds when admins exist → FAIL (403) 3. Password login for existing admin with hashed password → PASS (200) - All 130 tests pass (107 existing + 23 new)
151 lines
5.5 KiB
Python
151 lines
5.5 KiB
Python
"""Auth helpers, JWT token creation/verification, and FastAPI dependencies."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from jose import JWTError, jwt
|
|
from bcrypt import hashpw, checkpw, gensalt
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import get_session
|
|
from app.user_models import User
|
|
|
|
# ── Password hashing ───────────────────────────────────────────
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""Hash a plaintext password using bcrypt."""
|
|
return hashpw(password.encode("utf-8"), gensalt()).decode("utf-8")
|
|
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
"""Verify a plaintext password against a bcrypt hash."""
|
|
return bool(checkpw(plain.encode("utf-8"), hashed.encode("utf-8")))
|
|
|
|
|
|
class AuthBearer(HTTPBearer):
|
|
"""HTTP Bearer scheme that doesn't auto-fail on missing token."""
|
|
def __init__(self, auto_error: bool = False):
|
|
super().__init__(auto_error=auto_error)
|
|
|
|
|
|
bearer = AuthBearer(auto_error=False)
|
|
|
|
|
|
# ── Token helpers ──────────────────────────────────────────────
|
|
|
|
def create_access_token(user_id: int, is_admin: bool) -> str:
|
|
"""Create a JWT access token for the given user."""
|
|
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.JWT_EXPIRE_MINUTES)
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"is_admin": is_admin,
|
|
"exp": expire,
|
|
}
|
|
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
|
|
|
|
|
|
def create_download_token(artifact_id: int) -> str:
|
|
"""Create a short-lived JWT for downloading a specific artifact."""
|
|
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.DOWNLOAD_TOKEN_MINUTES)
|
|
payload = {
|
|
"sub": "download",
|
|
"artifact_id": artifact_id,
|
|
"exp": expire,
|
|
}
|
|
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
|
|
|
|
|
|
def verify_download_token(token: str) -> int | None:
|
|
"""Verify a download token and return the artifact_id, or None on failure."""
|
|
try:
|
|
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
|
|
if payload.get("sub") != "download":
|
|
return None
|
|
artifact_id = payload.get("artifact_id")
|
|
if not isinstance(artifact_id, int):
|
|
return None
|
|
return artifact_id
|
|
except JWTError:
|
|
return None
|
|
|
|
|
|
def decode_token(token: str) -> dict:
|
|
"""Decode and verify a JWT token. Raises HTTPException 401 on failure."""
|
|
try:
|
|
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
|
|
user_id = payload.get("sub")
|
|
if user_id is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
return payload
|
|
except JWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
|
|
|
|
|
|
# ── FastAPI dependencies ──────────────────────────────────────
|
|
|
|
async def get_current_user(
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> User:
|
|
"""Validate JWT and return the current User."""
|
|
if credentials is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
|
|
|
payload = decode_token(credentials.credentials)
|
|
user_id = int(payload["sub"])
|
|
|
|
result = await session.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
|
|
return user
|
|
|
|
|
|
async def get_current_admin_user(
|
|
current_user: User = Depends(get_current_user),
|
|
) -> User:
|
|
"""Raise 403 if the current user is not an admin.
|
|
|
|
Checks both the legacy is_admin flag and the new role-based admin role.
|
|
"""
|
|
if not current_user.is_admin_effective:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
|
|
return current_user
|
|
|
|
|
|
# ── Google OIDC verification ─────────────────────────────────
|
|
|
|
_GOOGLE_CERTS_URL = "https://www.googleapis.com/oauth2/v3/certs"
|
|
_GOOGLE_TOKEN_INFO_URL = "https://oauth2.googleapis.com/tokeninfo"
|
|
|
|
|
|
async def verify_google_id_token(id_token: str) -> dict:
|
|
"""Verify a Google ID token and return (email, name, picture)."""
|
|
async with httpx.AsyncClient() as client:
|
|
# Use tokeninfo endpoint for simplicity (no signature parsing needed)
|
|
resp = await client.get(_GOOGLE_TOKEN_INFO_URL, params={"idtoken": id_token})
|
|
|
|
if resp.status_code != 200:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid Google ID token",
|
|
)
|
|
|
|
info = resp.json()
|
|
email = info.get("email")
|
|
if not email:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No email in Google token")
|
|
|
|
return {
|
|
"email": email,
|
|
"name": info.get("name", email.split("@")[0]),
|
|
"picture": info.get("picture", ""),
|
|
}
|