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

Merge the wilm delivery work into main

Browse Source 
Ben Baker's installed-software collection, software delivery (run_script and
update_app), agentless adoption tool and command-queue hardening, with his
authorship preserved.

Reviewed before promotion. Fixed: the endpoints plugin tests were never collected
by pytest, the SBOM snapshot table's key would exhaust on MySQL, ingest did not
truncate two columns strict MySQL would reject, and update_app's version branch
pinned a digest for a file that route never serves. Agent packaging moved off
require_role('admin') onto a new endpoints.agentpackage permission.

Contract 0.27.0. Full suite 2231, frontend 314.
This commit is contained in:
Cameron Proudlock
2026-08-30 19:41:15 -04:00
parent 8f4fe74dcb
commit 6680c3c5de
59 changed files with 5557 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

393
plugins/endpoints/api/routes.py
View File

@@ -6,7 +6,9 @@ service token, and are handled by EndpointsPlugin.apply_collector_payload.
These routes are the human-facing read surface, gated by user JWT + RBAC.
"""
import hashlib
import json
import os
import time
from datetime import datetime, timedelta, timezone
@@ -19,9 +21,10 @@ from shopdb.api import (
require_role, caller_holds, mint_service_token, list_service_tokens,
revoke_service_tokens,
AuditLog, collector_authorized, User,
Application, AppVersion, Script,
)
from ..models import EndpointState, EndpointEvent, EndpointCommand
from ..models import EndpointState, EndpointEvent, EndpointCommand, EndpointApplication
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for
from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title
@@ -674,6 +677,32 @@ def panel_interfaces(asset_id: int):
return success_response(rows)
@endpoints_bp.route('/asset/<int:asset_id>/applications', methods=['GET'])
@require_permission('endpoints.view')
def panel_applications(asset_id: int):
"""Table panel: the installed-software (SBOM) snapshot for this endpoint.
Reads the endpointapplications rows the collector last wrote for this asset
(a per-report snapshot, not history) and returns them sorted by name. This
is the raw discovered inventory: appid is carried through but not yet linked
to the applications catalog ("Managed Applications"), so no managed status
is surfaced here. An endpoint with no reported SBOM yields no rows and the
panel hides itself.
"""
rows = (EndpointApplication.query
.filter_by(assetid=asset_id)
.order_by(db.func.lower(EndpointApplication.displayname))
.all())
return success_response([
{
'displayname': row.displayname,
'displayversion': row.displayversion or '-',
'publisher': row.publisher or '-',
}
for row in rows
])
# =============================================================================
# Dashboard
# =============================================================================
@@ -912,6 +941,41 @@ _CLAIM_HOLD_SECONDS = 25
_CLAIM_POLL_INTERVAL = 1.0
def _reap_stale_commands(asset_id):
"""Expire in-flight commands an agent will never finish for this endpoint.
The queue is at-most-one-in-flight: a single PENDING or RUNNING row blocks
every new command. If the agent is killed, uninstalled, or goes offline
after a command is claimed (RUNNING) or before it claims one (PENDING), that
row would pin the queue forever and every operator action returns 409. Reap
rows past their staleness ceiling to STATUS_EXPIRED so the queue self-heals.
Caller commits (this runs inside the enqueue transaction).
"""
now = _naive_utcnow()
running_cutoff = now - timedelta(seconds=EndpointCommand.RUNNING_STALE_SECONDS)
pending_cutoff = now - timedelta(seconds=EndpointCommand.PENDING_STALE_SECONDS)
stale = (EndpointCommand.query
.filter(EndpointCommand.assetid == asset_id,
db.or_(
db.and_(
EndpointCommand.status == EndpointCommand.STATUS_RUNNING,
EndpointCommand.claimedat.isnot(None),
EndpointCommand.claimedat < running_cutoff),
db.and_(
EndpointCommand.status == EndpointCommand.STATUS_PENDING,
EndpointCommand.createdat < pending_cutoff)))
.all())
for command in stale:
command.status = EndpointCommand.STATUS_EXPIRED
command.completedat = now
if not command.resultmessage:
command.resultmessage = (
'Expired: agent did not report a result before the staleness '
'ceiling (agent offline, killed, or reinstalled mid-command).')
return stale
def _collector_auth_or_error():
"""Authorize an agent-facing command route on the endpoints collector token.
@@ -931,6 +995,184 @@ def _resolve_asset_for_hostname(hostname):
return computer.assetid if computer is not None else None
_HOSTED_PACKAGE_PREFIX = '/api/applications/package/'
def _hosted_package_filename(application):
"""Return the stored filename for a ShopDB-hosted installer, or None.
installpath is the core package URL (/api/applications/package/<filename>);
a site-typed share path (not our upload) has no servable file here.
"""
installpath = application.installpath or ''
if not installpath.startswith(_HOSTED_PACKAGE_PREFIX):
return None
return installpath[len(_HOSTED_PACKAGE_PREFIX):]
def _backfill_app_installer_sha256(application):
"""Compute and persist installersha256 for a pre-migration package.
Packages uploaded before the 7d36 migration have a null installersha256,
which used to force a re-upload (Bug B). When the hosted package file is
present on disk, hash it once, stamp the app, and commit so future resolves
hit the stored value. Returns the digest, or None when no servable file.
"""
filename = _hosted_package_filename(application)
if not filename:
return None
package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
package_path = os.path.join(package_dir, filename)
if not os.path.isfile(package_path):
return None
digest = hashlib.sha256()
with open(package_path, 'rb') as package_file:
for chunk in iter(lambda: package_file.read(65536), b''):
digest.update(chunk)
application.installersha256 = digest.hexdigest()
db.session.commit()
return application.installersha256
def _resolve_update_app(target):
"""Resolve an update_app target to a delivery descriptor.
Accepts either an app-level or a version-level source, so a push works
whether or not the site tracks AppVersion rows (the applications UI stores a
single app-level installer; versioning is optional):
- "app:<appid>" - the app's own uploaded installer (installersha256)
- "version:<id>" - a specific AppVersion's pinned package
- bare "<int>" - legacy: treated as an appversionid
Returns (descriptor, error_message). The descriptor carries what the agent
needs to fetch and run the pinned installer: a collector-authenticated
download URL, its SHA-256, and the silent-install switches. The agent
fetches the bytes with its existing collector credential; no bytes here.
"""
kind, _sep, raw = str(target or '').partition(':')
if not raw:
# No prefix: legacy bare appversionid.
kind, raw = 'version', kind
try:
ident = int(raw)
except (TypeError, ValueError):
return None, 'update_app target must be app:<id>, version:<id>, or an appversionid'
if kind == 'app':
application = db.session.get(Application, ident)
if application is None:
return None, f'Unknown appid: {ident}'
if not application.installpath:
return None, (f'Application {ident} has no uploaded installer package')
# Backfill installersha256 for packages uploaded before the 7d36 migration
# so app-level delivery resolves without a re-upload (Bug B).
sha256 = application.installersha256 or _backfill_app_installer_sha256(application)
if not sha256:
return None, (f'Application {ident} has no uploaded installer package')
# Carry the real installer filename (with extension) on the delivery URL so
# the agent's Path.GetExtension routes MSI through msiexec (Bug A rc2). The
# trailing filename is cosmetic for the agent; the app_id drives the lookup.
filename = _hosted_package_filename(application)
installpath = '/api/endpoints/apppackage/app/{}'.format(application.appid)
if filename:
installpath = '{}/{}'.format(installpath, filename)
return {
'source': 'app',
'appid': application.appid,
'appversionid': None,
'appname': application.appname,
'version': None,
'installpath': installpath,
'sha256': sha256,
'silentargs': None,
}, None
if kind == 'version':
version = db.session.get(AppVersion, ident)
if version is None:
return None, f'Unknown appversionid: {ident}'
application = db.session.get(Application, version.appid)
if application is None or not application.installpath:
return None, ('Application for appversionid {} has no installer package'
.format(ident))
# PIN THE FILE THAT IS ACTUALLY SERVED.
#
# /api/endpoints/apppackage/<appversionid> resolves the version to its
# APPLICATION and serves application.installpath - there is no
# version-specific artifact behind that route. So version.sha256 pinned
# a digest that need not describe the bytes delivered, and where it was
# null the descriptor carried no digest at all and the agent had nothing
# to verify against. Deriving it from the application, exactly as the
# app branch does, makes the pin describe the delivery.
sha256 = (application.installersha256
or _backfill_app_installer_sha256(application))
if not sha256:
return None, ('Application for appversionid {} has an installer '
'package with no recorded SHA-256, so the delivery '
'cannot be verified'.format(ident))
# Carry the real filename so the agent's Path.GetExtension routes an MSI
# through msiexec. Without it every MSI push failed, because this branch
# ended the URL at the id and the extension was simply absent.
filename = _hosted_package_filename(application)
installpath = '/api/endpoints/apppackage/{}'.format(version.appversionid)
if filename:
installpath = '/api/endpoints/apppackage/app/{}/{}'.format(
application.appid, filename)
return {
'source': 'version',
'appversionid': version.appversionid,
'appid': version.appid,
'appname': application.appname,
'version': version.version,
'installpath': installpath,
'sha256': sha256,
'silentargs': version.silentargs,
}, None
return None, f'Unknown update_app target kind: {kind!r}'
def _resolve_run_script(target):
"""Resolve a run_script target to a script delivery descriptor.
Accepts either "script:<scriptid>" or a bare script name. The descriptor
carries a collector-authenticated download URL, the body's SHA-256, and the
per-script run context (system or user). The agent fetches the body with
its collector credential and runs it; the body never crosses the wire.
"""
kind, _sep, raw = str(target or '').partition(':')
script = None
if kind == 'script' and raw:
try:
script = db.session.get(Script, int(raw))
except (TypeError, ValueError):
return None, 'run_script target script:<id> must be numeric'
else:
# Bare name (the whole target, prefix included when there was no colon).
name = (target or '').strip()
if name:
script = Script.query.filter_by(scriptname=name).first()
if script is None:
return None, f'Unknown script: {target!r}'
if not script.filename or not script.sha256:
return None, f'Script {script.scriptname!r} has no uploaded body'
if not script.isactive:
return None, f'Script {script.scriptname!r} is inactive'
return {
'scriptid': script.scriptid,
'scriptname': script.scriptname,
'installpath': '/api/endpoints/scriptbody/{}'.format(script.scriptid),
'sha256': script.sha256,
'runcontext': script.runcontext,
}, None
@endpoints_bp.route('/<int:asset_id>/commands', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_commands(asset_id: int):
@@ -989,12 +1231,33 @@ def enqueue_endpoint_command(asset_id: int):
ErrorCodes.VALIDATION_ERROR,
f'Action {action!r} does not accept a target')
# 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.
if action == 'update_app':
_descriptor, resolve_error = _resolve_update_app(target)
if resolve_error is not None:
return error_response(ErrorCodes.VALIDATION_ERROR, resolve_error)
# 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)
if script_error is not None:
return error_response(ErrorCodes.VALIDATION_ERROR, script_error)
state = EndpointState.query.filter_by(assetid=asset_id).first()
if state is None:
return error_response(ErrorCodes.NOT_FOUND,
f'No endpoint state for asset {asset_id}',
http_code=404)
# Self-heal a queue wedged by a command the agent will never finish (killed,
# uninstalled, or offline mid-flight) before enforcing at-most-one-in-flight,
# so a dead in-flight row cannot permanently 409 every operator action.
_reap_stale_commands(asset_id)
pending = (EndpointCommand.query
.filter(EndpointCommand.assetid == asset_id,
EndpointCommand.status.in_((
@@ -1060,12 +1323,33 @@ def claim_next_command():
command.status = EndpointCommand.STATUS_RUNNING
command.claimedat = _naive_utcnow()
db.session.commit()
return success_response({
claim = {
'endpointcommandid': command.endpointcommandid,
'action': command.action,
'target': command.target,
'targetkind': command.targetkind,
})
}
# update_app rides a resolved delivery descriptor: the agent gets
# the package URL, its pinned SHA-256, and the version's silent-
# install switches, and fetches the bytes with its collector
# credential. If the version was deleted between enqueue and claim,
# send the command through with no descriptor and let the agent
# fail closed rather than 500 the poll.
if command.action == 'update_app':
descriptor, _resolve_error = _resolve_update_app(command.target)
if descriptor is not None:
claim['app'] = descriptor
# run_script rides a resolved script descriptor: the agent gets a
# collector-authenticated body URL, the pinned SHA-256, and the
# per-script run context. Fetch happens agent-side with the
# collector credential. If the script vanished between enqueue and
# claim, send the command through with no descriptor and let the
# agent fail closed rather than 500 the poll.
if command.action == 'run_script':
script_descriptor, _script_error = _resolve_run_script(command.target)
if script_descriptor is not None:
claim['script'] = script_descriptor
return success_response(claim)
# Nothing queued: release the session so a concurrent enqueue is visible
# on the next poll, wait a beat, then re-check until the hold expires.
@@ -1114,3 +1398,106 @@ def report_command_result(command_id: int):
db.session.commit()
return success_response(command.to_dict(), message='Result recorded')
@endpoints_bp.route('/apppackage/<int:appversionid>', methods=['GET'])
def download_app_package(appversionid: int):
"""Agent-facing installer download (collector-token auth).
The core /api/applications/package route is gated on applications.view, a
permission the endpoint agent's collector token does not hold. This route
authorizes on the same endpoints collector token the agent already uses for
telemetry and command polling, so the whole app-delivery path rides one key.
Serves the exact bytes the update_app descriptor pinned by SHA-256.
"""
from shopdb.api import send_upload
error = _collector_auth_or_error()
if error is not None:
return error
version = db.session.get(AppVersion, appversionid)
if version is None:
return error_response(ErrorCodes.NOT_FOUND,
f'Unknown appversionid: {appversionid}', http_code=404)
application = db.session.get(Application, version.appid)
if application is None or not application.installpath:
return error_response(ErrorCodes.NOT_FOUND,
'No installer package for this application',
http_code=404)
# installpath is the core package URL (/api/applications/package/<filename>);
# take the stored filename and serve it from the core package directory. A
# site-typed share path (not our upload) has no servable file here.
prefix = '/api/applications/package/'
if not application.installpath.startswith(prefix):
return error_response(ErrorCodes.NOT_FOUND,
'Application installer is not a ShopDB-hosted package',
http_code=404)
filename = application.installpath[len(prefix):]
package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
return send_upload(package_dir, filename, as_attachment=True)
@endpoints_bp.route('/apppackage/app/<int:app_id>', methods=['GET'])
@endpoints_bp.route('/apppackage/app/<int:app_id>/<path:filename>', methods=['GET'])
def download_app_package_by_app(app_id: int, filename: str = None):
"""Agent-facing installer download by APP (collector-token auth).
The app-level counterpart to download_app_package: serves the application's
own uploaded installer (installpath + installersha256) so update_app can
resolve an install source by app alone, with no AppVersion row. Same
collector-token auth and same package directory as the per-version route.
The optional trailing <filename> segment lets the delivery URL carry the
real installer extension (so the agent routes MSI through msiexec); it is
validated against the stored filename and does not drive the lookup.
"""
from shopdb.api import send_upload
error = _collector_auth_or_error()
if error is not None:
return error
application = db.session.get(Application, app_id)
if application is None or not application.installpath:
return error_response(ErrorCodes.NOT_FOUND,
'No installer package for this application',
http_code=404)
prefix = '/api/applications/package/'
if not application.installpath.startswith(prefix):
return error_response(ErrorCodes.NOT_FOUND,
'Application installer is not a ShopDB-hosted package',
http_code=404)
stored_filename = application.installpath[len(prefix):]
if filename is not None and filename != stored_filename:
return error_response(ErrorCodes.NOT_FOUND,
'Requested installer filename does not match this application',
http_code=404)
package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
return send_upload(package_dir, stored_filename, as_attachment=True)
@endpoints_bp.route('/scriptbody/<int:script_id>', methods=['GET'])
def download_script_body(script_id: int):
"""Agent-facing script body download (collector-token auth).
The core /api/scripts/body route is gated on applications.view, which the
endpoint agent's collector token does not hold. This route authorizes on
the same endpoints collector token the agent already uses, keeping the
whole script-delivery path on one key. Serves the exact bytes the
run_script descriptor pinned by SHA-256.
"""
from shopdb.api import send_upload
error = _collector_auth_or_error()
if error is not None:
return error
script = db.session.get(Script, script_id)
if script is None or not script.filename:
return error_response(ErrorCodes.NOT_FOUND,
'No body for this script', http_code=404)
script_dir = os.path.join(current_app.instance_path, 'scripts')
return send_upload(script_dir, script.filename, as_attachment=True)