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

Add endpoint monitor catalog and warn-only drift reconciler

Browse Source 
Operators need to see when a machine's running services or processes fall outside the desired state defined by policy. This adds a reusable monitor catalog (shipped read-only rows plus user-built local rows), computes drift server-side on agent check-in, and surfaces a warning on the endpoint panels. Remediation stays manual for now; automation can follow once demand is proven. Includes migrations 0006-0008 and the monitors API client.
This commit is contained in:
benjamin.baker
2026-09-01 15:44:03 -04:00
parent 1a1bf0640f
commit 42db980ca9
23 changed files with 2054 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
plugins/endpoints/api/__init__.py
View File

@@ -5,5 +5,7 @@ from .routes import endpoints_bp
from . import adoption # noqa: F401
# Attaches the site agent generator packaging routes to endpoints_bp.
from . import siteagent # noqa: F401
# Attaches the monitor catalog CRUD and resolve routes to endpoints_bp.
from . import monitors # noqa: F401
__all__ = ['endpoints_bp']

204
plugins/endpoints/api/monitors.py Normal file
View File

@@ -0,0 +1,204 @@
"""Endpoint monitor definitions API (the monitor catalog CRUD + resolver).
A monitor definition is the reusable "what" an operator builds on the Settings
page: a named process or service with a bespoke display label and optional
launch/collector paths. Attaching a monitor to a PC type is a site-policy
concern (see plugins/computers policies); this surface owns the definitions
themselves plus the resolve-by-pctype read the agent uses at runtime.
Shipped rows (source='shipped') are product-owned: readable by everyone with
the view permission but not editable or deletable through this API.
The resolve-by-pctype read returns the monitor set a deployed agent should watch
for its PC type: the shipped defaults for 'Shopfloor PC' unioned with whatever
active site policies attach to that type. Shipped monitors are hardcoded to
'Shopfloor PC' as a proof of concept until they are authored through the policy
route like any user-built monitor.
"""
from flask import request
from flask_jwt_extended import current_user
from shopdb.api import (
db, success_response, error_response, ErrorCodes, require_permission,
require_role,
)
from ..models import EndpointMonitor
from .routes import endpoints_bp
# Shipped monitors default to this PC type until they are authored as policies.
SHIPPED_DEFAULT_PCTYPE = 'Shopfloor PC'
_VALID_KINDS = ('process', 'service')
def _validate_monitor_payload(body, kind):
"""Return an error message for an invalid create/update body, or None."""
displayname = (body.get('displayname') or '').strip()
if not displayname:
return 'displayname is required'
if kind not in _VALID_KINDS:
return "kind must be 'process' or 'service'"
if kind == 'process' and not (body.get('processname') or '').strip():
return 'processname is required for a process monitor'
if kind == 'service' and not (body.get('servicename') or '').strip():
return 'servicename is required for a service monitor'
return None
@endpoints_bp.route('/monitors', methods=['GET'])
@require_permission('endpoints.view')
def list_monitors():
"""List monitor definitions (shipped and local)."""
monitors = EndpointMonitor.query.order_by(
EndpointMonitor.source.desc(),
EndpointMonitor.displayname.asc()).all()
return success_response([monitor.to_dict() for monitor in monitors])
@endpoints_bp.route('/monitors', methods=['POST'])
@require_role('admin')
def create_monitor():
"""Create a user-defined (local) monitor definition."""
body = request.get_json(silent=True) or {}
kind = (body.get('kind') or '').strip().lower()
problem = _validate_monitor_payload(body, kind)
if problem:
return error_response(ErrorCodes.VALIDATION_ERROR, problem, http_code=400)
monitor = EndpointMonitor(
kind=kind,
displayname=(body.get('displayname') or '').strip(),
processname=(body.get('processname') or '').strip() or None,
servicename=(body.get('servicename') or '').strip() or None,
launchpath=(body.get('launchpath') or '').strip() or None,
datacollectorpath=(body.get('datacollectorpath') or '').strip() or None,
logcollectorpath=(body.get('logcollectorpath') or '').strip() or None,
source='local',
isactive=bool(body.get('isactive', True)),
)
db.session.add(monitor)
db.session.commit()
return success_response(monitor.to_dict(), message='Monitor created',
http_code=201)
@endpoints_bp.route('/monitors/<int:monitor_id>', methods=['PUT'])
@require_role('admin')
def update_monitor(monitor_id):
"""Update a local monitor definition. Shipped rows are read-only."""
monitor = db.session.get(EndpointMonitor, monitor_id)
if monitor is None:
return error_response(ErrorCodes.NOT_FOUND, 'Monitor not found',
http_code=404)
if monitor.source != 'local':
return error_response(ErrorCodes.FORBIDDEN,
'Shipped monitors cannot be edited', http_code=403)
body = request.get_json(silent=True) or {}
kind = (body.get('kind') or monitor.kind or '').strip().lower()
problem = _validate_monitor_payload(
{
'displayname': body.get('displayname', monitor.displayname),
'processname': body.get('processname', monitor.processname),
'servicename': body.get('servicename', monitor.servicename),
},
kind,
)
if problem:
return error_response(ErrorCodes.VALIDATION_ERROR, problem, http_code=400)
monitor.kind = kind
monitor.displayname = (body.get('displayname', monitor.displayname) or '').strip()
monitor.processname = (body.get('processname', monitor.processname) or '').strip() or None
monitor.servicename = (body.get('servicename', monitor.servicename) or '').strip() or None
monitor.launchpath = (body.get('launchpath', monitor.launchpath) or '').strip() or None
monitor.datacollectorpath = (body.get('datacollectorpath', monitor.datacollectorpath) or '').strip() or None
monitor.logcollectorpath = (body.get('logcollectorpath', monitor.logcollectorpath) or '').strip() or None
if 'isactive' in body:
monitor.isactive = bool(body.get('isactive'))
db.session.commit()
return success_response(monitor.to_dict(), message='Monitor updated')
@endpoints_bp.route('/monitors/<int:monitor_id>', methods=['DELETE'])
@require_role('admin')
def delete_monitor(monitor_id):
"""Delete a local monitor definition. Shipped rows are protected."""
monitor = db.session.get(EndpointMonitor, monitor_id)
if monitor is None:
return error_response(ErrorCodes.NOT_FOUND, 'Monitor not found',
http_code=404)
if monitor.source != 'local':
return error_response(ErrorCodes.FORBIDDEN,
'Shipped monitors cannot be deleted', http_code=403)
db.session.delete(monitor)
db.session.commit()
return success_response(None, message='Monitor deleted')
def _resolve_pctype_monitors(pctype):
"""Monitor definitions a PC type should watch: shipped defaults + policy.
Shipped monitors are attached to SHIPPED_DEFAULT_PCTYPE as a code rule;
policy-attached monitors come from the computers policy resolver by PC type.
Returns a list of EndpointMonitor rows (active only), de-duped by id.
"""
selected = {}
if pctype == SHIPPED_DEFAULT_PCTYPE:
for monitor in EndpointMonitor.query.filter_by(
source='shipped', isactive=True).all():
selected[monitor.endpointmonitorid] = monitor
computertypeid = _computertypeid_for(pctype)
if computertypeid is not None:
policy_ids = _policy_monitor_ids(computertypeid)
if policy_ids:
for monitor in EndpointMonitor.query.filter(
EndpointMonitor.endpointmonitorid.in_(policy_ids),
EndpointMonitor.isactive.is_(True)).all():
selected[monitor.endpointmonitorid] = monitor
return list(selected.values())
def _computertypeid_for(pctype):
"""Resolve a PC type name to its computertypeid, or None."""
try:
from plugins.computers.models import ComputerType
except Exception:
return None
row = ComputerType.query.filter_by(computertype=pctype).first()
return row.computertypeid if row else None
def _policy_monitor_ids(computertypeid):
"""Monitor ids attached to a PC type by effective policy, or empty set."""
try:
from plugins.computers.services.policyresolver import monitor_ids_for_pctype
except Exception:
return set()
return monitor_ids_for_pctype(computertypeid)
@endpoints_bp.route('/monitors/resolve', methods=['GET'])
@require_permission('endpoints.view')
def resolve_monitors():
"""Resolve the monitor set for a PC type (agent runtime fetch).
Query param pctype names the computer type. Returns the definitions the
agent should watch: shipped defaults for 'Shopfloor PC' plus any monitors
active site policies attach to that type.
"""
pctype = (request.args.get('pctype') or '').strip()
if not pctype:
return error_response(ErrorCodes.VALIDATION_ERROR,
'pctype is required', http_code=400)
monitors = _resolve_pctype_monitors(pctype)
return success_response({
'pctype': pctype,
'monitors': [monitor.to_dict() for monitor in monitors],
})

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

