Check node and worker ownership and require an org in the CLI

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 <ORG_UUID> \
   --pool-id <POOL_UUID> \
   --node-name gpu-1 \
   --advertise-url http://<worker-host>: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 <id> --advertise-url http://host:8080
+# Compute nodes (or set INFERIA_ORG_ID instead of --org-id)
+export INFERIA_ORG_ID=<org-id>
 inferiallm worker compose --pool-id <id> --node-name gpu-1 --advertise-url http://host:8080
 inferiallm worker token   --pool-id <id> --ttl-hours 1
 inferiallm worker list    --pool-id <id>

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")

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:

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"),
     }
 
 

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")

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)

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.
@@ -30,6 +34,30 @@ 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 as e:
+        logger.warning("org check for pool %s failed: %s", pool_id, e)
+        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 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)