ilia/punimtag
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: Update .gitignore and add role permissions management API

Browse Source 
This commit updates the .gitignore file to include Node.js related directories and files. Additionally, it introduces a new API for managing role-to-feature permissions, allowing for better control over user access levels. The API includes endpoints for listing and updating role permissions, ensuring that the permissions matrix is initialized and maintained. Documentation has been updated to reflect these changes.
This commit is contained in:
tanyar09
2025-11-26 15:00:28 -05:00
parent eed3b36dad
commit 7c35e4d8ec
34 changed files with 2961 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/web/api/auth.py
+26 -1
View File
@@ -10,6 +10,11 @@ from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from src.web.constants.roles import (
DEFAULT_ADMIN_ROLE,
DEFAULT_USER_ROLE,
ROLE_VALUES,
)
from src.web.db.session import get_db
from src.web.db.models import User
from src.web.utils.password import verify_password, hash_password
@@ -21,6 +26,7 @@ from src.web.schemas.auth import (
PasswordChangeRequest,
PasswordChangeResponse,
)
from src.web.services.role_permissions import fetch_role_permissions_map
router = APIRouter(prefix="/auth", tags=["auth"])
security = HTTPBearer()
@@ -110,6 +116,7 @@ def get_current_user_with_id(
full_name=username,
is_active=True,
is_admin=False,
role=DEFAULT_USER_ROLE,
)
db.add(user)
db.commit()
@@ -118,6 +125,13 @@ def get_current_user_with_id(
return {"username": username, "user_id": user.id}
def _resolve_user_role(user: User | None, is_admin_flag: bool) -> str:
"""Determine the role value for a user, ensuring it is valid."""
if user and user.role in ROLE_VALUES:
return user.role
return DEFAULT_ADMIN_ROLE if is_admin_flag else DEFAULT_USER_ROLE
@router.post("/login", response_model=TokenResponse)
def login(credentials: LoginRequest, db: Session = Depends(get_db)) -> TokenResponse:
"""Authenticate user and return tokens.
@@ -262,6 +276,7 @@ def get_current_user_info(
full_name=username,
is_active=True,
is_admin=True,
role=DEFAULT_ADMIN_ROLE,
)
db.add(user)
db.commit()
@@ -275,6 +290,7 @@ def get_current_user_info(
# Update existing user to be admin if no admins exist
if not user.is_admin:
user.is_admin = True
user.role = DEFAULT_ADMIN_ROLE
db.commit()
db.refresh(user)
is_admin = user.is_admin
@@ -285,7 +301,16 @@ def get_current_user_info(
else:
is_admin = user.is_admin if user else False
return UserResponse(username=username, is_admin=is_admin)
role_value = _resolve_user_role(user, is_admin)
permissions_map = fetch_role_permissions_map(db)
permissions = permissions_map.get(role_value, {})
return UserResponse(
username=username,
is_admin=is_admin,
role=role_value,
permissions=permissions,
)
@router.post("/change-password", response_model=PasswordChangeResponse)
src/web/api/pending_identifications.py
+13 -5
View File
@@ -10,9 +10,10 @@ from pydantic import BaseModel, ConfigDict
from sqlalchemy import text, func
from sqlalchemy.orm import Session
from src.web.constants.roles import DEFAULT_USER_ROLE
from src.web.db.session import get_auth_db, get_db
from src.web.db.models import Face, Person, PersonEncoding, User
from src.web.api.users import get_current_admin_user
from src.web.api.users import get_current_admin_user, require_feature_permission
from src.web.utils.password import hash_password
router = APIRouter(prefix="/pending-identifications", tags=["pending-identifications"])
@@ -43,6 +44,7 @@ def get_or_create_frontend_user(db: Session) -> User:
full_name="Frontend System User",
is_active=False, # Not an active user, just a system marker
is_admin=False,
role=DEFAULT_USER_ROLE,
password_change_required=False,
)
db.add(user)
@@ -144,7 +146,9 @@ class ClearDatabaseResponse(BaseModel):
@router.get("", response_model=PendingIdentificationsListResponse)
def list_pending_identifications(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_user: Annotated[
dict, Depends(require_feature_permission("user_identified"))
],
include_denied: bool = False,
db: Session = Depends(get_auth_db),
main_db: Session = Depends(get_db),
@@ -240,7 +244,9 @@ def list_pending_identifications(
@router.post("/approve-deny", response_model=ApproveDenyResponse)
def approve_deny_pending_identifications(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_user: Annotated[
dict, Depends(require_feature_permission("user_identified"))
],
request: ApproveDenyRequest,
auth_db: Session = Depends(get_auth_db),
main_db: Session = Depends(get_db),
@@ -401,7 +407,9 @@ def approve_deny_pending_identifications(
@router.get("/report", response_model=IdentificationReportResponse)
def get_identification_report(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_user: Annotated[
dict, Depends(require_feature_permission("user_identified"))
],
date_from: Optional[str] = Query(None, description="Filter by identification date (from) - YYYY-MM-DD"),
date_to: Optional[str] = Query(None, description="Filter by identification date (to) - YYYY-MM-DD"),
main_db: Session = Depends(get_db),
@@ -484,7 +492,7 @@ def get_identification_report(
@router.post("/clear-denied", response_model=ClearDatabaseResponse)
def clear_denied_identifications(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_admin: dict = Depends(get_current_admin_user),
auth_db: Session = Depends(get_auth_db),
) -> ClearDatabaseResponse:
"""Delete all denied pending identifications from the database.
src/web/api/pending_photos.py
+10 -6
View File
@@ -14,7 +14,7 @@ from sqlalchemy import text
from sqlalchemy.orm import Session
from src.web.db.session import get_auth_db, get_db
from src.web.api.users import get_current_admin_user
from src.web.api.users import get_current_admin_user, require_feature_permission
from src.web.api.auth import get_current_user
from src.web.services.photo_service import import_photo_from_path, calculate_file_hash
from src.web.settings import PHOTO_STORAGE_DIR
@@ -83,7 +83,9 @@ class ReviewResponse(BaseModel):
@router.get("", response_model=PendingPhotosListResponse)
def list_pending_photos(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_user: Annotated[
dict, Depends(require_feature_permission("user_uploaded"))
],
status_filter: Optional[str] = None,
auth_db: Session = Depends(get_auth_db),
) -> PendingPhotosListResponse:
@@ -244,7 +246,9 @@ def get_pending_photo_image(
@router.post("/review", response_model=ReviewResponse)
def review_pending_photos(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_user: Annotated[
dict, Depends(require_feature_permission("user_uploaded"))
],
request: ReviewRequest,
auth_db: Session = Depends(get_auth_db),
main_db: Session = Depends(get_db),
@@ -267,7 +271,7 @@ def review_pending_photos(
rejected_count = 0
duplicate_count = 0
errors = []
admin_user_id = current_admin.get("user_id")
admin_user_id = current_user.get("user_id")
now = datetime.utcnow()
# Base directories
@@ -453,7 +457,7 @@ class CleanupResponse(BaseModel):
@router.post("/cleanup-files", response_model=CleanupResponse)
def cleanup_shared_files(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_admin: dict = Depends(get_current_admin_user),
status_filter: Optional[str] = Query(None, description="Filter by status: 'approved', 'rejected', or None for both"),
auth_db: Session = Depends(get_auth_db),
) -> CleanupResponse:
@@ -526,7 +530,7 @@ def cleanup_shared_files(
@router.post("/cleanup-database", response_model=CleanupResponse)
def cleanup_pending_photos_database(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_admin: dict = Depends(get_current_admin_user),
status_filter: Optional[str] = Query(None, description="Filter by status: 'approved', 'rejected', or None for all"),
auth_db: Session = Depends(get_auth_db),
) -> CleanupResponse:
src/web/api/reported_photos.py
+9 -5
View File
@@ -12,7 +12,7 @@ from sqlalchemy.orm import Session
from src.web.db.session import get_auth_db, get_db
from src.web.db.models import Photo, PhotoTagLinkage
from src.web.api.users import get_current_admin_user
from src.web.api.users import get_current_admin_user, require_feature_permission
router = APIRouter(prefix="/reported-photos", tags=["reported-photos"])
@@ -87,7 +87,9 @@ class CleanupResponse(BaseModel):
@router.get("", response_model=ReportedPhotosListResponse)
def list_reported_photos(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_user: Annotated[
dict, Depends(require_feature_permission("user_reported"))
],
status_filter: Optional[str] = None,
auth_db: Session = Depends(get_auth_db),
main_db: Session = Depends(get_db),
@@ -176,7 +178,9 @@ def list_reported_photos(
@router.post("/review", response_model=ReviewResponse)
def review_reported_photos(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_user: Annotated[
dict, Depends(require_feature_permission("user_reported"))
],
request: ReviewRequest,
auth_db: Session = Depends(get_auth_db),
main_db: Session = Depends(get_db),
@@ -194,7 +198,7 @@ def review_reported_photos(
kept_count = 0
removed_count = 0
errors = []
admin_user_id = current_admin.get("user_id")
admin_user_id = current_user.get("user_id")
now = datetime.utcnow()
for decision in request.decisions:
@@ -299,7 +303,7 @@ def review_reported_photos(
@router.post("/cleanup", response_model=CleanupResponse)
def cleanup_reported_photos(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
current_admin: dict = Depends(get_current_admin_user),
status_filter: Annotated[
Optional[str],
Query(description="Use 'keep' to clear reviewed or 'remove' to clear dismissed records.")
src/web/api/role_permissions.py
+68
View File
@@ -0,0 +1,68 @@
"""Manage role-to-feature permissions."""
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from src.web.api.users import get_current_admin_user
from src.web.constants.role_features import ROLE_FEATURES, ROLE_FEATURE_KEYS
from src.web.constants.roles import ROLE_VALUES
from src.web.db.session import get_db
from src.web.schemas.role_permissions import (
RoleFeatureSchema,
RolePermissionsResponse,
RolePermissionsUpdateRequest,
)
from src.web.services.role_permissions import (
ensure_role_permissions_initialized,
fetch_role_permissions_map,
set_role_permissions,
)
router = APIRouter(prefix="/role-permissions", tags=["role-permissions"])
@router.get("", response_model=RolePermissionsResponse)
def list_role_permissions(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
db: Session = Depends(get_db),
) -> RolePermissionsResponse:
"""Return the current role/feature permission matrix."""
ensure_role_permissions_initialized(db)
permissions = fetch_role_permissions_map(db)
features = [RoleFeatureSchema(**feature) for feature in ROLE_FEATURES]
return RolePermissionsResponse(features=features, permissions=permissions)
@router.put("", response_model=RolePermissionsResponse)
def update_role_permissions(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
request: RolePermissionsUpdateRequest,
db: Session = Depends(get_db),
) -> RolePermissionsResponse:
"""Update permissions for the provided matrix."""
invalid_roles = set(request.permissions.keys()) - set(ROLE_VALUES)
if invalid_roles:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid role(s): {', '.join(sorted(invalid_roles))}",
)
for feature_map in request.permissions.values():
invalid_features = set(feature_map.keys()) - set(ROLE_FEATURE_KEYS)
if invalid_features:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid feature(s): {', '.join(sorted(invalid_features))}",
)
set_role_permissions(db, request.permissions)
permissions = fetch_role_permissions_map(db)
features = [RoleFeatureSchema(**feature) for feature in ROLE_FEATURES]
return RolePermissionsResponse(features=features, permissions=permissions)
src/web/api/users.py
+140 -11
View File
@@ -2,13 +2,22 @@
from __future__ import annotations
import logging
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from src.web.api.auth import get_current_user
from src.web.constants.roles import (
DEFAULT_ADMIN_ROLE,
DEFAULT_USER_ROLE,
ROLE_VALUES,
UserRole,
is_admin_role,
)
from src.web.db.session import get_auth_db, get_db
from src.web.db.models import User
from src.web.schemas.users import (
@@ -18,8 +27,38 @@ from src.web.schemas.users import (
UsersListResponse,
)
from src.web.utils.password import hash_password
from src.web.services.role_permissions import fetch_role_permissions_map
router = APIRouter(prefix="/users", tags=["users"])
logger = logging.getLogger(__name__)
def _normalize_role_and_admin(
role: str | None,
is_admin_flag: bool | None,
) -> tuple[str, bool]:
"""Normalize requested role/is_admin values into a consistent pair."""
selected_role = role or (DEFAULT_ADMIN_ROLE if is_admin_flag else DEFAULT_USER_ROLE)
if selected_role not in ROLE_VALUES:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid role '{selected_role}'",
)
derived_is_admin = is_admin_role(selected_role)
if is_admin_flag is not None and is_admin_flag != derived_is_admin:
logger.warning(
"Role/is_admin mismatch detected. Using role-derived admin flag.",
extra={"role": selected_role, "is_admin_flag": is_admin_flag},
)
return selected_role, derived_is_admin
def _ensure_role_set(user: User) -> None:
"""Guarantee that a User instance has a valid role value."""
if user.role in ROLE_VALUES:
return
fallback_role = DEFAULT_ADMIN_ROLE if user.is_admin else DEFAULT_USER_ROLE
user.role = fallback_role
def get_auth_db_optional() -> Session | None:
@@ -137,6 +176,7 @@ def get_current_admin_user(
password_hash=default_password_hash,
is_active=True,
is_admin=True,
role=DEFAULT_ADMIN_ROLE,
)
db.add(main_user)
db.commit()
@@ -144,6 +184,7 @@ def get_current_admin_user(
elif not main_user.is_admin:
# User exists but is not admin - make them admin for bootstrap
main_user.is_admin = True
main_user.role = DEFAULT_ADMIN_ROLE
db.add(main_user)
db.commit()
db.refresh(main_user)
@@ -162,6 +203,53 @@ def get_current_admin_user(
return {"username": username, "user_id": main_user.id}
def require_feature_permission(feature_key: str):
"""Return a dependency that enforces feature-level access via role permissions."""
def dependency(
current_user: Annotated[dict, Depends(get_current_user)],
db: Session = Depends(get_db),
) -> dict:
username = current_user["username"]
user = db.query(User).filter(User.username == username).first()
if not user:
default_password_hash = hash_password("changeme")
user = User(
username=username,
password_hash=default_password_hash,
is_active=True,
is_admin=False,
role=DEFAULT_USER_ROLE,
)
db.add(user)
db.commit()
db.refresh(user)
_ensure_role_set(user)
has_access = user.is_admin or is_admin_role(user.role)
if not has_access:
permissions_map = fetch_role_permissions_map(db)
role_permissions = permissions_map.get(user.role, {})
has_access = bool(role_permissions.get(feature_key))
if not has_access:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied for this feature",
)
return {
"username": username,
"user_id": user.id,
"role": user.role,
"is_admin": user.is_admin,
}
return dependency
@router.get("", response_model=UsersListResponse)
def list_users(
current_admin: Annotated[dict, Depends(get_current_admin_user)],
@@ -182,6 +270,8 @@ def list_users(
query = query.filter(User.is_admin == is_admin)
users = query.order_by(User.username.asc()).all()
for user in users:
_ensure_role_set(user)
items = [UserResponse.model_validate(u) for u in users]
return UsersListResponse(items=items, total=len(items))
@@ -215,6 +305,16 @@ def create_user(
# Hash the password before storing
password_hash = hash_password(request.password)
if request.role is None:
requested_role = None
elif isinstance(request.role, UserRole):
requested_role = request.role.value
else:
requested_role = str(request.role)
normalized_role, normalized_is_admin = _normalize_role_and_admin(
requested_role,
request.is_admin,
)
user = User(
username=request.username,
@@ -222,7 +322,8 @@ def create_user(
email=request.email,
full_name=request.full_name,
is_active=request.is_active,
is_admin=request.is_admin,
is_admin=normalized_is_admin,
role=normalized_role,
password_change_required=True, # Force password change on first login
)
db.add(user)
@@ -234,7 +335,7 @@ def create_user(
email=request.email,
full_name=request.full_name,
password_hash=password_hash,
is_admin=request.is_admin,
is_admin=normalized_is_admin,
)
return UserResponse.model_validate(user)
@@ -253,6 +354,7 @@ def get_user(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"User with ID {user_id} not found",
)
_ensure_role_set(user)
return UserResponse.model_validate(user)
@@ -271,12 +373,26 @@ def update_user(
detail=f"User with ID {user_id} not found",
)
if request.role is None:
desired_role = None
elif isinstance(request.role, UserRole):
desired_role = request.role.value
else:
desired_role = str(request.role)
if desired_role is None:
if request.is_admin is not None:
desired_role = DEFAULT_ADMIN_ROLE if request.is_admin else DEFAULT_USER_ROLE
elif user.role:
desired_role = user.role
else:
desired_role = DEFAULT_ADMIN_ROLE if user.is_admin else DEFAULT_USER_ROLE
normalized_role, normalized_is_admin = _normalize_role_and_admin(
desired_role,
request.is_admin,
)
# Prevent admin from removing their own admin status
if (
current_admin["username"] == user.username
and request.is_admin is not None
and not request.is_admin
):
if current_admin["username"] == user.username and not normalized_is_admin:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot remove your own admin status",
@@ -300,8 +416,8 @@ def update_user(
user.full_name = request.full_name
if request.is_active is not None:
user.is_active = request.is_active
if request.is_admin is not None:
user.is_admin = request.is_admin
user.is_admin = normalized_is_admin
user.role = normalized_role
db.add(user)
db.commit()
@@ -342,8 +458,21 @@ def delete_user(
detail="Cannot delete your own account",
)
db.delete(user)
db.commit()
try:
db.delete(user)
db.commit()
except IntegrityError as exc:
db.rollback()
constraint_name = "faces_identified_by_user_id_fkey"
if exc.orig and constraint_name in str(exc.orig):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=(
"This user has identified faces and cannot be deleted. "
"Set the user inactive instead."
),
) from exc
raise
return Response(status_code=status.HTTP_204_NO_CONTENT)
src/web/app.py
+77
View File
@@ -23,12 +23,15 @@ from src.web.api.pending_photos import router as pending_photos_router
from src.web.api.tags import router as tags_router
from src.web.api.users import router as users_router
from src.web.api.auth_users import router as auth_users_router
from src.web.api.role_permissions import router as role_permissions_router
from src.web.api.version import router as version_router
from src.web.settings import APP_TITLE, APP_VERSION
from src.web.constants.roles import DEFAULT_ADMIN_ROLE, DEFAULT_USER_ROLE, ROLE_VALUES
from src.web.db.base import Base, engine
from src.web.db.session import database_url
# Import models to ensure they're registered with Base.metadata
from src.web.db import models # noqa: F401
from src.web.db.models import RolePermission
from src.web.utils.password import hash_password
# Global worker process (will be set in lifespan)
@@ -262,6 +265,77 @@ def ensure_face_identified_by_user_id_column(inspector) -> None:
print("✅ Added identified_by_user_id column to faces table")
def ensure_user_role_column(inspector) -> None:
"""Ensure users table has a role column with valid values."""
if "users" not in inspector.get_table_names():
return
columns = {column["name"] for column in inspector.get_columns("users")}
dialect = engine.dialect.name
role_values = sorted(ROLE_VALUES)
placeholder_parts = ", ".join(
f":role_value_{index}" for index, _ in enumerate(role_values)
)
where_clause = (
"role IS NULL OR role = ''"
if not placeholder_parts
else f"role IS NULL OR role = '' OR role NOT IN ({placeholder_parts})"
)
params = {
f"role_value_{index}": value for index, value in enumerate(role_values)
}
params["admin_role"] = DEFAULT_ADMIN_ROLE
params["default_role"] = DEFAULT_USER_ROLE
with engine.connect() as connection:
with connection.begin():
if "role" not in columns:
if dialect == "postgresql":
connection.execute(
text(
f"ALTER TABLE users ADD COLUMN IF NOT EXISTS role TEXT "
f"NOT NULL DEFAULT '{DEFAULT_USER_ROLE}'"
)
)
else:
connection.execute(
text(
f"ALTER TABLE users ADD COLUMN role TEXT "
f"DEFAULT '{DEFAULT_USER_ROLE}'"
)
)
connection.execute(
text(
f"""
UPDATE users
SET role = CASE
WHEN is_admin THEN :admin_role
ELSE :default_role
END
WHERE {where_clause}
"""
),
params,
)
connection.execute(
text("CREATE INDEX IF NOT EXISTS idx_users_role ON users(role)")
)
print("✅ Ensured users.role column exists and is populated")
def ensure_role_permissions_table(inspector) -> None:
"""Ensure the role_permissions table exists for permission matrix."""
if "role_permissions" in inspector.get_table_names():
return
try:
print("🔄 Creating role_permissions table...")
RolePermission.__table__.create(bind=engine, checkfirst=True)
print("✅ Created role_permissions table")
except Exception as exc:
print(f"⚠️ Failed to create role_permissions table: {exc}")
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Lifespan context manager for startup and shutdown events."""
@@ -297,6 +371,8 @@ async def lifespan(app: FastAPI):
ensure_user_password_change_required_column(inspector)
ensure_user_email_unique_constraint(inspector)
ensure_face_identified_by_user_id_column(inspector)
ensure_user_role_column(inspector)
ensure_role_permissions_table(inspector)
except Exception as exc:
print(f"❌ Database initialization failed: {exc}")
raise
@@ -337,6 +413,7 @@ def create_app() -> FastAPI:
app.include_router(tags_router, prefix="/api/v1")
app.include_router(users_router, prefix="/api/v1")
app.include_router(auth_users_router, prefix="/api/v1")
app.include_router(role_permissions_router, prefix="/api/v1")
return app
src/web/constants/role_features.py
+42
View File
@@ -0,0 +1,42 @@
"""Feature definitions and default role permissions."""
from __future__ import annotations
from typing import Dict, Final, List, Set
from src.web.constants.roles import UserRole
ROLE_FEATURES: Final[List[dict[str, str]]] = [
{"key": "scan", "label": "Scan"},
{"key": "process", "label": "Process"},
{"key": "search_photos", "label": "Search Photos"},
{"key": "identify_people", "label": "Identify People"},
{"key": "auto_match", "label": "Auto-Match"},
{"key": "modify_people", "label": "Modify People"},
{"key": "tag_photos", "label": "Tag Photos"},
{"key": "faces_maintenance", "label": "Faces Maintenance"},
{"key": "user_identified", "label": "User Identified"},
{"key": "user_reported", "label": "User Reported"},
{"key": "user_uploaded", "label": "User Uploaded"},
{"key": "manage_users", "label": "Manage Users"},
{"key": "manage_roles", "label": "Manage Roles"},
]
ROLE_FEATURE_KEYS: Final[List[str]] = [feature["key"] for feature in ROLE_FEATURES]
DEFAULT_ROLE_FEATURE_MATRIX: Final[Dict[str, Set[str]]] = {
UserRole.ADMIN.value: set(ROLE_FEATURE_KEYS),
UserRole.MANAGER.value: set(ROLE_FEATURE_KEYS),
UserRole.MODERATOR.value: {"scan", "process", "manage_users"},
UserRole.REVIEWER.value: {"user_identified", "user_reported", "user_uploaded"},
UserRole.EDITOR.value: {"user_identified", "user_uploaded", "manage_users"},
UserRole.IMPORTER.value: {"user_uploaded"},
UserRole.VIEWER.value: {"user_identified", "user_reported"},
}
def get_default_permission(role: str, feature_key: str) -> bool:
"""Return the default allowed value for a role/feature pair."""
allowed_features = DEFAULT_ROLE_FEATURE_MATRIX.get(role, set())
return feature_key in allowed_features
src/web/constants/roles.py
+32
View File
@@ -0,0 +1,32 @@
"""Shared role definitions for backend user management."""
from __future__ import annotations
from enum import Enum
from typing import Final, Set
class UserRole(str, Enum):
"""Enumerated set of supported user roles."""
ADMIN = "admin"
MANAGER = "manager"
MODERATOR = "moderator"
REVIEWER = "reviewer"
EDITOR = "editor"
IMPORTER = "importer"
VIEWER = "viewer"
ROLE_VALUES: Final[Set[str]] = {role.value for role in UserRole}
ADMIN_ROLE_VALUES: Final[Set[str]] = {
UserRole.ADMIN.value,
}
DEFAULT_ADMIN_ROLE: Final[str] = UserRole.ADMIN.value
DEFAULT_USER_ROLE: Final[str] = UserRole.VIEWER.value
def is_admin_role(role: str) -> bool:
"""Return True when the provided role is considered an admin role."""
return role in ADMIN_ROLE_VALUES
src/web/db/models.py
+26
View File
@@ -21,6 +21,8 @@ from sqlalchemy import (
)
from sqlalchemy.orm import declarative_base, relationship
from src.web.constants.roles import DEFAULT_USER_ROLE
if TYPE_CHECKING:
pass
@@ -212,6 +214,13 @@ class User(Base):
full_name = Column(Text, nullable=False)
is_active = Column(Boolean, default=True, nullable=False)
is_admin = Column(Boolean, default=False, nullable=False, index=True)
role = Column(
Text,
nullable=False,
default=DEFAULT_USER_ROLE,
server_default=DEFAULT_USER_ROLE,
index=True,
)
password_change_required = Column(Boolean, default=True, nullable=False, index=True)
created_date = Column(DateTime, default=datetime.utcnow, nullable=False)
last_login = Column(DateTime, nullable=True)
@@ -221,5 +230,22 @@ class User(Base):
Index("idx_users_email", "email"),
Index("idx_users_is_admin", "is_admin"),
Index("idx_users_password_change_required", "password_change_required"),
Index("idx_users_role", "role"),
)
class RolePermission(Base):
"""Role-to-feature permission matrix."""
__tablename__ = "role_permissions"
id = Column(Integer, primary_key=True, autoincrement=True)
role = Column(Text, nullable=False, index=True)
feature_key = Column(Text, nullable=False, index=True)
allowed = Column(Boolean, nullable=False, default=False, server_default="0")
__table_args__ = (
UniqueConstraint("role", "feature_key", name="uq_role_feature"),
Index("idx_role_permissions_role_feature", "role", "feature_key"),
)
src/web/schemas/auth.py
+6
View File
@@ -2,8 +2,12 @@
from __future__ import annotations
from typing import Dict
from pydantic import BaseModel, ConfigDict
from src.web.constants.roles import DEFAULT_USER_ROLE, UserRole
class LoginRequest(BaseModel):
"""Login request payload."""
@@ -39,6 +43,8 @@ class UserResponse(BaseModel):
username: str
is_admin: bool = False
role: UserRole = DEFAULT_USER_ROLE
permissions: Dict[str, bool] = {}
class PasswordChangeRequest(BaseModel):
src/web/schemas/role_permissions.py
+42
View File
@@ -0,0 +1,42 @@
"""Schemas for role permissions management."""
from __future__ import annotations
from typing import Dict
from pydantic import BaseModel, ConfigDict, Field
from src.web.constants.role_features import ROLE_FEATURES
from src.web.constants.roles import UserRole
class RoleFeatureSchema(BaseModel):
"""Feature metadata visible in the UI."""
key: str
label: str
class RolePermissionsResponse(BaseModel):
"""Payload returned when listing role permissions."""
model_config = ConfigDict(protected_namespaces=())
features: list[RoleFeatureSchema]
permissions: Dict[UserRole, Dict[str, bool]]
class RolePermissionsUpdateRequest(BaseModel):
"""Payload for updating role permissions."""
model_config = ConfigDict(protected_namespaces=())
permissions: Dict[UserRole, Dict[str, bool]] = Field(
...,
description="Map of role -> {feature_key: allowed}",
)
@staticmethod
def build_feature_list() -> list[RoleFeatureSchema]:
return [RoleFeatureSchema(**feature) for feature in ROLE_FEATURES]
src/web/schemas/users.py
+11
View File
@@ -7,6 +7,8 @@ from typing import Optional
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from src.web.constants.roles import DEFAULT_USER_ROLE, UserRole
class UserResponse(BaseModel):
"""User DTO returned from API."""
@@ -19,6 +21,7 @@ class UserResponse(BaseModel):
full_name: Optional[str] = None
is_active: bool
is_admin: bool
role: UserRole
password_change_required: bool
created_date: datetime
last_login: Optional[datetime] = None
@@ -35,6 +38,10 @@ class UserCreateRequest(BaseModel):
full_name: str = Field(..., min_length=1, max_length=200, description="Full name (required)")
is_active: bool = True
is_admin: bool = False
role: UserRole = Field(
DEFAULT_USER_ROLE,
description="Role for feature-level access; also controls admin status where applicable",
)
give_frontend_permission: bool = Field(False, description="Create user in auth database for frontend access")
@@ -48,6 +55,10 @@ class UserUpdateRequest(BaseModel):
full_name: str = Field(..., min_length=1, max_length=200, description="Full name (required)")
is_active: Optional[bool] = None
is_admin: Optional[bool] = None
role: Optional[UserRole] = Field(
None,
description="Updated role; determines admin status when provided",
)
give_frontend_permission: Optional[bool] = Field(
None,
description="Create user in auth database for frontend access if True",
src/web/services/face_service.py
+3 -3
View File
@@ -1248,11 +1248,11 @@ def list_unidentified_faces(
)
else:
# Photos that have ANY of the specified tags
query = (
query.join(PhotoTagLinkage, Photo.id == PhotoTagLinkage.photo_id)
tagged_photo_ids_subquery = (
db.query(PhotoTagLinkage.photo_id)
.filter(PhotoTagLinkage.tag_id.in_(tag_ids))
.distinct()
)
query = query.filter(Face.photo_id.in_(tagged_photo_ids_subquery))
else:
# No matching tags found - return empty result
return [], 0
src/web/services/role_permissions.py
+73
View File
@@ -0,0 +1,73 @@
"""Role permission helpers for ensuring and updating access matrix."""
from __future__ import annotations
from collections import defaultdict
from typing import Dict
from sqlalchemy import select
from sqlalchemy.orm import Session
from src.web.constants.role_features import (
ROLE_FEATURE_KEYS,
get_default_permission,
)
from src.web.constants.roles import ROLE_VALUES
from src.web.db.models import RolePermission
def ensure_role_permissions_initialized(session: Session) -> None:
"""Seed permissions table once using default matrix if table is empty."""
has_permissions = session.execute(select(RolePermission.id)).first()
if has_permissions:
return
for role in ROLE_VALUES:
for feature_key in ROLE_FEATURE_KEYS:
permission = RolePermission(
role=role,
feature_key=feature_key,
allowed=get_default_permission(role, feature_key),
)
session.add(permission)
session.commit()
def fetch_role_permissions_map(session: Session) -> Dict[str, Dict[str, bool]]:
"""Return permissions map keyed by role then feature."""
ensure_role_permissions_initialized(session)
permissions = defaultdict(dict)
results = session.execute(select(RolePermission)).scalars().all()
for perm in results:
permissions[perm.role][perm.feature_key] = bool(perm.allowed)
return dict(permissions)
def set_role_permissions(session: Session, permissions: Dict[str, Dict[str, bool]]) -> None:
"""Update permissions based on provided map."""
ensure_role_permissions_initialized(session)
existing = {
(perm.role, perm.feature_key): perm
for perm in session.execute(select(RolePermission)).scalars().all()
}
updated = False
for role, feature_map in permissions.items():
for feature_key, allowed in feature_map.items():
key = (role, feature_key)
perm = existing.get(key)
if perm is None:
perm = RolePermission(role=role, feature_key=feature_key)
session.add(perm)
existing[key] = perm
if perm.allowed != bool(allowed):
perm.allowed = bool(allowed)
updated = True
if updated:
session.commit()