fix(auth): enforce role permissions on REST storage and settings mutations

.gitignore

@@ -7,6 +7,8 @@ venv/
 .worktrees/
 .coverage
 .coverage.*
+coverage/
+coverage.xml
 .pytest_cache/
 htmlcov/
 *.log

REQUIREMENTS.md

@@ -2,7 +2,7 @@
 
 > **Note:** This document is automatically generated and verified against the live test suite by `scripts/generate_requirements.py` and `tests/backend/test_requirements_sync.py`.
 
-**Test Verification Baseline:** **1052 Automated Tests** (699 Pytest Backend + 303 Vitest Frontend + 50 Playwright E2E).
+**Test Verification Baseline:** **1056 Automated Tests** (703 Pytest Backend + 303 Vitest Frontend + 50 Playwright E2E).
 
 ---
 
@@ -1192,6 +1192,12 @@ and leaves the prior indexed state intact without data loss._
 - `test_large_file_auto_summarization`
 - `test_large_file_disabled_summarization`
 
+#### `tests/test_rest_role_permissions.py` (4 tests)
+- `test_anonymous_and_invalid_key_requests_denied` - _Anonymous or invalid/revoked requests receive 401 Unauthorized._
+- `test_viewer_role_permissions` - _Viewer key can read resources but is denied content mutations & administrative settings (403)._
+- `test_editor_role_permissions` - _Editor key can perform content mutations and syncs, but is denied administrative settings (403)._
+- `test_admin_role_permissions` - _Admin key has full permissions across configuration, editor, and viewer endpoints._
+
 #### `tests/test_storage_api_routes.py` (4 tests)
 - `test_storage_upload_and_get_file`
 - `test_storage_upload_multipart_and_put`

app/api/routers/files.py

@@ -1,12 +1,13 @@
 import logging
 from typing import Optional
 from pydantic import BaseModel, Field
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, Query, Depends
 from fastapi.responses import JSONResponse
 
 from app.services.file_reader import get_file_reader_service
 from app.services.summarizer import get_summarizer_service
-from app.services.auth import ForbiddenError
+from app.services.auth import Role, ForbiddenError
+from app.api.routers.auth import require_role
 
 logger = logging.getLogger("contextcortex.api.files")
 router = APIRouter()
@@ -18,7 +19,10 @@ class FileSummarizePayload(BaseModel):
     force_refresh: bool = Field(False, description="Whether to bypass cache and regenerate summary")
 
 
