diff --git a/.env.example b/.env.example index 86df112..4628dd7 100644 --- a/.env.example +++ b/.env.example @@ -20,6 +20,8 @@ MINIO_PRESIGN_EXPIRY_SECONDS=3600 REDIS_URL=redis://redis:6379/0 # App +# Set to `production` to enforce fail-fast on default/`change-me` secrets at startup. +APP_ENV=development SECRET_KEY=change-me-super-secret-must-be-at-least-32-characters-long CORS_ALLOW_ORIGINS=http://localhost:5173,http://127.0.0.1:5173 SKIP_DB_MIGRATIONS=false @@ -31,6 +33,9 @@ REFRESH_TOKEN_EXPIRE_DAYS=7 # Observability PROMETHEUS_PORT=9090 +# Optional: when set, GET /metrics requires `Authorization: Bearer `. +# Leave empty to keep /metrics open (dev / trusted internal networks only). +METRICS_BEARER_TOKEN= # Grafana GRAFANA_ADMIN_USER=admin diff --git a/AUDIT.md b/AUDIT.md new file mode 100644 index 0000000..3781659 --- /dev/null +++ b/AUDIT.md @@ -0,0 +1,151 @@ +# VisionForge — Platform Audit (Bugs & Production Readiness) + +Date: 2026-06-09 +Scope: full stack — backend API/services, Celery jobs, cluster agent, frontend, deployment/infra — measured against the target product: a streamlined web UI covering the full CV lifecycle (dataset curation → annotation → versioning → analysis → training → evaluation → model versioning → ONNX export). + +All findings below were verified against the code at the referenced locations. + +--- + +## 0. Remediation Status (this branch) + +All bugs and production-readiness items in §2–§4 and §6 have been fixed on +`claude/cv-platform-audit-8god0a`. The §5 items that are net-new product +features (per-project MinIO/S3 selection, duplicate-detection UI, training +checkpoint/resume, the active-learning retraining loop) are intentionally left +as roadmap work — the embedding storage groundwork for similarity/dedup is in +place (pgvector column + index), but the search endpoints and UI are not. + +| Area | Fixed | +|---|---| +| Object-level authz across all resource routers (C1–C5) | ✅ shared `services/authz.py`; job endpoints authenticated | +| Training dispatch failure leaves job queued / cluster busy (C6) | ✅ fails job + releases cluster + 502; new test | +| Celery worker-death recovery (C7) | ✅ `acks_late`, `reject_on_worker_lost`, time limits + `maintenance.sweep_stale_jobs` beat task | +| Fail-fast on default secrets (H1, H2) | ✅ `settings.require_secure_setting` (APP_ENV=production) | +| `/auth/refresh` accepts deleted users (H3) | ✅ user existence re-checked | +| In-memory auth rate limiting (H4) | ✅ Redis-backed with bounded fallback | +| Snapshot asset count (H5) | ✅ counts the version being frozen | +| Embeddings unqueryable / pgvector unused (H6) | ✅ `Asset.embedding` vector column + ANN index migration | +| Frontend token refresh (H7) | ✅ single-flight 401 refresh-and-retry in `api.ts` | +| AL `resolved_at` (H8) | ✅ set on resolve | +| Agent token timing / heartbeat header (H9, M9) | ✅ `hmac.compare_digest`; bearer-header auth | +| Training failure `dir()` guard (M1), CORS (M2), password bounds (M3) | ✅ | +| Frontend polling after terminal state (M4), silent catches (M5), annotator desync (M6) | ✅ | +| Migration concurrency (M7), agent supervisor respawn (M8) | ✅ advisory lock; backoff respawn | +| label_status spelling (L1), superuser column (L2), nginx exposure (L3), compose limits/volumes (L4), error boundary (L5) | ✅ | + +Validation: backend 180 unit tests + new dispatch test pass; agent 20 tests +pass; frontend builds, ESLint clean, 17 vitest tests pass, no new TypeScript +errors; ruff/black clean; docker-compose config valid. + +--- + +## 1. Executive Summary + +The platform is further along than a typical prototype: annotation, dataset versioning/metrics, training with full hyperparameter + augmentation control, evaluation with per-class metrics/confusion matrices, model lineage, and ONNX export all work end-to-end from the UI. The biggest problems are: + +1. **Authorization is broken at the object level.** Most resource endpoints never check workspace membership, and two job endpoints require no authentication at all. Any logged-in user can read/modify other workspaces' projects; anyone can read job status. +2. **The job/cluster lifecycle leaks on failure.** Broker enqueue failures are silently swallowed (job stays "queued" forever, cluster stays "busy" forever), and Celery has no time limits or `acks_late`, so a worker crash mid-training permanently wedges the job and its cluster. +3. **Several advertised lifecycle features are missing**: per-project MinIO/S3 storage selection, dataset duplicate detection / similarity search (pgvector is installed but unused), dataset improvement suggestions, an asset browse/filter UI, and checkpoint/resume for training. +4. **Deployment defaults are not production-safe**: Vite dev server in compose, weak fallback secrets, in-memory rate limiting, plain-HTTP agent communication, migrations auto-run by every replica. + +--- + +## 2. Critical Bugs + +### 2.1 Authorization / IDOR + +| # | Location | Issue | +|---|---|---| +| C1 | `backend/src/app/main.py:156` | `GET /jobs/{job_id}/stream` (SSE) has **no auth dependency** — anyone can stream any job's status by guessing/leaking a job ID. | +| C2 | `backend/src/app/api/jobs.py:14` | `GET /api/jobs/{jobId}` has **no auth dependency** — same exposure as C1. | +| C3 | `backend/src/app/api/projects.py` (`get_project`, `update_project`) | No workspace-membership check: any authenticated user can read **and modify** any project (name, description, task_type). `list_projects` does filter by membership (lines 74–83), so the gap is specific to the detail/update endpoints. | +| C4 | `backend/src/app/api/workspaces.py` (`get_workspace`, `list_members`, `invite_member`) | Any authenticated user can read any workspace's metadata, enumerate its members (leaks emails + roles), and **invite themselves or others** into any workspace — a privilege-escalation path that defeats RBAC entirely. | +| C5 | `backend/src/app/api/datasets.py` (`get_dataset` and sibling asset/annotation routes) | Dataset/asset/annotation access is not validated against the owning project's workspace membership. | + +**Fix direction:** add a shared `require_workspace_member(db, user, workspace_id, min_role=...)` dependency and apply it to every project/dataset/asset/annotation/experiment/artifact/job route, resolving the workspace via the resource's project. Job endpoints should authenticate and verify the job's project is visible to the caller. + +### 2.2 Job / cluster lifecycle + +| # | Location | Issue | +|---|---|---| +| C6 | `backend/src/app/services/training_service.py:109–119` | If `celery_app.send_task(...)` raises (broker down, Celery missing), the exception is swallowed and the API still returns `status: "queued"`. The Job row stays `queued` forever, the reserved cluster stays `busy` forever (its `active_job_id` was set at line 105), and the frontend polls indefinitely. Note `onnx_service.py` already does this correctly (releases cluster + fails job on dispatch error) — training should match it. | +| C7 | `backend/src/app/jobs/celery_app.py:45–52` | No `task_time_limit` / `task_soft_time_limit`, no `task_acks_late`, no `task_reject_on_worker_lost`. If a worker dies mid-task, the message is already acked → the task is never redelivered, `update_job_status` is never called, the job is stuck `running` and the cluster stuck `busy` with no recovery path. There is also no periodic sweeper to fail jobs/release clusters whose heartbeat/progress has gone stale. | + +*Correction to an earlier internal finding:* cluster release **does** work on normal task completion — `jobs_service.update_job_status()` releases the cluster on any terminal status (`jobs_service.py:44–46`), and the training task reaches that on both success (`training.py:297`) and handled failure (`training.py:325`). The leaks are specifically (a) enqueue failure (C6) and (b) unhandled worker death (C7). + +--- + +## 3. High-Severity Bugs + +| # | Location | Issue | Fix | +|---|---|---|---| +| H1 | `backend/src/app/services/auth.py:16` | JWT `SECRET_KEY` falls back to a hardcoded default → token forgery in any deployment that forgets the env var. | Fail startup if `SECRET_KEY` is unset/default outside dev. | +| H2 | `backend/src/app/db/session.py:19`, `services/storage.py:45–46` | DB password defaults to `change-me`; MinIO credentials default to `minioadmin/minioadmin`. | Same fail-fast treatment as H1. | +| H3 | `backend/src/app/api/auth.py:116–133` | `/auth/refresh` mints a new access token **without checking the user still exists** (no DB lookup) — deleted/disabled users keep API access until refresh-token expiry. | Load the user by `user_id` and 401 if missing/disabled. | +| H4 | `backend/src/app/api/middleware.py:33–60` | Auth rate limiting is an in-process dict: useless across replicas, unbounded growth between prunes, racy. (Known limitation per CLAUDE.md, but it's a launch blocker for >1 replica.) | Redis-backed limiter. | +| H5 | `backend/src/app/services/dataset_service.py:50–51` | `snapshot_version()` counts **all** assets in the dataset, not the version's assets — locked version rows report inflated/incorrect `asset_count` as the dataset grows. Undermines trust in dataset version control. | Count assets scoped to the version being locked. | +| H6 | `backend/src/app/jobs/tasks/embeddings.py:108`, `db/session.py:8–10` | Embeddings are serialized as JSON into `Asset.meta_data`; the pgvector extension is registered but **no vector column exists**. Similarity search, duplicate detection, and embedding-based curation are impossible without a full-table Python scan. | Add `Asset.embedding: Vector(512)` + ivfflat/hnsw index, migrate, write vectors there. | +| H7 | `frontend/src/services/auth.ts:51–64`, `services/api.ts` | `refreshToken()` exists but is never called; on any 401 the client hard-logs the user out and redirects to login. Long annotation/training sessions will lose work context when the access token expires. | 401 interceptor: refresh once, retry, then logout. | +| H8 | `backend/src/app/api/al.py:310–313` | AL item resolution never sets `resolved_at` (model field exists) — breaks AL audit trail/metrics. | Set `resolved_at = datetime.now(timezone.utc)`. | +| H9 | `agent/src/vf_agent/server.py:35–39` + install path | Agent tokens travel over plain HTTP, and token comparison is not constant-time (`!=` instead of `hmac.compare_digest`). | `compare_digest`; document/require TLS or private-network for agents. | + +--- + +## 4. Medium / Low Bugs + +- **M1** `backend/src/app/jobs/tasks/training.py:312` — failure handler guards with `"ExperimentRun" in dir()`; this only works because the import at line 56 is function-local and only if execution got past it. If the task fails before line 56, the run row is never marked failed. Replace with a module-level import and drop the `dir()` check. +- **M2** `backend/src/app/main.py:74–81` — CORS allows all methods/headers with `allow_credentials=True`; tighten for production. +- **M3** `backend/src/app/api/auth.py:20–28` — no password length constraints on signup/login schemas (bcrypt 72-byte truncation is silent). +- **M4** `frontend/src/pages/experiments/[runId].tsx:240–245` — metrics polling continues every 3 s after the run reaches a terminal state; clear the interval when status leaves `running/queued`. +- **M5** `frontend/src/pages/evaluations/new.tsx:44–49` (and similar in artifacts/datasets pages) — `.catch(() => {})` silently swallows load failures, leaving forms with empty dropdowns and no error message. +- **M6** `frontend/src/pages/annotate/Annotator.tsx:686` — annotation delete removes the box from the canvas even if the server delete fails (silent desync). +- **M7** `backend/src/app/main.py:101–119` — migrations auto-run by every backend replica on startup; concurrent `alembic upgrade` across replicas can deadlock. Move to an init job and set `SKIP_DB_MIGRATIONS=1` on replicas. +- **M8** `agent/src/vf_agent/main.py:110–121` — supervisor exits the whole agent if any child (heartbeat/worker/server) dies; no respawn with backoff. +- **M9** `backend/src/app/api/clusters.py` heartbeat — `register_token` is sent in the JSON body rather than an `Authorization` header (more likely to be logged). +- **L1** Spelling drift `unlabelled` vs `unlabeled` between `models/asset.py:28` and `services/asset_service.py:74` — exact-match filters can silently miss assets. +- **L2** `backend/src/app/api/rbac.py:38–39` — superuser determined by env-var email comparison while `User.is_superuser` column exists unused. +- **L3** `/metrics`, `/docs`, `/openapi.json` are unauthenticated and proxied by `frontend/nginx.conf` — restrict in production. +- **L4** `docker-compose.yml` — no resource limits on any service; Prometheus/Grafana have no persistent volumes; Grafana's Postgres datasource uses `sslmode: disable` with a plaintext password. +- **L5** No React error boundary — any render exception white-screens the app. + +--- + +## 5. Feature-Gap Matrix vs Target Product + +| Lifecycle capability | Status | Notes | +|---|---|---| +| Upload images/video | ✅ end-to-end | Presigned PUT flow, per-file progress (`pages/datasets/upload.tsx`, `/api/ingest/upload-url`). | +| Browse/filter dataset assets | ⚠️ backend only | `/api/datasets/{id}/assets` filters by status/split/version, but there is **no gallery/browse UI** — only the annotate gateway. | +| Video frame extraction | ⚠️ backend only | Job + endpoint exist (`api/ops.py:144–194`); no UI trigger. | +| Annotation (box/polygon/keypoint/classification) | ✅ end-to-end | Full canvas editor with undo/redo, bulk save, optimistic locking, review queue. | +| Model-assisted prelabeling | ⚠️ backend only | Prelabel task exists; no UI to queue it (suggestions surface in review only). | +| Dataset version control | ✅ (with bug H5) | Snapshot/lock/version-scoped assets & metrics all work; snapshot counts are wrong. | +| Dataset analysis | ⚠️ partial | Class balance, coverage, geometry/resolution histograms, velocity: ✅ rich dashboard. **Missing:** duplicate detection, similarity search (blocked by H6), and any "suggested improvements" engine. | +| Storage selection (MinIO vs S3 per project) | ❌ not implemented | Single env-var-configured MinIO client (`services/storage.py`); no per-project/workspace config model, no boto3/AWS-credential path, no UI. | +| Training with hyperparameter + augmentation control | ✅ end-to-end | Schema-driven param groups incl. full augmentation set (mosaic, mixup, HSV, flips…), validated allow-list (`training/ultralytics_trainer.py`). | +| Live training view | ⚠️ partial | Per-epoch metrics persisted and charted, but via 3 s polling (plus M4 leak); no SSE/WebSocket push. Acceptable, not "live". | +| Checkpoint / resume | ❌ not implemented | Any interruption restarts from epoch 0. | +| Auto-eval on test split + post-hoc eval on chosen split | ✅ end-to-end | Per-class P/R/F1/AP, confusion matrix, mAP@50/95, FP/FN sample browser, threshold controls. | +| Model version control / registry | ✅ end-to-end | Artifacts with version/checksum/lineage tree (run → dataset version → class map → cluster → evals → siblings). | +| ONNX export | ✅ end-to-end | Opset/dynamic-axes options, onnxruntime validation, artifact row + download. | +| Active learning | ⚠️ partial | Uncertainty + diversity selection and resolve workflow work; no retraining feedback loop, `proposed_json` never populated, H8 bug. | +| Compute clusters | ✅ (with C6/C7 risks) | Discovery, heartbeats, per-cluster queues, live telemetry grid. | +| Admin / membership UI | ⚠️ partial | Members table read-only; invite flow stubbed in UI (and over-permissive on the backend, C4). | + +--- + +## 6. Production Deployment Blockers (priority order) + +1. **Object-level authorization** (C1–C5) — ship a workspace-membership dependency across all resource routers; authenticate job endpoints. +2. **Job/cluster failure handling** (C6, C7) — fail the job + release the cluster on enqueue error; add Celery `task_acks_late`, `task_reject_on_worker_lost`, time limits; add a stale-job sweeper that fails jobs and releases clusters after heartbeat/progress timeout. +3. **Fail-fast secrets** (H1, H2) — refuse to boot with default `SECRET_KEY` / DB / MinIO credentials outside dev mode. +4. **Production frontend + TLS** — compose currently ships the Vite dev server; wire `frontend/Dockerfile.prod` (nginx) into a prod compose/profile, terminate TLS in front of API and agents (H9). +5. **Redis-backed auth rate limiting** (H4) and migration init-job strategy (M7) before running >1 API replica. +6. **Token refresh on the frontend** (H3 + H7) — fix both halves together. + +## 7. Recommended Roadmap to the Full-Lifecycle Vision + +**Phase 1 — correctness & security (1–2 weeks):** items in §6, plus H5 snapshot counts, H8, M1–M6. +**Phase 2 — missing lifecycle features:** per-project storage backend (Workspace/Project `storage_config` + boto3 strategy + owner-only settings UI); pgvector embedding column + similarity/duplicate endpoints + dedup UI; asset gallery with filter/browse; UI triggers for frame extraction and prelabeling. +**Phase 3 — polish:** SSE/WebSocket live training metrics (the SSE job stream in `main.py` is a starting point — after C1 is fixed); checkpoint/resume; dataset "suggested improvements" (class-imbalance, low-coverage, duplicate-driven recommendations from existing metrics); AL retraining loop; admin invite flow with proper RBAC. diff --git a/agent/src/vf_agent/heartbeat.py b/agent/src/vf_agent/heartbeat.py index b1af5ad..74db39e 100644 --- a/agent/src/vf_agent/heartbeat.py +++ b/agent/src/vf_agent/heartbeat.py @@ -20,6 +20,8 @@ def _payload(ident: identity.Identity) -> dict[str, Any]: snap = discover.discover() return { + # Token also goes in the Authorization header (preferred); kept in the + # body for compatibility with platforms that predate header auth. "register_token": ident.register_token, "status": "online", "cpu_usage_pct": snap["cpu_usage_pct"], @@ -43,7 +45,8 @@ def send_once(ident: identity.Identity, *, client: httpx.Client | None = None) - own = client is None c = client or httpx.Client(timeout=HEARTBEAT_TIMEOUT_S) try: - resp = c.post(url, json=_payload(ident)) + headers = {"Authorization": f"Bearer {ident.register_token}"} + resp = c.post(url, json=_payload(ident), headers=headers) return resp.status_code except httpx.HTTPError as exc: logger.warning("heartbeat transport error: %s", exc) diff --git a/agent/src/vf_agent/main.py b/agent/src/vf_agent/main.py index dd3b937..8e08191 100644 --- a/agent/src/vf_agent/main.py +++ b/agent/src/vf_agent/main.py @@ -10,6 +10,7 @@ import subprocess import sys import time +from collections.abc import Callable from vf_agent import identity @@ -19,6 +20,12 @@ HTTP_HOST = os.getenv("VF_AGENT_HOST", "0.0.0.0") # noqa: S104 - intended; agent serves on LAN WAIT_FOR_IDENTITY_S = float(os.getenv("VF_AGENT_ADOPT_POLL", "2")) +# Respawn policy for child processes (HTTP server, heartbeat, celery worker). +RESPAWN_BACKOFF_INITIAL_S = 2.0 +RESPAWN_BACKOFF_MAX_S = 60.0 +RESPAWN_STABLE_AFTER_S = 300.0 # child stayed up this long -> reset its backoff +RESPAWN_MAX_CONSECUTIVE_FAILURES = 5 + def _spawn_http() -> subprocess.Popen[bytes]: cmd = [ @@ -65,14 +72,20 @@ def _spawn_celery(ident: identity.Identity) -> subprocess.Popen[bytes]: return subprocess.Popen(cmd, env=env) -def _wait_for_identity() -> identity.Identity: - logger.info("waiting for adoption (POST /adopt on the HTTP server)...") - while True: - ident = identity.load() - if ident is not None: - logger.info("adopted as cluster %s", ident.cluster_id) - return ident - time.sleep(WAIT_FOR_IDENTITY_S) +class _Child: + """A supervised child process with restart-with-backoff bookkeeping.""" + + def __init__(self, name: str, spawn: Callable[[], subprocess.Popen[bytes]]) -> None: + self.name = name + self.spawn = spawn + self.proc: subprocess.Popen[bytes] = spawn() + self.started_at = time.monotonic() + self.backoff_s = RESPAWN_BACKOFF_INITIAL_S + self.consecutive_failures = 0 + + def respawn(self) -> None: + self.proc = self.spawn() + self.started_at = time.monotonic() def run() -> int: @@ -85,38 +98,104 @@ def run() -> int: logger.error("VF_AGENT_TOKEN is not set; refusing to start") return 2 - http = _spawn_http() - ident = _wait_for_identity() - heartbeat = _spawn_heartbeat() - celery_proc = _spawn_celery(ident) - - children: list[subprocess.Popen[bytes]] = [http, heartbeat, celery_proc] + children: list[_Child] = [] + shutting_down = False def _shutdown(*_: object) -> None: + nonlocal shutting_down + shutting_down = True logger.info("shutting down agent") for child in children: - if child.poll() is None: - child.terminate() + if child.proc.poll() is None: + child.proc.terminate() for child in children: try: - child.wait(timeout=10) + child.proc.wait(timeout=10) except subprocess.TimeoutExpired: - child.kill() + child.proc.kill() signal.signal(signal.SIGTERM, _shutdown) signal.signal(signal.SIGINT, _shutdown) + def _supervise(child: _Child) -> None: + """Check one child; respawn with backoff if it died. + + Raises SystemExit after RESPAWN_MAX_CONSECUTIVE_FAILURES consecutive + failed respawns of the same child. + """ + ret = child.proc.poll() + if ret is None: + # Child has stayed up long enough: consider it healthy again. + if ( + child.consecutive_failures + and time.monotonic() - child.started_at >= RESPAWN_STABLE_AFTER_S + ): + logger.info( + "child %s stable for %ss; resetting respawn backoff", + child.name, + int(RESPAWN_STABLE_AFTER_S), + ) + child.consecutive_failures = 0 + child.backoff_s = RESPAWN_BACKOFF_INITIAL_S + return + + if shutting_down: + return + + child.consecutive_failures += 1 + if child.consecutive_failures > RESPAWN_MAX_CONSECUTIVE_FAILURES: + logger.error( + "child %s failed %s consecutive respawns (last exit code %s); " + "giving up and tearing down", + child.name, + RESPAWN_MAX_CONSECUTIVE_FAILURES, + ret, + ) + raise SystemExit(ret or 1) + + logger.error( + "child %s exited unexpectedly with code %s; respawning in %.0fs (attempt %s/%s)", + child.name, + ret, + child.backoff_s, + child.consecutive_failures, + RESPAWN_MAX_CONSECUTIVE_FAILURES, + ) + time.sleep(child.backoff_s) + child.backoff_s = min(child.backoff_s * 2, RESPAWN_BACKOFF_MAX_S) + if not shutting_down: + child.respawn() + exit_code = 0 try: - while True: + http = _Child("http", _spawn_http) + children.append(http) + + # Adoption gate: only the HTTP server runs until the backend POSTs + # /adopt. Keep supervising it (with respawn) while we wait. + logger.info("waiting for adoption (POST /adopt on the HTTP server)...") + ident: identity.Identity | None = None + while not shutting_down: + ident = identity.load() + if ident is not None: + logger.info("adopted as cluster %s", ident.cluster_id) + break + _supervise(http) + time.sleep(WAIT_FOR_IDENTITY_S) + if shutting_down or ident is None: + return exit_code + + adopted = ident + children.append(_Child("heartbeat", _spawn_heartbeat)) + children.append(_Child("celery", lambda: _spawn_celery(adopted))) + + while not shutting_down: for child in children: - ret = child.poll() - if ret is not None: - logger.error("child process exited with code %s; tearing down", ret) - exit_code = ret or 1 - raise SystemExit(exit_code) + _supervise(child) time.sleep(2) - except SystemExit: + return exit_code + except SystemExit as exc: + exit_code = int(exc.code) if isinstance(exc.code, int) else 1 _shutdown() return exit_code diff --git a/agent/src/vf_agent/server.py b/agent/src/vf_agent/server.py index 67c6350..f317dd3 100644 --- a/agent/src/vf_agent/server.py +++ b/agent/src/vf_agent/server.py @@ -10,6 +10,7 @@ from __future__ import annotations +import hmac import os from typing import Any @@ -35,7 +36,8 @@ def _require_token(authorization: str | None = Header(default=None)) -> None: if not authorization or not authorization.lower().startswith("bearer "): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing bearer token") presented = authorization.split(" ", 1)[1].strip() - if presented != expected: + # Constant-time comparison to avoid leaking token bytes via timing. + if not hmac.compare_digest(presented.encode("utf-8"), expected.encode("utf-8")): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid agent token") diff --git a/backend/src/app/api/al.py b/backend/src/app/api/al.py index 7f4616e..da36592 100644 --- a/backend/src/app/api/al.py +++ b/backend/src/app/api/al.py @@ -3,6 +3,7 @@ import json import os import random +from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel @@ -14,8 +15,11 @@ from app.models.alrun import ALRun from app.models.artifact import ModelArtifact from app.models.asset import Asset +from app.models.dataset_version import DatasetVersion +from app.models.project import Project from app.models.user import User -from app.services import inference_service +from app.models.workspace import Membership, Role +from app.services import authz, inference_service from app.services.active_learning_service import select_diverse, select_uncertain from app.services.asset_fetch import fetch_asset_bytes from app.services.embeddings_service import EmbeddingsService @@ -28,6 +32,15 @@ _INLINE_SCORE_CAP = int(os.getenv("VF_AL_INLINE_SCORE_CAP", "200")) +def _member_workspace_ids(db: Session, user: User) -> list[str]: + ws_ids = { + m.workspace_id + for m in db.scalars(select(Membership).where(Membership.user_id == user.id)).all() + } + ws_ids.add(authz.DEFAULT_WORKSPACE_ID) + return list(ws_ids) + + def _uncertainty_scores(db: Session, assets: list, model_id: str | None) -> list[float]: """Compute uncertainty scores for ``assets`` using ``model_id`` if available. @@ -149,13 +162,16 @@ def queue_uncertainty_scoring( can read cached scores instead of running inference inline. """ from app.jobs.celery_app import celery_app - from app.models.dataset_version import DatasetVersion from app.services.jobs_service import create_job, update_job_status - if not db.get(ModelArtifact, body.artifact_id): + artifact = db.get(ModelArtifact, body.artifact_id) + if not artifact: raise HTTPException(status_code=400, detail="artifact not found") - if not db.get(DatasetVersion, body.dataset_version_id): + version = db.get(DatasetVersion, body.dataset_version_id) + if not version: raise HTTPException(status_code=400, detail="dataset version not found") + authz.require_project_access(db, current_user, artifact.project_id, Role.DEVELOPER) + authz.require_dataset_access(db, current_user, version.dataset_id, Role.DEVELOPER) payload = { "artifactId": body.artifact_id, @@ -182,6 +198,10 @@ def select_samples( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_project_access(db, current_user, body.project_id, Role.DEVELOPER) + version = db.get(DatasetVersion, body.dataset_version_id) + if version is not None: + authz.require_dataset_access(db, current_user, version.dataset_id, Role.DEVELOPER) # Get unlabeled assets for this version assets = list( db.scalars( @@ -257,7 +277,12 @@ def list_runs( ): base = select(ALRun) if project_id: + authz.require_project_access(db, current_user, project_id, Role.VIEWER) base = base.where(ALRun.project_id == project_id) + elif not authz.is_superuser(db, current_user): + base = base.join(Project, ALRun.project_id == Project.id).where( + Project.workspace_id.in_(_member_workspace_ids(db, current_user)) + ) total = db.scalar(select(func.count()).select_from(base.subquery())) or 0 offset = (page - 1) * page_size runs = list( @@ -285,6 +310,9 @@ def get_al_items( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + run = db.get(ALRun, al_run_id) + if run is not None: + authz.require_project_access(db, current_user, run.project_id, Role.VIEWER) items = list(db.scalars(select(ALItem).where(ALItem.al_run_id == al_run_id)).all()) return [ { @@ -307,8 +335,12 @@ def resolve_item( item = db.get(ALItem, item_id) if not item or item.al_run_id != al_run_id: raise HTTPException(status_code=404, detail="AL item not found") + run = db.get(ALRun, item.al_run_id) + if run is not None: + authz.require_project_access(db, current_user, run.project_id, Role.ANNOTATOR) item.resolved_status = "resolved" item.resolved_by = current_user.id + item.resolved_at = datetime.now(timezone.utc) db.add(item) db.commit() db.refresh(item) diff --git a/backend/src/app/api/annotations.py b/backend/src/app/api/annotations.py index e96bb6c..3f98d69 100644 --- a/backend/src/app/api/annotations.py +++ b/backend/src/app/api/annotations.py @@ -5,9 +5,11 @@ from sqlalchemy.orm import Session from app.db.deps import get_current_user, get_db -from app.models.annotation import REVIEW_STATUSES +from app.models.annotation import REVIEW_STATUSES, Annotation +from app.models.asset import Asset from app.models.user import User -from app.services import inference_service, suggestion_service +from app.models.workspace import Role +from app.services import authz, inference_service, suggestion_service from app.services.annotation_service import ( AnnotationError, VersionConflictError, @@ -27,6 +29,23 @@ router = APIRouter(prefix="/api/annotations", tags=["annotations"]) +def _require_asset_dataset_access(db: Session, user: User, asset_id: str, min_role: Role) -> None: + """Enforce dataset-level access for an asset. + + Missing assets are left for the service layer to report (it already + returns its own 400/404s), so existing status codes are preserved. + """ + asset = db.get(Asset, asset_id) + if asset is not None: + authz.require_dataset_access(db, user, asset.dataset_id, min_role) + + +def _require_annotation_access(db: Session, user: User, annotation_id: str, min_role: Role) -> None: + ann = db.get(Annotation, annotation_id) + if ann is not None: + _require_asset_dataset_access(db, user, ann.asset_id, min_role) + + class AnnotationCreate(BaseModel): asset_id: str type: str # "box" | "polygon" | "keypoint" | "classification" @@ -72,6 +91,7 @@ def create( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + _require_asset_dataset_access(db, current_user, body.asset_id, Role.ANNOTATOR) try: ann = create_annotation( db, @@ -93,6 +113,7 @@ def update( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + _require_annotation_access(db, current_user, annotation_id, Role.ANNOTATOR) try: ann = update_annotation( db, @@ -116,6 +137,7 @@ def delete( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + _require_annotation_access(db, current_user, annotation_id, Role.ANNOTATOR) if not delete_annotation(db, annotation_id): raise HTTPException(status_code=404, detail="Annotation not found") @@ -133,6 +155,19 @@ def bulk( UI can highlight conflicts (HTTP 409 equivalents) without rolling back successful entries. """ + asset_ids = {c.asset_id for c in body.creates} + ann_ids = {u.id for u in body.updates} | set(body.deletes) + for ann_id in ann_ids: + ann = db.get(Annotation, ann_id) + if ann is not None: + asset_ids.add(ann.asset_id) + dataset_ids: set[str] = set() + for asset_id in asset_ids: + asset = db.get(Asset, asset_id) + if asset is not None: + dataset_ids.add(asset.dataset_id) + for dataset_id in dataset_ids: + authz.require_dataset_access(db, current_user, dataset_id, Role.ANNOTATOR) return bulk_save( db, author_id=current_user.id, @@ -154,6 +189,7 @@ def review( status_code=400, detail=f"review_status must be one of {list(REVIEW_STATUSES)}", ) + _require_annotation_access(db, current_user, annotation_id, Role.ANNOTATOR) try: ann = set_review( db, @@ -177,6 +213,7 @@ def history( ): import json as _json + _require_annotation_access(db, current_user, annotation_id, Role.VIEWER) rows = get_history(db, annotation_id) out = [] for h in rows: @@ -197,6 +234,7 @@ def list_for_asset( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + _require_asset_dataset_access(db, current_user, asset_id, Role.VIEWER) anns = get_asset_annotations(db, asset_id) return [_ann_dict(a) for a in anns] @@ -207,6 +245,7 @@ def mark_labeled( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + _require_asset_dataset_access(db, current_user, asset_id, Role.ANNOTATOR) mark_asset_labeled(db, asset_id) return {"status": "labeled"} @@ -223,6 +262,7 @@ def review_queue( current_user: User = Depends(get_current_user), ): """List annotations for a dataset's review queue.""" + authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) try: items, total = list_review_queue( db, @@ -245,6 +285,7 @@ def review_queue_summary( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) return review_summary(db, dataset_id=dataset_id, version_id=version_id) @@ -266,6 +307,7 @@ def suggest( accepted ones through the normal bulk path. 404 when no model is available, 502 when inference fails. """ + _require_asset_dataset_access(db, current_user, body.asset_id, Role.ANNOTATOR) try: artifact, suggestions = suggestion_service.suggest_annotations( db, @@ -305,6 +347,7 @@ def suggest_artifacts( current_user: User = Depends(get_current_user), ): """List models trained on this dataset, newest first (override dropdown).""" + authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) arts = suggestion_service.candidate_artifacts_for_dataset(db, dataset_id) return { "items": [ @@ -340,10 +383,14 @@ def queue_error_mining( from app.models.dataset_version import DatasetVersion from app.services.jobs_service import create_job, update_job_status - if not db.get(ModelArtifact, body.artifact_id): + artifact = db.get(ModelArtifact, body.artifact_id) + if not artifact: raise HTTPException(status_code=400, detail="artifact not found") - if not db.get(DatasetVersion, body.dataset_version_id): + version = db.get(DatasetVersion, body.dataset_version_id) + if not version: raise HTTPException(status_code=400, detail="dataset version not found") + authz.require_project_access(db, current_user, artifact.project_id, Role.DEVELOPER) + authz.require_dataset_access(db, current_user, version.dataset_id, Role.DEVELOPER) payload = { "artifactId": body.artifact_id, diff --git a/backend/src/app/api/artifacts.py b/backend/src/app/api/artifacts.py index 45e2480..459e0c9 100644 --- a/backend/src/app/api/artifacts.py +++ b/backend/src/app/api/artifacts.py @@ -15,15 +15,36 @@ from app.models.dataset_version import DatasetVersion from app.models.evaluation import Evaluation from app.models.experiment import ExperimentRun +from app.models.project import Project from app.models.user import User +from app.models.workspace import Membership, Role from app.schemas.common import Job -from app.services import inference_service +from app.services import authz, inference_service from app.services.onnx_service import OnnxDispatchError from app.services.onnx_service import export_onnx as svc_export_onnx router = APIRouter(prefix="/api", tags=["artifacts"]) +def _member_workspace_ids(db: Session, user: User) -> list[str]: + ws_ids = { + m.workspace_id + for m in db.scalars(select(Membership).where(Membership.user_id == user.id)).all() + } + ws_ids.add(authz.DEFAULT_WORKSPACE_ID) + return list(ws_ids) + + +def _require_artifact_access( + db: Session, user: User, model_id: str, min_role: Role +) -> ModelArtifact: + artifact = db.get(ModelArtifact, model_id) + if not artifact: + raise HTTPException(status_code=404, detail="Model not found") + authz.require_project_access(db, user, artifact.project_id, min_role) + return artifact + + def _artifact_dict(a: ModelArtifact) -> dict: return { "id": a.id, @@ -50,7 +71,12 @@ def list_models( ): base = select(ModelArtifact) if project_id: + authz.require_project_access(db, current_user, project_id, Role.VIEWER) base = base.where(ModelArtifact.project_id == project_id) + elif not authz.is_superuser(db, current_user): + base = base.join(Project, ModelArtifact.project_id == Project.id).where( + Project.workspace_id.in_(_member_workspace_ids(db, current_user)) + ) total = db.scalar(select(func.count()).select_from(base.subquery())) or 0 offset = (page - 1) * page_size artifacts = list( @@ -72,9 +98,7 @@ def get_model( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): - artifact = db.get(ModelArtifact, model_id) - if not artifact: - raise HTTPException(status_code=404, detail="Model not found") + artifact = _require_artifact_access(db, current_user, model_id, Role.VIEWER) return { "id": artifact.id, "projectId": artifact.project_id, @@ -105,9 +129,7 @@ def export_model( ): from app.services import cluster_service - artifact = db.get(ModelArtifact, model_id) - if not artifact: - raise HTTPException(status_code=404, detail="Model not found") + artifact = _require_artifact_access(db, current_user, model_id, Role.DEVELOPER) # Use the artifact's run_id as experiment_id for ONNX export experiment_id = artifact.run_id or model_id try: @@ -136,9 +158,7 @@ def download_model( Falls back to streaming from the local filesystem if the storage_path is a local path (e.g. tests, sqlite-only environments). """ - artifact = db.get(ModelArtifact, model_id) - if not artifact: - raise HTTPException(status_code=404, detail="Model not found") + artifact = _require_artifact_access(db, current_user, model_id, Role.VIEWER) if not artifact.storage_path: raise HTTPException(status_code=404, detail="artifact has no storage_path") @@ -208,9 +228,7 @@ async def predict_multipart( Use this from browser file inputs / curl. For JSON callers, see `/predict-json` which accepts a base64-encoded image. """ - artifact = db.get(ModelArtifact, model_id) - if not artifact: - raise HTTPException(status_code=404, detail="Model not found") + artifact = _require_artifact_access(db, current_user, model_id, Role.VIEWER) image_bytes = await file.read() return _predict_response(model_id, artifact, image_bytes, score_threshold) @@ -223,9 +241,7 @@ def predict_json( current_user: User = Depends(get_current_user), ): """Run inference on the artifact via a JSON body with a base64 image.""" - artifact = db.get(ModelArtifact, model_id) - if not artifact: - raise HTTPException(status_code=404, detail="Model not found") + artifact = _require_artifact_access(db, current_user, model_id, Role.VIEWER) try: image_bytes = base64.b64decode(body.image_base64) except Exception as exc: @@ -261,9 +277,7 @@ def get_lineage( from app.models.cluster import Cluster from app.models.dataset import ClassMap - artifact = db.get(ModelArtifact, model_id) - if not artifact: - raise HTTPException(status_code=404, detail="Model not found") + artifact = _require_artifact_access(db, current_user, model_id, Role.VIEWER) run = db.get(ExperimentRun, artifact.run_id) if artifact.run_id else None version = ( db.get(DatasetVersion, run.dataset_version_id) if run and run.dataset_version_id else None diff --git a/backend/src/app/api/assets.py b/backend/src/app/api/assets.py index ab5b3bf..a02acfa 100644 --- a/backend/src/app/api/assets.py +++ b/backend/src/app/api/assets.py @@ -8,9 +8,11 @@ from sqlalchemy.orm import Session from app.db.deps import get_current_user, get_db +from app.models.dataset_version import DatasetVersion from app.models.user import User +from app.models.workspace import Role from app.schemas.split import SplitConfig, SplitSummary -from app.services import split_service +from app.services import authz, split_service from app.services.annotation_service import get_asset_annotations from app.services.asset_service import ( confirm_upload, @@ -33,6 +35,20 @@ class ConfirmUploadRequest(BaseModel): height: int | None = None +def _require_version_dataset_access( + db: Session, + user: User, + dataset_id: str, + version_id: str, + min_role: Role, +) -> None: + """Enforce access against the version's *actual* owning dataset when it + resolves, falling back to the dataset id from the path/body otherwise.""" + version = db.get(DatasetVersion, version_id) + owner = version.dataset_id if version is not None else dataset_id + authz.require_dataset_access(db, user, owner, min_role) + + def _presign_download(uri: str) -> str: """Attempt to generate a presigned download URL; fall back to raw URI.""" try: @@ -56,6 +72,7 @@ def get( asset = get_asset(db, asset_id) if not asset: raise HTTPException(status_code=404, detail="Asset not found") + authz.require_dataset_access(db, current_user, asset.dataset_id, Role.VIEWER) download_url = _presign_download(asset.uri) meta: dict = {} if asset.meta_data: @@ -84,6 +101,9 @@ def get_annotations( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + asset = get_asset(db, asset_id) + if asset is not None: + authz.require_dataset_access(db, current_user, asset.dataset_id, Role.VIEWER) anns = get_asset_annotations(db, asset_id) return [ { @@ -110,6 +130,7 @@ def list_dataset_assets( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) assets, total = list_assets( db, dataset_id, @@ -148,6 +169,7 @@ def get_version_split( current_user: User = Depends(get_current_user), ): """Return persisted train/val/test counts and per-class breakdown for a version.""" + _require_version_dataset_access(db, current_user, dataset_id, version_id, Role.VIEWER) return split_service.get_split_summary(db, version_id) @@ -160,6 +182,7 @@ def assign_version_split( current_user: User = Depends(get_current_user), ): """Deterministically (re)assign and persist the split for every asset in a version.""" + _require_version_dataset_access(db, current_user, dataset_id, version_id, Role.DEVELOPER) try: return split_service.assign_splits( db, @@ -181,6 +204,7 @@ def dataset_stats( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) return get_dataset_stats(db, dataset_id, version_id=version_id) @@ -192,6 +216,7 @@ def dataset_metrics( current_user: User = Depends(get_current_user), ): """Detailed dataset health metrics for the metrics dashboard.""" + authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) return get_dataset_metrics(db, dataset_id, version_id=version_id) @@ -215,6 +240,7 @@ def get_asset_neighbors( asset = db.get(Asset, asset_id) if not asset: raise HTTPException(status_code=404, detail="Asset not found") + authz.require_dataset_access(db, current_user, asset.dataset_id, Role.VIEWER) base_filters = [Asset.dataset_id == asset.dataset_id] if asset.version_id: @@ -280,6 +306,10 @@ def confirm_asset_upload( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_dataset_access(db, current_user, body.dataset_id, Role.DEVELOPER) + version = db.get(DatasetVersion, body.version_id) + if version is not None and version.dataset_id != body.dataset_id: + authz.require_dataset_access(db, current_user, version.dataset_id, Role.DEVELOPER) asset = confirm_upload( db, dataset_id=body.dataset_id, diff --git a/backend/src/app/api/auth.py b/backend/src/app/api/auth.py index e5197b9..49ef8ee 100644 --- a/backend/src/app/api/auth.py +++ b/backend/src/app/api/auth.py @@ -2,7 +2,7 @@ from fastapi import APIRouter, Depends, HTTPException, status from fastapi.responses import Response -from pydantic import BaseModel, EmailStr +from pydantic import BaseModel, EmailStr, Field from sqlalchemy.orm import Session from app.db.deps import get_current_user, get_db @@ -19,13 +19,14 @@ class LoginRequest(BaseModel): email: EmailStr - password: str + # bcrypt only considers the first 72 bytes; cap input well below that. + password: str = Field(min_length=1, max_length=128) class SignupRequest(BaseModel): - name: str + name: str = Field(min_length=1, max_length=255) email: EmailStr - password: str + password: str = Field(min_length=8, max_length=128) acceptTerms: bool @@ -114,7 +115,7 @@ def signup(req: SignupRequest, db: Session = Depends(get_db)) -> dict: @router.post("/refresh", response_model=AccessTokenResponse) -def refresh_token(req: TokenRefreshRequest) -> dict: +def refresh_token(req: TokenRefreshRequest, db: Session = Depends(get_db)) -> dict: payload = auth_service.decode_token(req.refresh_token) if payload is None or payload.get("type") != "refresh": raise HTTPException( @@ -129,7 +130,15 @@ def refresh_token(req: TokenRefreshRequest) -> dict: detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) - access_token = auth_service.create_access_token(user_id, payload.get("email", "")) + # The user must still exist: refresh tokens outlive account deletion. + user = db.get(User, user_id) + if user is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="User no longer exists", + headers={"WWW-Authenticate": "Bearer"}, + ) + access_token = auth_service.create_access_token(user.id, user.email) return {"access_token": access_token, "token_type": "bearer"} diff --git a/backend/src/app/api/clusters.py b/backend/src/app/api/clusters.py index eadbb62..d50ecb0 100644 --- a/backend/src/app/api/clusters.py +++ b/backend/src/app/api/clusters.py @@ -4,6 +4,7 @@ from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Path, Query, Request +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy.orm import Session from app.db.deps import get_current_user, get_db @@ -22,6 +23,10 @@ router = APIRouter(prefix="/api/clusters", tags=["clusters"]) +# Bearer scheme for unattended agent heartbeats (carries the register_token, +# not a user JWT); optional so the body-token fallback still works. +_heartbeat_bearer = HTTPBearer(auto_error=False) + def _platform_api_url(request: Request) -> str: """Return the public base URL the agent should heartbeat back to.""" @@ -125,14 +130,17 @@ def heartbeat( payload: ClusterHeartbeat, cluster_id: str = Path(...), db: Session = Depends(get_db), + credentials: HTTPAuthorizationCredentials | None = Depends(_heartbeat_bearer), ): """Endpoint called by the cluster agent every N seconds with telemetry. - Authenticates via the cluster's `register_token` (returned on creation). - Does not require a logged-in user, since the agent runs unattended. + Authenticates via the cluster's `register_token` (returned on creation), + supplied as an `Authorization: Bearer` header or in the body. Does not + require a logged-in user, since the agent runs unattended. """ + auth_token = credentials.credentials if credentials else None try: - cluster = cluster_service.record_heartbeat(db, cluster_id, payload) + cluster = cluster_service.record_heartbeat(db, cluster_id, payload, auth_token=auth_token) except cluster_service.ClusterError as exc: raise HTTPException(status_code=403, detail=str(exc)) from exc if not cluster: diff --git a/backend/src/app/api/datasets.py b/backend/src/app/api/datasets.py index 0f5ccc8..5bb3d21 100644 --- a/backend/src/app/api/datasets.py +++ b/backend/src/app/api/datasets.py @@ -21,8 +21,10 @@ from app.models.asset import Asset from app.models.dataset import ClassMap, Dataset from app.models.dataset_version import DatasetVersion +from app.models.project import Project from app.models.user import User -from app.services import datumaro_service +from app.models.workspace import Membership, Role +from app.services import authz, datumaro_service from app.services.dataset_service import snapshot_version from app.services.project_dataset_service import create_dataset as svc_create_dataset @@ -32,6 +34,16 @@ _LABELED_STATUSES = ("labeled", "prelabeled") +def _member_workspace_ids(db: Session, user: User) -> list[str]: + """Workspace ids the user belongs to, plus the shared default workspace.""" + ws_ids = { + m.workspace_id + for m in db.scalars(select(Membership).where(Membership.user_id == user.id)).all() + } + ws_ids.add(authz.DEFAULT_WORKSPACE_ID) + return list(ws_ids) + + def _coverage_for_version(db: Session, version_id: str | None) -> tuple[float, str]: """Return (coverage_pct, status) for a dataset version. @@ -94,7 +106,12 @@ def list_datasets( ): base = select(Dataset) if project_id: + authz.require_project_access(db, current_user, project_id, Role.VIEWER) base = base.where(Dataset.project_id == project_id) + elif not authz.is_superuser(db, current_user): + base = base.join(Project, Dataset.project_id == Project.id).where( + Project.workspace_id.in_(_member_workspace_ids(db, current_user)) + ) total = db.scalar(select(func.count()).select_from(base.subquery())) or 0 offset = (page - 1) * page_size @@ -145,9 +162,7 @@ def get_dataset( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): - d = db.get(Dataset, dataset_id) - if not d: - raise HTTPException(status_code=404, detail="Dataset not found") + d = authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) versions = list( db.scalars( select(DatasetVersion) @@ -197,6 +212,7 @@ def create_dataset( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_project_access(db, current_user, project_id, Role.DEVELOPER) try: d, v = svc_create_dataset( db, @@ -238,9 +254,7 @@ def update_dataset_task_type( status_code=400, detail=f"task_type must be one of {list(VALID_TASK_TYPES)}", ) - d = db.get(Dataset, dataset_id) - if not d: - raise HTTPException(status_code=404, detail="Dataset not found") + d = authz.require_dataset_access(db, current_user, dataset_id, Role.DEVELOPER) d.task_type = body.task_type db.add(d) db.commit() @@ -253,6 +267,7 @@ def list_versions( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_dataset_access(db, current_user, dataset_id, Role.VIEWER) versions = list( db.scalars( select(DatasetVersion) @@ -280,9 +295,7 @@ def create_snapshot( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): - d = db.get(Dataset, dataset_id) - if not d: - raise HTTPException(status_code=404, detail="Dataset not found") + authz.require_dataset_access(db, current_user, dataset_id, Role.DEVELOPER) v = snapshot_version(db, dataset_id, notes=body.notes) return { "id": v.id, @@ -307,9 +320,7 @@ async def import_dataset( If `version_id` is omitted, a fresh draft version is created. The archive must contain images plus annotation files for the chosen format. """ - d = db.get(Dataset, dataset_id) - if not d: - raise HTTPException(status_code=404, detail="Dataset not found") + authz.require_dataset_access(db, current_user, dataset_id, Role.DEVELOPER) # Resolve target version if version_id: @@ -367,6 +378,7 @@ def export_dataset( current_user: User = Depends(get_current_user), ): """Export a dataset version as a zip archive in the requested format.""" + authz.require_dataset_access(db, current_user, dataset_id, Role.DEVELOPER) try: archive = datumaro_service.export_archive( db, dataset_id=dataset_id, version_id=version_id, fmt=fmt @@ -386,9 +398,7 @@ def update_classes( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): - d = db.get(Dataset, dataset_id) - if not d: - raise HTTPException(status_code=404, detail="Dataset not found") + d = authz.require_dataset_access(db, current_user, dataset_id, Role.DEVELOPER) if d.class_map_id: cm = db.get(ClassMap, d.class_map_id) if cm: diff --git a/backend/src/app/api/evaluations.py b/backend/src/app/api/evaluations.py index 32d670b..c07df09 100644 --- a/backend/src/app/api/evaluations.py +++ b/backend/src/app/api/evaluations.py @@ -1,10 +1,16 @@ from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Path, Query +from sqlalchemy import func, select from sqlalchemy.orm import Session from app.db.deps import get_current_user, get_db +from app.models.artifact import ModelArtifact +from app.models.dataset_version import DatasetVersion +from app.models.evaluation import Evaluation +from app.models.project import Project from app.models.user import User +from app.models.workspace import Membership, Role from app.schemas.evaluation import ( EvaluationCreate, EvaluationJobResponse, @@ -12,11 +18,20 @@ EvaluationOut, EvaluationSummary, ) -from app.services import cluster_service, evaluation_service +from app.services import authz, cluster_service, evaluation_service router = APIRouter(prefix="/api/evaluations", tags=["evaluations"]) +def _member_workspace_ids(db: Session, user: User) -> list[str]: + ws_ids = { + m.workspace_id + for m in db.scalars(select(Membership).where(Membership.user_id == user.id)).all() + } + ws_ids.add(authz.DEFAULT_WORKSPACE_ID) + return list(ws_ids) + + @router.get("", response_model=EvaluationListPage) def list_evaluations( artifact_id: str | None = Query(None), @@ -27,6 +42,41 @@ def list_evaluations( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + # Enforce access on every filter the caller supplied. + if project_id: + authz.require_project_access(db, current_user, project_id, Role.VIEWER) + if artifact_id: + artifact = db.get(ModelArtifact, artifact_id) + if artifact is not None: + authz.require_project_access(db, current_user, artifact.project_id, Role.VIEWER) + if dataset_version_id: + version = db.get(DatasetVersion, dataset_version_id) + if version is not None: + authz.require_dataset_access(db, current_user, version.dataset_id, Role.VIEWER) + + # Unfiltered listing: scope to the caller's workspaces (superusers see all). + if ( + not project_id + and not artifact_id + and not dataset_version_id + and not authz.is_superuser(db, current_user) + ): + q = ( + select(Evaluation) + .join(Project, Evaluation.project_id == Project.id) + .where(Project.workspace_id.in_(_member_workspace_ids(db, current_user))) + .order_by(Evaluation.created_at.desc()) + ) + total = db.scalar(select(func.count()).select_from(q.subquery())) or 0 + offset = max(0, (page - 1) * page_size) + rows = list(db.scalars(q.offset(offset).limit(page_size)).all()) + return EvaluationListPage( + items=[EvaluationSummary(**evaluation_service.summarize(r)) for r in rows], + total=int(total), + page=page, + page_size=page_size, + ) + rows, total = evaluation_service.list_evaluations( db, artifact_id=artifact_id, @@ -50,6 +100,14 @@ def create_evaluation( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + # Missing artifact/version is left to the service (it raises 400-mapped + # EvaluationError); when they resolve, enforce developer access. + artifact = db.get(ModelArtifact, payload.artifact_id) + if artifact is not None: + authz.require_project_access(db, current_user, artifact.project_id, Role.DEVELOPER) + version = db.get(DatasetVersion, payload.dataset_version_id) + if version is not None: + authz.require_dataset_access(db, current_user, version.dataset_id, Role.DEVELOPER) try: _, job = evaluation_service.create_evaluation(db, payload) except evaluation_service.EvaluationError as exc: @@ -68,4 +126,5 @@ def get_evaluation( row = evaluation_service.get_evaluation(db, evaluation_id) if not row: raise HTTPException(status_code=404, detail="Evaluation not found") + authz.require_project_access(db, current_user, row.project_id, Role.VIEWER) return EvaluationOut(**evaluation_service.to_dict(row)) diff --git a/backend/src/app/api/experiments.py b/backend/src/app/api/experiments.py index 94a3a59..73957bc 100644 --- a/backend/src/app/api/experiments.py +++ b/backend/src/app/api/experiments.py @@ -8,13 +8,25 @@ from app.db.deps import get_current_user, get_db from app.models.experiment import ExperimentRun as ExperimentModel +from app.models.project import Project from app.models.user import User +from app.models.workspace import Membership, Role from app.schemas.experiment import Experiment as ExperimentSchema from app.schemas.experiment import ExperimentCreate +from app.services import authz router = APIRouter(prefix="/api/experiments", tags=["experiments"]) +def _member_workspace_ids(db: Session, user: User) -> list[str]: + ws_ids = { + m.workspace_id + for m in db.scalars(select(Membership).where(Membership.user_id == user.id)).all() + } + ws_ids.add(authz.DEFAULT_WORKSPACE_ID) + return list(ws_ids) + + def _run_to_schema(e: ExperimentModel) -> ExperimentSchema: return ExperimentSchema( id=e.id, @@ -42,7 +54,12 @@ def list_runs( ): base = select(ExperimentModel) if project_id: + authz.require_project_access(db, current_user, project_id, Role.VIEWER) base = base.where(ExperimentModel.project_id == project_id) + elif not authz.is_superuser(db, current_user): + base = base.join(Project, ExperimentModel.project_id == Project.id).where( + Project.workspace_id.in_(_member_workspace_ids(db, current_user)) + ) total = db.scalar(select(func.count()).select_from(base.subquery())) or 0 offset = (page - 1) * page_size rows = list( @@ -67,6 +84,7 @@ def get_run( e = db.get(ExperimentModel, runId) if not e: raise HTTPException(status_code=404, detail="Run not found") + authz.require_project_access(db, current_user, e.project_id, Role.VIEWER) return _run_to_schema(e) @@ -76,6 +94,7 @@ def create_run( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_project_access(db, current_user, body.project_id, Role.DEVELOPER) run = ExperimentModel( project_id=body.project_id, dataset_version_id=body.dataset_version_id, @@ -100,6 +119,7 @@ def get_metrics( e = db.get(ExperimentModel, runId) if not e: raise HTTPException(status_code=404, detail="Run not found") + authz.require_project_access(db, current_user, e.project_id, Role.VIEWER) metrics: list = [] summary: dict | None = None plots: list = [] @@ -169,6 +189,8 @@ def get_plot( from fastapi.responses import StreamingResponse e = db.get(ExperimentModel, runId) + if e: + authz.require_project_access(db, current_user, e.project_id, Role.VIEWER) if not e or not e.metrics_json: raise HTTPException(status_code=404, detail="Run or plots not found") try: diff --git a/backend/src/app/api/jobs.py b/backend/src/app/api/jobs.py index 00e8393..a452d9f 100644 --- a/backend/src/app/api/jobs.py +++ b/backend/src/app/api/jobs.py @@ -3,18 +3,25 @@ from fastapi import APIRouter, Depends, HTTPException, Path from sqlalchemy.orm import Session -from app.db.deps import get_db +from app.db.deps import get_current_user, get_db from app.models.job import Job as JobModel +from app.models.user import User from app.schemas.common import Job as JobSchema +from app.services.authz import require_job_access router = APIRouter(prefix="/api", tags=["jobs"]) @router.get("/jobs/{jobId}", response_model=JobSchema) -def get_job(jobId: str = Path(...), db: Session = Depends(get_db)): +def get_job( + jobId: str = Path(...), + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user), +): job = db.get(JobModel, jobId) if not job: raise HTTPException(status_code=404, detail="Job not found") + require_job_access(db, current_user, job) return JobSchema( id=job.id, jobId=job.id, diff --git a/backend/src/app/api/middleware.py b/backend/src/app/api/middleware.py index 1056049..6ac2ba6 100644 --- a/backend/src/app/api/middleware.py +++ b/backend/src/app/api/middleware.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import time import traceback import uuid @@ -30,31 +31,95 @@ async def _mw(request, call_next: Callable): return _mw -# Rate limiting for auth endpoints (simple in-memory, replace with Redis in prod) +# Rate limiting for auth endpoints. Backed by Redis (shared across replicas) +# with a bounded in-memory fallback when Redis is unreachable. _auth_attempts: dict[str, list[float]] = {} _MAX_ATTEMPTS = 10 # per window _WINDOW_SECONDS = 60 +_FALLBACK_MAX_KEYS = 10_000 # cap fallback memory under attack +_REDIS_RETRY_SECONDS = 30.0 + +_redis_client: object | None = None +_redis_failed_at: float = 0.0 + + +def _get_redis(): + """Lazily connect to Redis; back off for a while after a failure.""" + global _redis_client, _redis_failed_at + if _redis_client is not None: + return _redis_client + if time.time() - _redis_failed_at < _REDIS_RETRY_SECONDS: + return None + try: + import redis + + client = redis.Redis.from_url( + os.getenv("REDIS_URL", "redis://localhost:6379/0"), + socket_connect_timeout=0.5, + socket_timeout=0.5, + ) + client.ping() + _redis_client = client + return client + except Exception: + _redis_failed_at = time.time() + return None + + +def _over_limit_redis(client, client_ip: str) -> bool: + # Fixed-window counter: one key per IP per window bucket. + bucket = int(time.time() // _WINDOW_SECONDS) + key = f"vf:auth_rl:{client_ip}:{bucket}" + pipe = client.pipeline() + pipe.incr(key) + pipe.expire(key, _WINDOW_SECONDS * 2) + count, _ = pipe.execute() + return int(count) > _MAX_ATTEMPTS + + +def _over_limit_memory(client_ip: str) -> bool: + now = time.time() + window_start = now - _WINDOW_SECONDS + if len(_auth_attempts) > _FALLBACK_MAX_KEYS: + for ip in list(_auth_attempts): + pruned = [t for t in _auth_attempts[ip] if t > window_start] + if pruned: + _auth_attempts[ip] = pruned + else: + del _auth_attempts[ip] + attempts = [t for t in _auth_attempts.get(client_ip, []) if t > window_start] + if len(attempts) >= _MAX_ATTEMPTS: + _auth_attempts[client_ip] = attempts + return True + attempts.append(now) + _auth_attempts[client_ip] = attempts + return False def auth_rate_limit_middleware(): - """Simple in-memory rate limiter for /auth/* endpoints.""" + """Rate limiter for /auth/* endpoints (Redis-backed, in-memory fallback).""" async def _mw(request, call_next: Callable): if request.url.path.startswith("/auth/"): client_ip = request.client.host if request.client else "unknown" - now = time.time() - window_start = now - _WINDOW_SECONDS - attempts = _auth_attempts.get(client_ip, []) - # Prune old entries - attempts = [t for t in attempts if t > window_start] - if len(attempts) >= _MAX_ATTEMPTS: + over_limit = False + redis_client = _get_redis() + if redis_client is not None: + try: + over_limit = _over_limit_redis(redis_client, client_ip) + except Exception: + global _redis_client, _redis_failed_at + _redis_client = None + _redis_failed_at = time.time() + over_limit = _over_limit_memory(client_ip) + else: + over_limit = _over_limit_memory(client_ip) + if over_limit: return JSONResponse( {"detail": "Too many requests. Please wait before trying again."}, status_code=429, headers={"Retry-After": str(_WINDOW_SECONDS)}, ) - attempts.append(now) - _auth_attempts[client_ip] = attempts return await call_next(request) return _mw diff --git a/backend/src/app/api/ops.py b/backend/src/app/api/ops.py index f343fbc..f68fee0 100644 --- a/backend/src/app/api/ops.py +++ b/backend/src/app/api/ops.py @@ -4,7 +4,10 @@ from sqlalchemy.orm import Session from app.db.deps import get_current_user, get_db +from app.models.dataset_version import DatasetVersion +from app.models.experiment import ExperimentRun from app.models.user import User +from app.models.workspace import Role from app.schemas.common import ( Job, OnnxExportRequest, @@ -12,7 +15,7 @@ UploadUrlRequest, UploadUrlResponse, ) -from app.services import cluster_service +from app.services import authz, cluster_service from app.services.ingest_service import get_presigned_upload from app.services.onnx_service import OnnxDispatchError from app.services.onnx_service import export_onnx as svc_export_onnx @@ -24,8 +27,12 @@ @router.post("/ingest/upload-url", response_model=UploadUrlResponse) def get_upload_url( payload: UploadUrlRequest, + db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + version = db.get(DatasetVersion, payload.datasetVersionId) + if version is not None: + authz.require_dataset_access(db, current_user, version.dataset_id, Role.DEVELOPER) data = get_presigned_upload(payload.datasetVersionId, payload.filename, payload.contentType) # Derive the objectKey using the same path template as storage.presign_put_url object_key = f"datasets/{payload.datasetVersionId}/{payload.filename}" @@ -42,6 +49,10 @@ def train( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + authz.require_project_access(db, current_user, payload.projectId, Role.DEVELOPER) + version = db.get(DatasetVersion, payload.datasetVersionId) + if version is not None: + authz.require_dataset_access(db, current_user, version.dataset_id, Role.DEVELOPER) try: job = launch_training( db, @@ -68,6 +79,9 @@ def export_onnx( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + run = db.get(ExperimentRun, payload.experimentId) + if run is not None: + authz.require_project_access(db, current_user, run.project_id, Role.DEVELOPER) try: job = svc_export_onnx( db, payload.experimentId, payload.dynamicAxes, cluster_id=payload.clusterId @@ -157,6 +171,7 @@ def trigger_frame_extraction( current_user: User = Depends(get_current_user), ): """Queue a frame-extraction Celery task for a video asset.""" + authz.require_dataset_access(db, current_user, dataset_id, Role.DEVELOPER) from app.jobs.celery_app import celery_app from app.models.asset import Asset from app.services.jobs_service import create_job, update_job_status diff --git a/backend/src/app/api/projects.py b/backend/src/app/api/projects.py index 3817008..fcc2b74 100644 --- a/backend/src/app/api/projects.py +++ b/backend/src/app/api/projects.py @@ -9,7 +9,8 @@ from app.models.dataset import Dataset from app.models.project import Project as ProjectModel from app.models.user import User -from app.models.workspace import Membership, Workspace +from app.models.workspace import Membership, Role, Workspace +from app.services.authz import require_project_access, require_workspace_access from app.services.project_dataset_service import ( VALID_TASK_TYPES, ) @@ -78,6 +79,7 @@ def list_projects( base = select(ProjectModel) if workspace_id: + require_workspace_access(db, current_user, workspace_id) base = base.where(ProjectModel.workspace_id == workspace_id) else: base = base.where(ProjectModel.workspace_id.in_(ws_ids)) @@ -114,9 +116,7 @@ def get_project( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): - p = db.get(ProjectModel, project_id) - if not p: - raise HTTPException(status_code=404, detail="Project not found") + p = require_project_access(db, current_user, project_id) datasets = list(db.scalars(select(Dataset).where(Dataset.project_id == project_id)).all()) return { "id": p.id, @@ -140,6 +140,7 @@ def create_project( ws = db.get(Workspace, workspace_id) if not ws: workspace_id = _DEFAULT_WORKSPACE_ID + require_workspace_access(db, current_user, workspace_id, Role.DEVELOPER) if payload.task_type is not None and payload.task_type not in VALID_TASK_TYPES: raise HTTPException( @@ -175,9 +176,7 @@ def update_project( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): - p = db.get(ProjectModel, project_id) - if not p: - raise HTTPException(status_code=404, detail="Project not found") + p = require_project_access(db, current_user, project_id, Role.DEVELOPER) if payload.task_type is not None and payload.task_type not in VALID_TASK_TYPES: raise HTTPException( status_code=400, @@ -222,6 +221,7 @@ def project_wizard( ws = db.get(Workspace, workspace_id) if not ws: workspace_id = _DEFAULT_WORKSPACE_ID + require_workspace_access(db, current_user, workspace_id, Role.DEVELOPER) import json as _json @@ -287,6 +287,7 @@ def list_project_datasets( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + require_project_access(db, current_user, project_id) datasets = list(db.scalars(select(Dataset).where(Dataset.project_id == project_id)).all()) return [ { diff --git a/backend/src/app/api/rbac.py b/backend/src/app/api/rbac.py index 33e2944..f91b4ad 100644 --- a/backend/src/app/api/rbac.py +++ b/backend/src/app/api/rbac.py @@ -1,7 +1,5 @@ from __future__ import annotations -import os - from fastapi import Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.orm import Session @@ -34,9 +32,11 @@ def dependency( current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> User: - # Superusers bypass all role checks - superuser_email = os.getenv("FIRST_SUPERUSER_EMAIL") or os.getenv("SUPERUSER_EMAIL") - if superuser_email and current_user.email == superuser_email: + # Superusers bypass all role checks (column-backed, with the seed + # admin's env-var email kept as a fallback for pre-column users). + from app.services.authz import is_superuser + + if is_superuser(db, current_user): return current_user membership = db.scalar( diff --git a/backend/src/app/api/workspaces.py b/backend/src/app/api/workspaces.py index 4051c0d..384c001 100644 --- a/backend/src/app/api/workspaces.py +++ b/backend/src/app/api/workspaces.py @@ -8,6 +8,7 @@ from app.db.deps import get_current_user, get_db from app.models.user import User from app.models.workspace import Membership, Role, Workspace +from app.services.authz import require_workspace_access router = APIRouter(prefix="/api/workspaces", tags=["workspaces"]) @@ -81,6 +82,7 @@ def get_workspace( ws = db.get(Workspace, workspace_id) if not ws: raise HTTPException(status_code=404, detail="Workspace not found") + require_workspace_access(db, current_user, workspace_id) return { "id": ws.id, "name": ws.name, @@ -95,6 +97,7 @@ def list_members( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): + require_workspace_access(db, current_user, workspace_id) members = list( db.scalars(select(Membership).where(Membership.workspace_id == workspace_id)).all() ) @@ -122,6 +125,7 @@ def invite_member( ws = db.get(Workspace, workspace_id) if not ws: raise HTTPException(status_code=404, detail="Workspace not found") + require_workspace_access(db, current_user, workspace_id, Role.ADMIN) # Look up user by email invited_user = db.scalars(select(User).where(User.email == body.email)).first() if not invited_user: diff --git a/backend/src/app/db/migrations/versions/0009_asset_embedding_pgvector.py b/backend/src/app/db/migrations/versions/0009_asset_embedding_pgvector.py new file mode 100644 index 0000000..017f40a --- /dev/null +++ b/backend/src/app/db/migrations/versions/0009_asset_embedding_pgvector.py @@ -0,0 +1,64 @@ +"""Add pgvector embedding column to assets; normalize label_status spelling. + +Embeddings were previously serialized into ``assets.meta_data`` JSON, which +made similarity search / duplicate detection impossible without a full-table +scan. This adds a real ``vector(512)`` column (ViT-B-32 dim) with an ANN +index. On databases without the pgvector extension (SQLite tests, stock +postgres images) the column degrades to text storage. + +Also rewrites the legacy UK spelling ``unlabelled`` to ``unlabeled`` so exact +status filters behave consistently. + +Revision ID: 0009_asset_embedding_pgvector +Revises: 0008_training_framework +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "0009_asset_embedding_pgvector" +down_revision = "0008_training_framework" +branch_labels = None +depends_on = None + +_DIM = 512 + + +def _pgvector_available(bind) -> bool: + if bind.dialect.name != "postgresql": + return False + row = bind.execute( + sa.text("SELECT 1 FROM pg_available_extensions WHERE name = 'vector'") + ).first() + return row is not None + + +def upgrade() -> None: + bind = op.get_bind() + if _pgvector_available(bind): + op.execute("CREATE EXTENSION IF NOT EXISTS vector") + op.execute(f"ALTER TABLE assets ADD COLUMN IF NOT EXISTS embedding vector({_DIM})") + # HNSW needs pgvector >= 0.5; fall back to ivfflat on older builds. + try: + op.execute( + "CREATE INDEX IF NOT EXISTS ix_assets_embedding " + "ON assets USING hnsw (embedding vector_cosine_ops)" + ) + except Exception: + op.execute( + "CREATE INDEX IF NOT EXISTS ix_assets_embedding " + "ON assets USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100)" + ) + else: + op.add_column("assets", sa.Column("embedding", sa.Text(), nullable=True)) + + op.execute("UPDATE assets SET label_status = 'unlabeled' WHERE label_status = 'unlabelled'") + + +def downgrade() -> None: + bind = op.get_bind() + if bind.dialect.name == "postgresql": + op.execute("DROP INDEX IF EXISTS ix_assets_embedding") + op.drop_column("assets", "embedding") diff --git a/backend/src/app/db/session.py b/backend/src/app/db/session.py index 92c7595..f3dba10 100644 --- a/backend/src/app/db/session.py +++ b/backend/src/app/db/session.py @@ -12,11 +12,15 @@ def _db_url() -> str: + from app.settings import require_secure_setting + url = os.getenv("DATABASE_URL") if url: return url user = os.getenv("POSTGRES_USER", "visionforge") - pwd = os.getenv("POSTGRES_PASSWORD", "change-me") + pwd = require_secure_setting( + "POSTGRES_PASSWORD", os.getenv("POSTGRES_PASSWORD", "change-me"), ("change-me",) + ) host = os.getenv("POSTGRES_HOST", "localhost") port = os.getenv("POSTGRES_PORT", "5432") db = os.getenv("POSTGRES_DB", "visionforge") diff --git a/backend/src/app/db/types.py b/backend/src/app/db/types.py new file mode 100644 index 0000000..bde8813 --- /dev/null +++ b/backend/src/app/db/types.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +import json + +from sqlalchemy.types import Text, TypeDecorator + + +class EmbeddingVector(TypeDecorator): + """A float-vector column: pgvector ``Vector`` on PostgreSQL, JSON text elsewhere. + + SQLite (used by tests) has no vector type, so values are JSON-encoded + there. On PostgreSQL the pgvector type enables ANN similarity search. + """ + + impl = Text + cache_ok = True + + def __init__(self, dim: int) -> None: + super().__init__() + self.dim = dim + + def load_dialect_impl(self, dialect): + if dialect.name == "postgresql": + try: + from pgvector.sqlalchemy import Vector + + return dialect.type_descriptor(Vector(self.dim)) + except ImportError: + pass + return dialect.type_descriptor(Text()) + + def process_bind_param(self, value, dialect): + if value is None or dialect.name == "postgresql": + return value + return json.dumps([float(v) for v in value]) + + def process_result_value(self, value, dialect): + if value is None or dialect.name == "postgresql": + return value + try: + return json.loads(value) + except (ValueError, TypeError): + return None diff --git a/backend/src/app/jobs/celery_app.py b/backend/src/app/jobs/celery_app.py index 2926244..7163dac 100644 --- a/backend/src/app/jobs/celery_app.py +++ b/backend/src/app/jobs/celery_app.py @@ -37,18 +37,35 @@ def send_task(self, *args, **kwargs): # pragma: no cover - used only when Celer "app.jobs.tasks.uncertainty", "app.jobs.tasks.error_mining", "app.jobs.tasks.al_uncertainty", + "app.jobs.tasks.maintenance", ], ) celery_app.conf.task_routes = { "app.jobs.tasks.*": {"queue": "default"}, } + # Hard ceiling for any task; training defaults to 24h, everything inherits + # unless overridden per-task. acks_late + reject_on_worker_lost means a + # task whose worker dies is redelivered instead of silently vanishing + # (the visibility_timeout must exceed the longest task for Redis brokers). + _task_time_limit = int(os.getenv("CELERY_TASK_TIME_LIMIT", str(24 * 3600))) celery_app.conf.update( task_serializer="json", result_serializer="json", accept_content=["json"], task_track_started=True, - worker_prefetch_multiplier=int(os.getenv("CELERY_PREFETCH", "4")), + task_acks_late=True, + task_reject_on_worker_lost=True, + task_time_limit=_task_time_limit, + task_soft_time_limit=max(_task_time_limit - 300, 60), + broker_transport_options={"visibility_timeout": _task_time_limit + 3600}, + worker_prefetch_multiplier=int(os.getenv("CELERY_PREFETCH", "1")), broker_connection_retry_on_startup=True, + beat_schedule={ + "sweep-stale-jobs": { + "task": "app.jobs.tasks.maintenance.sweep_stale_jobs", + "schedule": float(os.getenv("VF_JOB_SWEEP_INTERVAL", "300")), + }, + }, ) else: celery_app = _DummyCelery() # type: ignore diff --git a/backend/src/app/jobs/tasks/embeddings.py b/backend/src/app/jobs/tasks/embeddings.py index a60bfd4..45598a1 100644 --- a/backend/src/app/jobs/tasks/embeddings.py +++ b/backend/src/app/jobs/tasks/embeddings.py @@ -1,6 +1,5 @@ from __future__ import annotations -import json import os import tempfile from pathlib import Path @@ -98,15 +97,9 @@ def generate_embeddings(payload: dict) -> dict: image_vectors.append(vec) - # Persist embedding into Asset.meta_data - meta: dict = {} - if asset.meta_data: - try: - meta = json.loads(asset.meta_data) - except Exception: - meta = {} - meta["embedding"] = vec - asset.meta_data = json.dumps(meta) + # Persist to the dedicated vector column (pgvector on + # Postgres) so similarity queries can use the ANN index. + asset.embedding = vec db.add(asset) # Commit in batches to avoid long transactions diff --git a/backend/src/app/jobs/tasks/frame_extraction.py b/backend/src/app/jobs/tasks/frame_extraction.py index 17acab3..3f6607f 100644 --- a/backend/src/app/jobs/tasks/frame_extraction.py +++ b/backend/src/app/jobs/tasks/frame_extraction.py @@ -151,7 +151,7 @@ def extract_frames(payload: dict) -> dict: "timestamp_seconds": frame_num / video_fps, } ), - label_status="unlabelled", + label_status="unlabeled", ) db.add(frame_asset) frame_assets.append( diff --git a/backend/src/app/jobs/tasks/maintenance.py b/backend/src/app/jobs/tasks/maintenance.py new file mode 100644 index 0000000..024e3cd --- /dev/null +++ b/backend/src/app/jobs/tasks/maintenance.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import os +from datetime import datetime, timedelta, timezone + +import structlog + +try: + from celery import shared_task # type: ignore +except Exception: # pragma: no cover + + def shared_task(*args, **kwargs): + def _wrap(fn): + return fn + + return _wrap + + +logger = structlog.get_logger(__name__) + +# A job is considered stuck when it has reported no progress for this long. +# Training commits progress at least once per epoch, so 6h is conservative. +_STALE_RUNNING_SECONDS = int(os.getenv("VF_JOB_STALE_RUNNING_SECONDS", str(6 * 3600))) +# A job that never left "queued" (e.g. enqueued while the broker was down, +# or its dedicated cluster never picked it up) is failed after this long. +_STALE_QUEUED_SECONDS = int(os.getenv("VF_JOB_STALE_QUEUED_SECONDS", str(24 * 3600))) + + +def _make_session(): + from app.db.session import SessionLocal + + return SessionLocal() + + +@shared_task(name="app.jobs.tasks.maintenance.sweep_stale_jobs") +def sweep_stale_jobs() -> dict: + """Fail stuck jobs and release their clusters (runs via celery beat). + + Safety net for the cases task-level error handling can't cover: a worker + killed mid-task, a broker outage after the job row was created, or an + agent that vanished while holding a cluster reservation. + """ + from app.models.cluster import Cluster + from app.models.job import Job + from app.services.jobs_service import update_job_status + + db = _make_session() + swept: list[str] = [] + released: list[str] = [] + try: + now = datetime.now(timezone.utc) + candidates = db.query(Job).filter(Job.status.in_(["queued", "running"])).all() # noqa: E501 + for job in candidates: + ref = job.updated_at or job.created_at + if ref is None: + continue + if ref.tzinfo is None: + ref = ref.replace(tzinfo=timezone.utc) + limit = _STALE_RUNNING_SECONDS if job.status == "running" else _STALE_QUEUED_SECONDS + if now - ref > timedelta(seconds=limit): + job.error_message = ( + f"Swept by maintenance: no progress for over {limit}s " f"while {job.status}" + ) + db.add(job) + db.commit() + # update_job_status also releases any cluster reserved for the job. + update_job_status(db, job.id, status="failed") + swept.append(job.id) + + # Release clusters whose reserved job is already terminal or gone. + busy = db.query(Cluster).filter(Cluster.active_job_id.isnot(None)).all() + for cluster in busy: + job = db.get(Job, cluster.active_job_id) + if job is None or job.status in ("succeeded", "failed", "cancelled"): + cluster.active_job_id = None + if cluster.status == "busy": + cluster.status = "online" + db.add(cluster) + db.commit() + released.append(cluster.id) + finally: + db.close() + + if swept or released: + logger.warning("maintenance:swept", jobs=swept, clusters=released) + return {"swept_jobs": swept, "released_clusters": released} diff --git a/backend/src/app/jobs/tasks/prelabels.py b/backend/src/app/jobs/tasks/prelabels.py index c932817..f220508 100644 --- a/backend/src/app/jobs/tasks/prelabels.py +++ b/backend/src/app/jobs/tasks/prelabels.py @@ -70,7 +70,9 @@ def apply_prelabels(payload: dict) -> dict: db.query(Asset) .filter( Asset.version_id == dataset_version_id, - Asset.label_status == "unlabelled", + # Accept the legacy UK spelling for rows predating the + # normalization migration. + Asset.label_status.in_(("unlabeled", "unlabelled")), ) .all() ) diff --git a/backend/src/app/jobs/tasks/training.py b/backend/src/app/jobs/tasks/training.py index 1627662..ab88bc6 100644 --- a/backend/src/app/jobs/tasks/training.py +++ b/backend/src/app/jobs/tasks/training.py @@ -309,7 +309,9 @@ def report(progress: float | None = None, epoch: dict | None = None) -> None: result = {"status": "failed", "error": error_msg} try: if experiment_id: - run_obj = db.get(ExperimentRun, experiment_id) if "ExperimentRun" in dir() else None + from app.models.experiment import ExperimentRun as _ExperimentRun + + run_obj = db.get(_ExperimentRun, experiment_id) if run_obj: run_obj.status = "failed" run_obj.completed_at = datetime.now(timezone.utc) diff --git a/backend/src/app/main.py b/backend/src/app/main.py index bfa4e2a..642a8f8 100644 --- a/backend/src/app/main.py +++ b/backend/src/app/main.py @@ -5,12 +5,15 @@ import os import time +import structlog from alembic import command as alembic_command from alembic.config import Config as AlembicConfig -from fastapi import Depends, FastAPI, Response +from fastapi import Depends, FastAPI, HTTPException, Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, StreamingResponse +from fastapi.security import HTTPAuthorizationCredentials from prometheus_client import CONTENT_TYPE_LATEST, generate_latest +from sqlalchemy import text from sqlalchemy.orm import Session from app.api.agents import router as agents_router @@ -26,11 +29,12 @@ from app.api.ops import router as ops_router from app.api.projects import router as projects_router from app.api.training import router as training_router -from app.db.deps import get_db +from app.db.deps import get_db, security from app.models.job import Job as JobModel from app.observability.logging import configure_logging from app.observability.metrics import metrics_middleware from app.services.auth import ensure_superuser +from app.services.authz import require_job_access # Import new routers (created by the new API implementation) try: @@ -75,8 +79,8 @@ CORSMiddleware, allow_origins=[o.strip() for o in _cors_origins if o.strip()], allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], + allow_headers=["Authorization", "Content-Type", "X-Request-ID"], expose_headers=["X-Request-ID"], ) @@ -98,19 +102,45 @@ def _should_init_db() -> bool: return True +# Arbitrary app-wide key for the Postgres advisory lock serializing migrations. +_MIGRATION_LOCK_KEY = 815_001 + + +def _upgrade_with_lock(cfg: AlembicConfig) -> None: + """Run ``alembic upgrade head``, serialized across replicas on Postgres. + + Without the advisory lock, several API replicas starting at once would + run migrations concurrently and can deadlock or double-apply DDL. + """ + from app.db.session import engine + + if engine.dialect.name != "postgresql": + alembic_command.upgrade(cfg, "head") + return + with engine.connect() as conn: + conn.execute(text("SELECT pg_advisory_lock(:key)"), {"key": _MIGRATION_LOCK_KEY}) + try: + alembic_command.upgrade(cfg, "head") + finally: + conn.execute(text("SELECT pg_advisory_unlock(:key)"), {"key": _MIGRATION_LOCK_KEY}) + + def _run_db_migrations(): + logger = structlog.get_logger(__name__) alembic_ini, migrations_dir = _alembic_paths() cfg = AlembicConfig(alembic_ini) cfg.set_main_option("script_location", migrations_dir) attempts = 0 while True: try: - alembic_command.upgrade(cfg, "head") + _upgrade_with_lock(cfg) break except Exception as e: # pragma: no cover attempts += 1 if attempts >= 10: + logger.error("db:migrations_failed", attempts=attempts, error=str(e)) raise e + logger.warning("db:migrations_retry", attempt=attempts, error=str(e)) time.sleep(2) @@ -141,25 +171,66 @@ def _run_db_migrations(): app.include_router(workspaces_router) +@app.exception_handler(Exception) +async def _unhandled_exception_handler(request: Request, exc: Exception): + # Dispatch failures surface as 502 so the UI can show "broker unavailable" + # instead of a phantom queued job. + from app.services.onnx_service import OnnxDispatchError + from app.services.training_service import TrainingDispatchError + + if isinstance(exc, (TrainingDispatchError, OnnxDispatchError)): + return JSONResponse({"detail": str(exc)}, status_code=502) + raise exc + + @app.get("/health", tags=["ops"]) def health(): return JSONResponse({"status": "ok", "version": "1.0.0"}) @app.get("/metrics", tags=["ops"]) -def metrics(): +def metrics(request: Request): + # When METRICS_BEARER_TOKEN is set, the scrape endpoint requires it; + # otherwise it stays open (dev / private-network deployments). + expected = os.getenv("METRICS_BEARER_TOKEN") + if expected: + import hmac as _hmac + + auth = request.headers.get("Authorization", "") + presented = auth.removeprefix("Bearer ").strip() + if not _hmac.compare_digest(presented, expected): + raise HTTPException(status_code=401, detail="Not authenticated") data = generate_latest() return Response(content=data, media_type=CONTENT_TYPE_LATEST) @app.get("/api/jobs/{job_id}/stream", tags=["jobs"]) -async def stream_job_status(job_id: str, db: Session = Depends(get_db)): +async def stream_job_status( + job_id: str, + token: str | None = None, + db: Session = Depends(get_db), + credentials: HTTPAuthorizationCredentials | None = Depends(security), +): """Server-Sent Events endpoint for real-time job status updates. - Connect with EventSource in the browser: - const es = new EventSource(`/api/jobs/${jobId}/stream`); + Auth: standard ``Authorization: Bearer`` header, or ``?token=`` query + parameter for EventSource clients (which cannot set headers): + const es = new EventSource(`/api/jobs/${jobId}/stream?token=${accessToken}`); es.onmessage = (e) => { const data = JSON.parse(e.data); ... }; """ + from app.services.auth import get_current_user_from_token + + raw_token = credentials.credentials if credentials else token + if not raw_token: + raise HTTPException( + status_code=401, + detail="Not authenticated", + headers={"WWW-Authenticate": "Bearer"}, + ) + user = get_current_user_from_token(raw_token, db) + job = db.get(JobModel, job_id) + if job is not None: + require_job_access(db, user, job) async def event_generator(): last_payload: dict | None = None diff --git a/backend/src/app/models/asset.py b/backend/src/app/models/asset.py index 658d056..dbf4582 100644 --- a/backend/src/app/models/asset.py +++ b/backend/src/app/models/asset.py @@ -6,6 +6,10 @@ from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base +from app.db.types import EmbeddingVector + +# ViT-B-32 (open-clip) output dimension; must match EmbeddingsService. +EMBEDDING_DIM = 512 def _uuid() -> str: @@ -25,5 +29,8 @@ class Asset(Base): width: Mapped[int | None] = mapped_column(Integer, nullable=True) height: Mapped[int | None] = mapped_column(Integer, nullable=True) meta_data: Mapped[str | None] = mapped_column(Text, nullable=True) # JSON - label_status: Mapped[str] = mapped_column(String(50), nullable=False, default="unlabelled") + embedding: Mapped[list[float] | None] = mapped_column( + EmbeddingVector(EMBEDDING_DIM), nullable=True + ) + label_status: Mapped[str] = mapped_column(String(50), nullable=False, default="unlabeled") created_at: Mapped[object] = mapped_column(DateTime(timezone=True), server_default=func.now()) diff --git a/backend/src/app/schemas/cluster.py b/backend/src/app/schemas/cluster.py index 1ca3f54..9da3ee8 100644 --- a/backend/src/app/schemas/cluster.py +++ b/backend/src/app/schemas/cluster.py @@ -94,10 +94,12 @@ class ClusterUpdate(BaseModel): class ClusterHeartbeat(BaseModel): """Telemetry payload sent periodically by the agent on a cluster. - The agent authenticates by including the cluster's `register_token`. + The agent authenticates by including the cluster's `register_token`, + either in this body field or as an ``Authorization: Bearer`` header + (preferred — keeps the secret out of request-body logs). """ - register_token: str + register_token: str | None = None status: ClusterStatus = "online" cpu_usage_pct: float = 0.0 ram_used_mb: int = 0 diff --git a/backend/src/app/services/auth.py b/backend/src/app/services/auth.py index ac73985..4426a0e 100644 --- a/backend/src/app/services/auth.py +++ b/backend/src/app/services/auth.py @@ -12,8 +12,12 @@ from app.db.session import SessionLocal from app.models.user import User from app.models.workspace import Workspace +from app.settings import require_secure_setting -SECRET_KEY: str = os.getenv("SECRET_KEY", "change-me-in-production-32-chars-min") +_DEFAULT_SECRET = "change-me-in-production-32-chars-min" +SECRET_KEY: str = require_secure_setting( + "SECRET_KEY", os.getenv("SECRET_KEY", _DEFAULT_SECRET), (_DEFAULT_SECRET,) +) ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 REFRESH_TOKEN_EXPIRE_DAYS = 7 @@ -169,9 +173,15 @@ def ensure_superuser() -> None: with SessionLocal() as db: existing = db.scalar(select(User).where(User.email == email)) if existing: + changed = False # Re-hash if the stored hash is not bcrypt if not existing.password_hash or not _is_bcrypt_hash(existing.password_hash): existing.password_hash = _hash_password(password) + changed = True + if not existing.is_superuser: + existing.is_superuser = True + changed = True + if changed: db.add(existing) db.commit() user = existing @@ -180,6 +190,7 @@ def ensure_superuser() -> None: email=email, name="Administrator", password_hash=_hash_password(password), + is_superuser=True, ) db.add(user) db.commit() diff --git a/backend/src/app/services/authz.py b/backend/src/app/services/authz.py new file mode 100644 index 0000000..b0d1f58 --- /dev/null +++ b/backend/src/app/services/authz.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +import os + +from fastapi import HTTPException, status +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models.dataset import Dataset +from app.models.project import Project +from app.models.user import User +from app.models.workspace import Membership, Role, Workspace + +# Shared workspace that every authenticated user can access (legacy projects +# and projects created without an explicit workspace land here). +DEFAULT_WORKSPACE_ID = "00000000-0000-0000-0000-000000000000" + +ROLE_ORDER: dict[Role, int] = { + Role.VIEWER: 0, + Role.ANNOTATOR: 1, + Role.DEVELOPER: 2, + Role.ADMIN: 3, + Role.OWNER: 4, +} + + +def is_superuser(db: Session, user: User) -> bool: + if getattr(user, "is_superuser", False): + return True + superuser_email = os.getenv("FIRST_SUPERUSER_EMAIL") or os.getenv("SUPERUSER_EMAIL") + return bool(superuser_email and user.email == superuser_email) + + +def _role_level(role: object) -> int: + if isinstance(role, Role): + return ROLE_ORDER.get(role, -1) + try: + return ROLE_ORDER.get(Role(str(role)), -1) + except ValueError: + return -1 + + +def get_membership(db: Session, user: User, workspace_id: str) -> Membership | None: + return db.scalar( + select(Membership).where( + Membership.user_id == user.id, + Membership.workspace_id == workspace_id, + ) + ) + + +def require_workspace_access( + db: Session, + user: User, + workspace_id: str | None, + min_role: Role = Role.VIEWER, +) -> None: + """Raise 403 unless ``user`` may act in ``workspace_id`` at ``min_role`` level. + + The default workspace is open to every authenticated user; workspace + creators and superusers always have owner-level access. + """ + if workspace_id is None or workspace_id == DEFAULT_WORKSPACE_ID: + return + if is_superuser(db, user): + return + ws = db.get(Workspace, workspace_id) + if ws is not None and ws.created_by == user.id: + return + membership = get_membership(db, user, workspace_id) + if membership is None: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Not a member of this workspace", + ) + if _role_level(membership.role) < ROLE_ORDER.get(min_role, 0): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Requires at least {min_role.value} role", + ) + + +def require_project_access( + db: Session, + user: User, + project_id: str, + min_role: Role = Role.VIEWER, +) -> Project: + """Load a project, enforcing workspace access. 404 if missing, 403 if denied.""" + project = db.get(Project, project_id) + if project is None: + raise HTTPException(status_code=404, detail="Project not found") + require_workspace_access(db, user, project.workspace_id, min_role) + return project + + +def require_job_access(db: Session, user: User, job: object) -> None: + """Enforce access to a Job row via the project referenced in its payload. + + Jobs created before this check (or maintenance jobs) may not reference a + project; those are visible to any authenticated user. + """ + import json + + project_id: str | None = None + payload_json = getattr(job, "payload_json", None) + if payload_json: + try: + payload = json.loads(payload_json) + project_id = payload.get("projectId") or payload.get("project_id") + except (ValueError, TypeError): + project_id = None + if not project_id: + return + project = db.get(Project, project_id) + if project is None: + return + require_workspace_access(db, user, project.workspace_id) + + +def require_dataset_access( + db: Session, + user: User, + dataset_id: str, + min_role: Role = Role.VIEWER, +) -> Dataset: + """Load a dataset, enforcing access on its owning project's workspace.""" + dataset = db.get(Dataset, dataset_id) + if dataset is None: + raise HTTPException(status_code=404, detail="Dataset not found") + require_project_access(db, user, dataset.project_id, min_role) + return dataset diff --git a/backend/src/app/services/cluster_service.py b/backend/src/app/services/cluster_service.py index 80af92c..eac4fa2 100644 --- a/backend/src/app/services/cluster_service.py +++ b/backend/src/app/services/cluster_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hmac import json from collections.abc import Iterable from datetime import datetime, timedelta, timezone @@ -254,11 +255,19 @@ def get_cluster(db: Session, cluster_id: str) -> Cluster | None: return db.get(Cluster, cluster_id) -def record_heartbeat(db: Session, cluster_id: str, payload: ClusterHeartbeat) -> Cluster | None: +def record_heartbeat( + db: Session, + cluster_id: str, + payload: ClusterHeartbeat, + *, + auth_token: str | None = None, +) -> Cluster | None: cluster = db.get(Cluster, cluster_id) if not cluster: return None - if payload.register_token != cluster.register_token: + # Prefer the Authorization header token; fall back to the body field. + presented = auth_token or payload.register_token or "" + if not presented or not hmac.compare_digest(presented, cluster.register_token or ""): raise ClusterError("invalid register_token") cluster.cpu_usage_pct = float(payload.cpu_usage_pct) diff --git a/backend/src/app/services/dataset_service.py b/backend/src/app/services/dataset_service.py index cf681bf..92bdd6f 100644 --- a/backend/src/app/services/dataset_service.py +++ b/backend/src/app/services/dataset_service.py @@ -35,7 +35,13 @@ def create_dataset( def snapshot_version(db: Session, dataset_id: str, notes: str | None = None) -> DatasetVersion: - """Create a new locked snapshot version for a dataset.""" + """Create a new locked snapshot version for a dataset. + + The snapshot freezes the dataset's current working set, so its + ``asset_count`` reflects the assets in the open (unlocked) version being + captured — not every asset that has ever belonged to the dataset across + all prior locked versions. + """ # Compute next version number versions = list( db.scalars( @@ -46,10 +52,17 @@ def snapshot_version(db: Session, dataset_id: str, notes: str | None = None) -> ) next_version = (versions[0].version + 1) if versions else 1 - # Get current asset count for this dataset - current_assets = ( - db.scalar(select(func.count(Asset.id)).where(Asset.dataset_id == dataset_id)) or 0 - ) + # Count assets in the working (unlocked) version being snapshotted. Fall + # back to the whole dataset only when no open version exists (legacy data). + open_version = next((v for v in versions if not v.locked), None) + if open_version is not None: + current_assets = ( + db.scalar(select(func.count(Asset.id)).where(Asset.version_id == open_version.id)) or 0 + ) + else: + current_assets = ( + db.scalar(select(func.count(Asset.id)).where(Asset.dataset_id == dataset_id)) or 0 + ) ver = DatasetVersion( dataset_id=dataset_id, diff --git a/backend/src/app/services/datumaro_service.py b/backend/src/app/services/datumaro_service.py index 3f7a0ba..f9d68b2 100644 --- a/backend/src/app/services/datumaro_service.py +++ b/backend/src/app/services/datumaro_service.py @@ -262,7 +262,7 @@ def _import_yolo( mime_type=_guess_mime(img_path.name), width=w, height=h, - label_status="labeled" if label_path else "unlabelled", + label_status="labeled" if label_path else "unlabeled", meta_data=json.dumps({"filename": img_path.name}), ) db.add(asset) @@ -337,7 +337,7 @@ def _import_via_datumaro( mime_type=_guess_mime(uri), width=w, height=h, - label_status="labeled" if item.annotations else "unlabelled", + label_status="labeled" if item.annotations else "unlabeled", meta_data=json.dumps({"filename": str(path) if path else item.id}), ) db.add(asset) diff --git a/backend/src/app/services/storage.py b/backend/src/app/services/storage.py index df91424..1c600d8 100644 --- a/backend/src/app/services/storage.py +++ b/backend/src/app/services/storage.py @@ -41,9 +41,13 @@ def get_bytes(client: Minio, object_key: str, bucket: str | None = None) -> byte def get_minio_client() -> Minio: + from app.settings import require_secure_setting + endpoint = os.getenv("MINIO_ENDPOINT", "localhost:9000") access_key = os.getenv("MINIO_ACCESS_KEY", "minioadmin") - secret_key = os.getenv("MINIO_SECRET_KEY", "minioadmin") + secret_key = require_secure_setting( + "MINIO_SECRET_KEY", os.getenv("MINIO_SECRET_KEY", "minioadmin"), ("minioadmin",) + ) secure = os.getenv("MINIO_SECURE", "false").lower() == "true" return Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure) diff --git a/backend/src/app/services/training_service.py b/backend/src/app/services/training_service.py index 47e1972..a2a1d0f 100644 --- a/backend/src/app/services/training_service.py +++ b/backend/src/app/services/training_service.py @@ -16,6 +16,14 @@ class TaskTypeMismatch(Exception): """Raised when a training run's task disagrees with its dataset's task_type.""" +class TrainingDispatchError(Exception): + """Raised when the training Celery task can't be dispatched. + + The job row and run are marked ``failed`` and any cluster reservation is + released before this is raised, so nothing is left permanently queued/busy. + """ + + def launch_training( db: Session, project_id: str, @@ -106,8 +114,10 @@ def launch_training( db.add(cluster) db.commit() - # Enqueue async job; if broker not available, we still return a job id for contract. - # When a cluster is selected, route to its dedicated queue so only its agent picks it up. + # Enqueue async job. When a cluster is selected, route to its dedicated + # queue so only its agent picks it up. If dispatch fails (broker down), + # fail the job/run and release the cluster — otherwise the job sits + # "queued" forever and the cluster stays "busy" (see onnx_service). queue = f"cluster.{cluster_id}" if cluster_id else None try: send_kwargs: dict[str, Any] = {"args": [payload]} @@ -115,8 +125,26 @@ def launch_training( send_kwargs["queue"] = queue celery_app.send_task("app.jobs.tasks.training.train_task", **send_kwargs) job_id = job_row.id - except Exception: - job_id = job_row.id + except Exception as exc: + from app.services.jobs_service import update_job_status + + if cluster_id: + try: + cluster_service.release_cluster(db, cluster_id) + except Exception: + pass + try: + update_job_status(db, job_row.id, status="failed", progress=0.0) + except Exception: + pass + try: + run.status = "failed" + run.metrics_json = json.dumps({"error": f"task dispatch failed: {exc}"}) + db.add(run) + db.commit() + except Exception: + db.rollback() + raise TrainingDispatchError(f"failed to dispatch training task: {exc}") from exc return { "id": job_id, diff --git a/backend/src/app/settings.py b/backend/src/app/settings.py new file mode 100644 index 0000000..5702156 --- /dev/null +++ b/backend/src/app/settings.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +import os + + +def is_production() -> bool: + """True when the deployment is explicitly marked production via APP_ENV.""" + return os.getenv("APP_ENV", "development").strip().lower() in ("production", "prod") + + +def require_secure_setting(name: str, value: str, insecure_values: tuple[str, ...]) -> str: + """Fail fast on known-insecure default secrets in production. + + In development the insecure default is allowed so `docker-compose up` + keeps working out of the box. + """ + if is_production() and (not value or value in insecure_values): + raise RuntimeError( + f"{name} is unset or uses an insecure default. " + f"Set a strong value before running with APP_ENV=production." + ) + return value diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index cb7155e..f94dd69 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -3,6 +3,7 @@ import sys from collections.abc import Generator +import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker @@ -63,3 +64,61 @@ def override_get_db() -> Generator: app.dependency_overrides[get_db] = override_get_db client = TestClient(app) + + +# Many tests attach datasets/runs to a project id of "p" and exercise endpoints +# as a fake user. Object-level authz resolves a resource -> project -> workspace, +# so that project must exist in an accessible workspace. The default workspace +# (all-zero UUID) is open to every authenticated user, so seed "p" there. +DEFAULT_WORKSPACE_ID = "00000000-0000-0000-0000-000000000000" +try: + from app.models.project import Project as _Project + + with TestingSessionLocal() as _db: + if not _db.get(_Project, "p"): + _db.add( + _Project( + id="p", + name="Test Project", + slug="test-project-p", + workspace_id=DEFAULT_WORKSPACE_ID, + ) + ) + _db.commit() +except Exception: + pass + + +@pytest.fixture(autouse=True) +def _hermetic_celery(monkeypatch): + """Unit tests must not reach a real broker. + + Patch ``celery_app.send_task`` to a no-op so job-launch services create + their Job row and return "queued" without enqueueing (the dispatch-failure + path is covered explicitly where it matters). + """ + from unittest.mock import MagicMock + + from app.jobs import celery_app as _celery_mod + + monkeypatch.setattr( + _celery_mod.celery_app, "send_task", MagicMock(return_value=None), raising=False + ) + yield + + +def ensure_project(project_id: str, *, workspace_id: str = DEFAULT_WORKSPACE_ID) -> None: + """Create a project (in the default, open workspace) if it doesn't exist.""" + from app.models.project import Project as _Project + + with TestingSessionLocal() as db: + if not db.get(_Project, project_id): + db.add( + _Project( + id=project_id, + name=f"Project {project_id[:8]}", + slug=f"proj-{project_id[:12]}", + workspace_id=workspace_id, + ) + ) + db.commit() diff --git a/backend/tests/unit/test_api_experiments.py b/backend/tests/unit/test_api_experiments.py index b120d3f..6eb8918 100644 --- a/backend/tests/unit/test_api_experiments.py +++ b/backend/tests/unit/test_api_experiments.py @@ -14,7 +14,7 @@ from app.main import app from app.models.experiment import ExperimentRun from app.models.user import User -from tests.conftest import TestingSessionLocal, client +from tests.conftest import TestingSessionLocal, client, ensure_project def _fake_user() -> User: @@ -25,6 +25,7 @@ def _fake_user() -> User: def _mk_run(project_id: str, *, metrics_json: str | None = None) -> str: + ensure_project(project_id) db = TestingSessionLocal() try: run = ExperimentRun( @@ -44,6 +45,7 @@ def _mk_run(project_id: str, *, metrics_json: str | None = None) -> str: def test_create_run(): project_id = uuid.uuid4().hex + ensure_project(project_id) r = client.post( "/api/experiments/runs", json={"project_id": project_id, "name": "My Run", "params": {"lr": 0.01}}, diff --git a/backend/tests/unit/test_api_ops.py b/backend/tests/unit/test_api_ops.py index 43a5009..55cda31 100644 --- a/backend/tests/unit/test_api_ops.py +++ b/backend/tests/unit/test_api_ops.py @@ -86,7 +86,17 @@ def test_extract_frames_rejects_non_video_asset(): def test_extract_frames_rejects_mismatched_dataset(): _, asset_id = _seed_asset(mime="video/mp4", uri="x/y.mp4") - r = client.post(f"/api/datasets/wrong-dataset/assets/{asset_id}/extract-frames") + # A real but different dataset (so authz passes and the endpoint's own + # "asset does not belong to dataset" guard is what rejects the request). + other_db = TestingSessionLocal() + try: + other = Dataset(id=str(uuid.uuid4()), project_id="p", name="other") + other_db.add(other) + other_db.commit() + other_id = other.id + finally: + other_db.close() + r = client.post(f"/api/datasets/{other_id}/assets/{asset_id}/extract-frames") assert r.status_code == 400 diff --git a/backend/tests/unit/test_services_training.py b/backend/tests/unit/test_services_training.py index 99c2bb7..5633f24 100644 --- a/backend/tests/unit/test_services_training.py +++ b/backend/tests/unit/test_services_training.py @@ -6,7 +6,12 @@ from app.db.base import Base from app.models.experiment import ExperimentRun -from app.services.training_service import TaskTypeMismatch, launch_training +from app.models.job import Job +from app.services.training_service import ( + TaskTypeMismatch, + TrainingDispatchError, + launch_training, +) def _session(): @@ -66,3 +71,25 @@ def test_launch_training_rejects_unsupported_task_for_framework(): ) finally: db.close() + + +def test_launch_training_fails_job_when_dispatch_fails(monkeypatch): + """If the broker can't be reached, the job is failed (not left queued).""" + from app.jobs import celery_app as _celery_mod + + def _boom(*args, **kwargs): + raise RuntimeError("broker down") + + monkeypatch.setattr(_celery_mod.celery_app, "send_task", _boom, raising=False) + + db = _session() + try: + with pytest.raises(TrainingDispatchError): + launch_training(db, project_id="p1", dataset_version_id="dv1", task="detect") + # No phantom queued jobs left behind. + jobs = db.query(Job).all() + assert jobs and all(j.status == "failed" for j in jobs) + runs = db.query(ExperimentRun).all() + assert runs and all(r.status == "failed" for r in runs) + finally: + db.close() diff --git a/deploy/grafana/provisioning/datasources/datasource.yml b/deploy/grafana/provisioning/datasources/datasource.yml index fcdbae5..bef8018 100644 --- a/deploy/grafana/provisioning/datasources/datasource.yml +++ b/deploy/grafana/provisioning/datasources/datasource.yml @@ -15,8 +15,12 @@ datasources: database: visionforge user: visionforge secureJsonData: + # Interpolated by Grafana provisioning from the container environment; + # POSTGRES_PASSWORD is passed through in docker-compose.yml. password: "${POSTGRES_PASSWORD}" jsonData: + # `disable` is acceptable only for the dev compose stack (dev postgres + # has no TLS). In production this MUST be `require` or `verify-full`. sslmode: disable maxOpenConns: 5 maxIdleConns: 2 diff --git a/docker-compose.yml b/docker-compose.yml index 604f2fa..07e83fd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -18,6 +18,11 @@ services: timeout: 5s retries: 10 start_period: 10s + deploy: + resources: + limits: + cpus: "2" + memory: 2G minio: image: minio/minio:latest @@ -36,6 +41,11 @@ services: timeout: 5s retries: 5 start_period: 10s + deploy: + resources: + limits: + cpus: "1" + memory: 1G redis: image: redis:7 @@ -46,6 +56,11 @@ services: interval: 5s timeout: 3s retries: 5 + deploy: + resources: + limits: + cpus: "1" + memory: 512M backend: build: ./backend @@ -74,6 +89,11 @@ services: timeout: 5s retries: 5 start_period: 40s + deploy: + resources: + limits: + cpus: "2" + memory: 2G worker: build: ./backend @@ -86,36 +106,66 @@ services: condition: service_healthy redis: condition: service_healthy - command: ["celery", "-A", "app.jobs.celery_app:celery_app", "worker", "-l", "info"] + # -B embeds celery beat in the worker so periodic maintenance tasks + # (e.g. stale-heartbeat degrade) run without a separate beat container. + command: ["celery", "-A", "app.jobs.celery_app:celery_app", "worker", "-B", "-l", "info"] + deploy: + resources: + limits: + cpus: "4" + memory: 8G + # NOTE: This is the Vite DEV server — for local development only. + # For production, build the static bundle and serve it via nginx using + # frontend/Dockerfile.prod (build with `-f frontend/Dockerfile.prod`), + # which uses frontend/nginx.conf. frontend: build: ./frontend ports: - "5173:5173" - # NOTE: This runs the Vite dev server and is intended for local development. - # For production, build the static bundle and serve it via nginx using - # frontend/Dockerfile.prod (build with `-f frontend/Dockerfile.prod`). environment: VITE_API_URL: http://localhost:8000 command: ["npm", "run", "dev", "--", "--host"] + deploy: + resources: + limits: + cpus: "1" + memory: 1G prometheus: image: prom/prometheus:latest volumes: - ./deploy/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml + - prometheus:/prometheus ports: - "9090:9090" + deploy: + resources: + limits: + cpus: "1" + memory: 512M grafana: image: grafana/grafana:latest environment: GF_SECURITY_ADMIN_USER: ${GRAFANA_ADMIN_USER} GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_ADMIN_PASSWORD} + # Interpolated by Grafana provisioning ($POSTGRES_PASSWORD in + # deploy/grafana/provisioning/datasources/datasource.yml). + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD} volumes: - ./deploy/grafana/provisioning:/etc/grafana/provisioning + - grafana:/var/lib/grafana ports: - "3000:3000" + deploy: + resources: + limits: + cpus: "1" + memory: 512M volumes: pgdata: minio: + prometheus: + grafana: diff --git a/frontend/nginx.conf b/frontend/nginx.conf index 45a11ed..6f8fb1a 100644 --- a/frontend/nginx.conf +++ b/frontend/nginx.conf @@ -24,8 +24,14 @@ server { return 200 "ok\n"; } - # Proxy API + auth + ops traffic to the backend. - location ~ ^/(api|auth|health|metrics|docs|redoc|openapi.json) { + # Proxy API + auth + health traffic to the backend. + # + # Intentionally NOT proxied: /metrics, /docs, /redoc, /openapi.json. + # Exposing them here would make Prometheus metrics and the API schema + # available unauthenticated on the public edge. Ops should scrape + # /metrics directly from the backend on the internal network (e.g. the + # compose/cluster network where Prometheus already reaches backend:8000). + location ~ ^/(api|auth|health) { proxy_pass http://visionforge_backend; proxy_http_version 1.1; proxy_set_header Host $host; diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index 1a063f7..2f493d8 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -1,47 +1,48 @@ -import React, { useState, useEffect } from "react"; -import { BrowserRouter, Routes, Route, Link } from "react-router-dom"; -import { AuthProvider } from "@/services/auth-store"; -import ProtectedRoute from "@/components/layout/ProtectedRoute"; -import LoginPage from "@/pages/auth/Login"; -import AppShell from "@/components/layout/AppShell"; -import AnnotatorPage from "./pages/annotate/Annotator"; -import AdminUsersPage from "./pages/admin/users"; -import ProjectsIndex from "./pages/projects/index"; -import ProjectsCreate from "./pages/projects/create"; -import ProjectDashboard from "./pages/projects/[projectId]/index"; -import DatasetUpload from "./pages/datasets/upload"; -import DatasetVersion from "./pages/datasets/version"; -import DatasetsIndex from "./pages/datasets/index"; -import DatasetNew from "./pages/datasets/new"; -import DatasetDetail from "./pages/datasets/[datasetId]/index"; -import DatasetMetrics from "./pages/datasets/[datasetId]/metrics"; -import DatasetAnnotateGateway from "./pages/datasets/[datasetId]/annotate"; -import DatasetReviewQueue from "./pages/datasets/[datasetId]/review"; -import ExperimentsIndex from "./pages/experiments/index"; -import ExperimentsNew from "./pages/experiments/new"; -import ExperimentDetail from "./pages/experiments/[runId]"; -import ArtifactsIndex from "./pages/artifacts/index"; -import ArtifactsExport from "./pages/artifacts/export"; -import ArtifactsLineage from "./pages/artifacts/lineage"; -import ActiveLearningIndex from "./pages/active-learning/index"; -import ActiveLearningNew from "./pages/active-learning/new"; -import ALRunDetail from "./pages/active-learning/[alRunId]"; -import ClustersIndex from "./pages/clusters/index"; -import ClustersNew from "./pages/clusters/new"; -import EvaluationsIndex from "./pages/evaluations/index"; -import EvaluationsNew from "./pages/evaluations/new"; -import EvaluationDetail from "./pages/evaluations/[evalId]"; -import ArtifactsPredict from "./pages/artifacts/predict"; -import DatasetImport from "./pages/datasets/import"; -import { apiGet } from "@/services/api"; -import Spinner from "@/components/ui/Spinner"; +import React, { useState, useEffect } from 'react'; +import { BrowserRouter, Routes, Route, Link } from 'react-router-dom'; +import { AuthProvider } from '@/services/auth-store'; +import ErrorBoundary from '@/components/common/ErrorBoundary'; +import ProtectedRoute from '@/components/layout/ProtectedRoute'; +import LoginPage from '@/pages/auth/Login'; +import AppShell from '@/components/layout/AppShell'; +import AnnotatorPage from './pages/annotate/Annotator'; +import AdminUsersPage from './pages/admin/users'; +import ProjectsIndex from './pages/projects/index'; +import ProjectsCreate from './pages/projects/create'; +import ProjectDashboard from './pages/projects/[projectId]/index'; +import DatasetUpload from './pages/datasets/upload'; +import DatasetVersion from './pages/datasets/version'; +import DatasetsIndex from './pages/datasets/index'; +import DatasetNew from './pages/datasets/new'; +import DatasetDetail from './pages/datasets/[datasetId]/index'; +import DatasetMetrics from './pages/datasets/[datasetId]/metrics'; +import DatasetAnnotateGateway from './pages/datasets/[datasetId]/annotate'; +import DatasetReviewQueue from './pages/datasets/[datasetId]/review'; +import ExperimentsIndex from './pages/experiments/index'; +import ExperimentsNew from './pages/experiments/new'; +import ExperimentDetail from './pages/experiments/[runId]'; +import ArtifactsIndex from './pages/artifacts/index'; +import ArtifactsExport from './pages/artifacts/export'; +import ArtifactsLineage from './pages/artifacts/lineage'; +import ActiveLearningIndex from './pages/active-learning/index'; +import ActiveLearningNew from './pages/active-learning/new'; +import ALRunDetail from './pages/active-learning/[alRunId]'; +import ClustersIndex from './pages/clusters/index'; +import ClustersNew from './pages/clusters/new'; +import EvaluationsIndex from './pages/evaluations/index'; +import EvaluationsNew from './pages/evaluations/new'; +import EvaluationDetail from './pages/evaluations/[evalId]'; +import ArtifactsPredict from './pages/artifacts/predict'; +import DatasetImport from './pages/datasets/import'; +import { apiGet } from '@/services/api'; +import Spinner from '@/components/ui/Spinner'; function StatReadout({ label, value, loading, href }) { return (
{label}
- {loading ? : value ?? 0} + {loading ? : (value ?? 0)}
{href && ( +
{label} {dot} {labelOut} @@ -84,10 +82,10 @@ function SystemStatusPanel({ system }) {
System Status
- + - - + +
); @@ -101,18 +99,14 @@ function HomeDashboard() { useEffect(() => { let cancelled = false; Promise.all([ - apiGet("/api/projects?page=1&page_size=1"), - apiGet("/api/datasets?page=1&page_size=1"), - apiGet("/api/artifacts/models?page=1&page_size=1"), + apiGet('/api/projects?page=1&page_size=1'), + apiGet('/api/datasets?page=1&page_size=1'), + apiGet('/api/artifacts/models?page=1&page_size=1'), ]) .then(([projects, datasets, models]) => { if (cancelled) return; const total = (resp) => - typeof resp?.total === "number" - ? resp.total - : Array.isArray(resp) - ? resp.length - : 0; + typeof resp?.total === 'number' ? resp.total : Array.isArray(resp) ? resp.length : 0; setStats({ projects: total(projects), datasets: total(datasets), @@ -130,9 +124,9 @@ function HomeDashboard() { let cancelled = false; let timer; const tick = () => { - apiGet("/api/system/status") + apiGet('/api/system/status') .then((data) => !cancelled && setSystem(data)) - .catch(() => !cancelled && setSystem({ status: "down", components: {} })); + .catch(() => !cancelled && setSystem({ status: 'down', components: {} })); }; tick(); timer = setInterval(tick, 15000); @@ -149,7 +143,9 @@ function HomeDashboard() {

- Manage datasets, annotate frames,
and iterate on models. + Manage datasets, annotate frames, +
+ and iterate on models.

End-to-end CV workflow: ingest → annotate → train → export → deploy. @@ -176,9 +172,24 @@ function HomeDashboard() {

Platform Overview
- - - + + +
@@ -186,11 +197,15 @@ function HomeDashboard() {
Quick Access
{[ - { to: "/projects", label: "PROJECTS", desc: "Manage project workspaces" }, - { to: "/datasets", label: "DATASETS", desc: "Browse and upload datasets" }, - { to: "/experiments", label: "EXPERIMENTS", desc: "Training runs and metrics" }, - { to: "/artifacts", label: "ARTIFACTS", desc: "Model exports and lineage" }, - { to: "/active-learning", label: "ACTIVE LEARN", desc: "Select informative samples to label" }, + { to: '/projects', label: 'PROJECTS', desc: 'Manage project workspaces' }, + { to: '/datasets', label: 'DATASETS', desc: 'Browse and upload datasets' }, + { to: '/experiments', label: 'EXPERIMENTS', desc: 'Training runs and metrics' }, + { to: '/artifacts', label: 'ARTIFACTS', desc: 'Model exports and lineage' }, + { + to: '/active-learning', + label: 'ACTIVE LEARN', + desc: 'Select informative samples to label', + }, ].map(({ to, label, desc }) => ( - - - - } /> - - - - } - /> - {/* Projects */} - } /> - } /> - } /> - {/* Datasets */} - } /> - } /> - } /> - } /> - } /> - } /> - } /> - } /> - {/* Experiments */} - } /> - } /> - } /> - {/* Artifacts */} - } /> - } /> - } /> - {/* Active Learning */} - } /> - } /> - } /> - {/* Clusters */} - } /> - } /> - {/* Evaluations */} - } /> - } /> - } /> - {/* Predict + import */} - } /> - } /> - {/* Annotate */} - } /> - {/* Admin */} - } /> - - - - + + + + + + } /> + + + + } + /> + {/* Projects */} + + + + } + /> + + + + } + /> + + + + } + /> + {/* Datasets */} + + + + } + /> + + + + } + /> + + + + } + /> + + + + } + /> + + + + } + /> + + + + } + /> + + + + } + /> + + + + } + /> + {/* Experiments */} + + + + } + /> + + + + } + /> + + + + } + /> + {/* Artifacts */} + + + + } + /> + + + + } + /> + + + + } + /> + {/* Active Learning */} + + + + } + /> + + + + } + /> + + + + } + /> + {/* Clusters */} + + + + } + /> + + + + } + /> + {/* Evaluations */} + + + + } + /> + + + + } + /> + + + + } + /> + {/* Predict + import */} + + + + } + /> + + + + } + /> + {/* Annotate */} + + + + } + /> + {/* Admin */} + + + + } + /> + + + + + ); } diff --git a/frontend/src/components/common/ErrorBoundary.tsx b/frontend/src/components/common/ErrorBoundary.tsx new file mode 100644 index 0000000..2c52052 --- /dev/null +++ b/frontend/src/components/common/ErrorBoundary.tsx @@ -0,0 +1,57 @@ +import React from 'react'; + +interface ErrorBoundaryProps { + children: React.ReactNode; +} + +interface ErrorBoundaryState { + error: Error | null; +} + +/** + * Top-level error boundary so unhandled render errors show a recoverable + * fallback instead of a white screen. + * + * NOTE: this is intentionally a class component despite the project's + * functional-components convention — React error boundaries require + * getDerivedStateFromError / componentDidCatch, which have no hook equivalent. + */ +export default class ErrorBoundary extends React.Component { + state: ErrorBoundaryState = { error: null }; + + static getDerivedStateFromError(error: Error): ErrorBoundaryState { + return { error }; + } + + componentDidCatch(error: Error, info: React.ErrorInfo) { + console.error('Unhandled render error:', error, info.componentStack); + } + + render() { + if (this.state.error) { + return ( +
+
+

+ Something went wrong +

+

+ {this.state.error.message || 'An unexpected error occurred.'} +

+ +
+
+ ); + } + return this.props.children; + } +} diff --git a/frontend/src/components/datasets/ArchiveImporter.tsx b/frontend/src/components/datasets/ArchiveImporter.tsx index 6f1097b..92a7125 100644 --- a/frontend/src/components/datasets/ArchiveImporter.tsx +++ b/frontend/src/components/datasets/ArchiveImporter.tsx @@ -47,7 +47,7 @@ export default function ArchiveImporter({ useEffect(() => { apiGet<{ formats: string[] }>('/api/datasets/formats') .then((r) => r.formats?.length && setFormats(r.formats)) - .catch(() => {}); + .catch((err) => console.warn('Failed to load import formats, using defaults', err)); }, []); async function onSubmit(e: React.FormEvent) { diff --git a/frontend/src/pages/annotate/Annotator.tsx b/frontend/src/pages/annotate/Annotator.tsx index 2f6f43c..7298cea 100644 --- a/frontend/src/pages/annotate/Annotator.tsx +++ b/frontend/src/pages/annotate/Annotator.tsx @@ -679,17 +679,32 @@ export default function AnnotatorPage() { // ------------------------------------------------------------------------- function deleteAnnotation(idx: number) { + const ann = annotationsRef.current[idx]; + if (!ann) return; pushHistory(); setAnnotations((prev) => { - const ann = prev[idx]; - if (!ann) return prev; - if (ann.id) apiDelete(`/api/annotations/${ann.id}`).catch(() => {}); const updated = prev.filter((_, i) => i !== idx); annotationsRef.current = updated; return updated; }); setSelectedAnnotationIdx(null); setDirty(true); + if (ann.id) { + apiDelete(`/api/annotations/${ann.id}`).catch((err) => { + console.warn('Failed to delete annotation on server, restoring it locally', err); + // Server delete failed — put the annotation back so local state stays in + // sync with the server, and surface a brief error in the status bar. + setAnnotations((prev) => { + const restored = [...prev]; + restored.splice(Math.min(idx, restored.length), 0, ann); + annotationsRef.current = restored; + return restored; + }); + setStatus( + `Delete failed: ${err instanceof Error ? err.message : 'error'} — annotation restored`, + ); + }); + } } function setAnnotationClass(idx: number, className: string) { diff --git a/frontend/src/pages/artifacts/predict.tsx b/frontend/src/pages/artifacts/predict.tsx index 7911f93..e95b30c 100644 --- a/frontend/src/pages/artifacts/predict.tsx +++ b/frontend/src/pages/artifacts/predict.tsx @@ -40,7 +40,12 @@ export default function ArtifactsPredict() { useEffect(() => { if (!modelId) return; - apiGet(`/api/artifacts/models/${modelId}`).then(setModel).catch(() => {}); + apiGet(`/api/artifacts/models/${modelId}`) + .then(setModel) + .catch((err) => { + console.warn('Failed to load model artifact', err); + setError(err instanceof Error ? err.message : 'Failed to load model'); + }); }, [modelId]); useEffect(() => { @@ -209,10 +214,8 @@ export default function ArtifactsPredict() { {result.result.top_class_index != null && (
Top index:{' '} - - {result.result.top_class_index} - {' '} - ({((result.result.top_score || 0) * 100).toFixed(1)}%) + {result.result.top_class_index} ( + {((result.result.top_score || 0) * 100).toFixed(1)}%)
)}
diff --git a/frontend/src/pages/datasets/[datasetId]/index.tsx b/frontend/src/pages/datasets/[datasetId]/index.tsx index 7feac76..bd5f11a 100644 --- a/frontend/src/pages/datasets/[datasetId]/index.tsx +++ b/frontend/src/pages/datasets/[datasetId]/index.tsx @@ -347,7 +347,7 @@ function FrameExtractionCard({ datasetId, versionId }: { datasetId: string; vers a.uri.toLowerCase().endsWith('.webm'); setVideos(items.filter(isVideo)); }) - .catch(() => {}); + .catch((err) => console.warn('Failed to load video assets for frame extraction', err)); return () => { cancelled = true; }; diff --git a/frontend/src/pages/datasets/[datasetId]/review.tsx b/frontend/src/pages/datasets/[datasetId]/review.tsx index 0bacb78..7f2dfa5 100644 --- a/frontend/src/pages/datasets/[datasetId]/review.tsx +++ b/frontend/src/pages/datasets/[datasetId]/review.tsx @@ -97,7 +97,7 @@ export default function DatasetReviewQueue() { useEffect(() => { apiGet<{ items: ArtifactLite[] }>('/api/artifacts/models?page=1&page_size=200') .then((r) => setArtifacts(r.items || [])) - .catch(() => {}); + .catch((err) => console.warn('Failed to load model artifacts for mining', err)); }, []); function buildQuery(extra: Record = {}) { @@ -195,12 +195,11 @@ export default function DatasetReviewQueue() {
-
@@ -337,7 +334,9 @@ export default function DatasetReviewQueue() { {error && } {loading ? ( -
+
+ +
) : annotations.length === 0 ? ( {ann.class_name} )} - {ann.flagged && ( - FLAGGED - )} - {ann.review_status === 'approved' && ( - APPROVED - )} - {ann.review_status === 'rejected' && ( - REJECTED - )} + {ann.flagged && FLAGGED} + {ann.review_status === 'approved' && APPROVED} + {ann.review_status === 'rejected' && REJECTED}
asset {asset.id.slice(0, 12)}… · v{ann.version} @@ -403,12 +396,7 @@ export default function DatasetReviewQueue() {
))}
- + )}
diff --git a/frontend/src/pages/datasets/import.tsx b/frontend/src/pages/datasets/import.tsx index eccdf65..e67ba46 100644 --- a/frontend/src/pages/datasets/import.tsx +++ b/frontend/src/pages/datasets/import.tsx @@ -43,15 +43,17 @@ export default function DatasetImport() { const [file, setFile] = useState(null); const [result, setResult] = useState(null); const [error, setError] = useState(null); + const [loadError, setLoadError] = useState(null); const [loading, setLoading] = useState(false); useEffect(() => { apiGet<{ items: DatasetSummary[] }>('/api/datasets?page=1&page_size=200') .then((d) => setDatasets(d.items || [])) - .catch(() => {}); + .catch((err) => setLoadError(err instanceof Error ? err.message : 'Failed to load datasets')); + // Optional: the hardcoded default format list keeps the form usable. apiGet<{ formats: string[] }>('/api/datasets/formats') .then((r) => setFormats(r.formats)) - .catch(() => {}); + .catch((err) => console.warn('Failed to load import formats, using defaults', err)); }, []); async function onSubmit(e: React.FormEvent) { @@ -98,18 +100,15 @@ export default function DatasetImport() { Upload a zip in COCO, YOLO, Pascal VOC, CVAT, LabelMe, or Datumaro format.

- + ← DATASETS {result ? ( - Imported {result.asset_count} assets and {result.annotation_count} annotations - ({result.classes.length} classes) into version{' '} + Imported {result.asset_count} assets and {result.annotation_count} annotations ( + {result.classes.length} classes) into version{' '} {result.version_id.slice(0, 8)}. {result.warnings.length > 0 && (
@@ -117,9 +116,7 @@ export default function DatasetImport() {
)}
- + @@ -127,6 +124,7 @@ export default function DatasetImport() { ) : (
+ {loadError && Failed to load datasets: {loadError}}