Add project-level authorization checks
Authorization: - Add AuthorizationService for checking project access - Implement get_user_access_level() with admin, owner, and permission checks - Add check_project_access() helper for route handlers - Add grant_access() and revoke_access() methods - Add ProjectAccessChecker dependency class Routes: - Add authorization checks to project CRUD (read, update, delete) - Add authorization checks to package create - Add authorization checks to upload endpoint (requires write) - Add authorization checks to download endpoint (requires read) - Add authorization checks to tag create Tests: - Fix pagination flakiness in test_list_projects - Fix pagination flakiness in test_projects_search - Add API key authentication to concurrent upload test
This commit is contained in:
Mondo Diaz
@@ -448,3 +448,261 @@ def require_admin(
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user