diff --git a/umbratrans_web/compute.py b/umbratrans_web/compute.py index 2cb9a8b..5f8ce8c 100644 --- a/umbratrans_web/compute.py +++ b/umbratrans_web/compute.py @@ -821,6 +821,27 @@ class ComputeRepository: return invalid node_key = generate_token(32) prefix = node_key[:8] + # A paired node gets the whole node scope set, not just enough to heartbeat. + # + # It was read-only, and that made every node write path dead. `node_agent.py` + # already calls claim, renew, complete, fail and decompose, and no credential + # pairing could ever issue was permitted any of them: the agent looked ready + # and the endpoints looked finished, and the whole surface was unusable. The + # same shape of defect as a form with no method attribute - everything + # present, nothing working. + # + # Granting it here is right because the pairing code *is* the authorization: + # minted by an authenticated administrator, single-use, expiring, bound to one + # node name, and rate-limited per source address. An administrator who mints a + # code has already decided to put that machine on the fleet. + # + # It does not weaken the boundary that matters. `NODE_SCOPES` contains nothing + # that reads project content, so a node credential cannot become a session + # token under another name. And the grant is per credential, with a node able + # to hold several, so a deliberately monitoring-only machine is issued a narrow + # credential and this one revoked - still possible, rather than made + # impossible. + paired_scopes = sorted(NODE_SCOPES) connection.execute( """ INSERT INTO public.node_credentials @@ -829,7 +850,7 @@ class ComputeRepository: """, (node_id, prefix, self._passwords.hash_password(node_key), agent_version or "unknown", - ["compute:heartbeat", "compute:read"]), + paired_scopes), ) # The plaintext key exists in this dict and nowhere else. It is not logged, not # stored, and cannot be recovered afterwards. @@ -839,7 +860,7 @@ class ComputeRepository: "node_name": matched["node_name"], "key_prefix": prefix, "key": node_key, - "scopes": ["compute:heartbeat", "compute:read"], + "scopes": paired_scopes, } @@ -1303,11 +1324,41 @@ class ComputeRepository: candidates and choosing in Python, because between choosing and claiming another node may have taken the row - and a job placed on a node that cannot run it fails later, further from the cause. + + The node's own status is checked in the same transaction, and that placement is + deliberate. ``set_node_status`` promises that draining "stops new claims and lets + running work finish", and it did not: this method never looked at the node, so a + drained machine kept being handed work by an agent that was still running, and + the operator control did nothing except change a label. Checking outside the + transaction would leave a window in which an administrator drains a node between + the check and the claim, and the one job that slips through is exactly the one + that makes the behaviour look intermittent. + + ``effective_status`` rather than the stored column, so a node whose agent has + died stops being offered work instead of taking a lease it will never renew. + Note what is *not* consulted: ``is_schedulable``. That also requires a device + that is not busy, and claiming a job is what makes a device busy - gating on it + here would mean no node could ever claim anything. """ now = _now() expires = now + timedelta(seconds=lease_seconds) with self._connection() as connection, self._tx(connection): + # Locked, so an administrator draining this node cannot land between the + # check below and the claim. FOR UPDATE, not SKIP LOCKED: this row is not a + # candidate to be skipped, it is the subject of the transaction. + status_row = connection.execute( + "SELECT status, last_heartbeat_at FROM public.compute_nodes " + "WHERE id = %s FOR UPDATE", + (node_id,), + ).fetchone() + if status_row is None: + return None + if effective_status( + status_row["status"], status_row["last_heartbeat_at"], now=now + ) != "online": + return None + row = connection.execute( """ UPDATE public.jobs SET diff --git a/umbratrans_web/tests/test_compute.py b/umbratrans_web/tests/test_compute.py index 453899b..99e09d2 100644 --- a/umbratrans_web/tests/test_compute.py +++ b/umbratrans_web/tests/test_compute.py @@ -189,6 +189,169 @@ def test_online_is_not_set_by_an_administrator(compute, organization) -> None: assert "earned by a heartbeat" in str(error.value) +# ---------------------------------------------------------------------- +# A paired node can actually do the work the agent asks it to do +# ---------------------------------------------------------------------- + + +@integration +def test_a_node_paired_over_http_can_claim_and_complete_work(compute, organization) -> None: + """The whole node write path, end to end, on a credential pairing actually issues. + + Every layer of this already existed - the endpoints, the agent's calls to them, the + scope checks - and none of it was reachable, because pairing issued a read-only + credential. The individual tests above each pass in isolation; this one is what says + the path is connected. + + Written against the HTTP surface rather than the repository because the failure was + never in the repository: it was in what pairing granted, and only the request path + shows what a node actually holds. + """ + + import io + from contextlib import redirect_stderr + + from fastapi.testclient import TestClient + + from umbratrans_web.app import create_app + from umbratrans_web.config import Environment, Settings + from umbratrans_web.repository import InMemoryAccountRepository + + with redirect_stderr(io.StringIO()): + app = create_app( + settings=Settings(environment=Environment.TEST, require_csrf=True), + repository=InMemoryAccountRepository(), + ) + client = TestClient(app) + # The app would otherwise build a repository of its own from settings; point it at + # the one this test already has, so the HTTP requests and the assertions afterwards + # are talking about the same rows. + app.state.compute_repository = compute + + name = f"e2e-{uuid4().hex[:8]}" + minted = compute.create_pairing_code(organization, node_name=name) + paired = client.post( + "/api/node/pair", json={"code": minted["code"], "node_name": name, "agent_version": "test/1"} + ) + assert paired.status_code == 200, paired.text + key = paired.json()["key"] + bearer = {"Authorization": f"Bearer {key}"} + + # It cannot take work until it has spoken. A node that has never authenticated + # claiming a job is the case I24 exists to prevent, and the credential granting + # `compute:claim` must not become a way around it. + compute.submit_job(organization, kind="sleep", requested_by="test") + early = client.post("/api/node/jobs/claim", headers=bearer) + assert early.status_code == 200, early.text + assert early.json() is None, "a node that has never authenticated was handed work over HTTP" + + beat = client.post( + "/api/node/heartbeat", + headers=bearer, + json={ + "agent_version": "test/1", + "devices": [_device(_node(compute, organization, "e2e-device-holder"))], + }, + ) + assert beat.status_code == 200, beat.text + + claimed = client.post("/api/node/jobs/claim", headers=bearer) + assert claimed.status_code == 200, claimed.text + assert claimed.json() is not None, "a paired, heartbeating node could not claim work" + job_id = claimed.json()["job"]["id"] + + done = client.post( + f"/api/node/jobs/{job_id}/complete", headers=bearer, json={"result": {"ok": True}} + ) + assert done.status_code == 200, done.text + + events = [event["event"] for event in compute.job_events(UUID(job_id))] + assert "completed" in events, f"the completion is not in the job's trail: {events}" + + +# ---------------------------------------------------------------------- +# A node that has been switched off must not be handed work +# ---------------------------------------------------------------------- +# +# `set_node_status` documents that draining "stops new claims and lets running work +# finish", and the add-node page offers drain, return and retire as the operator's +# controls. Neither was true of the claim path: `claim_job` selected any queued row +# matching the requested kinds and never looked at the node at all. +# +# So the switches in the UI were decorative for the one thing they exist to do. An +# administrator drained a machine before rebooting it, and the agent - which had not +# gone away - kept taking new work, which is the opposite of what draining is for. +# +# Gated on the *effective* status, which is what `Node.status` already holds, so a node +# whose agent has died is also skipped: its leases would simply expire, and the work +# would be done twice. + + +@pytest.mark.parametrize("switched_to", ["draining", "retired", "offline"]) +@integration +def test_a_node_that_has_been_switched_off_is_handed_no_work( + compute, organization, switched_to: str +) -> None: + node = _online( + compute, + _node(compute, organization, f"switched-{switched_to}"), + [_device(_node(compute, organization, "d"))], + ) + compute.submit_job(organization, kind="sleep", requested_by="test") + compute.set_node_status(node.id, switched_to) + + claimed = compute.claim_job(node.id) + + assert claimed is None, ( + f"a node set to {switched_to!r} was handed a job. The administrator's control " + "does not work: the claim path never looks at the node's status." + ) + + +@integration +def test_the_job_a_switched_off_node_was_not_handed_is_still_claimable(compute, organization) -> None: + """Refusing to hand over work must not consume or lose it. + + Stated separately because "returns nothing" and "quietly took the job and dropped it" + both produce ``None`` at the call site, and only one of them is correct. + """ + + switched = _online( + compute, + _node(compute, organization, "switched-then-freed"), + [_device(_node(compute, organization, "device-holder-a"))], + ) + working = _online( + compute, + _node(compute, organization, "still-working"), + [_device(_node(compute, organization, "device-holder-b"))], + ) + compute.submit_job(organization, kind="sleep", requested_by="test") + compute.set_node_status(switched.id, "draining") + + assert compute.claim_job(switched.id) is None + claimed = compute.claim_job(working.id) + + assert claimed is not None, "the refused job vanished instead of staying queued" + assert claimed.job.status == RUNNING + + +@integration +def test_a_node_that_has_never_heartbeated_is_handed_no_work(compute, organization) -> None: + """``provisioning`` means the agent has not proved it can reach the controller. + + The same reasoning as I24, applied to the claim path rather than to the scheduler. + A node that has never spoken is the node most likely to swallow work silently. + """ + + node = _node(compute, organization, "never-spoke") + compute.submit_job(organization, kind="sleep", requested_by="test") + + assert compute.claim_job(node.id) is None, ( + "a node that has never authenticated was handed work" + ) + + # ---------------------------------------------------------------------- # I25: one job, one lease # ---------------------------------------------------------------------- diff --git a/umbratrans_web/tests/test_node_pairing.py b/umbratrans_web/tests/test_node_pairing.py index 3473715..9b66b15 100644 --- a/umbratrans_web/tests/test_node_pairing.py +++ b/umbratrans_web/tests/test_node_pairing.py @@ -29,7 +29,7 @@ from datetime import timedelta import pytest -from umbratrans_web.compute import PAIRING_ATTEMPT_BUDGET, ComputeRepository +from umbratrans_web.compute import NODE_SCOPES, PAIRING_ATTEMPT_BUDGET, ComputeRepository _SCRATCH = os.environ.get("UMBRATRANS_TEST_DSN", "").strip() _DISPOSABLE = os.environ.get("UMBRATRANS_TEST_DATABASE_IS_DISPOSABLE") == "1" @@ -128,12 +128,45 @@ def test_redeeming_mints_a_node_and_a_credential(pairing_world) -> None: assert result["ok"] is True, result assert result["key"], "no key was issued" assert result["key"].startswith(result["key_prefix"]) - # Read-only scopes: enough to prove it is alive, nothing more. - assert result["scopes"] == ["compute:heartbeat", "compute:read"] resolved = repository.resolve_node_credential(result["key"]) assert resolved is not None, "the issued credential does not resolve" assert resolved[0] == uuid.UUID(str(result["node_id"])) + # The scopes stored are the scopes the credential actually resolves to, so this + # compares the authority rather than the response's own claim about it. + assert sorted(resolved[2]) == sorted(result["scopes"]) + + +def test_a_paired_node_can_do_work_and_still_cannot_read_content(pairing_world) -> None: + """Both halves, because only one of them being right is a real failure. + + A node that cannot claim is not a compute node: the agent calls claim, complete and + decompose, so a read-only credential leaves the entire write surface dead while every + endpoint looks finished. A node that can read project content is a session token with + a different name, which is the boundary `NODE_SCOPES` exists to hold. + + Written as two assertions on behaviour rather than one equality against a literal + list, because the literal list is an implementation detail that a future scope would + change without either of these being wrong. + """ + + repository, organization = pairing_world + name = f"working-{uuid.uuid4().hex[:8]}" + minted = _mint(repository, organization, name) + result = repository.redeem_pairing_code(minted["code"], name, agent_version="test/1") + + granted = set(result["scopes"]) + + assert {"compute:claim", "compute:complete", "compute:heartbeat", "compute:read"} <= granted, ( + "a paired node cannot take or finish work, so the agent's whole write path is " + f"unusable. It was granted {sorted(granted)}." + ) + assert granted <= set(NODE_SCOPES), ( + f"a node credential carries scopes outside the node set: {sorted(granted - set(NODE_SCOPES))}" + ) + assert not any( + "project" in scope or "content" in scope or "read:all" in scope for scope in granted + ), f"a node credential can read content: {sorted(granted)}" def test_a_paired_node_starts_in_provisioning_not_online(pairing_world) -> None: diff --git a/umbratrans_web/tests/test_node_tenancy.py b/umbratrans_web/tests/test_node_tenancy.py index 31e1afe..8d013bd 100644 --- a/umbratrans_web/tests/test_node_tenancy.py +++ b/umbratrans_web/tests/test_node_tenancy.py @@ -131,6 +131,12 @@ def _queue(repository, organization, kind=OWN_KIND, tracked=None): def test_a_node_claims_work_for_another_organization(world) -> None: + # The node claims as a real one does: online because it heartbeated. Claiming + # worked from a freshly created node until `claim_job` began checking status, and + # these tests were relying on that - a node that has never authenticated taking work + # is the state I24 exists to prevent. The property under test here is the tenancy, + # which is unaffected by bringing the node online first. + """The case that made the change necessary. A node in tenant A claims a job belonging to tenant B. Before the change this @@ -138,7 +144,7 @@ def test_a_node_claims_work_for_another_organization(world) -> None: """ repository, node_org, work_org, queued = world - node = _node(repository, node_org) + node = _go_online(repository, _node(repository, node_org)) _queue(repository, work_org, tracked=queued) claimed = repository.claim_job(node.id, kinds=[OWN_KIND]) @@ -183,8 +189,8 @@ def test_two_nodes_in_different_tenants_both_get_work(world) -> None: import concurrent.futures repository, node_org, work_org, queued = world - first = _node(repository, node_org) - second = _node(repository, work_org) + first = _go_online(repository, _node(repository, node_org)) + second = _go_online(repository, _node(repository, work_org)) _queue(repository, work_org, tracked=queued) _queue(repository, work_org, tracked=queued) @@ -210,7 +216,7 @@ def test_a_node_still_filters_by_kind(world) -> None: """ repository, node_org, work_org, queued = world - node = _node(repository, node_org) + node = _go_online(repository, _node(repository, node_org)) _queue(repository, work_org, kind="translate", tracked=queued) _queue(repository, work_org, kind="speak", tracked=queued) @@ -222,7 +228,7 @@ def test_a_node_still_filters_by_kind(world) -> None: def test_a_node_filtered_by_kind_gets_nothing_when_none_match(world) -> None: repository, node_org, work_org, queued = world - node = _node(repository, node_org) + node = _go_online(repository, _node(repository, node_org)) _queue(repository, work_org, kind="translate", tracked=queued) assert repository.claim_job(node.id, kinds=["assemble"]) is None @@ -285,7 +291,7 @@ def test_a_claimed_job_reports_its_own_tenant_not_the_nodes(world) -> None: """The separation that matters: the node borrows the job, the job keeps its tenant.""" repository, node_org, work_org, queued = world - node = _node(repository, node_org) + node = _go_online(repository, _node(repository, node_org)) queued = _queue(repository, work_org, tracked=queued) claimed = repository.claim_job(node.id, kinds=[OWN_KIND])