ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add one-click agent self-upgrade from the endpoint detail page

Browse Source 
An operator could see that an endpoint was behind the published build but
had no way to act on it. update_app resolves against Application rows, and
the agent lives in the version store, so there was nothing to push.

Bridge the two: the staged build is packaged and registered as a reserved
Application row, and a managed PowerShell script is seeded to install it.
The endpoint downloads that package over the existing collector-authenticated
app-package route, so no parallel delivery path is introduced.

The new route prepares the delivery and then DELEGATES to
enqueue_endpoint_command as a digest-pinned run_script. It deliberately does
not implement its own enqueue: a second path would duplicate, and eventually
drift from, the endpoint lock, the Configuration Policy gate, stale-command
reaping and the at-most-one-in-flight rule. A fleet-wide agent swap is the
last action that should skip a quarantine check. Tests pin that delegation
rather than only the happy path.

Because it lands as run_script, the caller needs endpoints.runscript on top
of endpoints.agentupdate. That is intended: this delivers arbitrary SYSTEM
PowerShell by the same mechanism as any other script push. The audit event
is written only on an actual 201, so a refused push does not record an
upgrade that never happened.

Replacing the agent is self-modifying the control plane - the process
receiving the command is the one being swapped. The script therefore
re-launches itself detached and exits immediately, because the agent runs
run_script as a direct child and holds its stdout/stderr pipes while
waiting on it. Expect the command to report queued rather than success:
the process that would report completion is stopped mid-run. Judge the
result by the version the endpoint reports afterward.

Tray relaunch does not use the installer's `explorer.exe <path>` trick.
That works in install-shopdbagent.ps1 only because the installer runs as
the user - note its own `-ne "SYSTEM"` guard. This script always runs as
SYSTEM in session 0, where the hand-off reaches no shell and exits 0, so
Start-Process reported success for a relaunch that never happened. It also
cannot descend into the session with a token-duplicating launch:
UserContextSpool documents that WTSQueryUserToken is blocked by endpoint
protection on these hosts. Instead a one-shot scheduled task targets the
pinned ShopFloor account, matching how the user-context spool already
resolves that user, and the tray must run as that account or the
mapped-drive probe reports the wrong (empty) drive table. The relaunch is
verified by polling for the process rather than assumed, and stays
best-effort: the Run key still starts the tray at next logon, and a late
tray is not worth rolling back a good agent upgrade.

panel_status now carries agentbehind/agentversion/agentcurrent so the UI
gates the action on a flag rather than parsing "(behind X)" out of a label.
The button is confirmed by typing UPDATE, naming both versions and the
service interruption, matching how run_script and system_restart are gated.

Verified on live endpoints: upgraded 1.13.0.0 and 1.15.0.0 machines to
1.16.0.0. run_script delivery, digest fetch and system context all predate
1.13.0.0, so no minimum agent version is declared for this path.
This commit is contained in:
benjamin.baker
2026-09-09 22:56:21 -04:00
parent d32293ccb2
commit 8be1c2379f
8 changed files with 957 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
tests/test_plugins/test_endpoints_agent_upgrade_route.py Normal file
View File

