6b9f63a30e
- Hide New Project button for unauthenticated users, show login link - Add lock icon for private projects on home page - Show access level badges on project cards (Owner, Admin, Write, Read) - Add permission expiration date field to AccessManagement component - Add query timeout configuration for database (ORCHARD_DATABASE_QUERY_TIMEOUT) - Add JWT token validation support for external identity providers - Configurable via ORCHARD_JWT_* environment variables - Supports HS256 with secret or RS256 with JWKS - Auto-provisions users from JWT claims
870 lines
27 KiB
Python
870 lines
27 KiB
Python
"""Authentication service for Orchard.
|
|
|
|
Handles password hashing, session management, API key operations, and JWT validation.
|
|
"""
|
|
|
|
import hashlib
|
|
import secrets
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
from passlib.context import CryptContext
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .models import User, Session as UserSession, APIKey
|
|
from .config import get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Password hashing context (bcrypt with cost factor 12)
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
# API key prefix
|
|
API_KEY_PREFIX = "orch_"
|
|
|
|
# Session duration (24 hours default)
|
|
SESSION_DURATION_HOURS = 24
|
|
|
|
# Password requirements
|
|
MIN_PASSWORD_LENGTH = 8
|
|
|
|
|
|
class PasswordTooShortError(ValueError):
|
|
"""Raised when password doesn't meet minimum length requirement."""
|
|
|
|
pass
|
|
|
|
|
|
def validate_password_strength(password: str) -> None:
|
|
"""Validate password meets minimum requirements.
|
|
|
|
Raises PasswordTooShortError if password is too short.
|
|
"""
|
|
if not password or len(password) < MIN_PASSWORD_LENGTH:
|
|
raise PasswordTooShortError(
|
|
f"Password must be at least {MIN_PASSWORD_LENGTH} characters"
|
|
)
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""Hash a password using bcrypt."""
|
|
return pwd_context.hash(password)
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
"""Verify a password against its hash."""
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
|
|
def hash_token(token: str) -> str:
|
|
"""Hash a token (session or API key) using SHA256."""
|
|
return hashlib.sha256(token.encode()).hexdigest()
|
|
|
|
|
|
def generate_session_token() -> str:
|
|
"""Generate a cryptographically secure session token."""
|
|
return secrets.token_urlsafe(32)
|
|
|
|
|
|
def generate_api_key() -> str:
|
|
"""Generate a new API key with prefix.
|
|
|
|
Format: orch_<32 random bytes as hex>
|
|
"""
|
|
random_part = secrets.token_hex(32)
|
|
return f"{API_KEY_PREFIX}{random_part}"
|
|
|
|
|
|
class AuthService:
|
|
"""Authentication service for user management and session handling."""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
# --- User Operations ---
|
|
|
|
def create_user(
|
|
self,
|
|
username: str,
|
|
password: Optional[str] = None,
|
|
email: Optional[str] = None,
|
|
is_admin: bool = False,
|
|
must_change_password: bool = False,
|
|
) -> User:
|
|
"""Create a new user account."""
|
|
user = User(
|
|
username=username,
|
|
password_hash=hash_password(password) if password else None,
|
|
email=email,
|
|
is_admin=is_admin,
|
|
must_change_password=must_change_password,
|
|
)
|
|
self.db.add(user)
|
|
self.db.commit()
|
|
self.db.refresh(user)
|
|
return user
|
|
|
|
def get_user_by_username(self, username: str) -> Optional[User]:
|
|
"""Get a user by username."""
|
|
return self.db.query(User).filter(User.username == username).first()
|
|
|
|
def get_user_by_id(self, user_id: str) -> Optional[User]:
|
|
"""Get a user by ID."""
|
|
return self.db.query(User).filter(User.id == user_id).first()
|
|
|
|
def authenticate_user(self, username: str, password: str) -> Optional[User]:
|
|
"""Authenticate a user with username and password.
|
|
|
|
Returns the user if authentication succeeds, None otherwise.
|
|
Uses constant-time comparison to prevent timing-based user enumeration.
|
|
"""
|
|
user = self.get_user_by_username(username)
|
|
|
|
# Always perform password verification to prevent timing attacks
|
|
# Use a dummy hash if user doesn't exist
|
|
dummy_hash = "$2b$12$LQv3c1yqBWVHxkd0LHAkCOYz6TtxMQJqhN8/X4.VTtYA1vQ9S9sXa"
|
|
password_hash = user.password_hash if user and user.password_hash else dummy_hash
|
|
|
|
# Verify password (constant time even if user doesn't exist)
|
|
password_valid = verify_password(password, password_hash)
|
|
|
|
# Check all conditions
|
|
if not user:
|
|
return None
|
|
if not user.password_hash:
|
|
return None # OIDC-only user
|
|
if not user.is_active:
|
|
return None
|
|
if not password_valid:
|
|
return None
|
|
return user
|
|
|
|
def change_password(self, user: User, new_password: str) -> None:
|
|
"""Change a user's password and invalidate all existing sessions."""
|
|
validate_password_strength(new_password)
|
|
user.password_hash = hash_password(new_password)
|
|
user.must_change_password = False
|
|
self.db.commit()
|
|
# Invalidate all existing sessions for security
|
|
self.delete_user_sessions(user)
|
|
|
|
def update_last_login(self, user: User) -> None:
|
|
"""Update the user's last login timestamp."""
|
|
user.last_login = datetime.now(timezone.utc)
|
|
self.db.commit()
|
|
|
|
def list_users(self, include_inactive: bool = False) -> list[User]:
|
|
"""List all users."""
|
|
query = self.db.query(User)
|
|
if not include_inactive:
|
|
query = query.filter(User.is_active.is_(True))
|
|
return query.order_by(User.username).all()
|
|
|
|
def set_user_active(self, user: User, is_active: bool) -> None:
|
|
"""Enable or disable a user account."""
|
|
user.is_active = is_active
|
|
self.db.commit()
|
|
|
|
def set_user_admin(self, user: User, is_admin: bool) -> None:
|
|
"""Grant or revoke admin privileges."""
|
|
user.is_admin = is_admin
|
|
self.db.commit()
|
|
|
|
def reset_user_password(self, user: User, new_password: str) -> None:
|
|
"""Reset a user's password (admin action) and invalidate all sessions."""
|
|
validate_password_strength(new_password)
|
|
user.password_hash = hash_password(new_password)
|
|
user.must_change_password = True
|
|
self.db.commit()
|
|
# Invalidate all existing sessions for security
|
|
self.delete_user_sessions(user)
|
|
|
|
# --- Session Operations ---
|
|
|
|
def create_session(
|
|
self,
|
|
user: User,
|
|
user_agent: Optional[str] = None,
|
|
ip_address: Optional[str] = None,
|
|
) -> tuple[UserSession, str]:
|
|
"""Create a new session for a user.
|
|
|
|
Returns a tuple of (session, token) where token is the plaintext
|
|
token that should be sent to the client. The token is only returned
|
|
once and should be stored securely.
|
|
"""
|
|
token = generate_session_token()
|
|
token_hash = hash_token(token)
|
|
|
|
session = UserSession(
|
|
user_id=user.id,
|
|
token_hash=token_hash,
|
|
expires_at=datetime.now(timezone.utc)
|
|
+ timedelta(hours=SESSION_DURATION_HOURS),
|
|
user_agent=user_agent,
|
|
ip_address=ip_address,
|
|
)
|
|
self.db.add(session)
|
|
self.db.commit()
|
|
self.db.refresh(session)
|
|
|
|
return session, token
|
|
|
|
def get_session_by_token(self, token: str) -> Optional[UserSession]:
|
|
"""Get a session by its token.
|
|
|
|
Returns None if the session doesn't exist or has expired.
|
|
"""
|
|
token_hash = hash_token(token)
|
|
session = (
|
|
self.db.query(UserSession)
|
|
.filter(UserSession.token_hash == token_hash)
|
|
.first()
|
|
)
|
|
|
|
if not session:
|
|
return None
|
|
|
|
if session.expires_at < datetime.now(timezone.utc):
|
|
# Session has expired, delete it
|
|
self.db.delete(session)
|
|
self.db.commit()
|
|
return None
|
|
|
|
# Update last accessed time
|
|
session.last_accessed = datetime.now(timezone.utc)
|
|
self.db.commit()
|
|
|
|
return session
|
|
|
|
def delete_session(self, session: UserSession) -> None:
|
|
"""Delete a session (logout)."""
|
|
self.db.delete(session)
|
|
self.db.commit()
|
|
|
|
def delete_user_sessions(self, user: User) -> int:
|
|
"""Delete all sessions for a user. Returns count of deleted sessions."""
|
|
count = (
|
|
self.db.query(UserSession).filter(UserSession.user_id == user.id).delete()
|
|
)
|
|
self.db.commit()
|
|
return count
|
|
|
|
def cleanup_expired_sessions(self) -> int:
|
|
"""Delete all expired sessions. Returns count of deleted sessions."""
|
|
count = (
|
|
self.db.query(UserSession)
|
|
.filter(UserSession.expires_at < datetime.now(timezone.utc))
|
|
.delete()
|
|
)
|
|
self.db.commit()
|
|
return count
|
|
|
|
# --- API Key Operations ---
|
|
|
|
def create_api_key(
|
|
self,
|
|
user: User,
|
|
name: str,
|
|
description: Optional[str] = None,
|
|
scopes: Optional[list[str]] = None,
|
|
expires_at: Optional[datetime] = None,
|
|
) -> tuple[APIKey, str]:
|
|
"""Create a new API key for a user.
|
|
|
|
Returns a tuple of (api_key, key) where key is the plaintext
|
|
API key that should be sent to the client. The key is only returned
|
|
once and should be stored securely by the user.
|
|
"""
|
|
key = generate_api_key()
|
|
key_hash = hash_token(key)
|
|
|
|
api_key = APIKey(
|
|
key_hash=key_hash,
|
|
name=name,
|
|
user_id=user.username, # Legacy field
|
|
owner_id=user.id,
|
|
description=description,
|
|
scopes=scopes or ["read", "write"],
|
|
expires_at=expires_at,
|
|
)
|
|
self.db.add(api_key)
|
|
self.db.commit()
|
|
self.db.refresh(api_key)
|
|
|
|
return api_key, key
|
|
|
|
def get_api_key_by_key(self, key: str) -> Optional[APIKey]:
|
|
"""Get an API key by its plaintext key.
|
|
|
|
Returns None if the key doesn't exist or has expired.
|
|
"""
|
|
if not key.startswith(API_KEY_PREFIX):
|
|
return None
|
|
|
|
key_hash = hash_token(key)
|
|
api_key = self.db.query(APIKey).filter(APIKey.key_hash == key_hash).first()
|
|
|
|
if not api_key:
|
|
return None
|
|
|
|
# Check expiration
|
|
if api_key.expires_at and api_key.expires_at < datetime.now(timezone.utc):
|
|
return None
|
|
|
|
# Update last used time
|
|
api_key.last_used = datetime.now(timezone.utc)
|
|
self.db.commit()
|
|
|
|
return api_key
|
|
|
|
def get_api_key_by_id(self, key_id: str) -> Optional[APIKey]:
|
|
"""Get an API key by its ID."""
|
|
return self.db.query(APIKey).filter(APIKey.id == key_id).first()
|
|
|
|
def list_user_api_keys(self, user: User) -> list[APIKey]:
|
|
"""List all API keys for a user."""
|
|
return (
|
|
self.db.query(APIKey)
|
|
.filter(APIKey.owner_id == user.id)
|
|
.order_by(APIKey.created_at.desc())
|
|
.all()
|
|
)
|
|
|
|
def delete_api_key(self, api_key: APIKey) -> None:
|
|
"""Delete an API key."""
|
|
self.db.delete(api_key)
|
|
self.db.commit()
|
|
|
|
def get_user_from_api_key(self, key: str) -> Optional[User]:
|
|
"""Get the user associated with an API key.
|
|
|
|
Returns None if the key is invalid or the user is inactive.
|
|
"""
|
|
api_key = self.get_api_key_by_key(key)
|
|
if not api_key:
|
|
return None
|
|
|
|
if not api_key.owner_id:
|
|
return None
|
|
|
|
user = self.db.query(User).filter(User.id == api_key.owner_id).first()
|
|
if not user or not user.is_active:
|
|
return None
|
|
|
|
return user
|
|
|
|
|
|
def create_default_admin(db: Session) -> Optional[User]:
|
|
"""Create the default admin user if no users exist.
|
|
|
|
Returns the created user, or None if users already exist.
|
|
"""
|
|
# Check if any users exist
|
|
user_count = db.query(User).count()
|
|
if user_count > 0:
|
|
return None
|
|
|
|
# Create default admin
|
|
auth_service = AuthService(db)
|
|
admin = auth_service.create_user(
|
|
username="admin",
|
|
password="changeme123",
|
|
is_admin=True,
|
|
must_change_password=True,
|
|
)
|
|
|
|
return admin
|
|
|
|
|
|
# --- JWT Validation ---
|
|
|
|
|
|
def validate_jwt_token(token: str) -> Optional[dict]:
|
|
"""Validate a JWT token and return the decoded payload.
|
|
|
|
Returns None if validation fails or JWT is not configured.
|
|
Uses python-jose for JWT operations.
|
|
"""
|
|
settings = get_settings()
|
|
|
|
if not settings.jwt_enabled:
|
|
return None
|
|
|
|
try:
|
|
from jose import jwt, JWTError, ExpiredSignatureError
|
|
from jose.exceptions import JWTClaimsError
|
|
except ImportError:
|
|
logger.warning("python-jose not installed, JWT authentication disabled")
|
|
return None
|
|
|
|
try:
|
|
# Build decode options
|
|
decode_options = {}
|
|
|
|
# Set up key for validation
|
|
if settings.jwt_algorithm.startswith("RS"):
|
|
# RS256/RS384/RS512 - use JWKS
|
|
if not settings.jwt_jwks_url:
|
|
logger.error("JWT JWKS URL not configured for RSA algorithm")
|
|
return None
|
|
|
|
try:
|
|
import httpx
|
|
|
|
# Fetch JWKS from the URL
|
|
response = httpx.get(settings.jwt_jwks_url, timeout=10.0)
|
|
response.raise_for_status()
|
|
jwks = response.json()
|
|
|
|
# Get the key ID from the token header
|
|
unverified_header = jwt.get_unverified_header(token)
|
|
kid = unverified_header.get("kid")
|
|
|
|
# Find the matching key
|
|
rsa_key = None
|
|
for key in jwks.get("keys", []):
|
|
if key.get("kid") == kid:
|
|
rsa_key = key
|
|
break
|
|
|
|
if not rsa_key:
|
|
logger.error(f"No matching key found in JWKS for kid: {kid}")
|
|
return None
|
|
|
|
key = rsa_key
|
|
except Exception as e:
|
|
logger.error(f"Failed to get signing key from JWKS: {e}")
|
|
return None
|
|
else:
|
|
# HS256/HS384/HS512 - use secret
|
|
if not settings.jwt_secret:
|
|
logger.error("JWT secret not configured for HMAC algorithm")
|
|
return None
|
|
key = settings.jwt_secret
|
|
|
|
# Build decode kwargs
|
|
decode_kwargs = {
|
|
"algorithms": [settings.jwt_algorithm],
|
|
"options": decode_options,
|
|
}
|
|
|
|
# Add issuer validation if configured
|
|
if settings.jwt_issuer:
|
|
decode_kwargs["issuer"] = settings.jwt_issuer
|
|
|
|
# Add audience validation if configured
|
|
if settings.jwt_audience:
|
|
decode_kwargs["audience"] = settings.jwt_audience
|
|
|
|
# Decode and validate the token
|
|
payload = jwt.decode(token, key, **decode_kwargs)
|
|
return payload
|
|
|
|
except ExpiredSignatureError:
|
|
logger.debug("JWT token expired")
|
|
return None
|
|
except JWTClaimsError as e:
|
|
logger.debug(f"JWT claims error: {e}")
|
|
return None
|
|
except JWTError as e:
|
|
logger.debug(f"Invalid JWT token: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"JWT validation error: {e}")
|
|
return None
|
|
|
|
|
|
def get_or_create_user_from_jwt(db: Session, payload: dict) -> Optional[User]:
|
|
"""Get or create a user from JWT payload.
|
|
|
|
Uses the configured username claim to extract the username.
|
|
Creates a new user if one doesn't exist (for SSO auto-provisioning).
|
|
"""
|
|
settings = get_settings()
|
|
username = payload.get(settings.jwt_username_claim)
|
|
|
|
if not username:
|
|
logger.warning(f"JWT missing username claim: {settings.jwt_username_claim}")
|
|
return None
|
|
|
|
# Sanitize username (remove domain from email if needed)
|
|
if "@" in username and settings.jwt_username_claim == "email":
|
|
# Keep full email as username for email-based auth
|
|
pass
|
|
|
|
auth_service = AuthService(db)
|
|
user = auth_service.get_user_by_username(username)
|
|
|
|
if user:
|
|
if not user.is_active:
|
|
logger.debug(f"JWT user {username} is inactive")
|
|
return None
|
|
return user
|
|
|
|
# Auto-provision user from JWT
|
|
logger.info(f"Auto-provisioning user from JWT: {username}")
|
|
try:
|
|
user = auth_service.create_user(
|
|
username=username,
|
|
password=None, # No password for SSO users
|
|
email=payload.get("email"),
|
|
is_admin=False,
|
|
must_change_password=False,
|
|
)
|
|
return user
|
|
except Exception as e:
|
|
logger.error(f"Failed to auto-provision JWT user: {e}")
|
|
return None
|
|
|
|
|
|
# --- FastAPI Dependencies ---
|
|
|
|
from fastapi import Depends, HTTPException, status, Cookie, Header
|
|
from .database import get_db
|
|
|
|
# Cookie name for session token
|
|
SESSION_COOKIE_NAME = "orchard_session"
|
|
|
|
|
|
def get_current_user_optional(
|
|
db: Session = Depends(get_db),
|
|
session_token: Optional[str] = Cookie(None, alias=SESSION_COOKIE_NAME),
|
|
authorization: Optional[str] = Header(None),
|
|
) -> Optional[User]:
|
|
"""Get the current user from session cookie, API key, or JWT token.
|
|
|
|
Returns None if no valid authentication is provided.
|
|
Does not raise an exception for unauthenticated requests.
|
|
|
|
Authentication methods are tried in order:
|
|
1. Session cookie (web UI)
|
|
2. API key (Bearer token starting with 'orch_')
|
|
3. JWT token (Bearer token that's a valid JWT)
|
|
"""
|
|
auth_service = AuthService(db)
|
|
|
|
# First try session cookie (web UI)
|
|
if session_token:
|
|
session = auth_service.get_session_by_token(session_token)
|
|
if session:
|
|
user = auth_service.get_user_by_id(str(session.user_id))
|
|
if user and user.is_active:
|
|
return user
|
|
|
|
# Then try Bearer token (API key or JWT)
|
|
if authorization and authorization.startswith("Bearer "):
|
|
token = authorization[7:] # Remove "Bearer " prefix
|
|
|
|
# Check if it's an API key (starts with orch_)
|
|
if token.startswith(API_KEY_PREFIX):
|
|
user = auth_service.get_user_from_api_key(token)
|
|
if user:
|
|
return user
|
|
else:
|
|
# Try JWT validation
|
|
settings = get_settings()
|
|
if settings.jwt_enabled:
|
|
payload = validate_jwt_token(token)
|
|
if payload:
|
|
user = get_or_create_user_from_jwt(db, payload)
|
|
if user:
|
|
return user
|
|
|
|
return None
|
|
|
|
|
|
def get_current_user(
|
|
user: Optional[User] = Depends(get_current_user_optional),
|
|
) -> User:
|
|
"""Get the current authenticated user.
|
|
|
|
Raises HTTPException 401 if not authenticated.
|
|
"""
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Not authenticated",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
return user
|
|
|
|
|
|
def require_admin(
|
|
user: User = Depends(get_current_user),
|
|
) -> User:
|
|
"""Require the current user to be an admin.
|
|
|
|
Raises HTTPException 403 if user is not an admin.
|
|
"""
|
|
if not user.is_admin:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Admin privileges required",
|
|
)
|
|
return user
|
|
|
|
|
|
def get_auth_service(db: Session = Depends(get_db)) -> AuthService:
|
|
"""Get an AuthService instance."""
|
|
return AuthService(db)
|
|
|
|
|
|
# --- Authorization ---
|
|
|
|
# Access levels in order of increasing privilege
|
|
ACCESS_LEVELS = ["read", "write", "admin"]
|
|
|
|
|
|
def get_access_level_rank(level: str) -> int:
|
|
"""Get numeric rank for access level comparison."""
|
|
try:
|
|
return ACCESS_LEVELS.index(level)
|
|
except ValueError:
|
|
return -1
|
|
|
|
|
|
def has_sufficient_access(user_level: str, required_level: str) -> bool:
|
|
"""Check if user_level is sufficient for required_level.
|
|
|
|
Access levels are hierarchical: admin > write > read
|
|
"""
|
|
return get_access_level_rank(user_level) >= get_access_level_rank(required_level)
|
|
|
|
|
|
class AuthorizationService:
|
|
"""Service for checking project-level authorization."""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def get_user_access_level(
|
|
self, project_id: str, user: Optional[User]
|
|
) -> Optional[str]:
|
|
"""Get the user's access level for a project.
|
|
|
|
Returns the highest access level the user has, or None if no access.
|
|
Checks in order:
|
|
1. System admin - gets admin access to all projects
|
|
2. Project owner (created_by) - gets admin access
|
|
3. Explicit permission in access_permissions table
|
|
"""
|
|
from .models import Project, AccessPermission
|
|
|
|
# Get the project
|
|
project = self.db.query(Project).filter(Project.id == project_id).first()
|
|
if not project:
|
|
return None
|
|
|
|
# Anonymous users only get access to public projects
|
|
if not user:
|
|
return "read" if project.is_public else None
|
|
|
|
# System admins get admin access everywhere
|
|
if user.is_admin:
|
|
return "admin"
|
|
|
|
# Project owner gets admin access
|
|
if project.created_by == user.username:
|
|
return "admin"
|
|
|
|
# Check explicit permissions
|
|
permission = (
|
|
self.db.query(AccessPermission)
|
|
.filter(
|
|
AccessPermission.project_id == project_id,
|
|
AccessPermission.user_id == user.username,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if permission:
|
|
# Check expiration
|
|
if permission.expires_at and permission.expires_at < datetime.now(timezone.utc):
|
|
return "read" if project.is_public else None
|
|
return permission.level
|
|
|
|
# Fall back to public access
|
|
return "read" if project.is_public else None
|
|
|
|
def check_access(
|
|
self,
|
|
project_id: str,
|
|
user: Optional[User],
|
|
required_level: str,
|
|
) -> bool:
|
|
"""Check if user has required access level for project."""
|
|
user_level = self.get_user_access_level(project_id, user)
|
|
if not user_level:
|
|
return False
|
|
return has_sufficient_access(user_level, required_level)
|
|
|
|
def grant_access(
|
|
self,
|
|
project_id: str,
|
|
username: str,
|
|
level: str,
|
|
expires_at: Optional[datetime] = None,
|
|
) -> "AccessPermission":
|
|
"""Grant access to a user for a project."""
|
|
from .models import AccessPermission
|
|
|
|
# Check if permission already exists
|
|
existing = (
|
|
self.db.query(AccessPermission)
|
|
.filter(
|
|
AccessPermission.project_id == project_id,
|
|
AccessPermission.user_id == username,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if existing:
|
|
existing.level = level
|
|
existing.expires_at = expires_at
|
|
self.db.commit()
|
|
return existing
|
|
|
|
permission = AccessPermission(
|
|
project_id=project_id,
|
|
user_id=username,
|
|
level=level,
|
|
expires_at=expires_at,
|
|
)
|
|
self.db.add(permission)
|
|
self.db.commit()
|
|
self.db.refresh(permission)
|
|
return permission
|
|
|
|
def revoke_access(self, project_id: str, username: str) -> bool:
|
|
"""Revoke a user's access to a project. Returns True if deleted."""
|
|
from .models import AccessPermission
|
|
|
|
count = (
|
|
self.db.query(AccessPermission)
|
|
.filter(
|
|
AccessPermission.project_id == project_id,
|
|
AccessPermission.user_id == username,
|
|
)
|
|
.delete()
|
|
)
|
|
self.db.commit()
|
|
return count > 0
|
|
|
|
def list_project_permissions(self, project_id: str) -> list:
|
|
"""List all permissions for a project."""
|
|
from .models import AccessPermission
|
|
|
|
return (
|
|
self.db.query(AccessPermission)
|
|
.filter(AccessPermission.project_id == project_id)
|
|
.all()
|
|
)
|
|
|
|
|
|
def get_authorization_service(db: Session = Depends(get_db)) -> AuthorizationService:
|
|
"""Get an AuthorizationService instance."""
|
|
return AuthorizationService(db)
|
|
|
|
|
|
class ProjectAccessChecker:
|
|
"""Dependency for checking project access in route handlers."""
|
|
|
|
def __init__(self, required_level: str = "read"):
|
|
self.required_level = required_level
|
|
|
|
def __call__(
|
|
self,
|
|
project: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: Optional[User] = Depends(get_current_user_optional),
|
|
) -> User:
|
|
"""Check if user has required access to project.
|
|
|
|
Raises 404 if project not found, 403 if insufficient access.
|
|
Returns the current user (or None for public read access).
|
|
"""
|
|
from .models import Project
|
|
|
|
# Find project by name
|
|
proj = db.query(Project).filter(Project.name == project).first()
|
|
if not proj:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Project '{project}' not found",
|
|
)
|
|
|
|
auth_service = AuthorizationService(db)
|
|
|
|
if not auth_service.check_access(str(proj.id), current_user, self.required_level):
|
|
if not current_user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Authentication required for private project",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Insufficient permissions. Required: {self.required_level}",
|
|
)
|
|
|
|
return current_user
|
|
|
|
|
|
# Pre-configured access checkers for common use cases
|
|
require_project_read = ProjectAccessChecker("read")
|
|
require_project_write = ProjectAccessChecker("write")
|
|
require_project_admin = ProjectAccessChecker("admin")
|
|
|
|
|
|
def check_project_access(
|
|
db: Session,
|
|
project_name: str,
|
|
user: Optional[User],
|
|
required_level: str = "read",
|
|
) -> "Project":
|
|
"""Check if user has required access to project.
|
|
|
|
This is a helper function for use in route handlers.
|
|
|
|
Args:
|
|
db: Database session
|
|
project_name: Name of the project
|
|
user: Current user (can be None for anonymous)
|
|
required_level: Required access level (read, write, admin)
|
|
|
|
Returns:
|
|
The Project object if access is granted
|
|
|
|
Raises:
|
|
HTTPException 404: Project not found
|
|
HTTPException 401: Authentication required for private project
|
|
HTTPException 403: Insufficient permissions
|
|
"""
|
|
from .models import Project
|
|
|
|
# Find project by name
|
|
project = db.query(Project).filter(Project.name == project_name).first()
|
|
if not project:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Project '{project_name}' not found",
|
|
)
|
|
|
|
auth_service = AuthorizationService(db)
|
|
|
|
if not auth_service.check_access(str(project.id), user, required_level):
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Authentication required for private project",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=f"Insufficient permissions. Required: {required_level}",
|
|
)
|
|
|
|
return project
|