Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 2 additions & 7 deletions apps/backend/main.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,12 @@
from __future__ import annotations
import logging
import os

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from prometheus_client import make_asgi_app

from libs.config.settings import settings
from apps.backend.routes import alerts, ingest, snapshot, cameras, feedback
from apps.backend.routes import router as feature_routers

logging.basicConfig(
level = logging.INFO,
Expand All @@ -30,11 +29,7 @@ def create_app() -> FastAPI:
allow_headers = ["*"],
)

app.include_router(ingest.router, prefix="/ingest", tags=["ingest"])
app.include_router(alerts.router, prefix="/alerts", tags=["alerts"])
app.include_router(snapshot.router, prefix="/snapshot", tags=["snapshot"])
app.include_router(cameras.router, prefix="/cameras", tags=["cameras"])
app.include_router(feedback.router, prefix="/feedback", tags=["feedback"])
app.include_router(feature_routers)

# Prometheus metrics scrape endpoint
metrics_app = make_asgi_app()
Expand Down
32 changes: 32 additions & 0 deletions apps/backend/routes/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
"""
Auto-discover and register all routers in this package.
Each module may expose one or more APIRouter instances.
"""
from __future__ import annotations
import importlib
import pkgutil
import logging

from fastapi import APIRouter

logger = logging.getLogger(__name__)

_discovered_router = APIRouter()


def _autodiscover():
package_dir = __path__[0]
for _, module_name, _ in pkgutil.iter_modules([package_dir]):
if module_name in ("__init__",):
continue
module = importlib.import_module(f"{__name__}.{module_name}")
for attr_name in dir(module):
obj = getattr(module, attr_name)
if isinstance(obj, APIRouter):
_discovered_router.include_router(obj)
logger.debug("Discovered router %s from %s", attr_name, module_name)


_autodiscover()

router = _discovered_router
2 changes: 1 addition & 1 deletion apps/backend/routes/alerts.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

logger = logging.getLogger("eagle.alerts")

router = APIRouter()
router = APIRouter(prefix="/alerts", tags=["alerts"])

def _parse_alert(raw: str) -> AlertResponse | None:
try:
Expand Down
2 changes: 1 addition & 1 deletion apps/backend/routes/feedback.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@

logger = logging.getLogger("eagle.feedback")

router = APIRouter()
router = APIRouter(prefix="/feedback", tags=["feedback"])

@router.post("/{alert_id}", response_model=FeedbackResponse)
def record_feedback(
Expand Down
2 changes: 1 addition & 1 deletion apps/backend/routes/ingest.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@

logger = logging.getLogger("eagle.ingest")

router = APIRouter()
router = APIRouter(prefix="/ingest", tags=["ingest"])

def _decode_frame(b64: str | None) -> np.ndarray:
"""Decode base64 JPEG to BGR numpy array. Returns blank frame if None."""
Expand Down
2 changes: 1 addition & 1 deletion apps/backend/routes/snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

logger = logging.getLogger("eagle.snapshot")

router = APIRouter()
router = APIRouter(prefix="/snapshot", tags=["snapshot"])


@router.get("", tags=["snapshot"])
Expand Down
165 changes: 165 additions & 0 deletions apps/backend/routes/themes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,165 @@
"""
Detection Overlay Theme Management API

Endpoints:
GET /themes - List all themes (presets + custom)
GET /themes/{name} - Get a specific theme
POST /themes - Save a custom theme
DELETE /themes/{name} - Delete a custom theme
"""
from __future__ import annotations
import logging
import json
import os

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/themes", tags=["themes"])

THEME_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "config", "themes")
THEMES_FILE = os.path.join(THEME_DIR, "custom_themes.json")

os.makedirs(THEME_DIR, exist_ok=True)

