diff --git a/api/ai_readiness.py b/api/ai_readiness.py index 9f42969..b4322fb 100644 --- a/api/ai_readiness.py +++ b/api/ai_readiness.py @@ -308,12 +308,16 @@ def generate_data_card( - Class balance (outlier/calibration ratios) - Temporal coverage (runs per month) - Known limitations / bias warnings + + Every statistic covers only the runs the caller can see (row-level + security on etcher_runs: their projects plus open projects; admins see + all). Without a user id an admin call is a system job and reads all runs. """ - from data_loader_pg import get_pg_worker_connection + from data_loader_pg import _scoped_connection from psycopg2.extras import RealDictCursor try: - conn = get_pg_worker_connection() + conn = _scoped_connection(nanohub_user_id, is_admin) with conn.cursor(cursor_factory=RealDictCursor) as cur: # ── Overview statistics ────────────────────────────────────── diff --git a/api/data_loader_pg.py b/api/data_loader_pg.py index a31121e..e017e30 100644 --- a/api/data_loader_pg.py +++ b/api/data_loader_pg.py @@ -5504,21 +5504,26 @@ def json_number(json_column: str, key: str) -> str: return None -def get_etcher_run_stats_pg() -> Dict[str, Any]: +def get_etcher_run_stats_pg(viewer: Any = None) -> Dict[str, Any]: """ Total run count and most recent run timestamp for the etcher. - Always uses the worker connection so the equipment detail card shows - the *true* fleet-wide count, independent of the caller's RLS scope. - The etcher_runs table has no equipment_id column today — every row in it - belongs to the etcher — so a plain COUNT(*) is the correct answer. + For a non-admin ``viewer`` (a PlatformUser) the count covers only the + runs they can see (row-level security: their projects plus open + projects). Admins and internal callers get the fleet-wide count from the + worker connection. The etcher_runs table has no equipment_id column + today — every row in it belongs to the etcher — so a plain COUNT(*) is + the correct answer. Returns ``{"total_runs": int, "last_data_at": iso_string | None}``. Returns ``{"total_runs": None, "last_data_at": None}`` on error so the UI can render "—" without crashing. """ try: - conn = get_pg_worker_connection() + if viewer is not None and getattr(viewer, "role", "") != "admin": + conn = get_pg_connection(viewer) + else: + conn = get_pg_worker_connection() with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( "SELECT count(*) AS total_runs, max(run_date) AS last_data_at FROM etcher_runs" diff --git a/api/metadata_pg.py b/api/metadata_pg.py index 4b8b97e..72c48b7 100644 --- a/api/metadata_pg.py +++ b/api/metadata_pg.py @@ -1194,6 +1194,48 @@ def equipment_visible_to_user( return True +def equipment_manageable_by(record: dict[str, Any], user: PlatformUser) -> bool: + """The owner, a trusted maintainer or an admin (same rule as editing).""" + if user.role == "admin" or (record.get("owner_id") or "") == user.id: + return True + maintainer_ids = set(record.get("_maintainer_ids") or []) + maintainer_ids.update( + maintainer.get("user_id") + for maintainer in record.get("maintainers") or [] + if isinstance(maintainer, dict) + ) + return user.id in maintainer_ids + + +_CONNECTION_FIELDS = ("connection_host", "connection_port", "connection_database") + + +def redact_equipment_for_viewer( + item: dict[str, Any], + user: PlatformUser, + *, + manageable: bool | None = None, +) -> dict[str, Any]: + """Hide the data-source connection (host, port, database) from users who + can see the equipment but not manage it. ``manageable`` overrides the check + when the caller already knows (serialized list items carry no maintainers). + """ + if manageable is None: + manageable = equipment_manageable_by(item, user) + if manageable: + return item + redacted = dict(item) + for field in _CONNECTION_FIELDS: + if field in redacted: + redacted[field] = "" + config = redacted.get("config_json") + if isinstance(config, dict) and "source_connection" in config: + redacted["config_json"] = { + key: value for key, value in config.items() if key != "source_connection" + } + return redacted + + def approved_equipment_available_to_user( record: dict[str, Any], user: PlatformUser, @@ -1256,20 +1298,25 @@ def _serialize_equipment( } -def _fetch_equipment_run_metrics(_conn=None) -> dict[str, dict[str, Any]]: - """Aggregate fleet-wide run counts and latest-data timestamps per equipment. +def _fetch_equipment_run_metrics( + viewer: PlatformUser | None = None, +) -> dict[str, dict[str, Any]]: + """Aggregate run counts and latest-data timestamps per equipment. Combines the canonical etcher_runs (attributed via the project's equipment) with the generic equipment_runs table (attributed via the run's own equipment_id) so manually uploaded data for any registered equipment is reflected in the inventory metrics. - Runs on the superuser connection so these non-sensitive aggregate counts are - complete (fleet-wide) regardless of the caller's project membership. The - RLS-scoped connection passed by callers would otherwise hide runs in - private/shared projects, even from members and admins, and undercount. + For a non-admin ``viewer`` the counts cover only the runs that user can + see (row-level security: their projects plus open projects), so private + projects' activity isn't revealed. Admins and internal callers (no + viewer) get the fleet-wide counts from the worker connection. """ - conn = get_pg_worker_connection() + if viewer is not None and viewer.role != "admin": + conn = get_pg_connection(viewer) + else: + conn = get_pg_worker_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( @@ -1359,8 +1406,8 @@ def list_equipment_pg(user: PlatformUser) -> list[dict[str, Any]]: for arow in access_rows: maintainer_map.setdefault(arow["equipment_id"], []).append(arow["user_id"]) - metrics = _fetch_equipment_run_metrics(conn) conn.commit() + metrics = _fetch_equipment_run_metrics(user) visible: list[dict[str, Any]] = [] for row in rows: @@ -1368,14 +1415,23 @@ def list_equipment_pg(user: PlatformUser) -> list[dict[str, Any]]: record["_maintainer_ids"] = maintainer_map.get(record["domain_id"], []) if equipment_visible_to_user(record, user): visible.append( - _serialize_equipment(record, metrics.get(record["domain_id"])) + redact_equipment_for_viewer( + _serialize_equipment(record, metrics.get(record["domain_id"])), + user, + manageable=equipment_manageable_by(record, user), + ) ) return visible finally: conn.close() -def get_equipment_pg(domain_id: str) -> dict[str, Any] | None: +def get_equipment_pg( + domain_id: str, + viewer: PlatformUser | None = None, +) -> dict[str, Any] | None: + """One equipment record. Pass ``viewer`` when the record is returned to a + user, so its run counts cover only runs that user can see.""" conn = get_pg_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: @@ -1457,8 +1513,8 @@ def get_equipment_pg(domain_id: str) -> dict[str, Any] | None: for r in maintainer_detail_rows ] - metrics = _fetch_equipment_run_metrics(conn) conn.commit() + metrics = _fetch_equipment_run_metrics(viewer) return _serialize_equipment(record, metrics.get(domain_id)) finally: conn.close() @@ -2456,9 +2512,12 @@ def _visible_experiment_type_filter( """ ``AND ...`` clause limiting ``experiment_types`` (referenced as ``alias``) to templates ``user`` may see or use: their own, shared or published ones, - their organization's, and templates attached to a project they can read. - Row-level security enforces the same rule in the database; this keeps it - when enforcement is off. + templates attached to a project they can read, and templates used by an + experiment they can read (theirs, in a project they can read, or an + execution request on equipment they own or maintain). A private template + is not visible to the rest of the owner's organization. Row-level security + (app_experiment_type_readable) enforces the same rule in the database; + this keeps it when enforcement is off. """ if user.role == "admin": return "", [] @@ -2467,7 +2526,6 @@ def _visible_experiment_type_filter( AND ( {alias}.owner_id = %s OR {alias}.visibility IN ('shared', 'published') - OR ({alias}.owner_org <> '' AND {alias}.owner_org = %s) OR EXISTS ( SELECT 1 FROM project_experiments vis_pe @@ -2483,9 +2541,39 @@ def _visible_experiment_type_filter( ) ) ) + OR EXISTS ( + SELECT 1 + FROM experiment_definitions vis_e + LEFT JOIN projects vis_ep ON vis_ep.id = vis_e.project_id + LEFT JOIN equipment_metadata vis_em ON vis_em.domain_id = vis_e.equipment_id + WHERE vis_e.type_id = {alias}.id + AND ( + vis_e.owner_id = %s + OR vis_ep.access_mode = 'open' + OR EXISTS ( + SELECT 1 + FROM project_members vis_epm + WHERE vis_epm.project_id = vis_e.project_id + AND vis_epm.nanohub_user_id = %s + ) + OR ( + vis_e.project_id IS NOT NULL + AND COALESCE(vis_e.execution_request_id, '') <> '' + AND ( + vis_em.owner_id = %s + OR EXISTS ( + SELECT 1 + FROM equipment_access vis_ea + WHERE vis_ea.equipment_id = vis_e.equipment_id + AND vis_ea.user_id = %s + ) + ) + ) + ) + ) ) """, - [user.id, user.organization, user.id], + [user.id, user.id, user.id, user.id, user.id, user.id], ) @@ -2581,7 +2669,7 @@ def list_experiment_types_pg( params.append(equipment_id) where_sql = f"WHERE {' AND '.join(conditions)}" if conditions else "WHERE true" # Same rule as everywhere else templates are used (and as RLS): own, - # shared/published, same organization, or attached to a readable project. + # shared/published, or attached to a readable project. visibility_sql, visibility_params = _visible_experiment_type_filter(user) where_sql += visibility_sql params.extend(visibility_params) @@ -7004,17 +7092,48 @@ def list_experiment_definitions_pg( conn.close() -def list_execution_queue_pg(user: PlatformUser) -> list[dict[str, Any]]: - """List project-run execution requests visible to a facility manager.""" - if user.role not in {"admin", "equipment_owner"}: - raise PermissionError("Facility execution queue access required") +_MANAGED_EQUIPMENT_SQL = """( + em.owner_id = %s + OR EXISTS ( + SELECT 1 FROM equipment_access ea + WHERE ea.equipment_id = em.domain_id AND ea.user_id = %s + ) +)""" + + +def user_manages_any_equipment_pg(user: PlatformUser) -> bool: + """Whether ``user`` owns or is a trusted maintainer of any equipment, and + so has an execution queue (admins always do).""" + if user.role == "admin": + return True + conn = get_pg_connection(user) + try: + with conn.cursor() as cur: + cur.execute( + """ + SELECT + EXISTS (SELECT 1 FROM equipment_metadata WHERE owner_id = %s) + OR EXISTS (SELECT 1 FROM equipment_access WHERE user_id = %s) + """, + (user.id, user.id), + ) + row = cur.fetchone() + conn.commit() + finally: + conn.close() + return bool(row and row[0]) + +def list_execution_queue_pg(user: PlatformUser) -> list[dict[str, Any]]: + """List project-run execution requests for the equipment ``user`` owns + or is a trusted maintainer of (admins: all equipment). The same + organization is not enough. Mirrors app_manages_equipment.""" conn = get_pg_connection(user) params: list[Any] = [] conditions = ["e.project_id IS NOT NULL", "e.execution_request_id <> ''"] if user.role != "admin": - conditions.append("(em.owner_id = %s OR em.owner_org = %s)") - params.extend([user.id, user.organization]) + conditions.append(_MANAGED_EQUIPMENT_SQL) + params.extend([user.id, user.id]) where_sql = " AND ".join(conditions) try: @@ -7144,8 +7263,6 @@ def update_execution_queue_status_pg( scheduled_for: str, user: PlatformUser, ) -> dict[str, Any]: - if user.role not in {"admin", "equipment_owner"}: - raise PermissionError("Facility execution queue access required") if status not in EXECUTION_STATUSES: raise ValueError("Invalid execution status") @@ -7156,25 +7273,20 @@ def update_execution_queue_status_pg( """ SELECT e.id::text, - em.owner_id AS equipment_owner_id, - em.owner_org AS equipment_owner_org + COALESCE({managed}, false) AS manages_equipment FROM experiment_definitions e LEFT JOIN equipment_metadata em ON em.domain_id = e.equipment_id WHERE e.id = %s AND e.project_id IS NOT NULL - """, - (experiment_id,), + """.format(managed=_MANAGED_EQUIPMENT_SQL), + (user.id, user.id, experiment_id), ) existing = cur.fetchone() if not existing: return {"status": "not_found", "id": experiment_id} - if user.role != "admin": - allowed = ( - existing.get("equipment_owner_id") == user.id - or existing.get("equipment_owner_org") == user.organization - ) - if not allowed: - raise PermissionError("Not authorized to update this execution request") + if user.role != "admin" and not existing.get("manages_equipment"): + # Only the equipment's owner or a trusted maintainer. + raise PermissionError("Not authorized to update this execution request") cur.execute( """ @@ -7579,7 +7691,18 @@ def create_upload_record_pg( conn.close() -def list_uploads_pg(user: PlatformUser) -> list[dict[str, Any]]: +def list_uploads_pg( + user: PlatformUser, + *, + include_shared_db3: bool = False, +) -> list[dict[str, Any]]: + """Uploads the caller may see: their own (admins see all). + + Uploads hold raw files that are not yet in any project, so other users, + including those in the same organization, never see them. The temporary + DB3 pilot is the one exception: its participants share one queue, so + ``include_shared_db3`` adds every .db3 upload for them. + """ conn = get_pg_connection() try: with conn.cursor(cursor_factory=RealDictCursor) as cur: @@ -7615,13 +7738,11 @@ def list_uploads_pg(user: PlatformUser) -> list[dict[str, Any]]: uploads: list[dict[str, Any]] = [] for row in rows: record = dict(row) - if user.role != "admin": - same_owner = record.get("owner_id") == user.id - same_org = ( - record.get("owner_org") - and record.get("owner_org") == user.organization - ) - if not same_owner and not same_org: + if user.role != "admin" and record.get("owner_id") != user.id: + shared_db3 = include_shared_db3 and str( + record.get("filename") or "" + ).lower().endswith(".db3") + if not shared_db3: continue uploads.append( diff --git a/api/ml_engine.py b/api/ml_engine.py index b97566c..4713e6a 100644 --- a/api/ml_engine.py +++ b/api/ml_engine.py @@ -664,8 +664,13 @@ def compute_feature_importance( gini_imp = rf.feature_importances_ # Permutation importance — prefer the active registry model over a - # fresh per-request fit (cheaper, reproducible, snapshot-pinned). - registry_bundle = _registry_model_for(primary_target, valid_features) + # fresh per-request fit (cheaper, reproducible, snapshot-pinned), unless + # the request's data scope differs from the registry's training data. + registry_bundle = ( + None + if (optimization_context or {}).get("force_live_fit") + else _registry_model_for(primary_target, valid_features) + ) if use_gpr and registry_bundle is not None and registry_bundle.get("model_kind") == "gpr": est = registry_bundle["estimator"] stored_scaler = registry_bundle.get("scaler") diff --git a/api/ml_response_cache.py b/api/ml_response_cache.py index ada9772..e04abbc 100644 --- a/api/ml_response_cache.py +++ b/api/ml_response_cache.py @@ -21,13 +21,17 @@ from psycopg2.extras import Json, RealDictCursor -from data_loader_pg import get_pg_migration_connection, get_pg_worker_connection +from data_loader_pg import ( + get_pg_connection, + get_pg_migration_connection, + get_pg_worker_connection, +) from security import PlatformUser logger = logging.getLogger("dt.api.ml.cache") -_SCHEMA_VERSION = "ml-response-cache-v2" +_SCHEMA_VERSION = "ml-response-cache-v3" _DEFAULT_TTL_SECONDS = 6 * 60 * 60 _locks: dict[str, threading.Lock] = {} _locks_guard = threading.Lock() @@ -117,46 +121,83 @@ def _ensure_table() -> None: _table_ready = True -def _data_signature(project_id: str) -> str: - """Cheap signature for the training data behind the ML pages.""" - conn = get_pg_worker_connection() +def _data_signature(project_id: str, user: PlatformUser) -> str: + """Signature of the runs behind this user's ML results. + + Read on the user's row-level-security connection (admins: the worker + connection), so it covers exactly the runs the user can see. Besides + counts it folds in an order-independent fingerprint of every visible run's + id, project and outlier/calibration flags: a run moving into a project the + user can't see, or being re-flagged, changes the key even when the counts + and latest ids stay the same. + """ + conn = ( + get_pg_worker_connection() + if user.role == "admin" + else get_pg_connection(user) + ) try: with conn.cursor(cursor_factory=RealDictCursor) as cur: - if project_id: - cur.execute( - """ - SELECT - count(*)::int AS total_runs, - count(*) FILTER ( - WHERE COALESCE(is_outlier, false) = false - AND COALESCE(is_calibration_recipe, false) = false - )::int AS clean_runs, - COALESCE(max(idruns), 0)::int AS max_run_id, - COALESCE(max(created_at)::text, '') AS latest_created_at - FROM etcher_runs - WHERE project_id = %s - """, - (project_id,), - ) - else: - cur.execute( - """ - SELECT - count(*)::int AS total_runs, - count(*) FILTER ( - WHERE COALESCE(is_outlier, false) = false - AND COALESCE(is_calibration_recipe, false) = false - )::int AS clean_runs, - COALESCE(max(idruns), 0)::int AS max_run_id, - COALESCE(max(created_at)::text, '') AS latest_created_at - FROM etcher_runs - """ - ) - row = cur.fetchone() or {} + cur.execute( + """ + SELECT + count(*)::int AS total_runs, + COALESCE(max(idruns), 0)::bigint AS max_run_id, + COALESCE(max(created_at)::text, '') AS latest_created_at, + COALESCE(sum(hashtext(concat_ws( + ':', idruns, project_id, + COALESCE(is_outlier, false), + COALESCE(is_calibration_recipe, false) + ))::bigint), 0)::text AS fingerprint + FROM etcher_runs + WHERE (%s = '' OR project_id = %s) + """, + (project_id, project_id), + ) + row = {f"etcher_{k}": v for k, v in dict(cur.fetchone() or {}).items()} + # Generic equipment runs feed the ML pages too. + cur.execute( + """ + SELECT + count(*)::int AS total_runs, + COALESCE(max(id), 0)::bigint AS max_run_id, + COALESCE(max(created_at)::text, '') AS latest_created_at, + COALESCE(sum(hashtext(concat_ws( + ':', id, project_id, + COALESCE(is_outlier, false), + COALESCE(is_calibration_recipe, false) + ))::bigint), 0)::text AS fingerprint + FROM equipment_runs + WHERE (%s = '' OR project_id = %s) + """, + (project_id, project_id), + ) + row.update({f"equipment_{k}": v for k, v in dict(cur.fetchone() or {}).items()}) + conn.commit() + finally: + conn.close() + return _stable_hash({"project_id": project_id, **row}) + + +def _access_signature(user: PlatformUser) -> str: + """Which projects the user can see right now, and whether each is open. + + Part of the cache key, so removing a member or making a project private + invalidates that user's cached results at once instead of after the TTL. + Read on the user's row-level-security connection: it lists exactly the + projects their results may draw on. + """ + if user.role == "admin": + return "admin" + conn = get_pg_connection(user) + try: + with conn.cursor() as cur: + cur.execute("SELECT id, COALESCE(access_mode, '') FROM projects ORDER BY id") + rows = cur.fetchall() conn.commit() finally: conn.close() - return _stable_hash({"project_id": project_id, **dict(row)}) + return _stable_hash([[str(project_id), mode] for project_id, mode in rows]) def _read_cache(cache_key: str, *, include_expired: bool = False) -> dict[str, Any] | None: @@ -283,7 +324,8 @@ def cached_ml_response( try: _ensure_table() - signature = _data_signature(project_id) + signature = _data_signature(project_id, user) + access = _access_signature(user) except Exception as exc: logger.warning("ML response cache unavailable; computing %s live: %s", endpoint, exc) return compute() @@ -296,6 +338,7 @@ def cached_ml_response( "project_id": project_id, "constraints_hash": constraints_hash, "data_signature": signature, + "access_signature": access, } ) diff --git a/api/routers/admin.py b/api/routers/admin.py index 0bbbeef..6337354 100644 --- a/api/routers/admin.py +++ b/api/routers/admin.py @@ -12,6 +12,7 @@ update_execution_queue_status_pg, update_review_status_pg, sync_config_snapshot, + user_manages_any_equipment_pg, ) from security import ( PlatformUser, @@ -349,10 +350,19 @@ def update_review_status( raise HTTPException(status_code=500, detail="Registration update failed") +def _has_facility_queue(user: PlatformUser) -> bool: + """Admins, equipment owners and anyone who owns or is a trusted + maintainer of equipment have an execution queue. What each sees in it is + limited to their own equipment (list_execution_queue_pg).""" + if user.role in {"admin", "equipment_owner"}: + return True + return user_manages_any_equipment_pg(user) + + def require_facility_queue_user( user: PlatformUser = Depends(get_platform_user), ) -> PlatformUser: - if user.role not in {"admin", "equipment_owner"}: + if not _has_facility_queue(user): raise HTTPException(status_code=403, detail="Facility execution queue access required") return user @@ -363,6 +373,18 @@ class ExecutionStatusPayload(BaseModel): scheduled_for: str = "" +@router.get("/execution-queue/access") +def get_execution_queue_access( + user: PlatformUser = Depends(get_platform_user), +) -> Dict[str, bool]: + """Whether the caller has an execution queue (shows the page in the menu).""" + try: + return {"allowed": _has_facility_queue(user)} + except Exception as e: + logger.error(f"Error checking execution queue access: {e}") + return {"allowed": False} + + @router.get("/execution-queue") def get_execution_queue( user: PlatformUser = Depends(require_facility_queue_user), diff --git a/api/routers/dataset_v2.py b/api/routers/dataset_v2.py index 34b4843..c0b26a1 100644 --- a/api/routers/dataset_v2.py +++ b/api/routers/dataset_v2.py @@ -84,7 +84,6 @@ get_platform_user, require_admin, require_ingestion_token, - resolve_request_user, ) logger = logging.getLogger("dt.api.dataset_v2") @@ -1862,13 +1861,15 @@ def run_metadata( @router.get("/snapshots") async def list_snapshots( limit: int = 20, - platform_user: PlatformUser = Depends(get_platform_user), + platform_user: PlatformUser = Depends(require_admin), ): """ - List recent dataset snapshots. + List recent dataset snapshots (admins only). Each snapshot records the SHA-256 hash of the full clean dataset at a point in time. Use this to verify which data version trained a model. + Snapshots describe the whole platform's data, not one user's projects, + so other users can't read them. """ try: from ai_readiness import get_latest_snapshots @@ -1887,7 +1888,8 @@ async def get_data_card( Generate an automated data card (dataset documentation). Returns statistics, class balance, feature distributions, temporal - coverage, and bias warnings for the current dataset. + coverage, and bias warnings for the runs the caller can see: their own + projects plus open projects (admins see every run). Useful for: - Documenting a model's training data in publications diff --git a/api/routers/equipment.py b/api/routers/equipment.py index f2d9024..05673c5 100644 --- a/api/routers/equipment.py +++ b/api/routers/equipment.py @@ -33,6 +33,7 @@ delete_equipment_pg, delete_experiment_type_pg, equipment_visible_to_user, + redact_equipment_for_viewer, fork_experiment_type_pg, get_equipment_pg, get_experiment_type_versions_pg, @@ -257,18 +258,18 @@ def _legacy_item_visible(item: dict, user: PlatformUser) -> bool: return True -def _live_run_stats_for(domain_id: str) -> dict: +def _live_run_stats_for(domain_id: str, user: PlatformUser) -> dict: """ - Fleet-wide run stats for legacy json_config equipment that don't have a - durable equipment_metadata row yet. Today only the etcher has a runs - table, so non-etcher equipment returns {"total_runs": None, - "last_data_at": None} and the UI will render "—". + Run stats for legacy json_config equipment that don't have a durable + equipment_metadata row yet, over the runs this user can see. Today only + the etcher has a runs table, so non-etcher equipment returns + {"total_runs": None, "last_data_at": None} and the UI will render "—". """ if domain_id != "etcher": return {"total_runs": None, "last_data_at": None} try: from data_loader_pg import get_etcher_run_stats_pg - return get_etcher_run_stats_pg() + return get_etcher_run_stats_pg(user) except Exception as exc: logger.warning("Failed to compute run stats for %s: %s", domain_id, exc) return {"total_runs": None, "last_data_at": None} @@ -305,26 +306,29 @@ def get_equipment_detail( """ Get a single equipment's metadata by domain_id. """ - item = get_equipment_pg(domain_id) + item = get_equipment_pg(domain_id, viewer=user) if item: if not equipment_visible_to_user(item, user): raise HTTPException(status_code=403, detail="Not authorized to view this equipment") - return item + return redact_equipment_for_viewer(item, user) config = load_domain_config(domain_id) if config: legacy = summarize_domain_config(config, config_file=f"{domain_id}.json") if not _legacy_item_visible(legacy, user): raise HTTPException(status_code=403, detail="Not authorized to view this equipment") - stats = _live_run_stats_for(domain_id) - return { - **legacy, - "source": "json_config", - "registered_at": None, - "updated_at": None, - "total_runs": stats["total_runs"], - "last_data_at": stats["last_data_at"], - } + stats = _live_run_stats_for(domain_id, user) + return redact_equipment_for_viewer( + { + **legacy, + "source": "json_config", + "registered_at": None, + "updated_at": None, + "total_runs": stats["total_runs"], + "last_data_at": stats["last_data_at"], + }, + user, + ) raise HTTPException(status_code=404, detail="Equipment not found") diff --git a/api/routers/ml.py b/api/routers/ml.py index c1b6e5a..0a140bc 100644 --- a/api/routers/ml.py +++ b/api/routers/ml.py @@ -89,12 +89,19 @@ def ml_runtime_status( def _user_training_context(user: PlatformUser, project_id: Optional[str] = None) -> dict: + is_admin = user.role == "admin" + project = (project_id or "").strip() or None return { "nanohub_user_id": user.id, - "is_admin": user.role == "admin", - "project_id": (project_id or "").strip() or None, + "is_admin": is_admin, + "project_id": project, "include_global_history": False, "disable_csv_fallback": True, + # The registry model is trained on every project's runs. Predictions + # from it would reflect runs this user can't see, so only an admin + # looking at all projects (the same data it was trained on) uses it; + # everyone else gets a model fitted on their visible runs only. + "force_live_fit": not (is_admin and project is None), } diff --git a/api/routers/upload.py b/api/routers/upload.py index 8219e28..a1ed833 100644 --- a/api/routers/upload.py +++ b/api/routers/upload.py @@ -38,6 +38,7 @@ from metadata_pg import ( approved_equipment_available_to_user, create_upload_record_pg, + equipment_visible_to_user, get_equipment_pg, get_upload_target_experiment_pg, list_uploads_pg, @@ -723,11 +724,11 @@ def _load_upload_record(upload_id: str, user: PlatformUser) -> dict[str, Any]: # and the platform administrator can help one another. No same-org # fallback applies to proprietary DB3 exports. _assert_db3_user_allowed(user) - elif user.role != "admin": - same_owner = record["owner_id"] == user.id - same_org = record["owner_org"] and record["owner_org"] == user.organization - if not same_owner and not same_org: - raise HTTPException(status_code=403, detail="Not authorized to process this upload") + elif user.role != "admin" and record["owner_id"] != user.id: + # Only the uploader (or an admin) may see, attach to or process an + # upload; the same organization is not enough. 404 rather than 403, + # so other users' upload ids aren't confirmed. + raise HTTPException(status_code=404, detail="Upload not found") return record @@ -1399,11 +1400,11 @@ def get_upload_capabilities( def list_uploads( user: PlatformUser = Depends(get_platform_user), ): - """List all previously uploaded files.""" + """List the caller's uploads (admins: all; DB3 pilot: the shared queue).""" db3_allowed = _db3_user_is_allowed(user) if db3_allowed: recover_stale_db3_uploads_pg() - uploads = list_uploads_pg(user) + uploads = list_uploads_pg(user, include_shared_db3=db3_allowed) if db3_allowed: return uploads return [ @@ -2839,13 +2840,20 @@ def generate_upload_template( the combined template regardless of the requested kind. """ kind = _normalize_kind(kind) + # Same visibility as the equipment page; 404 so hidden equipment (another + # organization's, or pending registrations of other owners) isn't confirmed. equipment = get_equipment_pg(equipment_id) if not equipment: legacy_config = load_domain_config(equipment_id) if not legacy_config: raise HTTPException(status_code=404, detail="Equipment not found") + owner_org = legacy_config.get("owner", {}).get("organization", "") + if user.role != "admin" and owner_org and owner_org != user.organization: + raise HTTPException(status_code=404, detail="Equipment not found") config = legacy_config else: + if not equipment_visible_to_user(equipment, user): + raise HTTPException(status_code=404, detail="Equipment not found") config = _effective_equipment_config(equipment) if equipment_id == "etcher": diff --git a/api/scripts/add_row_level_security.sql b/api/scripts/add_row_level_security.sql index 603ad80..7d825ad 100644 --- a/api/scripts/add_row_level_security.sql +++ b/api/scripts/add_row_level_security.sql @@ -143,22 +143,28 @@ AS $$ ) $$; --- Mirrors metadata_pg.list_execution_queue_pg: facility managers act on --- execution requests for equipment they (or their organization) own. +-- Mirrors metadata_pg.list_execution_queue_pg: the equipment's owner and its +-- trusted maintainers (any role) act on execution requests for that +-- equipment only; the same organization is not enough. CREATE OR REPLACE FUNCTION app_manages_equipment(p_equipment_id text) RETURNS boolean LANGUAGE sql STABLE SECURITY DEFINER SET search_path = public, pg_temp AS $$ SELECT app_is_admin() OR ( - COALESCE(app_current_user_role() = 'equipment_owner', false) - AND EXISTS ( - SELECT 1 - FROM equipment_metadata em - WHERE em.domain_id = p_equipment_id - AND ( - em.owner_id = app_current_user_id() - OR (COALESCE(em.owner_org, '') <> '' AND em.owner_org = app_current_user_org()) - ) + app_current_user_id() IS NOT NULL + AND ( + EXISTS ( + SELECT 1 + FROM equipment_metadata em + WHERE em.domain_id = p_equipment_id + AND em.owner_id = app_current_user_id() + ) + OR EXISTS ( + SELECT 1 + FROM equipment_access ea + WHERE ea.equipment_id = p_equipment_id + AND ea.user_id = app_current_user_id() + ) ) ) $$; @@ -261,9 +267,10 @@ AS $$ ), false) $$; --- Template read rule: list_experiment_types_pg (owner, shared, published, --- same organization), plus templates attached to a project or used by an --- experiment the caller can read, so joined names do not disappear. +-- Template read rule: list_experiment_types_pg (owner, shared, published), +-- plus templates attached to a project or used by an experiment the caller +-- can read, so joined names do not disappear. A private template is not +-- visible to the rest of the owner's organization. -- Published templates are readable without a signed-in user. CREATE OR REPLACE FUNCTION app_experiment_type_readable( p_type_id text, @@ -279,7 +286,6 @@ AS $$ app_is_admin() OR p_owner_id = app_current_user_id() OR p_visibility = 'shared' - OR (COALESCE(p_owner_org, '') <> '' AND p_owner_org = app_current_user_org()) OR EXISTS ( SELECT 1 FROM project_experiments pe diff --git a/api/security.py b/api/security.py index a50f25e..2acd331 100644 --- a/api/security.py +++ b/api/security.py @@ -1,3 +1,4 @@ +import hmac import os import logging from dataclasses import dataclass @@ -31,14 +32,11 @@ def get_expected_ingestion_token() -> str: return get_ingestion_token() -def _parse_bearer_user(authorization: Optional[str]) -> Optional[str]: - if not authorization: - return None - - scheme, _, value = authorization.partition(" ") - if scheme != "Bearer" or not value: - return None - return value.strip() or None +def _tokens_match(received: Optional[str], expected: str) -> bool: + """Constant-time comparison, so response timing can't reveal the token.""" + if received is None or not expected: + return False + return hmac.compare_digest(received.encode("utf-8"), expected.encode("utf-8")) def require_system_token( @@ -52,7 +50,7 @@ def require_system_token( detail="Platform authentication is not configured", ) from exc - if x_system_token != expected_token: + if not _tokens_match(x_system_token, expected_token): # Never place either the received credential or the configured secret # in logs. Operators only need the fact that validation failed. logger.warning("System token mismatch") @@ -70,7 +68,7 @@ def require_ingestion_token( detail="Ingestion authentication is not configured", ) from exc - if x_ingestion_token != expected_token: + if not _tokens_match(x_ingestion_token, expected_token): raise HTTPException(status_code=403, detail="Invalid ingestion token") @@ -101,16 +99,3 @@ def require_admin(user: PlatformUser = Depends(get_platform_user)) -> PlatformUs if user.role != "admin": raise HTTPException(status_code=403, detail="Admin access required") return user - - -def resolve_request_user( - x_system_token: Optional[str] = Header(None, alias="X-System-Token"), - x_user_id: str = Header("", alias="X-User-Id"), - x_user_email: str = Header("", alias="X-User-Email"), - authorization: Optional[str] = Header(None), -) -> Optional[str]: - if x_system_token is not None: - require_system_token(x_system_token) - return x_user_id or x_user_email or None - - return _parse_bearer_user(authorization) diff --git a/api/tests/test_data_scoping.py b/api/tests/test_data_scoping.py new file mode 100644 index 0000000..483197a --- /dev/null +++ b/api/tests/test_data_scoping.py @@ -0,0 +1,364 @@ +"""Users see their own data plus open projects, and nothing beyond that. + +Covers uploads (owner only), the upload template, equipment connection +details, the data card and snapshots, ML results built from the global model, +the ML response cache key, and the system-token check. +""" + +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +import numpy as np +import pandas as pd +from fastapi.testclient import TestClient + +os.environ.setdefault("DT_SYSTEM_TOKEN", "test-system-token") +os.environ.setdefault("INGESTION_TOKEN", "test-ingestion-token") +os.environ.setdefault("PG_PASS", "test-api-password") +os.environ.setdefault("PG_SUPER_PASS", "test-super-password") + +API_DIR = Path(__file__).resolve().parents[1] +if str(API_DIR) not in sys.path: + sys.path.insert(0, str(API_DIR)) +TESTS_DIR = Path(__file__).resolve().parent +if str(TESTS_DIR) not in sys.path: + sys.path.insert(0, str(TESTS_DIR)) + +import main +import metadata_pg +import ml_engine +import ml_response_cache +import security +from routers import admin, equipment, ml, upload +from security import PlatformUser +from test_phase3_security import ( + ADMIN_HEADERS, + OWNER_HEADERS, + RESEARCHER_HEADERS, + FakeConnection, +) + + +def _user(user_id: str, role: str = "researcher", org: str = "Birck") -> PlatformUser: + return PlatformUser(id=user_id, email=f"{user_id}@example.com", name=user_id, role=role, organization=org) + + +def _upload_row(upload_id: str, owner_id: str, filename: str = "runs.csv") -> dict: + return { + "id": upload_id, + "equipment_id": "etcher_01", + "filename": filename, + "stored_as": "stored", + "size_bytes": 1, + "row_count": 1, + "columns_json": [], + "errors_json": [], + "status": "success", + "kind": "all", + "owner_id": owner_id, + "owner_org": "Birck", + "uploaded_at": None, + "experiment_type_id": "", + "attachment_count": 0, + } + + +class UploadScopingTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.client = TestClient(main.app) + + def _list(self, user, **kwargs): + rows = [ + _upload_row("u-own", "researcher-1"), + _upload_row("u-colleague", "someone-else"), + _upload_row("u-colleague-db3", "someone-else", "pilot.db3"), + ] + connection = FakeConnection([{"contains": "FROM data_uploads", "fetchall": rows}]) + with patch.object(metadata_pg, "get_pg_connection", return_value=connection): + return [u["id"] for u in metadata_pg.list_uploads_pg(user, **kwargs)] + + def test_same_organization_is_not_enough_to_see_an_upload(self): + self.assertEqual(self._list(_user("researcher-1")), ["u-own"]) + + def test_db3_pilot_participants_keep_the_shared_db3_queue(self): + self.assertEqual( + self._list(_user("researcher-1"), include_shared_db3=True), + ["u-own", "u-colleague-db3"], + ) + + def test_admin_sees_every_upload(self): + self.assertEqual(len(self._list(_user("admin-1", role="admin"))), 3) + + def test_colleague_cannot_open_someone_elses_upload(self): + row = ("u-colleague", "etcher_01", "runs.csv", "/srv/x", "success", "someone-else", "Birck", [], "all") + connection = FakeConnection([{"contains": "FROM data_uploads", "fetchone": row}]) + with patch.object(upload, "get_pg_connection", return_value=connection): + response = self.client.get( + "/api/data/uploads/u-colleague/attachments", headers=RESEARCHER_HEADERS + ) + self.assertEqual(response.status_code, 404) + + +class TemplateVisibilityTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.client = TestClient(main.app) + + def test_template_for_hidden_equipment_is_not_found(self): + hidden = {"domain_id": "other_tool", "owner_id": "x", "owner_org": "Elsewhere", "status": "approved"} + with patch.object(upload, "get_equipment_pg", return_value=hidden): + response = self.client.get("/api/data/template/other_tool", headers=RESEARCHER_HEADERS) + self.assertEqual(response.status_code, 404) + + def test_template_for_someone_elses_pending_equipment_is_not_found(self): + pending = {"domain_id": "new_tool", "owner_id": "x", "owner_org": "Birck", "status": "pending"} + with patch.object(upload, "get_equipment_pg", return_value=pending): + response = self.client.get("/api/data/template/new_tool", headers=RESEARCHER_HEADERS) + self.assertEqual(response.status_code, 404) + + +class EquipmentConnectionTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.client = TestClient(main.app) + + def _item(self, maintainers=()): + return { + "domain_id": "etcher_01", + "owner_id": "owner-1", + "owner_org": "Birck", + "status": "approved", + "connection_host": "db.internal", + "connection_port": "5432", + "connection_database": "tooldb", + "config_json": {"source_connection": {"host": "db.internal"}, "outputs": [{"name": "rate"}]}, + "maintainers": [{"user_id": m} for m in maintainers], + } + + def _detail(self, headers, item): + with patch.object(equipment, "get_equipment_pg", return_value=item): + response = self.client.get("/api/equipment/detail/etcher_01", headers=headers) + self.assertEqual(response.status_code, 200) + return response.json() + + def test_viewer_does_not_see_connection_details(self): + body = self._detail(RESEARCHER_HEADERS, self._item()) + self.assertEqual( + [body["connection_host"], body["connection_port"], body["connection_database"]], + ["", "", ""], + ) + self.assertNotIn("source_connection", body["config_json"]) + self.assertEqual(body["config_json"]["outputs"], [{"name": "rate"}]) + + def test_owner_maintainer_and_admin_see_connection_details(self): + self.assertEqual(self._detail(OWNER_HEADERS, self._item())["connection_host"], "db.internal") + self.assertEqual(self._detail(ADMIN_HEADERS, self._item())["connection_host"], "db.internal") + body = self._detail(RESEARCHER_HEADERS, self._item(maintainers=["researcher-1"])) + self.assertEqual(body["config_json"]["source_connection"], {"host": "db.internal"}) + + def test_equipment_list_redacts_using_maintainer_ids(self): + item = self._item() + item.pop("maintainers") + viewer = metadata_pg.redact_equipment_for_viewer(item, _user("researcher-1"), manageable=False) + self.assertEqual(viewer["connection_host"], "") + self.assertEqual(item["connection_host"], "db.internal") # the original is untouched + self.assertTrue( + metadata_pg.equipment_manageable_by({"owner_id": "x", "_maintainer_ids": ["researcher-1"]}, _user("researcher-1")) + ) + + +class EquipmentRunCountTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.client = TestClient(main.app) + + def test_viewer_run_counts_come_from_their_scoped_connection(self): + rows = [{"equipment_id": "etcher_01", "total_runs": 3, "last_data_at": None}] + scoped = FakeConnection([{"contains": "FROM equipment_runs", "fetchall": rows}]) + with patch.object(metadata_pg, "get_pg_connection", return_value=scoped) as opened, patch.object( + metadata_pg, "get_pg_worker_connection", side_effect=AssertionError("fleet-wide counts used") + ): + metrics = metadata_pg._fetch_equipment_run_metrics(_user("researcher-1")) + self.assertEqual(metrics["etcher_01"]["total_runs"], 3) + self.assertEqual(opened.call_args.args[0].id, "researcher-1") + + def test_admin_and_internal_callers_get_fleet_wide_counts(self): + for viewer in (None, _user("admin-1", role="admin")): + worker = FakeConnection([{"contains": "FROM equipment_runs", "fetchall": []}]) + with patch.object(metadata_pg, "get_pg_worker_connection", return_value=worker), patch.object( + metadata_pg, "get_pg_connection", side_effect=AssertionError("scoped connection used") + ): + metadata_pg._fetch_equipment_run_metrics(viewer) + + def test_legacy_equipment_detail_is_redacted_and_scoped(self): + legacy = { + "domain_id": "etcher", + "owner_id": "owner-1", + "owner_org": "Birck", + "connection_host": "db.internal", + "config_json": {"source_connection": {"host": "db.internal"}}, + } + with patch.object(equipment, "get_equipment_pg", return_value=None), patch.object( + equipment, "load_domain_config", return_value={"domain": {"id": "etcher"}} + ), patch.object(equipment, "summarize_domain_config", return_value=legacy), patch.object( + equipment, "_live_run_stats_for", return_value={"total_runs": 1, "last_data_at": None} + ) as stats: + response = self.client.get("/api/equipment/detail/etcher", headers=RESEARCHER_HEADERS) + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertEqual(body["connection_host"], "") + self.assertNotIn("source_connection", body["config_json"]) + self.assertEqual(stats.call_args.args[1].id, "researcher-1") + + +class DatasetAggregateTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.client = TestClient(main.app) + + def test_snapshots_are_admin_only(self): + with patch("ai_readiness.get_latest_snapshots", return_value=[]): + self.assertEqual( + self.client.get("/api/dataset/v2/snapshots", headers=RESEARCHER_HEADERS).status_code, 403 + ) + self.assertEqual( + self.client.get("/api/dataset/v2/snapshots", headers=ADMIN_HEADERS).status_code, 200 + ) + + def test_data_card_reads_through_the_users_scoped_connection(self): + import ai_readiness + import data_loader_pg + + class Stop(Exception): + pass + + with patch.object(data_loader_pg, "_scoped_connection", side_effect=Stop) as scoped, patch.object( + data_loader_pg, "get_pg_worker_connection", side_effect=AssertionError("worker connection used") + ): + with self.assertRaises(Stop): + ai_readiness.generate_data_card(nanohub_user_id="researcher-1", is_admin=False) + scoped.assert_called_once_with("researcher-1", False) + + +class MlScopingTests(unittest.TestCase): + def test_only_an_admin_viewing_all_projects_may_use_the_global_model(self): + self.assertFalse(ml._user_training_context(_user("admin-1", role="admin"))["force_live_fit"]) + self.assertTrue(ml._user_training_context(_user("admin-1", role="admin"), "p1")["force_live_fit"]) + self.assertTrue(ml._user_training_context(_user("researcher-1"))["force_live_fit"]) + + def test_feature_importance_skips_the_global_model_when_forced(self): + features = ml_engine.ETCHER_FEATURES + rng = np.random.RandomState(0) + df = pd.DataFrame(rng.rand(12, len(features)), columns=features) + X = df[features].to_numpy() + ys = {ml_engine.ETCHER_PRIMARY_TARGET: X.sum(axis=1)} + with patch.object(ml_engine, "_get_xy", return_value=(X, ys, [], df)), patch.object( + ml_engine, "_registry_model_for", side_effect=AssertionError("global model used") + ): + result = ml_engine.compute_feature_importance( + domain_config={"features": features, "targets": [ml_engine.ETCHER_PRIMARY_TARGET], "ml_model": "random_forest"}, + optimization_context={"force_live_fit": True}, + ) + self.assertIsNone(result["model_version"]) + + +class MlCacheKeyTests(unittest.TestCase): + def _key(self, user, access): + written = [] + with patch.dict(os.environ, {"ML_ENDPOINT_CACHE_DISABLED": "false", "ML_ENDPOINT_CACHE_TTL_SECONDS": "60"}), \ + patch.object(ml_response_cache, "_ensure_table"), \ + patch.object(ml_response_cache, "_data_signature", return_value="data"), \ + patch.object(ml_response_cache, "_access_signature", return_value=access), \ + patch.object(ml_response_cache, "_read_cache", return_value=None), \ + patch.object(ml_response_cache, "_write_cache", side_effect=lambda **kw: written.append(kw["cache_key"])): + ml_response_cache.cached_ml_response(endpoint="parity", user=user, compute=lambda: {"ok": True}) + return written[0] + + def test_losing_access_to_a_project_changes_the_cache_key(self): + user = _user("researcher-1") + self.assertNotEqual(self._key(user, "projects-a-b"), self._key(user, "projects-a")) + + def test_admin_access_signature_needs_no_query(self): + with patch.object(ml_response_cache, "get_pg_connection", side_effect=AssertionError("queried")): + self.assertEqual(ml_response_cache._access_signature(_user("admin-1", role="admin")), "admin") + + +class MlDataSignatureTests(unittest.TestCase): + def _signature(self, user, fingerprint): + ops = [ + {"contains": "FROM etcher_runs", "fetchone": {"total_runs": 2, "max_run_id": 9, "latest_created_at": "t", "fingerprint": fingerprint}}, + {"contains": "FROM equipment_runs", "fetchone": {"total_runs": 0, "max_run_id": 0, "latest_created_at": "", "fingerprint": "0"}}, + ] + connection = FakeConnection(ops) + with patch.object(ml_response_cache, "get_pg_connection", return_value=connection) as opened, patch.object( + ml_response_cache, "get_pg_worker_connection", side_effect=AssertionError("worker connection used") + ): + signature = ml_response_cache._data_signature("", user) + self.assertIs(opened.call_args.args[0], user) + return signature + + def test_signature_reads_only_the_users_visible_runs(self): + self._signature(_user("researcher-1"), "100") + + def test_a_run_moving_project_changes_the_signature(self): + # Same counts and latest ids; only a run's project (fingerprint) changed. + user = _user("researcher-1") + self.assertNotEqual(self._signature(user, "100"), self._signature(user, "250")) + + +class ExecutionQueueAccessTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.client = TestClient(main.app) + + def test_trusted_maintainer_with_any_role_has_a_queue(self): + with patch.object(admin, "user_manages_any_equipment_pg", return_value=True), patch.object( + admin, "list_execution_queue_pg", return_value=[] + ) as listed: + self.assertEqual( + self.client.get("/api/admin/execution-queue/access", headers=RESEARCHER_HEADERS).json(), + {"allowed": True}, + ) + self.assertEqual(self.client.get("/api/admin/execution-queue", headers=RESEARCHER_HEADERS).status_code, 200) + self.assertEqual(listed.call_args.args[0].id, "researcher-1") + + def test_users_without_equipment_have_no_queue(self): + with patch.object(admin, "user_manages_any_equipment_pg", return_value=False): + self.assertEqual( + self.client.get("/api/admin/execution-queue/access", headers=RESEARCHER_HEADERS).json(), + {"allowed": False}, + ) + self.assertEqual(self.client.get("/api/admin/execution-queue", headers=RESEARCHER_HEADERS).status_code, 403) + + def test_admins_and_equipment_owners_keep_the_page(self): + with patch.object(admin, "user_manages_any_equipment_pg", side_effect=AssertionError("queried")): + for headers in (ADMIN_HEADERS, OWNER_HEADERS): + self.assertEqual( + self.client.get("/api/admin/execution-queue/access", headers=headers).json(), + {"allowed": True}, + ) + + +class SystemTokenTests(unittest.TestCase): + def test_token_comparison(self): + self.assertTrue(security._tokens_match("abc", "abc")) + self.assertFalse(security._tokens_match("abd", "abc")) + self.assertFalse(security._tokens_match(None, "abc")) + self.assertFalse(security._tokens_match("", "")) + + def test_wrong_token_is_rejected(self): + client = TestClient(main.app) + headers = {**RESEARCHER_HEADERS, "X-System-Token": "wrong"} + self.assertEqual(client.get("/api/data/uploads", headers=headers).status_code, 403) + + def test_header_trusting_helper_is_gone(self): + self.assertFalse(hasattr(security, "resolve_request_user")) + self.assertFalse(hasattr(security, "_parse_bearer_user")) + + +if __name__ == "__main__": + unittest.main() diff --git a/api/tests/test_row_level_security_pg.py b/api/tests/test_row_level_security_pg.py index 4e26017..e0280b6 100644 --- a/api/tests/test_row_level_security_pg.py +++ b/api/tests/test_row_level_security_pg.py @@ -712,6 +712,120 @@ def test_facility_manager_sees_execution_request_with_project_name(self): 0, ) + def _insert_user(self, person: PlatformUser) -> None: + import data_loader_pg + + conn = data_loader_pg.get_pg_migration_connection() + try: + with conn.cursor() as cur: + cur.execute( + "INSERT INTO users (id, name, email, role, organization) VALUES (%s, %s, %s, %s, %s)", + (person.id, person.name, person.email, person.role, person.organization), + ) + conn.commit() + finally: + conn.close() + + def _set_equipment_org(self, org: str) -> None: + import data_loader_pg + + conn = data_loader_pg.get_pg_migration_connection() + try: + with conn.cursor() as cur: + cur.execute( + "UPDATE equipment_metadata SET owner_org = %s WHERE domain_id = %s", + (org, self.equipment_id), + ) + conn.commit() + finally: + conn.close() + + def test_execution_queue_is_for_the_equipment_owner_and_maintainers_only(self): + from metadata_pg import ( + grant_equipment_access_pg, + list_execution_queue_pg, + update_execution_queue_status_pg, + user_manages_any_equipment_pg, + ) + + # Another equipment owner in the same organization as the equipment. + frank = _user(f"frank-{self.suffix}", "equipment_owner", "Facility") + # A researcher who becomes a trusted maintainer of the equipment. + maya = _user(f"maya-{self.suffix}", "researcher", "LabM") + for person in (frank, maya): + self._insert_user(person) + self._set_equipment_org("Facility") + self.addCleanup(self._set_equipment_org, "") + + project_id = self._create_project(self.alice) + type_id = self._create_template(self.alice) + experiment_id = self._create_experiment(self.alice, project_id, type_id)["experiment_id"] + + # Same organization is not enough. + self.assertNotIn(experiment_id, {row["id"] for row in list_execution_queue_pg(frank)}) + # Row-level security hides the request from Frank, so it is "not found". + self.assertEqual( + update_execution_queue_status_pg( + experiment_id=experiment_id, status="approved", note="", scheduled_for="", user=frank + )["status"], + "not_found", + ) + self.assertFalse(user_manages_any_equipment_pg(frank)) + self.assertFalse(user_manages_any_equipment_pg(maya)) + + grant_equipment_access_pg(equipment_id=self.equipment_id, user_id=maya.id, granted_by=self.olivia.id) + self.assertTrue(user_manages_any_equipment_pg(maya)) + self.assertIn(experiment_id, {row["id"] for row in list_execution_queue_pg(maya)}) + updated = update_execution_queue_status_pg( + experiment_id=experiment_id, status="approved", note="", scheduled_for="", user=maya + ) + self.assertEqual(updated["execution_status"], "approved") + self.assertIn(experiment_id, {row["id"] for row in list_execution_queue_pg(self.olivia)}) + + # Alice's template is private and the project is private, but the + # equipment's owner and maintainer can read the request's template; + # Frank (same organization, neither) cannot. + from metadata_pg import list_experiment_types_pg + + self.assertIn(type_id, {t["id"] for t in list_experiment_types_pg(user=maya)}) + self.assertIn(type_id, {t["id"] for t in list_experiment_types_pg(user=self.olivia)}) + self.assertNotIn(type_id, {t["id"] for t in list_experiment_types_pg(user=frank)}) + + def test_private_template_is_hidden_from_the_owners_organization(self): + from metadata_pg import list_experiment_types_pg + + dana = _user(f"dana-{self.suffix}", "pi", "LabA") # same organization as Alice + self._insert_user(dana) + private_id = self._create_template(self.alice, visibility="private") + shared_id = self._create_template(self.alice, visibility="shared") + + dana_sees = {t["id"] for t in list_experiment_types_pg(user=dana)} + self.assertNotIn(private_id, dana_sees) + self.assertIn(shared_id, dana_sees) + self.assertEqual( + self._count_as(dana, "SELECT count(*) FROM experiment_types WHERE id = %s", (private_id,)), 0 + ) + self.assertIn(private_id, {t["id"] for t in list_experiment_types_pg(user=self.alice)}) + + # A member of a project that uses the template can still read it. + project_id = self._create_project(self.alice) + self._add_member(project_id, dana, self.alice) + from metadata_pg import add_project_experiment_pg + + add_project_experiment_pg(project_id=project_id, type_id=private_id, user=self.alice) + self.assertEqual( + self._count_as(dana, "SELECT count(*) FROM experiment_types WHERE id = %s", (private_id,)), 1 + ) + + def test_ml_cache_signature_query_runs_under_row_level_security(self): + import ml_response_cache + + project_id = self._create_project(self.alice) + alice_signature = ml_response_cache._data_signature("", self.alice) + self.assertEqual(alice_signature, ml_response_cache._data_signature("", self.alice)) + self.assertEqual(len(ml_response_cache._data_signature(project_id, self.bob)), 64) + self.assertEqual(len(ml_response_cache._access_signature(self.bob)), 64) + # ── template deletion ─────────────────────────────────────────────── def test_template_delete_detaches_references_in_other_projects(self): diff --git a/web/lib/auth-context.tsx b/web/lib/auth-context.tsx index d097e58..740b187 100644 --- a/web/lib/auth-context.tsx +++ b/web/lib/auth-context.tsx @@ -22,6 +22,7 @@ import { createContext, useContext, useEffect, + useState, type ReactNode, } from "react"; import { useSession } from "next-auth/react"; @@ -222,8 +223,34 @@ export function AuthProvider({ children }: { children: ReactNode }) { permissions = EMPTY_PERMISSIONS; } + // Trusted maintainers of equipment have an execution queue whatever their + // role, so ask the API instead of going by role alone. + const sessionUserId = isAuthenticated ? session?.user?.id ?? "" : ""; + const queueByRole = permissions.visibleNavItems.includes("execution_queue"); + const [maintainerQueue, setMaintainerQueue] = useState<{ userId: string; allowed: boolean }>({ + userId: "", + allowed: false, + }); + useEffect(() => { + if (!sessionUserId || queueByRole) return; + let cancelled = false; + fetch("/api/dt/admin/execution-queue/access", { cache: "no-store" }) + .then((response) => (response.ok ? response.json() : null)) + .then((data: { allowed?: boolean } | null) => { + if (!cancelled) setMaintainerQueue({ userId: sessionUserId, allowed: data?.allowed === true }); + }) + .catch(() => undefined); + return () => { + cancelled = true; + }; + }, [sessionUserId, queueByRole]); + const hasPermission = (navItemId: string) => - permissions.visibleNavItems.includes(navItemId); + permissions.visibleNavItems.includes(navItemId) || + (navItemId === "execution_queue" && + maintainerQueue.allowed && + maintainerQueue.userId === sessionUserId && + sessionUserId !== ""); return ( = { "create and fork unit-experiment templates and process templates", "request experiment runs and upload run data", "request a role change", + "if a trusted maintainer of equipment: review that equipment's execution queue", ], equipment_owner: [ "everything a researcher can do", "register equipment and edit equipment they own or maintain", - "review the facility execution queue for their equipment", + "review the facility execution queue for equipment they own or maintain", ], pi: [ "everything a researcher can do", @@ -683,33 +684,38 @@ export const ASSISTANT_TOOLS: ToolDef[] = [ name: "get_data_card", title: "Data card", description: - "Auto-generated data card for the user's visible dataset: statistics, class balance, feature distributions, temporal coverage and bias warnings.", + "Auto-generated data card over the etcher runs the user can see (their projects plus open projects): statistics, class balance, feature distributions, temporal coverage and bias warnings.", inputSchema: NO_ARGS, }, () => "/dataset/v2/data-card", "build data card", ), - getTool( - { - name: "list_dataset_snapshots", - title: "Dataset snapshots", - description: "Recent SHA-256 dataset snapshots (which data version trained which model).", - inputSchema: { - type: "object", - properties: { limit: { type: "number", description: "1-50, default 10." } }, - additionalProperties: false, + { + // Snapshots describe the whole platform's data, so they are admin-only + // (the backend enforces this too). + ...getTool( + { + name: "list_dataset_snapshots", + title: "Dataset snapshots", + description: "Recent SHA-256 snapshots of the whole platform dataset (which data version trained which model).", + inputSchema: { + type: "object", + properties: { limit: { type: "number", description: "1-50, default 10." } }, + additionalProperties: false, + }, }, - }, - (args) => `/dataset/v2/snapshots${query({ limit: clampInt(num(args, "limit"), 10, 1, 50) })}`, - "list snapshots", - ), + (args) => `/dataset/v2/snapshots${query({ limit: clampInt(num(args, "limit"), 10, 1, 50) })}`, + "list snapshots", + ), + roles: ["admin"], + }, // Uploads getTool( { name: "list_my_uploads", title: "Upload history", - description: "Data files the user uploaded, with validation and processing status.", + description: "Data files the user uploaded (only their own), with validation and processing status.", inputSchema: NO_ARGS, }, () => "/data/uploads", @@ -862,13 +868,13 @@ export const ASSISTANT_TOOLS: ToolDef[] = [ { name: "get_execution_queue", title: "Execution queue", - description: "Facility execution queue of experiment requests (for admins and owners of the equipment).", + description: + "Facility execution queue of experiment requests for equipment the user owns or is a trusted maintainer of (admins: all equipment). Others get a permission error.", inputSchema: NO_ARGS, }, () => "/admin/execution-queue", "load execution queue", ), - roles: ["admin", "equipment_owner"], }, { ...getTool(