diff --git a/api/main.py b/api/main.py index be3a1adc..7e161b86 100644 --- a/api/main.py +++ b/api/main.py @@ -1827,12 +1827,100 @@ def is_same_flags(old_node, new_node): return False +# Rejected (or admin-forced) changes to completed nodes are published on this +# channel, so they can be monitored via /events?channel=node-integrity. +NODE_INTEGRITY_CHANNEL = "node-integrity" + + +def _done_node_update_allowed(existing: Node, new_state, new_result) -> bool: + """Check if an update may be applied to a node that is already done""" + if existing.state != "done": + return True + if new_state != "done": + return False + if new_result == existing.result: + return True + # A node closed without a final result (e.g. by node_timeout) can still + # receive the late result from its lab. + return existing.result in (None, "incomplete") + + +async def _check_done_node_update( + existing: Node, new_state, new_result, user: User, force: Optional[bool] +): + """Reject state/result changes to done nodes and report every attempt + + The first final result wins: once a node is done, its state and result + can only be changed by an admin with force=true. Both rejected and + forced changes are published on NODE_INTEGRITY_CHANNEL. + """ + # PATCH passes StateValues/ResultValues enums, PUT plain strings + new_state = getattr(new_state, "value", new_state) + new_result = getattr(new_result, "value", new_result) + if _done_node_update_allowed(existing, new_state, new_result): + return + forced = bool(force and user.is_superuser) + op = "done_update_forced" if forced else "done_update_rejected" + metrics.add(f"node_{op}_total", 1) + data = { + "op": op, + "id": str(existing.id), + "kind": existing.kind, + "name": existing.name, + "path": existing.path, + "group": existing.group, + "owner": existing.owner, + "runtime": (existing.data or {}).get("runtime"), + "platform": (existing.data or {}).get("platform"), + "state": existing.state, + "result": existing.result, + "attempted_state": new_state, + "attempted_result": new_result, + "user": user.username, + } + try: + await pubsub.publish_cloudevent( + NODE_INTEGRITY_CHANNEL, data, {"owner": existing.owner} + ) + except Exception as exc: + # Reporting must not turn the rejection into a server error + print(f"Failed to publish {op} event for {existing.id}: {exc}") + if not forced: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=( + f"Node {existing.id} is already done with result " + f"'{existing.result}', refusing to change it to " + f"state '{new_state}' result '{new_result}'" + ), + ) + + +async def _check_done_hierarchy_update( + hierarchy: Hierarchy, root: Node, user: User, force: Optional[bool] +): + """Run _check_done_node_update on every existing node of a hierarchy""" + node = hierarchy.node + existing = root + if existing is None and node.id: + existing = await db.find_by_id(Node, node.id) + if existing is not None: + await _check_done_node_update( + existing, node.state, node.result, user, force + ) + for child in hierarchy.child_nodes: + await _check_done_hierarchy_update(child, None, user, force) + + @app.put("/node/{node_id}", response_model=Node, response_model_by_alias=False) async def put_node( node_id: str, node: Node, user: str = Depends(authorize_user), noevent: Optional[bool] = Query(None), + force: Optional[bool] = Query( + None, description="Admin only: allow changing a done node" + ), ): """Update an already added node""" metrics.add("http_requests_total", 1) @@ -1847,6 +1935,10 @@ async def put_node( # [TODO] Remove translation below once we can use it in the pipeline node = _translate_version_fields(node) + await _check_done_node_update( + node_from_id, node.state, node.result, user, force + ) + # Sanity checks # Note: do not update node ownership fields, don't update 'state' # until we've checked the state transition is valid. @@ -1913,6 +2005,9 @@ async def patch_node( patch: NodePatchRequest, user: str = Depends(authorize_user), noevent: Optional[bool] = Query(None), + force: Optional[bool] = Query( + None, description="Admin only: allow changing a done node" + ), ): """Partial update of an existing node""" metrics.add("http_requests_total", 1) @@ -1930,6 +2025,14 @@ async def patch_node( detail="No fields to update", ) + await _check_done_node_update( + node_from_id, + update_data.get("state", node_from_id.state), + update_data.get("result", node_from_id.result), + user, + force, + ) + # Handle state transition separately new_state = update_data.pop("state", None) @@ -2041,6 +2144,9 @@ async def put_nodes( nodes: Hierarchy, authorization: str | None = Header(default=None), user: str = Depends(authorize_user), + force: Optional[bool] = Query( + None, description="Admin only: allow changing done nodes" + ), ): """Add a hierarchy of nodes to an existing root node""" metrics.add("http_requests_total", 1) @@ -2052,6 +2158,8 @@ async def put_nodes( status_code=status.HTTP_404_NOT_FOUND, detail=f"Node not found with id: {node_id}", ) + # Check the whole hierarchy before writing any of it + await _check_done_hierarchy_update(nodes, node_from_id, user, force) submitter = node_from_id.submitter treeid = node_from_id.treeid diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index e73e6d90..20506b5a 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -89,6 +89,7 @@ def mock_get_current_admin_user(request: Request): # Mock dependency callables for getting current user app.dependency_overrides[get_current_user] = mock_get_current_user app.dependency_overrides[get_current_superuser] = mock_get_current_admin_user +versioned_app.dependency_overrides.update(app.dependency_overrides) @pytest.fixture(scope="session") diff --git a/tests/unit_tests/test_node_done_update.py b/tests/unit_tests/test_node_done_update.py new file mode 100644 index 00000000..dc3108f6 --- /dev/null +++ b/tests/unit_tests/test_node_done_update.py @@ -0,0 +1,206 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later + +"""Unit tests for rejecting state/result changes to done nodes""" + +import json + +import pytest +from fastapi import HTTPException +from kernelci.api.models import Node + +from api.main import NODE_INTEGRITY_CHANNEL, _check_done_node_update +from api.models import User +from tests.unit_tests.conftest import BEARER_TOKEN + +NODE_ID = "6ae228345c0836a8c2f041a8" +HEADERS = {"Accept": "application/json", "Authorization": BEARER_TOKEN} + + +def _make_node(state="done", result="pass", node_id=NODE_ID): + return Node( + id=node_id, + kind="job", + name="baseline-arm64", + path=["checkout", "kbuild-gcc-14-arm64", "baseline-arm64"], + group="baseline-arm64", + parent="6ae2138e5c0836a8c2eeeac6", + state=state, + result=result, + owner="bob", + data={"runtime": "sasha-lab", "platform": "jetson-tx1-devkit"}, + ) + + +def _make_user(is_superuser=False): + return User( + id="65265305c74695807499037f", + username="bob", + hashed_password="hash", + email="bob@kernelci.org", + is_active=True, + is_superuser=is_superuser, + is_verified=True, + ) + + +def _node_payload(node, **update): + payload = json.loads(node.model_dump_json()) + payload.update(update) + return payload + + +@pytest.fixture +def mock_integrity_publish(mocker): + """Capture events published by the done-node guard""" + pubsub = mocker.AsyncMock() + mocker.patch("api.main.pubsub", pubsub) + return pubsub.publish_cloudevent + + +def test_put_node_rejects_result_flip( + mock_db_find_by_id, mock_db_update, mock_integrity_publish, test_client +): + """PUT changing pass to fail on a done node is rejected and reported""" + node = _make_node() + mock_db_find_by_id.return_value = node + + response = test_client.put( + f"node/{NODE_ID}", + headers=HEADERS, + json=_node_payload(node, result="fail"), + ) + + assert response.status_code == 409 + mock_db_update.assert_not_called() + mock_integrity_publish.assert_awaited_once() + channel, data, _ = mock_integrity_publish.await_args.args + assert channel == NODE_INTEGRITY_CHANNEL + assert data["op"] == "done_update_rejected" + assert data["id"] == NODE_ID + assert data["result"] == "pass" + assert data["attempted_result"] == "fail" + assert data["runtime"] == "sasha-lab" + + +def test_put_node_allows_same_result( + mock_db_find_by_id, + mock_db_update, + mock_publish_cloudevent, + mock_integrity_publish, + test_client, +): + """PUT keeping the result of a done node (e.g. adding data) is allowed""" + node = _make_node() + mock_db_find_by_id.return_value = node + mock_db_update.return_value = node + + response = test_client.put( + f"node/{NODE_ID}", + headers=HEADERS, + json=_node_payload(node), + ) + + assert response.status_code == 200 + channels = [c.args[0] for c in mock_integrity_publish.await_args_list] + assert NODE_INTEGRITY_CHANNEL not in channels + + +def test_put_node_allows_late_result_after_timeout( + mock_db_find_by_id, mock_db_update, mock_integrity_publish, test_client +): + """A node closed as incomplete can still get its final result""" + node = _make_node(result="incomplete") + mock_db_find_by_id.return_value = node + mock_db_update.return_value = _make_node(result="fail") + + response = test_client.put( + f"node/{NODE_ID}", + headers=HEADERS, + json=_node_payload(node, result="fail"), + ) + + assert response.status_code == 200 + channels = [c.args[0] for c in mock_integrity_publish.await_args_list] + assert NODE_INTEGRITY_CHANNEL not in channels + + +def test_patch_node_rejects_result_flip( + mock_db_find_by_id, mock_db_update, mock_integrity_publish, test_client +): + """PATCH changing the result of a done node is rejected""" + mock_db_find_by_id.return_value = _make_node() + + response = test_client.patch( + f"node/{NODE_ID}", + headers=HEADERS, + json={"result": "fail"}, + ) + + assert response.status_code == 409 + mock_db_update.assert_not_called() + mock_integrity_publish.assert_awaited_once() + + +def test_put_nodes_rejects_hierarchy_overwrite( + mocker, mock_db_find_by_id, mock_integrity_publish, test_client +): + """A hierarchy re-submitted with a different root result is rejected""" + node = _make_node() + mock_db_find_by_id.return_value = node + create_hierarchy = mocker.patch( + "api.db.Database.create_hierarchy", side_effect=mocker.AsyncMock() + ) + + response = test_client.put( + f"nodes/{NODE_ID}", + headers=HEADERS, + json={ + "node": _node_payload(node, result="fail"), + "child_nodes": [], + }, + ) + + assert response.status_code == 409 + create_hierarchy.assert_not_called() + mock_integrity_publish.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_force_by_admin_is_allowed_and_reported(mock_integrity_publish): + """An admin can force the change; it is still reported""" + await _check_done_node_update( + _make_node(), "done", "fail", _make_user(is_superuser=True), True + ) + + mock_integrity_publish.assert_awaited_once() + assert ( + mock_integrity_publish.await_args.args[1]["op"] == "done_update_forced" + ) + + +@pytest.mark.asyncio +async def test_force_by_regular_user_is_rejected(mock_integrity_publish): + """force=true is ignored for non-admin users""" + with pytest.raises(HTTPException) as exc: + await _check_done_node_update( + _make_node(), "done", "fail", _make_user(), True + ) + + assert exc.value.status_code == 409 + assert ( + mock_integrity_publish.await_args.args[1]["op"] + == "done_update_rejected" + ) + + +@pytest.mark.asyncio +async def test_publish_failure_still_rejects(mock_integrity_publish): + """A failing event publish must not turn the 409 into a 500""" + mock_integrity_publish.side_effect = RuntimeError("redis down") + + with pytest.raises(HTTPException) as exc: + await _check_done_node_update( + _make_node(), "done", "fail", _make_user(), None + ) + + assert exc.value.status_code == 409