Pin the approved content digest on an endpoint command

Delivery descriptors for run_script and update_app are resolved when the
agent claims a command, not when an operator queues it, and the command
row recorded no digest: the enqueue path resolved the descriptor purely
to validate it and threw the result away.

So what an operator approved and what an endpoint runs were two separate
lookups against mutable content, minutes or hours apart. Anyone able to
replace a script body or an installer package could do so in the gap.
The claim then re-resolves, the descriptor faithfully describes the new
bytes, the agent verifies correctly, and it runs as LocalSystem. The
audit trail names the operator who queued the original.

The gate on the verb and the gate on the payload were also different
grants: pushing needs endpoints.deploy, but replacing what gets pushed
needs only applications.edit.

pinnedsha256 records the digest as it stood at enqueue, and the claim
refuses a descriptor that has moved, failing the push with an audit row
instead of shipping it. Null for actions that deliver no content, and
for rows queued before this migration, which claim as before. A
descriptor that vanished entirely is not treated as tampering: that path
already sends the command through undescribed so the agent fails closed.

Separately, a result may now only finalize a command that is actually
running. That needs no agent change and closes two things: reporting on
a command that was never claimed, and overwriting a recorded failure
with a success.

The agent now sends its hostname on the result callback so the server
can check it owns the command it is finalizing. The shared collector key
does not say who is calling, so without it any agent can post a result
for any command id. The server treats the parameter as optional for
agents that predate this and enforces it when present, so there is no
flag day. This is a floor, not a fix: only a per-endpoint credential
makes the caller's identity provable, and that is tracked separately.

The .NET side is unbuilt here and needs a build plus a win11 VM run.

Scope: product
(cherry picked from commit d64b7122ab)

CHANGELOG.md

@@ -167,6 +167,21 @@ ADR-007 and ADR-002.
   set `TRUSTED_PROXY_COUNT`** or its whole floor shares one rate-limit bucket;
   INSTALL-WINDOWS-IIS.md now says so.
 
+- **What an operator approved and what an endpoint ran were two different
+  lookups.** Delivery descriptors for `run_script` and `update_app` are resolved
+  when the agent claims a command, not when the command is queued, and the
+  command row recorded no digest - the enqueue path resolved the descriptor
+  purely to validate it and threw the result away. Anyone able to replace a
+  script body or an installer package could do so in the gap, and the
+  re-resolved descriptor would then faithfully describe the new bytes: the agent
+  verifies correctly and runs something nobody approved, with the audit trail
+  naming the operator who queued the original. The approved digest is now pinned
+  on the command at enqueue and the claim refuses a descriptor that no longer
+  matches, failing the push instead of shipping it. Separately, an agent may
+  only finalize a command that is actually running, so a result cannot be posted
+  for a command that was never claimed nor overwrite a recorded failure with a
+  success.
+
 - **The published site's API documentation page had no spec to show.**
   `shopdb/core/api/docs.py` reads `docs/openapi.json` at runtime to serve
   `/api/docs`, and publication excluded the whole of `docs/` because the prose

agent/ShopDbEndpointAgent.Core/CommandClient.cs

@@ -64,10 +64,17 @@ public sealed class CommandClient : IDisposable
     // result is logged and swallowed so a transient network blip does not crash
     // the worker; the command simply stays 'running' server-side until a later
     // operator retry supersedes it.
+    // hostname rides along so the server can check this agent owns the command
+    // it is finalizing. The shared collector key does not say who is calling,
+    // so without it any agent can post a result for any command id. The server
+    // treats the parameter as optional for older agents and enforces it when
+    // present, so shipping this ahead of a fleet update is safe.
     public async Task ReportResultAsync(
-        int commandId, CommandResult result, CancellationToken token)
+        int commandId, string hostname, CommandResult result,
+        CancellationToken token)
     {
-        var url = $"api/endpoints/commands/{commandId}/result";
+        var url = $"api/endpoints/commands/{commandId}/result?hostname="
+            + Uri.EscapeDataString(hostname);
         using var response = await http.PostAsJsonAsync(
             url, result, JsonOptions, token);
         if (!response.IsSuccessStatusCode)

agent/ShopDbEndpointAgent.Core/CommandWorker.cs

@@ -130,7 +130,8 @@ public sealed class CommandWorker : BackgroundService
             };
         }
 
-        await client.ReportResultAsync(claim.EndpointCommandId, result, token);
+        await client.ReportResultAsync(
+            claim.EndpointCommandId, hostname, result, token);
         logger.LogInformation("Command {Id} finished succeeded={Succeeded}",
             claim.EndpointCommandId, result.Succeeded);
 