@@ -0,0 +1,281 @@
"""The agent self-upgrade enqueue route.
Replacing the agent is self-modifying the control plane: the process that
receives the command is the one being swapped. The route therefore sits behind
endpoints.agentupdate, and - more importantly - it does NOT implement its own
enqueue. It prepares the delivery and delegates to enqueue_endpoint_command so
the endpoint lock, the Configuration Policy gate, stale-command reaping and the
at-most-one-in-flight rule all still apply.
These tests pin that delegation. A parallel enqueue path that quietly skipped a
lock or the in-flight rule would be invisible in normal use and would show up
as a fleet-wide agent swap landing on a quarantined machine.
The store is isolated per test: create_app('testing') resolves instance_path to
the real repo instance/, so without redirection these would stage binaries into
the developer's actual agent store.
"""
import os
import tempfile
import pytest
from werkzeug.security import generate_password_hash
from plugins.endpoints.api import agentstore, agentregistry, agentupgradescript
UPGRADE_URL = '/api/endpoints/{}/agentupgrade'
_PE_BYTES = b'MZ' + b'\x00' * 4096
@pytest.fixture(autouse=True)
def isolate_store(tmp_path, monkeypatch):
"""Redirect the agent store, package output and script bodies."""
siteagent = tmp_path / 'siteagent'
packages = tmp_path / 'applicationpackages'
scripts = tmp_path / 'scripts'
for directory in (siteagent, packages, scripts):
directory.mkdir()
monkeypatch.setattr(agentstore, '_siteagent_dir', lambda: str(siteagent))
monkeypatch.setattr(agentregistry, 'application_package_location',
lambda: (str(packages), '/api/applications/package/'))
monkeypatch.setattr(agentupgradescript, 'script_body_location',
lambda: (str(scripts), '/api/scripts/body/'))
return tmp_path
def _publish_agent(version='1.16.0.0'):
"""Stage a build so there is something to upgrade to."""
agentstore.store_version('agent', version, _PE_BYTES, make_current=True)
def _endpoint(db, hostname, agentversion='1.15.0.0'):
from shopdb.api import Asset, AssetType
from plugins.computers.models import Computer
from plugins.endpoints.models import EndpointState
assettype = AssetType.query.filter_by(assettype='computer').first()
if assettype is None:
assettype = AssetType(assettype='computer', pluginname='computers',
tablename='computers')
db.session.add(assettype)
db.session.flush()
asset = Asset(assetnumber=hostname, name=hostname,
assettypeid=assettype.assettypeid, isactive=True)
db.session.add(asset)
db.session.flush()
db.session.add(Computer(assetid=asset.assetid, hostname=hostname))
db.session.add(EndpointState(assetid=asset.assetid, hostname=hostname,
agentversion=agentversion))
db.session.commit()
return asset.assetid
def _login(db, client, username, permissions=None, admin=False):
"""A user holding exactly the named permissions (or the admin role)."""
from shopdb.core.models import User, Role, Permission
Permission.seed()
db.session.flush()
rolename = 'admin' if admin else username + 'role'
role = Role.query.filter_by(rolename=rolename).first()
if role is None:
role = Role(rolename=rolename)
db.session.add(role)
db.session.flush()
if not admin:
for name in (permissions or []):
perm = Permission.query.filter_by(name=name).first()
if perm is not None and perm not in role.permissions:
role.permissions.append(perm)
db.session.commit()
user = User(username=username, email=username + '@test.local',
passwordhash=generate_password_hash('testpass'))
user.roles.append(role)
db.session.add(user)
db.session.commit()
response = client.post('/api/auth/login',
json={'username': username, 'password': 'testpass'})
token = response.get_json()['data']['access_token']
return {'Authorization': f'Bearer {token}'}
def test_an_upgrade_queues_as_a_pinned_run_script(client, db, app):
"""The push must ride the normal script channel, digest-pinned.
Landing as run_script is what makes the body verifiable on the endpoint;
a bespoke action would need its own pinning and would not get it.
"""
from plugins.endpoints.models import EndpointCommand
_publish_agent()
assetid = _endpoint(db, 'UPGRADE01')
headers = _login(db, client, 'upgradeadmin', admin=True)
response = client.post(UPGRADE_URL.format(assetid), json={}, headers=headers)
assert response.status_code == 201
command = EndpointCommand.query.filter_by(assetid=assetid).first()
assert command is not None
assert command.action == 'run_script'
seeded = agentupgradescript.seed_upgrade_script()
assert command.target == 'script:{}'.format(seeded['scriptid'])
# Pinned to the exact body approved at enqueue, so a later edit cannot
# swap what the endpoint runs.
assert command.pinnedsha256 == seeded['sha256']
def test_the_upgrade_requires_the_agentupdate_permission(client, db, app):
"""endpoints.command alone must not be able to swap the SYSTEM agent."""
_publish_agent()
assetid = _endpoint(db, 'UPGRADE02')
headers = _login(db, client, 'plainoperator',
permissions=['endpoints.view', 'endpoints.command'])
response = client.post(UPGRADE_URL.format(assetid), json={},
headers=headers)
assert response.status_code == 403
def test_an_upgrade_is_refused_when_no_build_is_published(client, db, app):
"""Nothing staged means there is no payload; refuse before queueing.
Queueing anyway would send an endpoint to fetch a package that does not
exist and fail opaquely minutes later.
"""
from plugins.endpoints.models import EndpointCommand
assetid = _endpoint(db, 'UPGRADE03')
headers = _login(db, client, 'nobuildadmin', admin=True)
response = client.post(UPGRADE_URL.format(assetid), json={},
headers=headers)
assert response.status_code == 400
assert 'No agent build is published' in \
response.get_json()['data']['error']['message']
assert EndpointCommand.query.filter_by(assetid=assetid).count() == 0
def test_an_endpoint_lock_still_blocks_the_upgrade(client, db, app):
"""The delegate's lock check must not be bypassed by this route.
A quarantined machine is reserved by a person; a fleet agent swap is
exactly the kind of action that must respect that.
"""
from plugins.endpoints.models import EndpointCommand
_publish_agent()
assetid = _endpoint(db, 'UPGRADE04')
# Lock as one user, then act as a different non-admin holder of the grant.
owner = _login(db, client, 'lockowner', admin=True)
locked = client.post(f'/api/endpoints/{assetid}/lock',
json={'reason': 'bench repair'}, headers=owner)
assert locked.status_code in (200, 201)
headers = _login(db, client, 'upgradeoperator',
permissions=['endpoints.view', 'endpoints.command',
'endpoints.agentupdate',
'endpoints.runscript'])
response = client.post(UPGRADE_URL.format(assetid), json={},
headers=headers)
assert response.status_code == 403
assert 'locked' in \
response.get_json()['data']['error']['message'].lower()
assert EndpointCommand.query.filter_by(assetid=assetid).count() == 0
def test_a_second_upgrade_cannot_stack_on_an_in_flight_command(client, db, app):
"""at-most-one-in-flight must still hold through this route."""
from plugins.endpoints.models import EndpointCommand
_publish_agent()
assetid = _endpoint(db, 'UPGRADE05')
headers = _login(db, client, 'stackadmin', admin=True)
first = client.post(UPGRADE_URL.format(assetid), json={}, headers=headers)
assert first.status_code == 201
second = client.post(UPGRADE_URL.format(assetid), json={}, headers=headers)
assert second.status_code == 409
assert EndpointCommand.query.filter_by(assetid=assetid).count() == 1
def test_the_upgrade_refreshes_the_package_before_queueing(client, db, app):
"""A newer staged build must be what gets pushed.
If the route queued against a previously registered row, an operator who
uploaded a new agent and immediately pushed would ship the old one.
"""
from shopdb.api import Application
_publish_agent('1.16.0.0')
assetid = _endpoint(db, 'UPGRADE06')
headers = _login(db, client, 'refreshadmin', admin=True)
client.post(UPGRADE_URL.format(assetid), json={}, headers=headers)
row = Application.query.filter_by(
appname=agentregistry.AGENT_APPNAME).first()
firstsha = row.installersha256
# Stage a different build and push to a second endpoint.
agentstore.store_version('agent', '1.17.0.0', b'MZ' + b'\x01' * 4096,
make_current=True)
second = _endpoint(db, 'UPGRADE07')
response = client.post(UPGRADE_URL.format(second), json={}, headers=headers)
assert response.status_code == 201
db.session.expire_all()
row = Application.query.filter_by(
appname=agentregistry.AGENT_APPNAME).first()
assert row.installersha256 != firstsha
def test_a_failed_push_is_not_audited_as_an_upgrade(client, db, app):
"""A refused push must not leave an AgentUpgrade event behind."""
from shopdb.api import AuditLog
_publish_agent()
assetid = _endpoint(db, 'UPGRADE08')
headers = _login(db, client, 'auditadmin', admin=True)
client.post(UPGRADE_URL.format(assetid), json={}, headers=headers)
before = AuditLog.query.filter_by(entitytype='AgentUpgrade').count()
# Second push is refused by the in-flight rule.
blocked = client.post(UPGRADE_URL.format(assetid), json={}, headers=headers)
assert blocked.status_code == 409
after = AuditLog.query.filter_by(entitytype='AgentUpgrade').count()
assert after == before
def test_the_seeded_script_body_is_written_and_pinned(client, db, app,
isolate_store):
"""The agent verifies the body by digest, so both must exist and agree."""
import hashlib
_publish_agent()
assetid = _endpoint(db, 'UPGRADE09')
headers = _login(db, client, 'bodyadmin', admin=True)
response = client.post(UPGRADE_URL.format(assetid), json={},
headers=headers)
assert response.status_code == 201
seeded = agentupgradescript.seed_upgrade_script()
path = os.path.join(str(isolate_store / 'scripts'),
'script-{}.ps1'.format(seeded['scriptid']))
assert os.path.isfile(path)
with open(path, 'rb') as handle:
body = handle.read()
assert hashlib.sha256(body).hexdigest() == seeded['sha256']
# The delivery path must be fully resolved, not a literal placeholder.
assert b'{{AGENT_APPID}}' not in body