PRESETS = {
"default": {
"name": "Default",
"description": "Standard surveillance overlay",
"boundingBoxColors": {"palette": ["#ef4444", "#3b82f6", "#22c55e", "#f59e0b", "#a855f7"], "opacity": 0.8, "borderWidth": 4},
"label": {"visible": True, "fontSize": 12, "fontFamily": "monospace", "backgroundColor": "rgba(0,0,0,0.6)", "textColor": "#ffffff"},
"confidence": {"visible": True, "mode": "percentage", "color": "#ffffff"},
"trackingTrails": {"enabled": True, "maxLength": 20, "trailOpacity": 0.5, "trailWidth": 2},
"accessibility": {"highContrast": False, "reducedMotion": False, "largeText": False},
},
"professional": {
"name": "Professional",
"description": "Clean, muted palette for long shifts",
"boundingBoxColors": {"palette": ["#f87171", "#60a5fa", "#4ade80", "#fbbf24", "#c084fc"], "opacity": 0.6, "borderWidth": 3},
"label": {"visible": True, "fontSize": 11, "fontFamily": "sans-serif", "backgroundColor": "rgba(30,30,30,0.7)", "textColor": "#e0e0e0"},
"confidence": {"visible": True, "mode": "percentage", "color": "#d1d5db"},
"trackingTrails": {"enabled": True, "maxLength": 15, "trailOpacity": 0.4, "trailWidth": 2},
"accessibility": {"highContrast": False, "reducedMotion": False, "largeText": False},
},
"highContrast": {
"name": "High Contrast",
"description": "Maximum visibility for accessibility",
"boundingBoxColors": {"palette": ["#ff0000", "#00ff00", "#0000ff", "#ffff00", "#ff00ff"], "opacity": 1.0, "borderWidth": 5},
"label": {"visible": True, "fontSize": 14, "fontFamily": "sans-serif", "backgroundColor": "rgba(0,0,0,0.9)", "textColor": "#ffffff", "fontWeight": "bold"},
"confidence": {"visible": True, "mode": "bar", "color": "#00ff00"},
"trackingTrails": {"enabled": True, "maxLength": 25, "trailOpacity": 0.8, "trailWidth": 3},
"accessibility": {"highContrast": True, "reducedMotion": False, "largeText": True},
},
"minimal": {
"name": "Minimal",
"description": "Subtle overlay with reduced visual clutter",
"boundingBoxColors": {"palette": ["#dc2626", "#2563eb", "#16a34a", "#d97706", "#9333ea"], "opacity": 0.4, "borderWidth": 2},
"label": {"visible": True, "fontSize": 10, "fontFamily": "monospace", "backgroundColor": "transparent", "textColor": "#ffffff"},
"confidence": {"visible": False, "mode": "off", "color": "#ffffff"},
"trackingTrails": {"enabled": False, "maxLength": 10, "trailOpacity": 0.3, "trailWidth": 1},
"accessibility": {"highContrast": False, "reducedMotion": True, "largeText": False},
},
"nightMode": {
"name": "Night Mode",
"description": "Dimmed palette for dark environments",
"boundingBoxColors": {"palette": ["#ff6b6b", "#74b9ff", "#55efc4", "#ffeaa7", "#a29bfe"], "opacity": 0.5, "borderWidth": 3},
"label": {"visible": True, "fontSize": 12, "fontFamily": "monospace", "backgroundColor": "rgba(10,10,20,0.8)", "textColor": "#a0a0b0"},
"confidence": {"visible": True, "mode": "percentage", "color": "#a0a0b0"},
"trackingTrails": {"enabled": True, "maxLength": 15, "trailOpacity": 0.3, "trailWidth": 2},
"accessibility": {"highContrast": False, "reducedMotion": False, "largeText": False},
},
"colorBlind": {
"name": "Color Blind Friendly",
"description": "Pattern-based cues for color vision deficiency",
"boundingBoxColors": {"palette": ["#e69f00", "#56b4e9", "#009e73", "#f0e442", "#0072b2"], "opacity": 0.85, "borderWidth": 4},
"label": {"visible": True, "fontSize": 13, "fontFamily": "sans-serif", "backgroundColor": "rgba(0,0,0,0.7)", "textColor": "#ffffff"},
"confidence": {"visible": True, "mode": "bar", "color": "#56b4e9"},
"trackingTrails": {"enabled": True, "maxLength": 20, "trailOpacity": 0.6, "trailWidth": 3},
"accessibility": {"highContrast": True, "reducedMotion": False, "largeText": True},
},
}


class ThemeRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
boundingBoxColors: dict
label: dict
confidence: dict
trackingTrails: dict
accessibility: dict


class ThemeResponse(BaseModel):
name: str
boundingBoxColors: dict
label: dict
confidence: dict
trackingTrails: dict
accessibility: dict
isCustom: bool = False


def _load_custom_themes() -> dict:
if not os.path.exists(THEMES_FILE):
return {}
try:
with open(THEMES_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return {}


def _save_custom_themes(themes: dict):
os.makedirs(os.path.dirname(THEMES_FILE), exist_ok=True)
with open(THEMES_FILE, "w", encoding="utf-8") as f:
json.dump(themes, f, indent=2)

Comment on lines +104 to +118

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift

Race condition and data loss hazard during concurrent file operations.

The endpoints for reading and writing custom themes are standard synchronous def functions, which FastAPI runs in a threadpool. Concurrent requests to POST /themes or DELETE /themes will cause race conditions because the file reads and writes are not synchronized.

Furthermore, if a read occurs while a write is truncating the file, json.load(f) will raise an exception. Because _load_custom_themes catches all exceptions and silently returns {}, the subsequent save operation will overwrite the file with just the single new or remaining theme, permanently wiping out all other custom themes.

Please use a threading.Lock to synchronize file access and remove the blanket except Exception clause so that file corruption fails loudly rather than causing silent data loss.

🔒️ Proposed fix to synchronize file operations
 import json
 import os
+import threading
 
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/themes", tags=["themes"])
 
 THEME_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "config", "themes")
 THEMES_FILE = os.path.join(THEME_DIR, "custom_themes.json")