docs/PROJECT-MAP.md

@@ -12,7 +12,7 @@ never by editing this file.
 | series | value | governed by |
 |---|---|---|
 | product `__version__` | `0.12.0` | ADR-007 |
-| plugin contract `__contract_version__` | `0.27.0` | ADR-002 |
+| plugin contract `__contract_version__` | `0.28.0` | ADR-002 |
 
 They move independently. A contract bump is not a release.
 
@@ -27,7 +27,7 @@ with `flask plugin upgrade-all`. Both are needed on a deploy.
 | backups | `backups0003clearlastseen` |
 | computers | `computers0004policies` |
 | employees | `employees0002photo` |
-| endpoints | `endpoints0004apps` |
+| endpoints | `endpoints0005cmdpin` |
 | geenforce | `geenforce0004minlib` |
 | guides | `guides0001baseline` |
 | knowledgebase | `knowledgebase0001anchor` |
@@ -90,6 +90,6 @@ Manifest-less directories under `plugins/` are core frontend surface and always
 
 ## Size
 
-- test functions defined: **1544** (parametrised cases collect higher)
+- test functions defined: **1560** (parametrised cases collect higher)
 - documented API paths: **345** (`docs/openapi.json`, regenerate with `scripts/gen_openapi.py`)
 

plugins/endpoints/api/routes.py

@@ -7,6 +7,7 @@ These routes are the human-facing read surface, gated by user JWT + RBAC.
 """
 
 import hashlib
+import hmac
 import json
 import os
 import time
@@ -941,6 +942,43 @@ _CLAIM_HOLD_SECONDS = 25
 _CLAIM_POLL_INTERVAL = 1.0
 
 
+def _pin_holds(command, descriptor):
+    """Does the descriptor still carry the digest this command was queued for?
+
+    True when nothing was pinned (actions that deliver no content), or when the
+    descriptor is missing entirely - a vanished script or version already sends
+    the command through with no descriptor so the agent fails closed, and that
+    path is not a content swap.
+    """
+    if not command.pinnedsha256:
+        return True
+    if descriptor is None:
+        return True
+    return hmac.compare_digest(
+        str(descriptor.get('sha256') or ''), command.pinnedsha256)
+
+
+def _fail_on_pin_drift(command, action):
+    """Refuse a claim whose content changed after the operator approved it."""
+    command.status = EndpointCommand.STATUS_FAILED
+    command.resultmessage = (
+        'Refused: the {} content changed after this command was queued'
+        .format(action))
+    command.completedat = _naive_utcnow()
+    db.session.commit()
+    current_app.logger.warning(
+        'Endpoint command %s (%s, asset %s) refused: pinned sha256 %s no '
+        'longer matches the resolved descriptor',
+        command.endpointcommandid, action, command.assetid,
+        command.pinnedsha256)
+    AuditLog.log('refused', 'EndpointCommand',
+                 entityid=command.endpointcommandid,
+                 entityname=action,
+                 details='content digest changed after enqueue')
+    db.session.commit()
+    return ('', 204)
+
+
 def _reap_stale_commands(asset_id):
     """Expire in-flight commands an agent will never finish for this endpoint.
 
@@ -1234,18 +1272,25 @@ def enqueue_endpoint_command(asset_id: int):
     # update_app pins a specific application version; verify it resolves to a
     # real installer now so a broken push is rejected at enqueue rather than
     # failing opaquely on the endpoint minutes later.
+    # Digest of the content this push was APPROVED against. Kept on the row so
+    # the claim path can refuse a descriptor that has changed since; see
+    # EndpointCommand.pinnedsha256.
+    pinnedsha256 = None
+
     if action == 'update_app':
-        _descriptor, resolve_error = _resolve_update_app(target)
+        descriptor, resolve_error = _resolve_update_app(target)
         if resolve_error is not None:
             return error_response(ErrorCodes.VALIDATION_ERROR, resolve_error)
+        pinnedsha256 = (descriptor or {}).get('sha256')
 
     # run_script pins a ShopDB-hosted script; verify it resolves to a real
     # uploaded body now so a broken push is rejected at enqueue rather than
     # failing opaquely on the endpoint later.
     if action == 'run_script':
-        _script, script_error = _resolve_run_script(target)
+        script, script_error = _resolve_run_script(target)
         if script_error is not None:
             return error_response(ErrorCodes.VALIDATION_ERROR, script_error)
+        pinnedsha256 = (script or {}).get('sha256')
 
     state = EndpointState.query.filter_by(assetid=asset_id).first()
     if state is None:
@@ -1277,6 +1322,7 @@ def enqueue_endpoint_command(asset_id: int):
         targetkind=targetkind if target else None,
         status=EndpointCommand.STATUS_PENDING,
         requestedby=getattr(current_user, 'userid', None),
+        pinnedsha256=pinnedsha256,
     )
     db.session.add(command)
 
@@ -1337,6 +1383,8 @@ def claim_next_command():
             # fail closed rather than 500 the poll.
             if command.action == 'update_app':
                 descriptor, _resolve_error = _resolve_update_app(command.target)
+                if not _pin_holds(command, descriptor):
+                    return _fail_on_pin_drift(command, 'update_app')
                 if descriptor is not None:
                     claim['app'] = descriptor
             # run_script rides a resolved script descriptor: the agent gets a
@@ -1347,6 +1395,8 @@ def claim_next_command():
             # agent fail closed rather than 500 the poll.
             if command.action == 'run_script':
                 script_descriptor, _script_error = _resolve_run_script(command.target)
+                if not _pin_holds(command, script_descriptor):
+                    return _fail_on_pin_drift(command, 'run_script')
                 if script_descriptor is not None:
                     claim['script'] = script_descriptor
             return success_response(claim)
@@ -1376,6 +1426,33 @@ def report_command_result(command_id: int):
         return error_response(ErrorCodes.NOT_FOUND,
                               f'No command {command_id}', http_code=404)
 
