From 897b5b408ead09dee246bc4f40bd83a585f3e581 Mon Sep 17 00:00:00 2001 From: Priyanshu-u07 Date: Mon, 19 Oct 2026 23:17:30 +0530 Subject: [PATCH 1/3] Check node and worker ownership and require an org in the CLI Signed-off-by: Priyanshu-u07 --- src/cli/__init__.py | 28 ++ src/cli/node.py | 8 +- src/cli/worker.py | 2 + src/orchestration/api/admin_workers.py | 27 +- src/orchestration/api/nodes.py | 35 ++- src/orchestration/ownership.py | 26 +- .../orchestration/api/test_admin_workers.py | 66 +++-- .../orchestration/api/test_node_ownership.py | 265 ++++++++++++++++++ src/tests/orchestration/api/test_nodes.py | 36 ++- .../api/test_nodes_provisioning_endpoints.py | 31 +- 10 files changed, 451 insertions(+), 73 deletions(-) create mode 100644 src/tests/orchestration/api/test_node_ownership.py diff --git a/src/cli/__init__.py b/src/cli/__init__.py index d1e76d15..ae10edc5 100644 --- a/src/cli/__init__.py +++ b/src/cli/__init__.py @@ -866,6 +866,10 @@ def main(argv=None): default=None, help="Internal API key (default: INTERNAL_API_KEY env var)", ) + tok_p.add_argument( + "--org-id", default=None, + help="Organization the command acts in (default: INFERIA_ORG_ID env var)", + ) # worker compose — scaffold a ready-to-run docker-compose directory cmp_p = worker_sub.add_parser( @@ -911,6 +915,10 @@ def main(argv=None): default=None, help="Internal API key (default: INTERNAL_API_KEY env var)", ) + cmp_p.add_argument( + "--org-id", default=None, + help="Organization the command acts in (default: INFERIA_ORG_ID env var)", + ) # worker list — show workers in a pool list_w = worker_sub.add_parser( @@ -928,6 +936,10 @@ def main(argv=None): default=None, help="Internal API key (default: INTERNAL_API_KEY env var)", ) + list_w.add_argument( + "--org-id", default=None, + help="Organization the command acts in (default: INFERIA_ORG_ID env var)", + ) # --- node sub-command (node-centric API) ------------------------------- node_parser = sub.add_parser( @@ -956,23 +968,39 @@ def main(argv=None): nl_set.add_argument("kv", nargs="+", help="key=value pairs") nl_set.add_argument("--orchestration-url", default=None) nl_set.add_argument("--internal-api-key", default=None) + nl_set.add_argument( + "--org-id", default=None, + help="Organization the command acts in (default: INFERIA_ORG_ID env var)", + ) nl_del = nl_lab_sub.add_parser("del", help="Remove labels from a node") nl_del.add_argument("node_id") nl_del.add_argument("keys", nargs="+", help="label keys to remove") nl_del.add_argument("--orchestration-url", default=None) nl_del.add_argument("--internal-api-key", default=None) + nl_del.add_argument( + "--org-id", default=None, + help="Organization the command acts in (default: INFERIA_ORG_ID env var)", + ) nl_get = nl_lab_sub.add_parser("get", help="Print a node's labels as JSON") nl_get.add_argument("node_id") nl_get.add_argument("--orchestration-url", default=None) nl_get.add_argument("--internal-api-key", default=None) + nl_get.add_argument( + "--org-id", default=None, + help="Organization the command acts in (default: INFERIA_ORG_ID env var)", + ) # node rm n_rm = node_sub.add_parser("rm", help="Soft-delete a node") n_rm.add_argument("node_id") n_rm.add_argument("--orchestration-url", default=None) n_rm.add_argument("--internal-api-key", default=None) + n_rm.add_argument( + "--org-id", default=None, + help="Organization the command acts in (default: INFERIA_ORG_ID env var)", + ) # node pool … np = node_sub.add_parser("pool", help="Pool management") diff --git a/src/cli/node.py b/src/cli/node.py index bfa68139..54b2c823 100644 --- a/src/cli/node.py +++ b/src/cli/node.py @@ -69,9 +69,11 @@ def _internal_headers(args, org_id: str | None = None) -> dict: "X-Gateway-Request": "true", "Content-Type": "application/json", } + # Orchestration only acts on the org's own pools and nodes. org = org_id or getattr(args, "org_id", None) or os.getenv("INFERIA_ORG_ID") - if org: - headers["X-Organization-ID"] = org + if not org: + sys.exit("error: --org-id required (or set INFERIA_ORG_ID)") + headers["X-Organization-ID"] = org return headers @@ -91,8 +93,6 @@ def _parse_labels(items: Iterable[str] | None) -> dict[str, str]: def _cmd_list(args) -> None: - if not args.org_id and not os.getenv("INFERIA_ORG_ID"): - sys.exit("error: --org-id required (or set INFERIA_ORG_ID)") base = _orchestration_base(args) qs = "" if args.label: diff --git a/src/cli/worker.py b/src/cli/worker.py index b2e2d0d0..e7ea906b 100644 --- a/src/cli/worker.py +++ b/src/cli/worker.py @@ -69,6 +69,8 @@ def _internal_headers(args) -> dict: "X-Internal-API-Key": key, "X-Gateway-Request": "true", "Content-Type": "application/json", + # Orchestration only acts on the org's own pools and nodes. + "X-Organization-ID": _resolve(args, "org-id", "INFERIA_ORG_ID"), } diff --git a/src/orchestration/api/admin_workers.py b/src/orchestration/api/admin_workers.py index aa5a8a67..8a057219 100644 --- a/src/orchestration/api/admin_workers.py +++ b/src/orchestration/api/admin_workers.py @@ -32,7 +32,7 @@ ) from pydantic import BaseModel, Field -from orchestration.ownership import caller_org, owns_pool +from orchestration.ownership import owns_node, owns_pool, owns_pool_id from orchestration.workers.worker_controller.auth import ( WorkerAuth, ) @@ -198,7 +198,7 @@ async def mint_bootstrap_token( ): pool_repo = _pools() pool = await pool_repo.get(body.pool_id) - if pool is None: + if pool is None or not owns_pool(request, pool): raise HTTPException(status_code=404, detail="pool not found") if pool.get("lifecycle_state") in ("terminating", "terminated"): raise HTTPException( @@ -240,8 +240,11 @@ async def mint_bootstrap_token( @router.get("/pool/{pool_id}", response_model=ListResponse) async def list_workers( pool_id: str, + request: Request, _granted: bool = Depends(_need_perm("deployment:read")), ): + if not await owns_pool_id(request, pool_id, _deps.pool_repo): + raise HTTPException(status_code=404, detail="pool not found") rows = await _inventory().list_workers(pool_id=pool_id) live = set(_registry().list_nodes()) workers = [_row_to_view(r, connected=str(r["id"]) in live) for r in rows] @@ -251,11 +254,14 @@ async def list_workers( @router.delete("/{node_id}") async def revoke_worker( node_id: str, + request: Request, _granted: bool = Depends(_need_perm("deployment:delete")), ): - # Verify the node exists. Return 404 if not. + # Verify the node exists and is the caller's. Return 404 if not. node = await _inventory().get_node_by_id(node_id) - if node is None: + if node is None or not await owns_pool_id( + request, node.get("pool_id"), _deps.pool_repo, + ): raise HTTPException(status_code=404, detail="node not found") # AWS branch: route the EC2 tear-down through the provisioning @@ -374,11 +380,14 @@ async def revoke_worker( @router.get("/{node_id}/metrics", response_model=NodeMetricsResponse) async def get_node_metrics( node_id: str, + request: Request, _granted: bool = Depends(_need_perm("deployment:read")), ): """Live per-node resource metrics (CPU/mem/GPU/net/disk) from the worker heartbeat ring buffer. Empty buffer (node just connected, never sent metrics, or disconnected) -> {"latest": null, "samples": []} — not a 404.""" + if not await owns_node(request, node_id, _deps.inventory_repo, _deps.pool_repo): + raise HTTPException(status_code=404, detail="node not found") samples = _registry().get_metrics(node_id) latest = samples[-1] if samples else None return NodeMetricsResponse(latest=latest, samples=samples) @@ -455,15 +464,7 @@ async def _node_in_caller_org(ws: WebSocket, node_id: str) -> bool: The gateway checks the user's permission, not which nodes it covers, so a node id from another organisation must be refused here. """ - if not caller_org(ws) or _deps.inventory_repo is None or _deps.pool_repo is None: - return False - try: - node = await _deps.inventory_repo.get_node_by_id(node_id) - pool = await _deps.pool_repo.get(node["pool_id"]) if node else None - except Exception as e: - logger.warning("org check for node %s failed: %s", node_id, e) - return False - return owns_pool(ws, pool) + return await owns_node(ws, node_id, _deps.inventory_repo, _deps.pool_repo) @router.websocket("/{node_id}/logs") diff --git a/src/orchestration/api/nodes.py b/src/orchestration/api/nodes.py index 6233bb46..5cb8fe52 100644 --- a/src/orchestration/api/nodes.py +++ b/src/orchestration/api/nodes.py @@ -31,6 +31,7 @@ NodeNotFoundError, NodeTerminatedError, ) +from orchestration.ownership import owns_node, owns_pool_id from orchestration.state_machine.jobs.model import Phase logger = logging.getLogger("nodes_api") @@ -115,6 +116,13 @@ def _org_id_from_headers(authorization, x_organization_id) -> str: return x_organization_id +async def _require_own_node(request: Request, node_id: str = Path(...)) -> None: + """404 unless the node sits in one of the caller's pools: the same answer + as a missing node, so another org's id reveals nothing.""" + if not await owns_node(request, node_id, _deps.inventory_repo, _deps.pool_repo): + raise HTTPException(404, "node not found") + + # --------------------------------------------------------------------------- # Pydantic schemas. # --------------------------------------------------------------------------- @@ -263,21 +271,27 @@ async def list_nodes( @router.get("/xds-status") -async def list_xds_nodes(): - """Return all non-terminated nodes in xDS control-plane format. +async def list_xds_nodes(request: Request): + """The caller's non-terminated nodes in xDS control-plane format, with + engines derived from running deployments. - Internal endpoint for the envoyproxy xDS control plane. Returns every - node regardless of org, with engines derived from running deployments. - No auth — only accessible via internal API key. + The gateway exposes this route, so it is scoped like the other listings; + the xDS control plane reads the repository directly, not this route. """ rows = await _deps.inventory_repo.list_xds_nodes() - return {"nodes": rows} + owned: dict = {} + for row in rows: + pool_id = row.get("pool_id") + if pool_id not in owned: + owned[pool_id] = await owns_pool_id(request, pool_id, _deps.pool_repo) + return {"nodes": [row for row in rows if owned[row.get("pool_id")]]} @router.get("/{node_id}", response_model=NodeView) async def get_node( node_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:list")), + _owned: None = Depends(_require_own_node), ): row = await _deps.inventory_repo.get_node(node_id=node_id) if not row: @@ -290,6 +304,7 @@ async def get_node_log_stream( node_id: str = Path(...), request: Request = None, _granted: bool = Depends(_need_perm("deployment:list")), + _owned: None = Depends(_require_own_node), ): """WS connection info for streaming a node's live container logs. @@ -392,6 +407,7 @@ async def get_node_deploy_metrics( node_id: str = Path(...), deployment_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:list")), + _owned: None = Depends(_require_own_node), ): """Return the most recent per-deployment metrics from the worker heartbeat.""" metrics = await _deps.inventory_repo.get_deploy_metrics( @@ -407,6 +423,7 @@ async def patch_labels( body: PatchLabelsBody, node_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:update")), + _owned: None = Depends(_require_own_node), ): overlap = set(body.add.keys()) & set(body.remove) if overlap: @@ -430,6 +447,7 @@ async def patch_labels( async def delete_node( node_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:delete")), + _owned: None = Depends(_require_own_node), ): existing = await _deps.inventory_repo.get_node(node_id=node_id) if not existing: @@ -546,6 +564,7 @@ async def delete_node( async def retry_provisioning( node_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:create")), + _owned: None = Depends(_require_own_node), ): """Re-enqueue a failed provisioning job. @@ -728,6 +747,7 @@ class DepinDetails(BaseModel): async def get_depin_details( node_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:list")), + _owned: None = Depends(_require_own_node), ): """Per-node instance details for a DePIN (Nosana/Akash) node — the Instance Details tab. Reads the node's stored metadata (which ``NodeView`` drops) @@ -811,6 +831,7 @@ async def get_depin_details( async def get_provisioning( node_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:list")), + _owned: None = Depends(_require_own_node), ): row = await _deps.inventory_repo.get_node(node_id=node_id) if not row: @@ -938,6 +959,7 @@ async def get_provisioning_logs( after: int = Query(default=0, ge=0), limit: int = Query(default=500, ge=1, le=2000), _granted: bool = Depends(_need_perm("deployment:list")), + _owned: None = Depends(_require_own_node), ): row = await _deps.inventory_repo.get_node(node_id=node_id) if not row: @@ -965,6 +987,7 @@ async def get_provisioning_logs( async def get_ec2_console( node_id: str = Path(...), _granted: bool = Depends(_need_perm("deployment:list")), + _owned: None = Depends(_require_own_node), ): from datetime import datetime, timezone row = await _deps.inventory_repo.get_node(node_id=node_id) diff --git a/src/orchestration/ownership.py b/src/orchestration/ownership.py index 2f9c4656..6f37905a 100644 --- a/src/orchestration/ownership.py +++ b/src/orchestration/ownership.py @@ -30,6 +30,28 @@ def owns_pool(conn, pool) -> bool: """ return ( pool is not None - and pool["owner_type"] in ("organization", "user") - and owns(conn, pool["owner_id"]) + and pool.get("owner_type") in ("organization", "user") + and owns(conn, pool.get("owner_id")) ) + + +async def owns_pool_id(conn, pool_id, pool_repo) -> bool: + """owns_pool for a pool looked up by id. Any lookup failure owns nothing.""" + if not caller_org(conn) or pool_repo is None or not pool_id: + return False + try: + pool = await pool_repo.get(pool_id) + except Exception: + return False + return owns_pool(conn, pool) + + +async def owns_node(conn, node_id, inventory_repo, pool_repo) -> bool: + """Whether the node sits in a pool the caller's org owns.""" + if not caller_org(conn) or inventory_repo is None: + return False + try: + node = await inventory_repo.get_node_by_id(node_id) + except Exception: + return False + return bool(node) and await owns_pool_id(conn, node.get("pool_id"), pool_repo) diff --git a/src/tests/orchestration/api/test_admin_workers.py b/src/tests/orchestration/api/test_admin_workers.py index 39406500..07488eb2 100644 --- a/src/tests/orchestration/api/test_admin_workers.py +++ b/src/tests/orchestration/api/test_admin_workers.py @@ -39,6 +39,9 @@ def _patched_client_init(self, *args, **kwargs): SECRET = "test-secret-key-at-least-32-chars-long!" POOL_ID = "00000000-0000-0000-0000-000000000001" +# Every pool and worker here belongs to ORG; ownership is tested in +# test_node_ownership.py. +ORG = "org-1" NODE_ID = "11111111-2222-3333-4444-555555555555" @@ -59,7 +62,7 @@ async def mark_terminated_worker(self, *, node_id): async def get_node_by_id(self, node_id): for w in self.workers: if w["id"] == node_id: - return w + return {**w, "pool_id": w.get("pool_id") or POOL_ID} return None @@ -113,7 +116,7 @@ def app_and_deps(): registry = WorkerRegistry() inventory = FakeInventory() pool_repo = FakePoolRepo(pool={ - "id": POOL_ID, "pool_name": "p", "lifecycle_state": "running", + "id": POOL_ID, "pool_name": "p", "lifecycle_state": "running", "owner_type": "organization", "owner_id": ORG, }) admin_workers.configure( @@ -140,7 +143,7 @@ def test_happy_path(self, app_and_deps): r = client.post( "/v1/admin/workers/tokens", json={"pool_id": POOL_ID, "ttl_hours": 1}, - headers={"Authorization": "Bearer deployment:create"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:create"}, ) assert r.status_code == 200, r.text data = r.json() @@ -165,7 +168,7 @@ def test_default_ttl_is_one_hour(self, app_and_deps): r = client.post( "/v1/admin/workers/tokens", json={"pool_id": POOL_ID}, - headers={"Authorization": "Bearer deployment:create"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:create"}, ) assert r.status_code == 200 # exp ≈ now + 3600 (±60s tolerance for test runtime). @@ -178,7 +181,7 @@ def test_ttl_capped_at_24_hours(self, app_and_deps): r = client.post( "/v1/admin/workers/tokens", json={"pool_id": POOL_ID, "ttl_hours": 100}, - headers={"Authorization": "Bearer deployment:create"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:create"}, ) assert r.status_code == 422 @@ -188,7 +191,7 @@ def test_ttl_below_one_rejected(self, app_and_deps): r = client.post( "/v1/admin/workers/tokens", json={"pool_id": POOL_ID, "ttl_hours": 0}, - headers={"Authorization": "Bearer deployment:create"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:create"}, ) assert r.status_code == 422 @@ -198,7 +201,7 @@ def test_missing_permission_rejected(self, app_and_deps): r = client.post( "/v1/admin/workers/tokens", json={"pool_id": POOL_ID}, - headers={"Authorization": "Bearer some_other_perm"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer some_other_perm"}, ) assert r.status_code == 403 @@ -209,7 +212,7 @@ def test_pool_not_found(self, app_and_deps): r = client.post( "/v1/admin/workers/tokens", json={"pool_id": POOL_ID}, - headers={"Authorization": "Bearer deployment:create"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:create"}, ) assert r.status_code == 404 @@ -220,7 +223,7 @@ def test_pool_terminated_conflict(self, app_and_deps): r = client.post( "/v1/admin/workers/tokens", json={"pool_id": POOL_ID}, - headers={"Authorization": "Bearer deployment:create"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:create"}, ) assert r.status_code == 409 @@ -261,7 +264,7 @@ def test_returns_workers_with_connection_state(self, app_and_deps): client = TestClient(app) r = client.get( f"/v1/admin/workers/pool/{POOL_ID}", - headers={"Authorization": "Bearer deployment:read"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:read"}, ) assert r.status_code == 200, r.text workers = r.json()["workers"] @@ -277,7 +280,7 @@ def test_empty_pool(self, app_and_deps): client = TestClient(app) r = client.get( f"/v1/admin/workers/pool/{POOL_ID}", - headers={"Authorization": "Bearer deployment:read"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:read"}, ) assert r.status_code == 200 assert r.json() == {"workers": []} @@ -287,7 +290,7 @@ def test_missing_permission(self, app_and_deps): client = TestClient(app) r = client.get( f"/v1/admin/workers/pool/{POOL_ID}", - headers={"Authorization": "Bearer some_other_perm"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer some_other_perm"}, ) assert r.status_code == 403 @@ -313,7 +316,7 @@ def test_marks_terminated_and_closes_ws(self, app_and_deps): client = TestClient(app) r = client.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 204 assert NODE_ID in inventory.terminated @@ -325,11 +328,11 @@ def test_idempotent_on_already_terminated(self, app_and_deps): client = TestClient(app) r1 = client.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) r2 = client.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r1.status_code == 204 # Second call still succeeds because mark_terminated_worker is @@ -342,7 +345,7 @@ def test_unknown_node_returns_404(self, app_and_deps): client = TestClient(app) r = client.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 404 @@ -351,7 +354,7 @@ def test_missing_permission(self, app_and_deps): client = TestClient(app) r = client.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer some_other_perm"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer some_other_perm"}, ) assert r.status_code == 403 @@ -406,7 +409,7 @@ def aws_app_and_deps(): registry = WorkerRegistry() inventory = AwsFakeInventory() pool_repo = FakePoolRepo(pool={ - "id": POOL_ID, "pool_name": "p", "lifecycle_state": "running", + "id": POOL_ID, "pool_name": "p", "lifecycle_state": "running", "owner_type": "organization", "owner_id": ORG, }) _aw.configure( worker_auth=auth, @@ -458,7 +461,7 @@ def depin_app_and_deps(): registry = WorkerRegistry() inventory = DepinFakeInventory() pool_repo = FakePoolRepo(pool={ - "id": POOL_ID, "pool_name": "p", "lifecycle_state": "running", + "id": POOL_ID, "pool_name": "p", "lifecycle_state": "running", "owner_type": "organization", "owner_id": ORG, "provider_credential_name": "nosana-cred-1", }) _aw.configure( @@ -495,7 +498,7 @@ async def test_nosana_worker_deprovisions_then_terminates_204( ) as c: r = await c.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 204, r.text @@ -527,7 +530,7 @@ async def test_nosana_placeholder_pii_skips_deprovision( ) as c: r = await c.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 204, r.text fake_adapter.deprovision_node.assert_not_called() @@ -555,7 +558,7 @@ async def test_nosana_deprovision_failure_still_terminates( ) as c: r = await c.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 204, r.text fake_adapter.deprovision_node.assert_awaited_once() @@ -610,7 +613,7 @@ async def test_aws_worker_force_cancels_and_returns_202(self, aws_app_and_deps): ) as c: r = await c.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 202, r.text @@ -647,7 +650,7 @@ async def test_aws_worker_no_live_job_falls_back_to_soft_delete( ) as c: r = await c.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 204, r.text @@ -672,7 +675,7 @@ async def test_non_aws_worker_keeps_legacy_path(self, aws_app_and_deps): ) as c: r = await c.delete( f"/v1/admin/workers/{NODE_ID}", - headers={"Authorization": "Bearer deployment:delete"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:delete"}, ) assert r.status_code == 204 spawn.assert_not_called() @@ -687,13 +690,14 @@ async def test_non_aws_worker_keeps_legacy_path(self, aws_app_and_deps): class TestNodeMetrics: def test_returns_latest_and_samples(self, app_and_deps): - app, _auth, registry, *_ = app_and_deps + app, _auth, registry, inventory, _pool = app_and_deps + inventory.workers.append({"id": NODE_ID}) registry.record_metrics(NODE_ID, {"ts": "a", "cpu_pct": 1.0}) registry.record_metrics(NODE_ID, {"ts": "b", "cpu_pct": 2.0}) client = TestClient(app) r = client.get( f"/v1/admin/workers/{NODE_ID}/metrics", - headers={"Authorization": "Bearer deployment:read"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:read"}, ) assert r.status_code == 200, r.text data = r.json() @@ -701,11 +705,13 @@ def test_returns_latest_and_samples(self, app_and_deps): assert data["latest"]["ts"] == "b" def test_empty_buffer_returns_null_latest_not_404(self, app_and_deps): - app, *_ = app_and_deps + # A known node that has not reported yet: empty, not 404. + app, _auth, _reg, inventory, _pool = app_and_deps + inventory.workers.append({"id": NODE_ID}) client = TestClient(app) r = client.get( f"/v1/admin/workers/{NODE_ID}/metrics", - headers={"Authorization": "Bearer deployment:read"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer deployment:read"}, ) assert r.status_code == 200, r.text assert r.json() == {"latest": None, "samples": []} @@ -715,6 +721,6 @@ def test_requires_permission(self, app_and_deps): client = TestClient(app) r = client.get( f"/v1/admin/workers/{NODE_ID}/metrics", - headers={"Authorization": "Bearer some:other"}, + headers={"X-Organization-ID": ORG, "Authorization": "Bearer some:other"}, ) assert r.status_code == 403 diff --git a/src/tests/orchestration/api/test_node_ownership.py b/src/tests/orchestration/api/test_node_ownership.py new file mode 100644 index 00000000..9f328dbb --- /dev/null +++ b/src/tests/orchestration/api/test_node_ownership.py @@ -0,0 +1,265 @@ +"""Node and worker routes act only on the caller's own nodes and pools. + +The gateway proxies /v1/nodes/* and /v1/admin/workers/*, checking a user's +permission but not which nodes it covers. Each route here must refuse a node +or pool outside the caller's org with the same 404 as a missing one, before +doing any work, and the CLI must name its org. +""" +from __future__ import annotations + +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from orchestration.api import admin_workers, nodes as nodes_api +from orchestration.ownership import owns_node, owns_pool_id + +OWNER = "org-a" +POOL = "00000000-0000-0000-0000-0000000000aa" +NODE = "11111111-1111-1111-1111-111111111111" + + +class _Reached(Exception): + """Raised by any repository call after the ownership lookup.""" + + +class _Inventory: + """Answers the ownership lookup; any other call means the route acted.""" + + def __init__(self): + self.acted = [] + + async def get_node_by_id(self, node_id): + return {"id": node_id, "pool_id": POOL} if node_id == NODE else None + + def __getattr__(self, name): + async def _act(*_a, **_kw): + self.acted.append(name) + raise _Reached(name) + return _act + + +class _Pools: + def __init__(self, owner_type="organization", owner_id=OWNER): + self.row = {"id": POOL, "owner_type": owner_type, "owner_id": owner_id, + "lifecycle_state": "running"} + self.acted = [] + + async def get(self, pool_id): + return self.row if str(pool_id) == POOL else None + + def __getattr__(self, name): + async def _act(*_a, **_kw): + self.acted.append(name) + raise _Reached(name) + return _act + + +def _allow(_perm): + return lambda *_a, **_kw: True + + +@pytest.fixture +def nodes_app(): + inv, pools = _Inventory(), _Pools() + # Provisioning and events repos record too: some routes start there. + prov, events = _Inventory(), _Inventory() + nodes_api.configure( + inventory_repo=inv, pool_repo=pools, worker_auth=SimpleNamespace(), + control_plane_external_url="", adapters={}, require_permission=_allow, + provisioning_repo=prov, node_events_repo=events, + ) + app = FastAPI() + app.include_router(nodes_api.router) + app.state.acted = lambda: inv.acted + prov.acted + events.acted + return app, inv, pools + + +@pytest.fixture +def workers_app(): + inv, pools = _Inventory(), _Pools() + admin_workers.configure( + worker_auth=SimpleNamespace(), worker_registry=SimpleNamespace( + get_metrics=lambda _n: [], list_nodes=lambda: []), + inventory_repo=inv, pool_repo=pools, control_plane_external_url="", + require_permission=_allow, + ) + app = FastAPI() + app.include_router(admin_workers.router) + return app, inv, pools + + +def _call(app, method, path, org, body=None): + headers = {"Authorization": "Bearer t"} + if org is not None: + headers["X-Organization-ID"] = org + client = TestClient(app, raise_server_exceptions=False) + return client.request(method, path, json=body, headers=headers) + + +NOT_OWNER = pytest.mark.parametrize( + "org", ["org-b", "", "None", None], ids=["other-org", "empty", "None", "no-header"], +) + +NODE_ROUTES = [ + ("GET", f"/v1/nodes/{NODE}", None), + ("GET", f"/v1/nodes/{NODE}/log-stream", None), + ("GET", f"/v1/nodes/{NODE}/deploy-metrics/d1", None), + ("PATCH", f"/v1/nodes/{NODE}/labels", {"add": {"k": "v"}, "remove": []}), + ("DELETE", f"/v1/nodes/{NODE}", None), + ("POST", f"/v1/nodes/{NODE}/provisioning/retry", None), + ("GET", f"/v1/nodes/{NODE}/depin-details", None), + ("GET", f"/v1/nodes/{NODE}/provisioning", None), + ("GET", f"/v1/nodes/{NODE}/provisioning-logs", None), + ("GET", f"/v1/nodes/{NODE}/ec2-console", None), +] +NODE_IDS = ["get", "log-stream", "deploy-metrics", "labels", "delete", "retry", + "depin-details", "provisioning", "provisioning-logs", "ec2-console"] + + +@pytest.mark.parametrize("method, path, body", NODE_ROUTES, ids=NODE_IDS) +@NOT_OWNER +def test_another_orgs_node_is_not_found(nodes_app, method, path, body, org): + app, inv, _ = nodes_app + + r = _call(app, method, path, org, body) + + # 404, not 403: the same answer as a missing node reveals nothing. + assert r.status_code == 404, r.text + assert inv.acted == [], f"the route acted on a node the caller does not own: {inv.acted}" + + +@pytest.mark.parametrize("method, path, body", NODE_ROUTES, ids=NODE_IDS) +def test_the_owning_org_gets_through(nodes_app, method, path, body): + app, _, _ = nodes_app + + _call(app, method, path, OWNER, body) + + assert app.state.acted(), "the route never got past the ownership check" + + +@pytest.mark.parametrize("method, path, body", NODE_ROUTES[:1], ids=NODE_IDS[:1]) +def test_a_node_in_a_system_pool_belongs_to_no_org(nodes_app, method, path, body): + app, inv, pools = nodes_app + pools.row["owner_type"] = "system" + + assert _call(app, method, path, OWNER, body).status_code == 404 + assert inv.acted == [] + + +def test_xds_status_lists_only_the_callers_nodes(nodes_app): + app, inv, _ = nodes_app + rows = [{"id": "n1", "pool_id": POOL}, {"id": "n2", "pool_id": "someone-elses"}] + + async def _rows(): + return rows + + inv.list_xds_nodes = _rows + + assert _call(app, "GET", "/v1/nodes/xds-status", OWNER).json() == {"nodes": [rows[0]]} + assert _call(app, "GET", "/v1/nodes/xds-status", None).json() == {"nodes": []} + + +# ---- worker admin ---------------------------------------------------------- + +WORKER_ROUTES = [ + ("POST", "/v1/admin/workers/tokens", {"pool_id": POOL}), + ("GET", f"/v1/admin/workers/pool/{POOL}", None), + ("DELETE", f"/v1/admin/workers/{NODE}", None), + ("GET", f"/v1/admin/workers/{NODE}/metrics", None), +] +WORKER_IDS = ["mint-token", "list-workers", "revoke", "metrics"] + + +@pytest.mark.parametrize("method, path, body", WORKER_ROUTES, ids=WORKER_IDS) +@NOT_OWNER +def test_another_orgs_pool_or_worker_is_not_found(workers_app, method, path, body, org): + app, inv, pools = workers_app + + r = _call(app, method, path, org, body) + + assert r.status_code == 404, r.text + assert inv.acted == [] and pools.acted == [], (inv.acted, pools.acted) + + +@pytest.mark.parametrize("method, path, body", WORKER_ROUTES, ids=WORKER_IDS) +def test_the_owning_org_reaches_its_workers(workers_app, method, path, body): + app, inv, pools = workers_app + + r = _call(app, method, path, OWNER, body) + + assert r.status_code != 404 or inv.acted or pools.acted, r.text + + +def test_a_token_cannot_be_minted_for_a_system_pool(workers_app): + app, _, pools = workers_app + pools.row["owner_type"] = "system" + + r = _call(app, "POST", "/v1/admin/workers/tokens", OWNER, {"pool_id": POOL}) + + assert r.status_code == 404 + assert pools.acted == [] + + +# ---- the shared lookups fail closed ---------------------------------------- + + +def _conn(org): + return SimpleNamespace(headers={} if org is None else {"x-organization-id": org}) + + +class _Broken: + async def get(self, *_a): + raise RuntimeError("database down") + + async def get_node_by_id(self, *_a): + raise RuntimeError("database down") + + +@pytest.mark.asyncio +async def test_a_failed_lookup_owns_nothing(): + assert await owns_pool_id(_conn(OWNER), POOL, _Broken()) is False + assert await owns_node(_conn(OWNER), NODE, _Broken(), _Pools()) is False + + +@pytest.mark.asyncio +async def test_a_pool_row_without_owner_fields_owns_nothing(): + class _Bare: + async def get(self, _pool_id): + return {"id": POOL} + + assert await owns_pool_id(_conn(OWNER), POOL, _Bare()) is False + + +# ---- the CLI names its org -------------------------------------------------- + + +def _args(**kw): + base = {"internal_api_key": "k", "orchestration_url": None, "org_id": None} + base.update(kw) + return SimpleNamespace(**base) + + +@pytest.mark.parametrize("module", ["cli.node", "cli.worker"]) +def test_the_cli_refuses_to_call_without_an_org(monkeypatch, module): + import importlib + + mod = importlib.import_module(module) + monkeypatch.delenv("INFERIA_ORG_ID", raising=False) + + with pytest.raises(SystemExit): + mod._internal_headers(_args()) + + +@pytest.mark.parametrize("module", ["cli.node", "cli.worker"]) +def test_the_cli_sends_the_org_it_is_given(monkeypatch, module): + import importlib + + mod = importlib.import_module(module) + monkeypatch.delenv("INFERIA_ORG_ID", raising=False) + + assert mod._internal_headers(_args(org_id=OWNER))["X-Organization-ID"] == OWNER + monkeypatch.setenv("INFERIA_ORG_ID", "org-env") + assert mod._internal_headers(_args())["X-Organization-ID"] == "org-env" diff --git a/src/tests/orchestration/api/test_nodes.py b/src/tests/orchestration/api/test_nodes.py index 3fead8c9..e39a71dc 100644 --- a/src/tests/orchestration/api/test_nodes.py +++ b/src/tests/orchestration/api/test_nodes.py @@ -61,6 +61,12 @@ async def list_nodes(self, *, org_id, selector=None): async def get_node(self, *, node_id): return self.nodes.get(node_id) + async def get_node_by_id(self, node_id): + # Every node here is in ORG's pool; ownership is tested in + # test_node_ownership.py. + node = self.nodes.get(node_id) + return {**node, "pool_id": node.get("pool_id") or POOL} if node else None + async def set_labels(self, *, node_id, add, remove): node = self.nodes.get(node_id) if not node: @@ -107,6 +113,9 @@ class FakePoolRepo: async def ensure_default_pool(self, *, org_id): return POOL + async def get(self, pool_id): + return {"owner_type": "organization", "owner_id": ORG} + class FakeAdapter: def __init__(self, raises=None): @@ -639,7 +648,7 @@ async def test_aws_ready_node_force_cancels_through_reconciler(self): transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://t") as c: - r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t"}) + r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t", "X-Organization-ID": ORG}) assert r.status_code == 202, r.text assert r.json()["state"] == "terminating" @@ -695,7 +704,7 @@ async def test_delete_no_job_softdeletes_204(self): app = _configure_with_provisioning_repo(inventory, provisioning_repo) transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://t") as c: - r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t"}) + r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t", "X-Organization-ID": ORG}) assert r.status_code == 204 provisioning_repo.force_cancel.assert_not_called() assert inventory.nodes[NODE]["state"] == "terminated" @@ -723,7 +732,8 @@ async def ensure_default_pool(self, *, org_id): return POOL async def get(self, pool_id): - return {"id": pool_id, "provider_credential_name": self.cred} + return {"id": pool_id, "provider_credential_name": self.cred, + "owner_type": "organization", "owner_id": ORG} class DepinInventory(FakeInventory): @@ -767,7 +777,7 @@ async def test_nosana_node_deprovisions_then_softdeletes_204(self): ): transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://t") as c: - r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t"}) + r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t", "X-Organization-ID": ORG}) assert r.status_code == 204, r.text fake_adapter.deprovision_node.assert_awaited_once_with( provider_instance_id="job-addr-123", @@ -790,7 +800,7 @@ async def test_nosana_placeholder_skips_deprovision(self): ): transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://t") as c: - r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t"}) + r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t", "X-Organization-ID": ORG}) assert r.status_code == 204, r.text fake_adapter.deprovision_node.assert_not_called() assert inventory.nodes[NODE]["state"] == "terminated" @@ -811,7 +821,7 @@ async def test_nosana_deprovision_failure_still_softdeletes(self): ): transport = ASGITransport(app=app) async with httpx.AsyncClient(transport=transport, base_url="http://t") as c: - r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t"}) + r = await c.delete(f"/v1/nodes/{NODE}", headers={"Authorization": "Bearer t", "X-Organization-ID": ORG}) assert r.status_code == 204, r.text fake_adapter.deprovision_node.assert_awaited_once() assert inventory.nodes[NODE]["state"] == "terminated" @@ -891,7 +901,7 @@ def test_get_provisioning_includes_error_and_aws_metadata(): app = FastAPI() app.include_router(nodes_api.router) client = TestClient(app) - auth = {"Authorization": "Bearer test"} + auth = {"Authorization": "Bearer test", "X-Organization-ID": ORG} resp = client.get(f"/v1/nodes/{nid}/provisioning", headers=auth) assert resp.status_code == 200, resp.text body = resp.json() @@ -928,7 +938,7 @@ def test_get_provisioning_returns_404_when_node_missing(): app = FastAPI() app.include_router(nodes_api.router) client = TestClient(app) - auth = {"Authorization": "Bearer test"} + auth = {"Authorization": "Bearer test", "X-Organization-ID": ORG} resp = client.get( "/v1/nodes/00000000-0000-0000-0000-000000000000/provisioning", headers=auth, @@ -984,7 +994,7 @@ def test_retry_provisioning_on_failed_job_returns_200_and_requeues(): app = _configure_with_provisioning_repo(inventory, provisioning_repo) client = TestClient(app) - auth = {"Authorization": "Bearer test"} + auth = {"Authorization": "Bearer test", "X-Organization-ID": ORG} resp = client.post( f"/v1/nodes/{nid}/provisioning/retry", headers=auth, ) @@ -1019,7 +1029,7 @@ def test_retry_provisioning_on_non_failed_job_returns_409(): app = _configure_with_provisioning_repo(inventory, provisioning_repo) client = TestClient(app) - auth = {"Authorization": "Bearer test"} + auth = {"Authorization": "Bearer test", "X-Organization-ID": ORG} resp = client.post( f"/v1/nodes/{nid}/provisioning/retry", headers=auth, ) @@ -1036,7 +1046,7 @@ def test_retry_provisioning_on_missing_node_returns_404(): app = _configure_with_provisioning_repo(inventory, provisioning_repo) client = TestClient(app) - auth = {"Authorization": "Bearer test"} + auth = {"Authorization": "Bearer test", "X-Organization-ID": ORG} resp = client.post( f"/v1/nodes/{uuid.uuid4()}/provisioning/retry", headers=auth, ) @@ -1071,7 +1081,7 @@ def _delete_with_job(phase: Phase, *, node_state: str): app = _configure_with_provisioning_repo(inventory, provisioning_repo) client = TestClient(app) resp = client.delete( - f"/v1/nodes/{nid}", headers={"Authorization": "Bearer test"}, + f"/v1/nodes/{nid}", headers={"Authorization": "Bearer test", "X-Organization-ID": ORG}, ) return resp, inventory, provisioning_repo, nid @@ -1133,7 +1143,7 @@ def test_delete_missing_node_returns_404_with_provisioning_repo(): app = _configure_with_provisioning_repo(inventory, provisioning_repo) client = TestClient(app) - auth = {"Authorization": "Bearer test"} + auth = {"Authorization": "Bearer test", "X-Organization-ID": ORG} resp = client.delete(f"/v1/nodes/{uuid.uuid4()}", headers=auth) assert resp.status_code == 404 # We bailed before consulting the provisioning repo. diff --git a/src/tests/orchestration/api/test_nodes_provisioning_endpoints.py b/src/tests/orchestration/api/test_nodes_provisioning_endpoints.py index edb853da..b0376a29 100644 --- a/src/tests/orchestration/api/test_nodes_provisioning_endpoints.py +++ b/src/tests/orchestration/api/test_nodes_provisioning_endpoints.py @@ -31,13 +31,29 @@ def _patched_client_init(self, *args, **kwargs): from orchestration.api import nodes as nodes_api # noqa: E402 +# Every node here is in a pool ORG owns; ownership is tested in +# test_node_ownership.py. +ORG = str(uuid4()) + + def _user_ctx_header(): return { "authorization": "Bearer test", - "x-organization-id": str(uuid4()), + "x-organization-id": ORG, } +def _by_id(inv): + """get_node_by_id answering from the same row as the mocked get_node.""" + async def _get(node_id): + return await inv.get_node(node_id=node_id) + return _get + + +def _owned(pool_row=None): + return {**(pool_row or {}), "owner_type": "organization", "owner_id": ORG} + + @pytest.fixture def app_and_deps(): inv = MagicMock() @@ -51,6 +67,7 @@ def app_and_deps(): "provider_instance_id": "placeholder:" + str(pool_id), } inv.get_node = AsyncMock(return_value=node_row) + inv.get_node_by_id = _by_id(inv) prov = MagicMock() prov.summarize_phases = AsyncMock(return_value=[ {"phase": "prepare", "status": "succeeded", @@ -79,7 +96,8 @@ def app_and_deps(): }) app = FastAPI() nodes_api.configure( - inventory_repo=inv, pool_repo=MagicMock(), worker_auth=MagicMock(), + inventory_repo=inv, pool_repo=MagicMock(get=AsyncMock(return_value=_owned())), + worker_auth=MagicMock(), control_plane_external_url="", adapters={"aws": aws_adapter}, require_permission=lambda _: (lambda: True), provisioning_repo=prov, @@ -269,6 +287,7 @@ def _build_region_app( inv = MagicMock() inv.get_node = AsyncMock(return_value=node_row) + inv.get_node_by_id = _by_id(inv) prov = MagicMock() prov.get_by_node = AsyncMock(return_value=job) @@ -279,7 +298,7 @@ def _build_region_app( prov.current_phase = AsyncMock(return_value=None) pool_repo_mock = MagicMock() - pool_repo_mock.get = AsyncMock(return_value=pool_row) + pool_repo_mock.get = AsyncMock(return_value=_owned(pool_row)) aws_adapter = MagicMock() aws_adapter.get_logs = AsyncMock(return_value={"logs": ["line1"]}) @@ -307,9 +326,11 @@ def test_region_from_job_spec(app_and_deps): prov.get_by_node = AsyncMock(return_value=job) app = FastAPI() + inv = MagicMock(**{"get_node": AsyncMock(return_value=node_row)}) + inv.get_node_by_id = _by_id(inv) nodes_api.configure( - inventory_repo=MagicMock(**{"get_node": AsyncMock(return_value=node_row)}), - pool_repo=MagicMock(), + inventory_repo=inv, + pool_repo=MagicMock(get=AsyncMock(return_value=_owned())), worker_auth=MagicMock(), control_plane_external_url="", adapters={"aws": aws_adapter}, From adabe024e3afbfc1de0e465687b55fa1e6a9670c Mon Sep 17 00:00:00 2001 From: Priyanshu-u07 Date: Mon, 19 Oct 2026 23:42:21 +0530 Subject: [PATCH 2/3] Give the CLI pool tests an org Signed-off-by: Priyanshu-u07 --- src/tests/test_cli_node_pool.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tests/test_cli_node_pool.py b/src/tests/test_cli_node_pool.py index 3d5c048c..3940b31a 100644 --- a/src/tests/test_cli_node_pool.py +++ b/src/tests/test_cli_node_pool.py @@ -36,7 +36,7 @@ class A: image_tag = kw.get("image_tag") orchestration_url = kw.get("orchestration_url", "http://localhost:8080") internal_api_key = kw.get("internal_api_key", "test-key") - org_id = kw.get("org_id") + org_id = kw.get("org_id", "org-1") return A() From 351d0c69d17590552b4dd57b7bebc73d178858b4 Mon Sep 17 00:00:00 2001 From: Priyanshu-u07 Date: Tue, 20 Oct 2026 00:02:02 +0530 Subject: [PATCH 3/3] Log failed org lookups and show --org-id in the README Signed-off-by: Priyanshu-u07 --- README.md | 7 ++++--- src/orchestration/ownership.py | 10 ++++++++-- src/tests/orchestration/api/test_node_ownership.py | 4 +++- 3 files changed, 15 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 7f8693dd..0f43b7cc 100644 --- a/README.md +++ b/README.md @@ -168,10 +168,11 @@ sequenceDiagram ### Adding a self-hosted GPU node -On the control-plane host, scaffold a ready-to-run worker deployment: +On the control-plane host, scaffold a ready-to-run worker deployment. `node` and `worker` commands act only on your organisation's pools, so pass `--org-id` or set `INFERIA_ORG_ID`: ```bash inferiallm worker compose \ + --org-id \ --pool-id \ --node-name gpu-1 \ --advertise-url http://:8080 \ @@ -469,8 +470,8 @@ inferiallm providers list inferiallm providers add aws --name prod --type access_key_id --value AKIA... inferiallm providers remove aws --name prod -# Compute nodes -inferiallm node add worker --name gpu-1 --pool-id --advertise-url http://host:8080 +# Compute nodes (or set INFERIA_ORG_ID instead of --org-id) +export INFERIA_ORG_ID= inferiallm worker compose --pool-id --node-name gpu-1 --advertise-url http://host:8080 inferiallm worker token --pool-id --ttl-hours 1 inferiallm worker list --pool-id diff --git a/src/orchestration/ownership.py b/src/orchestration/ownership.py index 6f37905a..9f1f3234 100644 --- a/src/orchestration/ownership.py +++ b/src/orchestration/ownership.py @@ -6,6 +6,10 @@ """ from __future__ import annotations +import logging + +logger = logging.getLogger(__name__) + def caller_org(conn) -> str: """The forwarded org, or "" when there is none. @@ -41,7 +45,8 @@ async def owns_pool_id(conn, pool_id, pool_repo) -> bool: return False try: pool = await pool_repo.get(pool_id) - except Exception: + except Exception as e: + logger.warning("org check for pool %s failed: %s", pool_id, e) return False return owns_pool(conn, pool) @@ -52,6 +57,7 @@ async def owns_node(conn, node_id, inventory_repo, pool_repo) -> bool: return False try: node = await inventory_repo.get_node_by_id(node_id) - except Exception: + except Exception as e: + logger.warning("org check for node %s failed: %s", node_id, e) return False return bool(node) and await owns_pool_id(conn, node.get("pool_id"), pool_repo) diff --git a/src/tests/orchestration/api/test_node_ownership.py b/src/tests/orchestration/api/test_node_ownership.py index 9f328dbb..91f3cbd7 100644 --- a/src/tests/orchestration/api/test_node_ownership.py +++ b/src/tests/orchestration/api/test_node_ownership.py @@ -219,9 +219,11 @@ async def get_node_by_id(self, *_a): @pytest.mark.asyncio -async def test_a_failed_lookup_owns_nothing(): +async def test_a_failed_lookup_owns_nothing(caplog): assert await owns_pool_id(_conn(OWNER), POOL, _Broken()) is False assert await owns_node(_conn(OWNER), NODE, _Broken(), _Pools()) is False + # Logged, so a database outage is not just a run of unexplained 404s. + assert [r.levelname for r in caplog.records] == ["WARNING", "WARNING"] @pytest.mark.asyncio