+THEMES_LOCK = threading.Lock()
 
 os.makedirs(THEME_DIR, exist_ok=True)

# ... (skip to _load_custom_themes) ...

 def _load_custom_themes() -> dict:
-    if not os.path.exists(THEMES_FILE):
-        return {}
-    try:
+    with THEMES_LOCK:
+        if not os.path.exists(THEMES_FILE):
+            return {}
         with open(THEMES_FILE, "r", encoding="utf-8") as f:
             return json.load(f)
-    except Exception:
-        return {}
 
 
 def _save_custom_themes(themes: dict):
     os.makedirs(os.path.dirname(THEMES_FILE), exist_ok=True)
-    with open(THEMES_FILE, "w", encoding="utf-8") as f:
-        json.dump(themes, f, indent=2)
+    with THEMES_LOCK:
+        with open(THEMES_FILE, "w", encoding="utf-8") as f:
+            json.dump(themes, f, indent=2)
🧰 Tools
🪛 ast-grep (0.44.1)

[warning] 107-107: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(THEMES_FILE, "r", encoding="utf-8")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').

(open-filename-from-request)


[warning] 115-115: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(THEMES_FILE, "w", encoding="utf-8")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').

(open-filename-from-request)

🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/backend/routes/themes.py` around lines 104 - 118, Update
_load_custom_themes and _save_custom_themes to synchronize all custom-theme file
access with a shared threading.Lock, including the existence check, read, and
write operations. Replace _load_custom_themes’ blanket exception handling with
specific handling only for the intended missing-file case, allowing JSON or I/O
corruption errors to propagate instead of returning {} and risking data loss.


@router.get("", response_model=dict[str, ThemeResponse])
def list_themes():
customs = _load_custom_themes()
result = {}
for k, v in PRESETS.items():
result[k] = ThemeResponse(name=v.get("name", k), boundingBoxColors=v["boundingBoxColors"], label=v["label"], confidence=v["confidence"], trackingTrails=v["trackingTrails"], accessibility=v["accessibility"], isCustom=False)
for k, v in customs.items():
result[k] = ThemeResponse(name=v.get("name", k), boundingBoxColors=v["boundingBoxColors"], label=v["label"], confidence=v["confidence"], trackingTrails=v["trackingTrails"], accessibility=v["accessibility"], isCustom=True)
return result


@router.get("/{theme_name}", response_model=ThemeResponse)
def get_theme(theme_name: str):
all_themes = {**PRESETS, **_load_custom_themes()}
if theme_name not in all_themes:
raise HTTPException(status_code=404, detail=f"Theme {theme_name!r} not found")
t = all_themes[theme_name]
is_custom = theme_name in _load_custom_themes()
return ThemeResponse(name=t.get("name", theme_name), boundingBoxColors=t["boundingBoxColors"], label=t["label"], confidence=t["confidence"], trackingTrails=t["trackingTrails"], accessibility=t["accessibility"], isCustom=is_custom)


@router.post("", response_model=ThemeResponse, status_code=201)
def save_theme(body: ThemeRequest):
customs = _load_custom_themes()
customs[body.name] = {
"name": body.name,
"boundingBoxColors": body.boundingBoxColors,
"label": body.label,
"confidence": body.confidence,
"trackingTrails": body.trackingTrails,
"accessibility": body.accessibility,
}
_save_custom_themes(customs)
logger.info("Custom theme saved name=%s", body.name)
return ThemeResponse(name=body.name, boundingBoxColors=body.boundingBoxColors, label=body.label, confidence=body.confidence, trackingTrails=body.trackingTrails, accessibility=body.accessibility, isCustom=True)


@router.delete("/{theme_name}")
def delete_theme(theme_name: str):
customs = _load_custom_themes()
if theme_name not in customs:
raise HTTPException(status_code=404, detail=f"Custom theme {theme_name!r} not found")
del customs[theme_name]
_save_custom_themes(customs)
logger.info("Custom theme deleted name=%s", theme_name)
return {"deleted": theme_name}
Loading