@@ -87,7 +87,8 @@ def _format_uptime(seconds):
# Endpoint state
# =============================================================================
@endpoints_bp.route('', methods=['GET'])
@endpoints_bp.route('', methods=['GET'], strict_slashes=False)
@endpoints_bp.route('/', methods=['GET'], strict_slashes=False)
@require_permission('endpoints.fleet')
def list_endpoints():
"""List reported endpoint states.
@@ -396,6 +397,24 @@ def list_endpoint_events(asset_id: int):
# Asset panels (ADR-010) - shaped for the generic PluginAssetPanels renderer
# =============================================================================
def _drift_for_state(state):
"""Parse the cached warn-only reconciler drift list off an EndpointState.
driftjson is written on every check-in by the endpoints ingest: the monitors
site policy expects running that the latest snapshot did not report running.
Null means never computed; an empty list means computed with no drift.
Tolerant of a malformed blob so a bad cache never breaks the panel.
"""
raw = getattr(state, 'driftjson', None) if state else None
if not raw:
return []
try:
parsed = json.loads(raw)
except (TypeError, ValueError):
return []
return parsed if isinstance(parsed, list) else []
@endpoints_bp.route('/asset/<int:asset_id>/status', methods=['GET'])
@require_permission('endpoints.view')
def panel_status(asset_id: int):
@@ -425,6 +444,19 @@ def panel_status(asset_id: int):
state.lastreportedat.isoformat() + 'Z' if state.lastreportedat else None),
]
fields = [f for f in fields if f['value'] not in (None, '')]
# Warn-only reconciler summary: monitors site policy expects to be running
# that the last check-in did not report as running. Surfaced as a status
# field so an operator sees the drift on the panel; nothing is auto-started.
drift = _drift_for_state(state)
if drift:
names = ', '.join(entry.get('displayname') or entry.get('name')
for entry in drift)
fields.append(field(
'Needs attention',
'{} monitor(s) expected running but not reported: {}'.format(
len(drift), names)))
return success_response({'fields': fields})
@@ -446,28 +478,100 @@ def panel_telemetry_meta(asset_id: int):
return success_response({'lastreportedat': stamp})
def _pctype_for_asset(asset_id):
"""Resolve an asset's PC type name, or None.
Endpoints stays optional-on-computers: a missing plugin, a non-computer
asset, or an unset type all resolve to None so the caller falls back to
reported-only behaviour.
"""
try:
from plugins.computers.models import Computer, ComputerType
except Exception:
return None
row = (db.session.query(ComputerType.computertype)
.join(Computer,
Computer.computertypeid == ComputerType.computertypeid)
.filter(Computer.assetid == asset_id)
.first())
return row[0] if row else None
def _resolved_monitors_for_asset(asset_id):
"""Monitor definitions attached to this asset's PC type by policy/shipped.
Reuses the endpoints monitor resolver (shipped defaults plus policy-attached
monitors). Returns a list of EndpointMonitor rows, or an empty list when the
PC type is unknown or nothing resolves.
"""
pctype = _pctype_for_asset(asset_id)
if not pctype:
return []
try:
from .monitors import _resolve_pctype_monitors
return _resolve_pctype_monitors(pctype) or []
except Exception:
return []
def _row_action_dict(app, kind, name, running):
"""Build the inline Start/Stop action map for a services/processes row."""
targets = row_actions_for(app, kind, name)
if not targets:
return None
return {
'running': running,
'start': targets['start'],
'stop': targets['stop'],
'status': targets['status'],
'target': targets['target'],
'targetkind': targets['targetkind'],
}
@endpoints_bp.route('/asset/<int:asset_id>/services', methods=['GET'])
@require_permission('endpoints.view')
def panel_services(asset_id: int):
"""List panel: monitored services/processes with a run-state badge per row.
Reshapes the latest servicesjson snapshot into rows the list renderer maps
to name + status badge. Tolerant of a missing or unexpected snapshot shape:
an unknown state renders as an Unknown badge rather than failing.
to name + status badge, then MERGES the monitors that site policy (and the
shipped defaults) attach to this asset's PC type. A policy-attached monitor
the agent has not yet reported surfaces as an Unknown row with a Start
control, so an operator sees the intent immediately rather than waiting for
the next agent snapshot. Tolerant of a missing snapshot or plugin: an
unknown state renders as an Unknown badge rather than failing.
"""
state = EndpointState.query.filter_by(assetid=asset_id).first()
if not state or not state.servicesjson:
return success_response([])
maycommand = caller_holds('endpoints.command')
try:
services = json.loads(state.servicesjson)
except (TypeError, ValueError):
return success_response([])
services = []
if state and state.servicesjson:
try:
services = json.loads(state.servicesjson) or []
except (TypeError, ValueError):
services = []
rows = []
for entry in (services or []):
# Reported identities, keyed (kind, lowercased name), so a policy monitor
# the agent already reports is not duplicated as a synthetic row.
reported = set()
# Warn-only reconciler drift, keyed (kind, lowercased name), so a row that is
# expected running but not reported running can be flagged for the operator.
# kind may be None on an older snapshot, so match on name too.
drift_keys = set()
for entry in _drift_for_state(state):
name = (entry.get('name') or '').strip().lower()
if not name:
continue
drift_keys.add(((entry.get('kind') or '').lower() or None, name))
drift_keys.add((None, name))
def _is_drift(kind, name):
key = (name or '').strip().lower()
return (kind, key) in drift_keys or (None, key) in drift_keys
for entry in services:
if isinstance(entry, dict):
name = entry.get('name') or entry.get('service') or 'Unknown'
raw = (entry.get('state') or entry.get('status') or '').strip().lower()
@@ -476,9 +580,10 @@ def panel_services(asset_id: int):
else:
name, raw, app, kind = str(entry), '', None, None
running = _service_running(entry)
normkind = (kind or '').lower() or None
row = {
'name': name,
'kind': (kind or '').lower() or None,
'kind': normkind,
'statelabel': 'Running' if running else (raw.title() or 'Unknown'),
# Amber for paused and the four Windows transitional states. A
# service stuck in Start Pending is failing to start on a retry
@@ -493,17 +598,42 @@ def panel_services(asset_id: int):
# emission never consulted the caller, so a read-only viewer was shown
# controls on a shop-floor PC that 403 on click - wrong signal, and it
# gets worse the moment guests can read this panel.
targets = row_actions_for(app, kind, name) if maycommand else None
if targets:
row['action'] = {
'running': running,
'start': targets['start'],
'stop': targets['stop'],
'status': targets['status'],
'target': targets['target'],
'targetkind': targets['targetkind'],
}
action = _row_action_dict(app, kind, name, running) if maycommand else None
if action:
row['action'] = action
if not running and _is_drift(normkind, name):
row['drift'] = True
rows.append(row)
reported.add((normkind, name.strip().lower()))
# Merge policy/shipped monitors this PC type should watch. A monitor already
# reported by the agent stays as its live row; one not yet seen is appended
# as an Unknown row so the operator sees the intended coverage and can Start
# it. The monitor's displayname is the operator label (e.g. 'msedge.exe'
# -> 'Edge'); its process/service name is the wire target for actions.
for monitor in _resolved_monitors_for_asset(asset_id):
kind = (monitor.kind or '').lower() or None
osname = (monitor.processname if kind == 'process'
else monitor.servicename)
osname = (osname or '').strip()
if not osname:
continue
if (kind, osname.lower()) in reported:
continue
row = {
'name': monitor.displayname or osname,
'kind': kind,
'statelabel': 'Unknown',
'statecolor': 'gray',
}
action = _row_action_dict(None, kind, osname, False) if maycommand else None
if action:
row['action'] = action
if _is_drift(kind, osname):
row['drift'] = True
rows.append(row)
reported.add((kind, osname.lower()))
return success_response(rows)
@@ -1335,6 +1465,64 @@ def enqueue_endpoint_command(asset_id: int):
message='Command queued', http_code=201)
@endpoints_bp.route('/monitors/resolved', methods=['GET'])
def resolved_monitors_for_agent():
"""Agent watch list: the monitors this endpoint should report on.
Collector-token auth (not JWT), same as the command channel. The agent
calls this each telemetry cycle with its hostname and receives the process
and service monitors that site policy (and the shipped defaults) attach to
this endpoint's PC type, so a policy-attached monitor is enumerated and
reported without editing appsettings.json on the box. The shape mirrors the
agent's MonitorConfig (processes[]/services[] of {name, app}) so the agent
can bind it directly: name is the wire identity (process image or service
key), app is the operator label. An unknown hostname or a PC type with no
monitors returns empty lists, which the agent merges as "watch nothing extra".
"""
error = _collector_auth_or_error()
if error is not None:
return error
hostname = (request.args.get('hostname') or '').strip()
if not hostname:
return error_response(ErrorCodes.VALIDATION_ERROR,
'hostname is required')
asset_id = _resolve_asset_for_hostname(hostname)
if asset_id is None:
return error_response(ErrorCodes.NOT_FOUND,
f'Unknown hostname: {hostname!r}', http_code=404)
processes = []
services = []
for monitor in _resolved_monitors_for_asset(asset_id):
kind = (monitor.kind or '').lower()
if kind == 'process':
wirename = (monitor.processname or '').strip()
bucket = processes
elif kind == 'service':
wirename = (monitor.servicename or '').strip()
bucket = services
else:
continue
if not wirename:
continue
entry = {
'name': wirename,
'app': (monitor.displayname or wirename),
}
# Only process monitors carry a launch path, and only when set. The
# agent never trusts it blindly: it validates the path against an
# approved-root allowlist before it becomes launchable (fail closed).
if kind == 'process':
launchpath = (monitor.launchpath or '').strip()
if launchpath:
entry['launchpath'] = launchpath
bucket.append(entry)
return success_response({'processes': processes, 'services': services})
@endpoints_bp.route('/commands/next', methods=['GET'])
def claim_next_command():
"""Agent long-poll: claim the next pending command for a hostname.