api: reject state/result changes to done nodes

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
 

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

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