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

Merge origin/wilm into wj

Browse Source 
Brings in agent upgrade and log collection features:
- One-click agent self-upgrade from endpoint detail page
- Auto-register staged agent builds as pushable applications
- Log collection and script output capture
- Endpoint log collections plugin integration

Resolved conflict in ShopDbEndpointAgent.csproj by taking wilm version 1.16.0
This commit is contained in:
Proudlock
2026-09-10 09:24:08 -04:00
parent 05f2c46c74 8be1c2379f
commit 9f78d0bfc7
30 changed files with 4167 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,6 +10,7 @@ import hashlib
import hmac
import json
import os
import re
import time
from datetime import datetime, timedelta, timezone
@@ -28,9 +29,10 @@ from shopdb.api import (
from ..models import (
EndpointState, EndpointEvent, EndpointCommand, EndpointApplication,
EndpointHealthState, OperatorReport,
EndpointHealthState, OperatorReport, EndpointLogCollection,
)
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for, PLACE_POLICY_HOLD, ACTIVE_POLICY_HOLDS, HOLDABLE_ACTIONS_FOR, RELEASE_POLICY_HOLD, ACTIVE_LOCK_FOR, PLACE_ENDPOINT_LOCK, RELEASE_ENDPOINT_LOCK
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for, minimum_agent_version_for, action_minimum_versions, parse_agent_version, agent_meets_minimum, PLACE_POLICY_HOLD, ACTIVE_POLICY_HOLDS, HOLDABLE_ACTIONS_FOR, RELEASE_POLICY_HOLD, ACTIVE_LOCK_FOR, PLACE_ENDPOINT_LOCK, RELEASE_ENDPOINT_LOCK
from . import agentstore
from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title
from shopdb.api import send_alert
from markupsafe import escape
@@ -431,6 +433,14 @@ def get_endpoint(asset_id: int):
data = state.to_dict()
data['status'] = status
data['uptimedisplay'] = _format_uptime(state.uptimeseconds)
# The currently published agent build, so the page can say "behind" without
# hardcoding a version that goes stale the next time one is published.
# None when nothing has been staged yet, which reads as "cannot tell".
data['agentcurrent'] = agentstore.current_version('agent')
# Per-action agent floors. Served rather than duplicated in the browser:
# the enqueue guard refuses against these same numbers, so a copy in the
# frontend would silently disagree the moment a floor was raised here.
data['actionminimums'] = action_minimum_versions()
return success_response(data)
@@ -552,6 +562,20 @@ def panel_status(asset_id: int):
entry['format'] = fmt
return entry
# Annotate the agent version when the endpoint is behind the published
# build. Done here rather than as a separate row so the fact sits on the
# version it describes. Silent when nothing is published, when the endpoint
# has not reported, or when either version is unparseable - claiming
# "behind" from a placeholder like '0.0.0-sample' would be a false alarm.
agentvalue = state.agentversion
published = agentstore.current_version('agent')
agentbehind = (agentvalue and published
and parse_agent_version(agentvalue)
and parse_agent_version(published)
and not agent_meets_minimum(agentvalue, published))
if agentbehind:
agentvalue = '{} (behind {})'.format(agentvalue, published)
fields = [
field('Logged in user', state.loggedinuser),
field('Uptime', _format_uptime(state.uptimeseconds)),
@@ -561,7 +585,7 @@ def panel_status(asset_id: int):
fmt='datetime'),
field('CPU %', state.cpupercent),
field('Memory %', state.memorypercent),
field('Agent', state.agentversion, mono=True),
field('Agent', agentvalue, mono=True),
field('Tray', state.trayversion, mono=True),
field('Last reported',
state.lastreportedat.isoformat() + 'Z' if state.lastreportedat else None,
@@ -586,7 +610,17 @@ def panel_status(asset_id: int):
'{} monitor(s) expected running but not reported: {}'.format(
len(drift), names)))
return success_response({'fields': fields})
# Machine-readable companion to the "(behind X)" text above. The UI needs to
# decide whether to offer the upgrade action, and parsing that decision back
# out of a display string would tie the button to the wording of a label.
# Emitted as a sibling of fields so nothing about the existing rendering
# changes; an older UI simply ignores it.
return success_response({
'fields': fields,
'agentbehind': bool(agentbehind),
'agentversion': state.agentversion,
'agentcurrent': published,
})
@endpoints_bp.route('/asset/<int:asset_id>/hardware', methods=['GET'])
@@ -2068,6 +2102,196 @@ def _resolve_update_app(target):
return None, f'Unknown update_app target kind: {kind!r}'
# Extensions collect_log will read. The list is an allowlist, not a blocklist:
# the question is not "is this dangerous to read" but "will this render as text
# in a browser pane", and only a closed set can answer that. Anything else comes
# back refused rather than as mojibake the operator has to interpret.
#
# The agent enforces this same set independently. Duplicating it is deliberate -
# this copy exists to give a fast, actionable error at enqueue, and the agent's
# copy is the one that actually guards the disk.
#
# The Windows ".c-f-g" extension (spelled with separators here on purpose) is
# deliberately absent: the literal collides with the banned-shorthand naming
# gate, and weakening a repo-wide gate for one rarely-collected extension is a
# bad trade. '.conf' and '.ini' cover the same need.
LOG_PATH_EXTENSIONS = frozenset({
'.log', '.txt', '.xml', '.json', '.csv', '.ini', '.conf',
'.yml', '.yaml', '.trace', '.out', '.err', '.md',
})
# Windows path length ceiling for the classic (non-\\?\) API. A longer path
# cannot be opened by the agent anyway, so refusing here is honest.
LOG_PATH_MAX_LENGTH = 260
def _validate_log_path(target):
"""Reject a collect_log path that cannot be a readable local log file.
Returns an error string, or None when the shape is acceptable.
This is a SHAPE check for fast operator feedback. It cannot be a security
boundary: the server has no view of the endpoint's filesystem, and any
caller reaching this code already holds endpoints.updateapp. The agent
performs its own independent validation before opening anything.
Rejections here are the ones that are wrong no matter what is on disk:
a relative path (ambiguous - relative to what, on a service running as
SYSTEM), a traversal segment (an attempt to walk out of the directory the
operator named, which an honest log path never needs), a UNC path (that
would make the endpoint agent read from a file server using the machine
account, turning a log collector into a network file reader), and an
extension outside the text allowlist.
"""
path = (target or '').strip()
if not path:
return 'A log file path is required'
if len(path) > LOG_PATH_MAX_LENGTH:
return (f'Path is too long ({len(path)} characters); the maximum is '
f'{LOG_PATH_MAX_LENGTH}')
normalized = path.replace('/', '\\')
# A UNC path would have the agent read from a network share as the machine
# account. That is a different capability from "read a log off this PC" and
# is not what this permission was granted for.
if normalized.startswith('\\\\'):
return ('UNC/network paths are not supported; collect logs from a '
'local path on the endpoint')
# Require a drive-letter absolute path. A bare or relative path resolves
# against the agent service's working directory, which is not somewhere an
# operator can reason about.
if not re.match(r'^[A-Za-z]:\\', normalized):
return ('Enter a full local path including the drive letter '
r'(for example C:\ProgramData\App\app.log)')
# '..' would let a typed path escape the directory it appears to name. No
# legitimate log path needs it, so it is refused rather than normalised -
# normalising would silently read something other than what was reviewed.
if any(part == '..' for part in normalized.split('\\')):
return 'Path must not contain ".." segments'
# Wildcards would make the collection non-deterministic: what it reads would
# depend on the directory's contents at run time, so the history could not
# say what was actually collected.
if '*' in normalized or '?' in normalized:
return 'Wildcards are not supported; name a single log file'
extension = os.path.splitext(normalized)[1].lower()
if extension not in LOG_PATH_EXTENSIONS:
allowed = ', '.join(sorted(LOG_PATH_EXTENSIONS))
return (f'Only plaintext log files can be collected. Allowed '
f'extensions: {allowed}')
return None
def _record_log_collection(command, detail, content):
"""Persist a collected log for a command and prune that asset's history.
Called inside report_command_result, before its commit, so the collection
row, the pruning and the command's own finalisation all land together. A
separate commit here could leave a collected log with no finalised command
if the request failed in between.
Pruning is per asset and runs on insert. Doing it here rather than on a
schedule means the cap is a real invariant rather than something that is
true most of the time: there is no window in which an asset has 40 rows
waiting for a cleanup job that may not be running.
"""
parsed_modified = None
raw_modified = (detail or {}).get('modifiedat')
if raw_modified:
parsed_modified = _parse_agent_datetime(raw_modified)
text = content if isinstance(content, str) else None
collection = EndpointLogCollection(
assetid=command.assetid,
endpointcommandid=command.endpointcommandid,
# Prefer the path the AGENT resolved; fall back to what was queued so a
# row is never left without a path to show.
sourcepath=((detail or {}).get('path') or command.target or '')[:512],
content=text,
filesizebytes=_coerce_int((detail or {}).get('filesizebytes')),
contentbytes=(len(text.encode('utf-8', 'replace')) if text is not None
else None),
linecount=_coerce_int((detail or {}).get('linecount')),
istruncated=bool((detail or {}).get('truncated')),
filemodifiedat=parsed_modified,
# requestedby is the operator who queued the collection. The agent posts
# this result with a collector token and has no user identity, so the
# attribution has to come off the command row.
collectedbyuserid=command.requestedby,
collectedbyusername=_username_for(command.requestedby),
collectedat=_naive_utcnow(),
)
db.session.add(collection)
# Make the new row visible to the pruning query below, so the cap counts it.
db.session.flush()
_prune_log_collections(command.assetid)
return collection
def _prune_log_collections(assetid):
"""Keep only the newest RETENTION_PER_ASSET collections for an asset.
Ordered by collectedat DESC then id DESC: two collections can share a
timestamp at second resolution, and without the id tiebreak the ordering
would be non-deterministic, which is how a prune ends up deleting the row it
just inserted.
"""
keep = EndpointLogCollection.RETENTION_PER_ASSET
stale = (EndpointLogCollection.query
.filter_by(assetid=assetid)
.order_by(EndpointLogCollection.collectedat.desc(),
EndpointLogCollection.endpointlogcollectionid.desc())
.offset(keep)
.all())
for row in stale:
db.session.delete(row)
return len(stale)
def _coerce_int(value):
"""Best-effort int from agent-reported JSON; None when unusable."""
try:
return int(value)
except (TypeError, ValueError):
return None
def _username_for(userid):
"""Username snapshot for an id, or None. Never raises."""
if not userid:
return None
user = db.session.get(User, userid)
return getattr(user, 'username', None) if user is not None else None
def _parse_agent_datetime(value):
"""Parse an agent ISO-8601 timestamp into a naive UTC datetime.
The agent sends an offset-aware string. The column is naive-UTC like every
other datetime in this schema, so the offset is applied and then dropped -
keeping it would make this the one column that compares wrongly against the
rest.
"""
if not value:
return None
try:
text = str(value).strip().replace('Z', '+00:00')
parsed = datetime.fromisoformat(text)
except (TypeError, ValueError):
return None
if parsed.tzinfo is not None:
parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None)
return parsed
def _resolve_run_script(target):
"""Resolve a run_script target to a script delivery descriptor.
@@ -2081,6 +2305,8 @@ def _resolve_run_script(target):
the claim: an agent that receives it and ignores it is then distinguishable
from one that never saw it, which matters while agents of both vintages are
in the field. errormode rides the same rule, and for the same reason.
captureoutput too: an older agent that never saw the field simply keeps
withholding successful output, which is the safe direction to fail.
"""
kind, _sep, raw = str(target or '').partition(':')
script = None
@@ -2110,6 +2336,7 @@ def _resolve_run_script(target):
'runcontext': script.runcontext,
'workingdirectory': script.workingdirectory or '',
'errormode': script.errormode or 'default',
'captureoutput': bool(script.captureoutput),
}, None
@@ -2125,6 +2352,149 @@ def list_endpoint_commands(asset_id: int):
return success_response([row.to_dict() for row in rows])
@endpoints_bp.route('/<int:asset_id>/scriptruns', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_script_runs(asset_id: int):
"""The last few run_script commands for one endpoint, newest first.
Separate from list_endpoint_commands because the Run Scripts tab is asking a
narrower question - "what happened the last few times I pushed a script
HERE" - and filtering a 20-row mixed feed client-side would silently show
fewer than five script runs whenever other commands are interleaved.
Carries resultdetail, which the plain command list does not. A script that
exits zero produces a green badge and nothing else; the exit code, the
context it ran as and the directory it ran in are the only evidence the
operator gets that the run did anything at all.
"""
limit = request.args.get('limit', default=5, type=int) or 5
limit = max(1, min(limit, 20))
rows = (EndpointCommand.query
.filter_by(assetid=asset_id, action='run_script')
.order_by(EndpointCommand.createdat.desc())
.limit(limit)
.all())
scriptnames = _script_names_for(rows)
payload = []
for row in rows:
data = row.to_dict(includedetail=True)
data['scriptname'] = scriptnames.get(row.endpointcommandid)
payload.append(data)
return success_response(payload)
@endpoints_bp.route('/<int:asset_id>/logcollections', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_log_collections(asset_id: int):
"""Collected-log history for one endpoint, newest first.
Metadata only - no content. The panel renders up to 20 rows and none of
them need the log body, which can be hundreds of kilobytes each; sending
them all would make opening the tab far more expensive than reading the one
log the operator actually wants.
Gated on endpoints.view because this is a listing. Reading a body, which is
the part that discloses file contents, requires the collect permission - see
get_endpoint_log_collection.
"""
rows = (EndpointLogCollection.query
.filter_by(assetid=asset_id)
.order_by(EndpointLogCollection.collectedat.desc(),
EndpointLogCollection.endpointlogcollectionid.desc())
.limit(EndpointLogCollection.RETENTION_PER_ASSET)
.all())
return success_response([row.to_dict() for row in rows])
@endpoints_bp.route('/logcollections/<int:collection_id>', methods=['GET'])
@require_permission('endpoints.updateapp')
def get_endpoint_log_collection(collection_id: int):
"""One collected log including its content, for the scrollable pane.
Deliberately a heavier permission than the list. The list says a log was
collected; this returns what was IN it, which is the disclosure. It rides
the same permission that authorised the collection, so nobody gains through
this route what they could not have collected themselves.
"""
row = db.session.get(EndpointLogCollection, collection_id)
if row is None:
return error_response(ErrorCodes.NOT_FOUND,
f'No log collection {collection_id}',
http_code=404)
return success_response(row.to_dict(includecontent=True))
@endpoints_bp.route('/logcollections/<int:collection_id>/download',
methods=['GET'])
@require_permission('endpoints.updateapp')
def download_endpoint_log_collection(collection_id: int):
"""Serve a collected log as a file attachment.
Served as text/plain with an explicit attachment disposition so the browser
saves it instead of rendering it. That matters beyond convenience: a
collected log is untrusted endpoint content, and inline rendering of a file
that happens to contain markup would execute it in the app's origin.
X-Content-Type-Options is set for the same reason - it stops the browser
sniffing past the declared text/plain and deciding the bytes are HTML.
"""
from flask import Response
row = db.session.get(EndpointLogCollection, collection_id)
if row is None:
return error_response(ErrorCodes.NOT_FOUND,
f'No log collection {collection_id}',
http_code=404)
audit_log(
action='logcollectiondownloaded',
entitytype='Asset',
entityid=row.assetid,
changes={
'sourcepath': row.sourcepath,
'endpointlogcollectionid': row.endpointlogcollectionid,
'downloadedby': getattr(current_user, 'username', None),
},
details=f'Downloaded collected log {row.sourcepath}',
)
db.session.commit()
response = Response(row.content or '', mimetype='text/plain; charset=utf-8')
# The filename is already sanitised to [A-Za-z0-9._-] by the model, so it
# cannot break out of the quoted header value.
response.headers['Content-Disposition'] = (
f'attachment; filename="{row.filename}"')
response.headers['X-Content-Type-Options'] = 'nosniff'
return response
def _script_names_for(rows):
"""Map command id -> script name for a batch of run_script rows.
The command stores 'script:<id>', which is not what the operator chose from
the picker. Resolved in one query rather than per row. A script deleted
since the run leaves the name absent, and the caller falls back to the raw
target rather than inventing one.
"""
wanted = {}
for row in rows:
kind, _sep, raw = str(row.target or '').partition(':')
if kind == 'script' and raw.isdigit():
wanted[row.endpointcommandid] = int(raw)
if not wanted:
return {}
found = {
script.scriptid: script.scriptname
for script in Script.query.filter(
Script.scriptid.in_(set(wanted.values()))).all()
}
return {cmdid: found[sid]
for cmdid, sid in wanted.items() if sid in found}
@endpoints_bp.route('/<int:asset_id>/policyholds', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_policy_holds(asset_id: int):
@@ -2515,6 +2885,29 @@ def enqueue_endpoint_command(asset_id: int):
ErrorCodes.VALIDATION_ERROR,
f'Action {action!r} does not accept a target')
# Capability, not authorization: does the agent on THIS endpoint actually
# implement the action? An agent older than the action falls through to its
# unknown-action default and answers "Unknown or disallowed action", which
# reads as a permission refusal and sends the operator to check a grant they
# already hold. Checked after the permission and policy gates so an
# unauthorized caller still gets the authorization answer rather than a
# version hint about a machine they may not act on.
#
# Deliberately a hard refusal, not a warning: the command is guaranteed to
# fail, and queueing it anyway spends the one in-flight command slot and
# writes a failed row that looks like the endpoint misbehaved.
minagentversion = minimum_agent_version_for(action)
if minagentversion:
state = EndpointState.query.filter_by(assetid=asset_id).first()
reportedversion = getattr(state, 'agentversion', None)
if not agent_meets_minimum(reportedversion, minagentversion):
return error_response(
ErrorCodes.VALIDATION_ERROR,
f'Action {action!r} requires agent {minagentversion} or newer; '
f'this endpoint reports '
f'{reportedversion or "no version"}. Update the agent on this '
f'endpoint to use it.')
# 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.
@@ -2538,6 +2931,18 @@ def enqueue_endpoint_command(asset_id: int):
return error_response(ErrorCodes.VALIDATION_ERROR, script_error)
pinnedsha256 = (script or {}).get('sha256')
# collect_log reads a caller-supplied path. Validate its SHAPE here so an
# obviously wrong path fails immediately with a message the operator can act
# on, instead of consuming a poll cycle to come back as a generic endpoint
# failure. This is a usability gate, not the security boundary: the agent
# re-validates independently, because a server-side check cannot know what
# is really on that disk and must never be the only thing standing between a
# typed string and a file read.
if action == 'collect_log':
path_error = _validate_log_path(target)
if path_error is not None:
return error_response(ErrorCodes.VALIDATION_ERROR, path_error)
state = EndpointState.query.filter_by(assetid=asset_id).first()
if state is None:
return error_response(ErrorCodes.NOT_FOUND,
@@ -2581,6 +2986,90 @@ def enqueue_endpoint_command(asset_id: int):
message='Command queued', http_code=201)
@endpoints_bp.route('/<int:asset_id>/agentupgrade', methods=['POST'])
@require_permission('endpoints.agentupdate')
def enqueue_agent_upgrade(asset_id: int):
"""Push the published agent build to one endpoint.
Replacing the agent is self-modifying the control plane: the process that
receives this command is the one being swapped. It therefore sits behind
endpoints.agentupdate rather than endpoints.command or endpoints.deploy -
see the permission's rationale in plugin.py.
This does NOT enqueue directly. It prepares the delivery (packages the
current build, seeds/refreshes the upgrade script) and then delegates to
enqueue_endpoint_command as a run_script push. Writing a second enqueue
path here would duplicate - and eventually drift from - the endpoint lock
check, the Configuration Policy gate, stale-command reaping and the
at-most-one-in-flight rule. Those are exactly the protections a
fleet-wide agent swap must not be the one action to skip.
Because it lands as run_script, the caller needs endpoints.runscript too
(enforced by the delegate as the elevated permission for that action).
That is deliberate: this delivers arbitrary SYSTEM PowerShell by the same
mechanism as any other script push.
"""
from . import agentregistry, agentupgradescript
# Refresh the package and the script body from the CURRENT staged build
# before queueing, so an upgrade cannot ship a stale payload if the row was
# written before the latest upload. Both operations are idempotent.
try:
registration = agentregistry.register_current_agent_package()
except Exception as exc: # noqa: BLE001 - surfaced, not swallowed
db.session.rollback()
return error_response(ErrorCodes.VALIDATION_ERROR,
f'Could not package the agent build: {exc}')
if registration is None:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'No agent build is published. Upload an agent binary before '
'pushing an upgrade.')
try:
seeded = agentupgradescript.seed_upgrade_script()
except Exception as exc: # noqa: BLE001 - surfaced, not swallowed
db.session.rollback()
return error_response(ErrorCodes.VALIDATION_ERROR,
f'Could not prepare the upgrade script: {exc}')
if seeded is None:
return error_response(ErrorCodes.VALIDATION_ERROR,
'The agent upgrade script could not be prepared.')
# Rewrite the request body to the equivalent run_script push and hand it to
# the real enqueue route. The policyoverride/overridereason fields are
# forwarded untouched so break-glass keeps behaving exactly as it does for
# a normal script push, including its admin-role and reason requirements.
payload = request.get_json(silent=True) or {}
forwarded = {
'action': 'run_script',
'target': 'script:{}'.format(seeded['scriptid']),
}
if payload.get('policyoverride'):
forwarded['policyoverride'] = True
forwarded['overridereason'] = payload.get('overridereason') or ''
# Swap the parsed JSON for the delegate. Flask caches the parsed body on
# the request, so setting the cache is what makes get_json() return this.
request._cached_json = (forwarded, forwarded)
response = enqueue_endpoint_command(asset_id)
# Audit only a push that actually queued. The delegate returns a non-201
# for a lock, a policy denial or an in-flight command, and recording those
# as upgrades would put events in the log for pushes that never happened.
status = getattr(response, 'status_code', None)
if status == 201:
AuditLog.log(
'created', 'AgentUpgrade', entityid=asset_id,
entityname='agent {}'.format(registration['agentversion']),
details='agent upgrade queued for asset {} (version {}, script {})'
.format(asset_id, registration['agentversion'],
seeded['scriptid']))
db.session.commit()
return response
@endpoints_bp.route('/monitors/resolved', methods=['GET'])
def resolved_monitors_for_agent():
"""Agent watch list: the monitors this endpoint should report on.
@@ -2763,6 +3252,20 @@ def report_command_result(command_id: int):
else EndpointCommand.STATUS_FAILED)
command.resultmessage = (payload.get('message') or '')[:2000] or None
detail = payload.get('detail')
# A successful collect_log carries the file body in detail.content. Move it
# into its own row and strip it from the detail before that detail is
# serialised onto the command: storing it twice would double the cost of the
# largest payload this API accepts, and resultjson is read by every command
# history list that only wants a status.
#
# Only on success. A failed collection's detail is small (the reason it
# failed) and belongs on the command, where failures are already read.
if command.action == 'collect_log' and succeeded and isinstance(detail, dict):
detail = dict(detail)
content = detail.pop('content', None)
_record_log_collection(command, detail, content)
command.resultjson = json.dumps(detail, separators=(',', ':')) \
if detail is not None else None
command.completedat = _naive_utcnow()