-@router.get("/admin/api/files/read")
+@router.get(
+    "/admin/api/files/read",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_read_file(
     path: str = Query(..., description="Target file path (relative to repo/storage or absolute within indexed root)"),
     repo: Optional[str] = Query(None, description="Optional repo or storage namespace"),
@@ -57,7 +61,10 @@ async def api_read_file(
         return JSONResponse(status_code=500, content={"error": "Internal server error reading file."})
 
 
-@router.post("/admin/api/files/summarize")
+@router.post(
+    "/admin/api/files/summarize",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_summarize_file(payload: FileSummarizePayload):
     try:
         raw_path = payload.path.strip() if payload.path else ""

app/api/routers/graph.py

@@ -4,16 +4,21 @@
 import sqlite3
 import logging
 from typing import Optional
-from fastapi import APIRouter, Request
+from fastapi import APIRouter, Request, Depends
 from fastapi.responses import JSONResponse
 
+from app.services.auth import Role
+from app.api.routers.auth import require_role
 import app.services.topology as topo_service
 
 logger = logging.getLogger("contextcortex.api")
 
 router = APIRouter()
 
-@router.get("/admin/api/graph/topology")
+@router.get(
+    "/admin/api/graph/topology",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_graph_topology(
     repo: str,
     view_type: str = "files",
@@ -44,7 +49,10 @@ async def api_get_graph_topology(
         logger.error(f"Error generating topology graph: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to generate topology graph."})
 
-@router.get("/admin/api/graph/node-details")
+@router.get(
+    "/admin/api/graph/node-details",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_graph_node_details(id: str):
     try:
         details = topo_service.get_node_details(id)

app/api/routers/ingestion.py

@@ -1,15 +1,20 @@
 import logging
 from typing import Optional
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, Query, Depends
 from fastapi.responses import JSONResponse
 
+from app.services.auth import Role
+from app.api.routers.auth import require_role
 from app.services.database import get_db_connection
 from app.services.local_storage import get_local_storage_service
 
 logger = logging.getLogger("contextcortex.api.ingestion")
 router = APIRouter()
 
-@router.get("/admin/api/ingestion/catalog")
+@router.get(
+    "/admin/api/ingestion/catalog",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_ingestion_catalog(
     source_type: str = Query("all", description="Source type filter"),
     repo_name: Optional[str] = Query(None, description="Repository filter"),

app/api/routers/navigator.py

@@ -1,16 +1,21 @@
 import logging
 from typing import Optional
-from fastapi import APIRouter, Query
+from fastapi import APIRouter, Query, Depends
 from fastapi.responses import JSONResponse
 
+from app.services.auth import Role
+from app.api.routers.auth import require_role
 import app.services.navigator as nav_service
 
 logger = logging.getLogger("contextcortex.api.navigator")
 
 router = APIRouter()
 
 
-@router.get("/admin/api/navigator/tree")
+@router.get(
+    "/admin/api/navigator/tree",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_navigator_tree(
     repo: str = Query(..., description="Repository name or '__all__'")
 ):
@@ -27,7 +32,10 @@ async def api_get_navigator_tree(
         return JSONResponse(status_code=500, content={"error": "Failed to generate navigator tree."})
 
 
-@router.get("/admin/api/navigator/file-outline")
+@router.get(
+    "/admin/api/navigator/file-outline",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_file_outline(
     repo: str = Query(..., description="Repository name or '__all__'"),
     filepath: str = Query(..., description="Relative file path")
@@ -45,7 +53,10 @@ async def api_get_file_outline(
         return JSONResponse(status_code=500, content={"error": "Failed to retrieve file outline."})
 
 
-@router.get("/admin/api/navigator/symbol-impact")
+@router.get(
+    "/admin/api/navigator/symbol-impact",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_symbol_impact(
     repo: str = Query(..., description="Repository name or '__all__'"),
     symbol_id: int = Query(..., description="AST symbol ID")
@@ -63,7 +74,10 @@ async def api_get_symbol_impact(
         return JSONResponse(status_code=500, content={"error": "Failed to retrieve symbol impact."})
 
 
-@router.get("/admin/api/navigator/omni-search")
+@router.get(
+    "/admin/api/navigator/omni-search",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_omni_search(
     repo: str = Query(..., description="Repository name or '__all__'"),
     q: str = Query("", description="Search query string"),

app/api/routers/repositories.py

@@ -6,8 +6,10 @@
 import logging
 import asyncio
 from typing import Optional
-from fastapi import APIRouter, Request
+from fastapi import APIRouter, Request, Depends
 from fastapi.responses import JSONResponse, StreamingResponse
+from app.services.auth import Role
+from app.api.routers.auth import require_role
 
 from app.models.schemas import RepoConfig, LocalPathConfig, AutoSyncToggleRequest, KeepShallowToggleRequest, SearchRequest
 import app.services.database as db_service
@@ -21,7 +23,10 @@
 
 router = APIRouter()
 
-@router.get("/admin/api/repos")
+@router.get(
+    "/admin/api/repos",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_repos():
     try:
         with db_service.get_db_connection() as conn:
@@ -45,7 +50,10 @@ async def api_get_repos():
         logger.error(f"Error fetching repos: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to fetch repositories."})
 
-@router.post("/admin/api/repos")
+@router.post(
+    "/admin/api/repos",
+    dependencies=[Depends(require_role(Role.ADMIN))],
+)
 async def api_add_repo(repo: RepoConfig):
     if not repo.name or not repo.name.strip() or not repo.url or not repo.url.strip():
         return JSONResponse(status_code=400, content={"error": "Name and URL are required"})
@@ -74,7 +82,10 @@ async def api_add_repo(repo: RepoConfig):
         logger.error(f"Error adding repo: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to add repository."})
 
-@router.patch("/admin/api/repos/{repo_id}/auto-sync")
+@router.patch(
+    "/admin/api/repos/{repo_id}/auto-sync",
+    dependencies=[Depends(require_role(Role.ADMIN))],
+)
 async def api_toggle_repo_auto_sync(repo_id: int, payload: AutoSyncToggleRequest):
     try:
         updated = db_service.set_repo_auto_sync(repo_id, payload.auto_sync)
@@ -85,7 +96,10 @@ async def api_toggle_repo_auto_sync(repo_id: int, payload: AutoSyncToggleRequest
         logger.error(f"Error toggling auto-sync for repo {repo_id}: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to toggle auto-sync for repository."})
 
-@router.patch("/admin/api/repos/{repo_id}/keep-shallow")
+@router.patch(
+    "/admin/api/repos/{repo_id}/keep-shallow",
+    dependencies=[Depends(require_role(Role.ADMIN))],
+)
 async def api_toggle_repo_keep_shallow(repo_id: int, payload: KeepShallowToggleRequest):
     try:
         with db_service.get_db_connection() as conn:
@@ -100,8 +114,14 @@ async def api_toggle_repo_keep_shallow(repo_id: int, payload: KeepShallowToggleR
         logger.error(f"Error toggling keep-shallow for repo {repo_id}: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to toggle keep-shallow for repository."})
 
-@router.post("/admin/api/repos/{repo_id}/sync")
-@router.post("/admin/api/repos/sync/{repo_id}")
+@router.post(
+    "/admin/api/repos/{repo_id}/sync",
+    dependencies=[Depends(require_role(Role.EDITOR))],
+)
+@router.post(
+    "/admin/api/repos/sync/{repo_id}",
+    dependencies=[Depends(require_role(Role.EDITOR))],
+)
 async def api_sync_repo(repo_id: int):
     try:
         with db_service.get_db_connection() as conn:
@@ -122,11 +142,17 @@ async def api_sync_repo(repo_id: int):
 
 STREAM_KEEPALIVE_TIMEOUT = 15.0
 
-@router.get("/admin/api/repos/sync-status")
+@router.get(
+    "/admin/api/repos/sync-status",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_all_sync_status():
     return progress_tracker.get_snapshot()
 
-@router.get("/admin/api/repos/sync/stream")
+@router.get(
+    "/admin/api/repos/sync/stream",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_stream_repo_sync(request: Request):
     queue = progress_tracker.subscribe()
 
@@ -159,21 +185,30 @@ async def event_generator():
         }
     )
 
-@router.get("/admin/api/repos/{repo_id}/sync-status")
+@router.get(
+    "/admin/api/repos/{repo_id}/sync-status",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_repo_sync_status(repo_id: int):
     snapshot = progress_tracker.get_snapshot(repo_id)
     if not snapshot:
         return JSONResponse(status_code=404, content={"error": f"No active sync job for repo {repo_id}"})
     return snapshot
 
-@router.post("/admin/api/repos/{repo_id}/cancel-sync")
+@router.post(
+    "/admin/api/repos/{repo_id}/cancel-sync",
+    dependencies=[Depends(require_role(Role.EDITOR))],
+)
 async def api_cancel_repo_sync(repo_id: int):
     cancelled = progress_tracker.cancel_job(repo_id)
     if not cancelled:
         return JSONResponse(status_code=400, content={"error": f"Repo {repo_id} is not currently syncing"})
     return {"status": "cancelled", "repo_id": repo_id}
 
-@router.delete("/admin/api/repos/{repo_id}")
+@router.delete(
+    "/admin/api/repos/{repo_id}",
+    dependencies=[Depends(require_role(Role.ADMIN))],
+)
 async def api_delete_repo(repo_id: int):
     try:
         with db_service.get_db_connection() as conn:
@@ -206,7 +241,10 @@ async def api_delete_repo(repo_id: int):
         logger.error(f"Error deleting repo {repo_id}: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to delete repository."})
 
-@router.get("/admin/api/paths")
+@router.get(
+    "/admin/api/paths",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_get_paths():
     try:
         with db_service.get_db_connection() as conn:
@@ -216,7 +254,10 @@ async def api_get_paths():
         logger.error(f"Error getting paths: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to get paths."})
 
-@router.post("/admin/api/paths")
+@router.post(
+    "/admin/api/paths",
+    dependencies=[Depends(require_role(Role.ADMIN))],
+)
 async def api_add_path(config: LocalPathConfig):
     try:
         raw_path = config.path.strip() if config.path else ""
@@ -252,7 +293,10 @@ async def api_add_path(config: LocalPathConfig):
         logger.error(f"Error adding path: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to add path."})
 
-@router.delete("/admin/api/paths/{path_id}")
+@router.delete(
+    "/admin/api/paths/{path_id}",
+    dependencies=[Depends(require_role(Role.ADMIN))],
+)
 async def api_delete_path(path_id: int):
     try:
         with db_service.get_db_connection() as conn:
@@ -276,8 +320,14 @@ async def api_delete_path(path_id: int):
         logger.error(f"Error deleting path {path_id}: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to delete path."})
 
-@router.post("/admin/api/sync")
-@router.post("/admin/api/reindex")
+@router.post(
+    "/admin/api/sync",
+    dependencies=[Depends(require_role(Role.EDITOR))],
+)
+@router.post(
+    "/admin/api/reindex",
+    dependencies=[Depends(require_role(Role.EDITOR))],
+)
 async def api_trigger_sync():
     is_idx = idx_service.is_indexing
     if (is_idx() if callable(is_idx) else is_idx):
@@ -286,7 +336,10 @@ async def api_trigger_sync():
     threading.Thread(target=idx_service.run_full_indexing, daemon=True).start()
     return {"status": "success", "message": "Background ingestion pipeline dispatched."}
 
-@router.post("/admin/api/search/test")
+@router.post(
+    "/admin/api/search/test",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_test_search(payload: SearchRequest):
     try:
         query = payload.query.strip() if payload.query else ""
@@ -326,7 +379,10 @@ async def api_test_search(payload: SearchRequest):
         logger.error(f"Error testing search: {e}")
         return JSONResponse(status_code=500, content={"error": "Failed to execute search test."})
 
-@router.get("/admin/api/browse")
+@router.get(
+    "/admin/api/browse",
+    dependencies=[Depends(require_role(Role.VIEWER))],
+)
 async def api_browse_dir(path: str = "/"):
     cleaned = path.strip() if path else "/"
     if "\x00" in cleaned or any(part in ("..", ".") for part in cleaned.replace("\\", "/").split("/") if part):