+    # An agent may only finalize ITS OWN command. The collector key is shared
+    # fleet-wide, so without this any agent could post a result for any command
+    # id - marking an install succeeded on a machine that never ran it, which
+    # writes a false compliance record. Binding to the claimed hostname is a
+    # floor, not a fix: the credential still does not prove which endpoint is
+    # calling, and only a per-endpoint credential closes that.
+    # Optional because the shipped agent does not send it yet; enforced when it
+    # does, so an updated fleet gets the check without a flag day.
+    hostname = (request.args.get('hostname') or '').strip()
+    if hostname:
+        caller_asset_id = _resolve_asset_for_hostname(hostname)
+        if caller_asset_id != command.assetid:
+            return error_response(
+                ErrorCodes.FORBIDDEN,
+                'This command belongs to another endpoint', http_code=403)
+
+    # Only a claimed command can be finalized. Needs no agent change, and it
+    # closes two things the hostname check cannot: reporting on a command that
+    # was never claimed, and overwriting a terminal result - which would let a
+    # recorded failure be replaced by a success.
+    if command.status != EndpointCommand.STATUS_RUNNING:
+        return error_response(
+            ErrorCodes.VALIDATION_ERROR,
+            'Command {} is {}, not running'.format(
+                command_id, command.status),
+            http_code=409)
+
     payload = request.get_json(silent=True) or {}
     succeeded = bool(payload.get('succeeded'))
     command.status = (EndpointCommand.STATUS_SUCCEEDED if succeeded

plugins/endpoints/migrations/versions/0005_endpoint_command_pinnedsha256.py

@@ -0,0 +1,51 @@
+"""Pin the approved content digest on an endpoint command.
+
+Delivery descriptors (run_script, update_app) are re-resolved when the agent
+claims a command, not when an operator queues it. Between those two moments the
+underlying script body or installer package can be replaced by anyone who can
+edit the applications catalog, and the descriptor's SHA-256 would then describe
+the NEW bytes: the agent verifies correctly and runs something the operator
+never approved.
+
+pinnedsha256 records the digest as it stood at enqueue. The claim path compares
+and refuses the command if it has moved, so a swap fails the push instead of
+shipping it.
+
+Null for actions that deliver no content, and null on rows queued before this
+migration - those are treated as unpinned and claim as before.
+
+Idempotent: checks the live schema first so a db.create_all() site re-runs clean
+on strict MySQL 8.
+"""
+from alembic import op
+
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'endpoints0005cmdpin'
+down_revision = 'endpoints0004apps'
+branch_labels = None
+depends_on = None
+
+
+def _has_column(table_name, column_name):
+    from sqlalchemy import inspect
+
+    inspector = inspect(op.get_bind())
+    if not inspector.has_table(table_name):
+        return False
+    return any(col['name'] == column_name
+               for col in inspector.get_columns(table_name))
+
+
+def upgrade():
+    if not _has_column('endpointcommands', 'pinnedsha256'):
+        op.add_column('endpointcommands',
+                      sa.Column('pinnedsha256', sa.String(length=64),
+                                nullable=True))
+
+
+def downgrade():
+    if _has_column('endpointcommands', 'pinnedsha256'):
+        op.drop_column('endpointcommands', 'pinnedsha256')

plugins/endpoints/models/endpoint.py

@@ -172,6 +172,19 @@ class EndpointCommand(db.Model):
     # Who queued it (userid) for audit; the agent never sees this.
     requestedby = db.Column(db.Integer, nullable=True)
 
+    # The content digest as it stood WHEN AN OPERATOR APPROVED THIS PUSH.
+    #
+    # Delivery descriptors are re-resolved at claim time, minutes or hours
+    # later, so without this the bytes an operator approved and the bytes the
+    # agent runs are two different lookups against mutable content. Whoever can
+    # replace a script body or an installer package could swap it in the gap
+    # and the digest would faithfully describe the new bytes.
+    #
+    # Set at enqueue for run_script and update_app; null for actions that
+    # deliver no content. The claim path refuses a command whose descriptor no
+    # longer matches, so a swap fails the command instead of shipping.
+    pinnedsha256 = db.Column(db.String(64), nullable=True)
+
     # Agent-reported outcome. resultjson holds structured detail (for example
     # the observed MTConnect running state); resultmessage is a short summary.
     resultmessage = db.Column(db.Text, nullable=True)

tests/test_plugin_migrations.py

@@ -94,7 +94,7 @@ EXPECTED_HEAD_REVISION['warranty'] = 'warranty0002proof'
 # adds the endpointcommands queue plus the MTConnect state columns, 0003
 # adds the generic command target/targetkind columns, and 0004 adds the
 # endpointapplications SBOM snapshot table.
-EXPECTED_HEAD_REVISION['endpoints'] = 'endpoints0004apps'
+EXPECTED_HEAD_REVISION['endpoints'] = 'endpoints0005cmdpin'
 
 # Plugins built after the cutover: their 0001 baseline really creates tables the
 # core chain never owned.

tests/test_plugins/test_endpoint_delivery_descriptor.py

@@ -79,3 +79,73 @@ def test_a_package_with_no_recorded_digest_is_refused_not_delivered_unverified(a
 
     assert descriptor is None
     assert 'SHA-256' in error
+
+
+# ---------------------------------------------------------------------------
+# The digest is pinned when an operator approves the push, not when the agent
+# claims it. Between those two moments the bytes behind a descriptor can be
+# replaced by anyone who can edit the catalog, and the re-resolved descriptor
+# would faithfully describe the NEW bytes.
+# ---------------------------------------------------------------------------
+
+def test_an_unpinned_command_still_claims(app):
+    """Actions that deliver no content, and rows queued before the pin existed."""
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='mtconnect_start',
+                                  pinnedsha256=None)
+        assert _pin_holds(command, None) is True
+        assert _pin_holds(command, {'sha256': 'b' * 64}) is True
+
+
+def test_a_matching_digest_claims(app):
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='run_script',
+                                  pinnedsha256='c' * 64)
+        assert _pin_holds(command, {'sha256': 'c' * 64}) is True
+
+
+def test_a_swapped_digest_does_not_claim(app):
+    """The finding this exists for: approve one script, ship another.
+
+    An operator queues run_script against a reviewed body. Before the agent's
+    long poll returns, somebody with applications.edit replaces the body and
+    the digest is restamped. Without the pin the claim re-resolves, hands the
+    agent the attacker's digest, and the agent verifies it correctly.
+    """
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='run_script',
+                                  pinnedsha256='c' * 64)
+        assert _pin_holds(command, {'sha256': 'd' * 64}) is False
+
+
+def test_a_vanished_descriptor_is_not_treated_as_a_swap(app):
+    """A deleted script already sends the command through with no descriptor so
+    the agent fails closed; that path must not be reported as tampering."""
+    from plugins.endpoints.api.routes import _pin_holds
+    from plugins.endpoints.models import EndpointCommand
+
+    with app.app_context():
+        command = EndpointCommand(assetid=1, action='run_script',
+                                  pinnedsha256='c' * 64)
+        assert _pin_holds(command, None) is True
+
+
+def test_enqueue_records_the_digest_it_approved(app, hosted_app):
+    """The resolved descriptor used to be discarded at enqueue (`_descriptor`)."""
+    from plugins.endpoints.api.routes import _resolve_update_app
+
+    appid, appversionid = hosted_app
+    with app.app_context():
+        descriptor, error = _resolve_update_app(str(appversionid))
+        assert error is None
+        # This is the value the enqueue path now writes to pinnedsha256.
+        assert descriptor['sha256'] == 'a' * 64