diff --git a/frontend/src/api/index.js b/frontend/src/api/index.js index 6bc0f1b3..57fcc970 100644 --- a/frontend/src/api/index.js +++ b/frontend/src/api/index.js @@ -178,6 +178,25 @@ export const endpointsApi = { }, telemetryMeta(assetId) { return api.get(`/endpoints/asset/${assetId}/telemetrymeta`) + }, + // Monitor catalog: reusable process/service definitions. Shipped rows are + // product-owned and read-only; local rows are user-built on the Settings page. + monitors: { + list() { + return api.get('/endpoints/monitors') + }, + create(data) { + return api.post('/endpoints/monitors', data) + }, + update(monitorid, data) { + return api.put(`/endpoints/monitors/${monitorid}`, data) + }, + remove(monitorid) { + return api.delete(`/endpoints/monitors/${monitorid}`) + }, + resolve(pctype) { + return api.get('/endpoints/monitors/resolve', { params: { pctype } }) + } } } diff --git a/plugins/endpoints/api/__init__.py b/plugins/endpoints/api/__init__.py index 2fc14121..53667058 100644 --- a/plugins/endpoints/api/__init__.py +++ b/plugins/endpoints/api/__init__.py @@ -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'] diff --git a/plugins/endpoints/api/monitors.py b/plugins/endpoints/api/monitors.py new file mode 100644 index 00000000..4e8f7e0e --- /dev/null +++ b/plugins/endpoints/api/monitors.py @@ -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/', 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/', 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], + }) diff --git a/plugins/endpoints/api/routes.py b/plugins/endpoints/api/routes.py index 087181c2..6085822c 100644 --- a/plugins/endpoints/api/routes.py +++ b/plugins/endpoints/api/routes.py @@ -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//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//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. diff --git a/plugins/endpoints/commands.py b/plugins/endpoints/commands.py index fdf3f14c..31b05376 100644 --- a/plugins/endpoints/commands.py +++ b/plugins/endpoints/commands.py @@ -185,6 +185,36 @@ def requires_target(action): return action_targetkind(action) is not None +# Monitor attachment as a policy action. A monitor definition is bound to a PC +# type by carrying a ConfigurationPolicyAction whose name is 'monitor:'. +# Distinct names let one policy attach several monitors and fold naturally +# (first-match-wins per name), and they are validated in the policy layer but +# never enqueued as commands because ALLOWED_ACTIONS_FOR intersects with +# COMMAND_ALLOWLIST, which carries no monitor: entries. +MONITOR_ACTION_PREFIX = 'monitor:' + + +def monitor_action_name(endpointmonitorid): + """Build the policy action name that attaches a monitor definition.""" + return f'{MONITOR_ACTION_PREFIX}{endpointmonitorid}' + + +def is_monitor_action(action): + """True when action names a monitor attachment (monitor:).""" + return isinstance(action, str) and action.startswith(MONITOR_ACTION_PREFIX) + + +def monitor_id_from_action(action): + """The monitor id an action attaches, or None when it is not a monitor action.""" + if not is_monitor_action(action): + return None + raw = action[len(MONITOR_ACTION_PREFIX):] + try: + return int(raw) + except (TypeError, ValueError): + return None + + def ALLOWED_ACTIONS_FOR(pctype=None, assetid=None): """The action names an endpoint may run. @@ -211,6 +241,44 @@ def ALLOWED_ACTIONS_FOR(pctype=None, assetid=None): # behavior). This is intentionally permissive; deny-by-default would be # a separate, explicit setting. return base + # Attaching a monitor expresses intent to watch a process/service AND to + # control it: the panel renders that row's Start/Stop/Status buttons, so the + # gate must permit the matching generic primitives or every click 409s + # "not permitted by policy". Expand each monitor: to its kind's control + # actions before intersecting. + allowed = allowed | _monitor_control_actions(allowed) # Never widen past the code-level allowlist: policy can only gate within it. return base & allowed + +def _monitor_control_actions(action_names): + """Control primitives granted by the monitor: attachments in a policy. + + A process monitor grants process_start/stop/status; a service monitor grants + service_start/stop/status. The monitor's kind is read from the catalog, so a + definition's kind is the single source of truth for what its row can do. + """ + ids = set() + for name in action_names: + if isinstance(name, str) and name.startswith(MONITOR_ACTION_PREFIX): + monitor_id = monitor_id_from_action(name) + if monitor_id is not None: + ids.add(monitor_id) + if not ids: + return set() + + try: + from .models import EndpointMonitor + except Exception: + return set() + + granted = set() + for monitor in EndpointMonitor.query.filter( + EndpointMonitor.endpointmonitorid.in_(ids)).all(): + kind = (monitor.kind or '').lower() + if kind == 'process': + granted |= {'process_start', 'process_stop', 'process_status'} + elif kind == 'service': + granted |= {'service_start', 'service_stop', 'service_status'} + return granted + diff --git a/plugins/endpoints/frontend/routes.js b/plugins/endpoints/frontend/routes.js index bb9e2d28..0bf67bd9 100644 --- a/plugins/endpoints/frontend/routes.js +++ b/plugins/endpoints/frontend/routes.js @@ -18,6 +18,12 @@ export default [ component: () => import('./views/AgentKeySettings.vue'), meta: { requiresAuth: true, requiresAdmin: true, plugin: 'endpoints' } }, + { + path: 'settings/endpoints/monitors', + name: 'endpoints-monitors-settings', + component: () => import('./views/MonitorsSettings.vue'), + meta: { requiresAuth: true, requiresAdmin: true, plugin: 'endpoints' } + }, // TEMPORARY / DEV ONLY. A manual bench to exercise the update_app and // run_script delivery path end to end while Cameron builds the real UI. // Gated on endpoints.deploy inside the view; remove once the production diff --git a/plugins/endpoints/frontend/views/AgentKeySettings.vue b/plugins/endpoints/frontend/views/AgentKeySettings.vue index 539a1e04..e3816c1e 100644 --- a/plugins/endpoints/frontend/views/AgentKeySettings.vue +++ b/plugins/endpoints/frontend/views/AgentKeySettings.vue @@ -307,7 +307,7 @@

Agent configuration

@@ -319,11 +319,16 @@ stamped into the build above so a deployed agent needs no local edits.

- Coming soon. These controls arrive with the C# agent - runtime. Until then the agent uses its built-in defaults. + Monitored processes and services are defined as reusable monitors and + attached to PC types through site policies, so a deployed agent fetches + the right set for its type.

-
@@ -332,8 +337,10 @@ diff --git a/plugins/endpoints/migrations/versions/0006_endpoint_monitors.py b/plugins/endpoints/migrations/versions/0006_endpoint_monitors.py new file mode 100644 index 00000000..db0bc87c --- /dev/null +++ b/plugins/endpoints/migrations/versions/0006_endpoint_monitors.py @@ -0,0 +1,151 @@ +"""Reusable endpoint monitor definitions (the monitor catalog). + +Adds endpointmonitors: one row per process or service the agent can watch, with +a bespoke display label separate from the raw OS name and optional launch and +collector paths. A definition is the "what"; attaching it to a PC type is a +separate site-policy concern. + +Seeds the four monitors we ship today, lifted from the agent's appsettings.json +so the catalog matches what a deployed agent already watches: + + process dncMain.exe -> eDNC + process udc.exe -> Universal Data Collector + process agent.exe -> MTConnect + service MTConnect UDC Agent -> MTConnect + +Shipped rows are product-owned (source='shipped'): the CRUD API refuses to edit +or delete them. Launch paths mirror the agent's launchpaths block; collector and +log paths are left null until site paths are supplied. + +Idempotent: creation checks the live schema first so a db.create_all() site +(which already built the table) re-runs clean on strict MySQL 8. Seeding is a +separate check on row count, so a site that built the table without the shipped +rows still gets them. +""" +from alembic import op + +import sqlalchemy as sa + +from datetime import datetime + + +# revision identifiers, used by Alembic. +revision = 'endpoints0006monitors' +down_revision = 'endpoints0005cmdpin' +branch_labels = None +depends_on = None + + +def _has_table(table_name): + from sqlalchemy import inspect + + return inspect(op.get_bind()).has_table(table_name) + + +def upgrade(): + bind = op.get_bind() + + if not _has_table('endpointmonitors'): + op.create_table( + 'endpointmonitors', + sa.Column('endpointmonitorid', sa.Integer(), nullable=False), + sa.Column('kind', sa.String(length=16), nullable=False), + sa.Column('displayname', sa.String(length=128), nullable=False), + sa.Column('processname', sa.String(length=255), nullable=True), + sa.Column('servicename', sa.String(length=255), nullable=True), + sa.Column('launchpath', sa.String(length=512), nullable=True), + sa.Column('datacollectorpath', sa.String(length=512), nullable=True), + sa.Column('logcollectorpath', sa.String(length=512), nullable=True), + sa.Column('source', sa.String(length=16), nullable=False, + server_default='local'), + sa.Column('isactive', sa.Boolean(), nullable=False, + server_default=sa.true()), + sa.Column('createdat', sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint('endpointmonitorid'), + sa.UniqueConstraint('kind', 'displayname', + name='uq_endpointmonitors_kinddisplay'), + ) + op.create_index('idx_endpointmonitors_kind', 'endpointmonitors', ['kind']) + op.create_index('idx_endpointmonitors_source', 'endpointmonitors', ['source']) + + # Seed the shipped catalog independently of table creation so a + # db.create_all() site (which already built the table) still gets the rows. + existing = bind.execute( + sa.text('SELECT COUNT(*) FROM endpointmonitors')).scalar() + if existing: + return + + monitors = sa.table( + 'endpointmonitors', + sa.column('kind', sa.String), + sa.column('displayname', sa.String), + sa.column('processname', sa.String), + sa.column('servicename', sa.String), + sa.column('launchpath', sa.String), + sa.column('datacollectorpath', sa.String), + sa.column('logcollectorpath', sa.String), + sa.column('source', sa.String), + sa.column('isactive', sa.Boolean), + sa.column('createdat', sa.DateTime), + ) + + now = datetime.utcnow() + op.bulk_insert(monitors, [ + { + 'kind': 'process', + 'displayname': 'eDNC', + 'processname': 'dncMain.exe', + 'servicename': None, + 'launchpath': r'C:\Program Files (x86)\Dnc\bin\DncMain.exe', + 'datacollectorpath': None, + 'logcollectorpath': None, + 'source': 'shipped', + 'isactive': True, + 'createdat': now, + }, + { + 'kind': 'process', + 'displayname': 'Universal Data Collector', + 'processname': 'udc.exe', + 'servicename': None, + 'launchpath': r'C:\Program Files\UDC\UDC.exe', + 'datacollectorpath': None, + 'logcollectorpath': None, + 'source': 'shipped', + 'isactive': True, + 'createdat': now, + }, + { + 'kind': 'process', + 'displayname': 'MTConnect', + 'processname': 'agent.exe', + 'servicename': None, + 'launchpath': r'C:\ProgramData\udc\mtconnect_udc\agent\run.bat', + 'datacollectorpath': None, + 'logcollectorpath': None, + 'source': 'shipped', + 'isactive': True, + 'createdat': now, + }, + { + 'kind': 'service', + 'displayname': 'MTConnect', + 'processname': None, + 'servicename': 'MTConnect UDC Agent', + 'launchpath': None, + 'datacollectorpath': None, + 'logcollectorpath': None, + 'source': 'shipped', + 'isactive': True, + 'createdat': now, + }, + ]) + + +def downgrade(): + if not _has_table('endpointmonitors'): + return + + op.drop_index('idx_endpointmonitors_source', table_name='endpointmonitors') + op.drop_index('idx_endpointmonitors_kind', table_name='endpointmonitors') + op.drop_table('endpointmonitors') diff --git a/plugins/endpoints/migrations/versions/0007_shipped_monitor_policy.py b/plugins/endpoints/migrations/versions/0007_shipped_monitor_policy.py new file mode 100644 index 00000000..eb43a6cc --- /dev/null +++ b/plugins/endpoints/migrations/versions/0007_shipped_monitor_policy.py @@ -0,0 +1,234 @@ +"""Ship the shopfloor app monitors as a read-only Site Policy. + +Wraps the four shipped endpointmonitors (eDNC, Universal Data Collector, +MTConnect process, MTConnect UDC Agent service) in a product-owned policy bundle +so the agent stops carrying them as hardcoded appsettings.json defaults and reads +its watch-list from the server instead. + +Seeds two computers-plugin rows, both source='shipped' (read-only in the UI, the +CRUD API refuses to edit or delete them): + + ConfigurationPolicy "Shopfloor App Monitors" + scope {"type":"true"} (every endpoint), one action per shipped monitor, + named 'monitor:', mode='gate', desiredstate=NULL. + + Gate, not desired, on purpose. A monitor REPORTS a process/service state and + exposes Start/Stop in the Monitor UI; the state read happens on manual + refresh and on agent check-in, so attaching a monitor never generates fleet + traffic on its own and a policy edit does not push a collection. A 'desired' + running state is a separate, deliberate opt-in that keeps an app alive by + trying to START the process when a launch path is mapped - never something + the product ships switched on for every endpoint. + + BusinessPolicy "RPCA Shopfloor App Monitors" + the site-level wrapper, carrying the Configuration Policy as its one item. + +Monitors are looked up by identity (kind, displayname), never by a hardcoded id, +so the bundle survives id drift across installs. + +Lives in the endpoints chain, after 0006, because the monitor rows it references +are seeded there. The computers policy tables are created by the core baseline +before per-plugin chains run; this migration still guards on their presence and +no-ops cleanly if they are absent, and is idempotent on the policy/business-policy +names so re-running never duplicates the bundle. +""" +from alembic import op + +import sqlalchemy as sa + +from datetime import datetime + + +# revision identifiers, used by Alembic. +revision = 'endpoints0007shippedpolicy' +down_revision = 'endpoints0006monitors' +branch_labels = None +depends_on = None + + +CONFIG_POLICY_NAME = 'Shopfloor App Monitors' +BUSINESS_POLICY_NAME = 'RPCA Shopfloor App Monitors' + +# The shipped monitors this bundle attaches, by identity (kind, displayname). +_SHIPPED_MONITORS = ( + ('process', 'eDNC'), + ('process', 'Universal Data Collector'), + ('process', 'MTConnect'), + ('service', 'MTConnect'), +) + + +def _tables(bind): + from sqlalchemy import inspect + + return set(inspect(bind).get_table_names()) + + +def upgrade(): + bind = op.get_bind() + tables = _tables(bind) + + # The bundle lives in the computers policy tables and references the + # endpoints monitor catalog. If either side is missing (a partial site), + # do nothing rather than fail - a later run reconciles once both exist. + required = {'endpointmonitors', 'configurationpolicies', + 'configurationpolicyactions', 'businesspolicies', + 'businesspolicyitems'} + if not required.issubset(tables): + return + + # Idempotent: never seed the bundle twice. + existing_config = bind.execute( + sa.text('SELECT configurationpolicyid FROM configurationpolicies ' + 'WHERE name = :name'), + {'name': CONFIG_POLICY_NAME}).first() + if existing_config: + return + + # Resolve the shipped monitor ids by identity. Skip any that a site has + # not got; attach whatever shipped monitors are present. + monitor_ids = [] + for kind, displayname in _SHIPPED_MONITORS: + row = bind.execute( + sa.text('SELECT endpointmonitorid FROM endpointmonitors ' + 'WHERE kind = :kind AND displayname = :displayname'), + {'kind': kind, 'displayname': displayname}).first() + if row: + monitor_ids.append(row[0]) + + if not monitor_ids: + return + + now = datetime.utcnow() + + config_policies = sa.table( + 'configurationpolicies', + sa.column('name', sa.String), + sa.column('description', sa.Text), + sa.column('scopejson', sa.Text), + sa.column('source', sa.String), + sa.column('createddate', sa.DateTime), + sa.column('modifieddate', sa.DateTime), + sa.column('isactive', sa.Boolean), + ) + op.bulk_insert(config_policies, [{ + 'name': CONFIG_POLICY_NAME, + 'description': 'Product-shipped watch-list for the shopfloor apps ' + '(eDNC, Universal Data Collector, MTConnect). Read-only.', + 'scopejson': '{"type": "true"}', + 'source': 'shipped', + 'createddate': now, + 'modifieddate': now, + 'isactive': True, + }]) + + config_policy_id = bind.execute( + sa.text('SELECT configurationpolicyid FROM configurationpolicies ' + 'WHERE name = :name'), + {'name': CONFIG_POLICY_NAME}).scalar() + + actions = sa.table( + 'configurationpolicyactions', + sa.column('configurationpolicyid', sa.Integer), + sa.column('action', sa.String), + sa.column('mode', sa.String), + sa.column('desiredstate', sa.Text), + sa.column('params', sa.Text), + sa.column('intervalminutes', sa.Integer), + sa.column('sortorder', sa.Integer), + sa.column('createddate', sa.DateTime), + sa.column('modifieddate', sa.DateTime), + sa.column('isactive', sa.Boolean), + ) + op.bulk_insert(actions, [{ + 'configurationpolicyid': config_policy_id, + 'action': f'monitor:{monitorid}', + 'mode': 'gate', + 'desiredstate': None, + 'params': None, + 'intervalminutes': None, + 'sortorder': index, + 'createddate': now, + 'modifieddate': now, + 'isactive': True, + } for index, monitorid in enumerate(monitor_ids)]) + + business_policies = sa.table( + 'businesspolicies', + sa.column('name', sa.String), + sa.column('description', sa.Text), + sa.column('priority', sa.Integer), + sa.column('source', sa.String), + sa.column('createddate', sa.DateTime), + sa.column('modifieddate', sa.DateTime), + sa.column('isactive', sa.Boolean), + ) + op.bulk_insert(business_policies, [{ + 'name': BUSINESS_POLICY_NAME, + 'description': 'Product-shipped Site Policy that applies the shopfloor ' + 'app monitors to every endpoint. Read-only.', + 'priority': 100, + 'source': 'shipped', + 'createddate': now, + 'modifieddate': now, + 'isactive': True, + }]) + + business_policy_id = bind.execute( + sa.text('SELECT businesspolicyid FROM businesspolicies ' + 'WHERE name = :name'), + {'name': BUSINESS_POLICY_NAME}).scalar() + + items = sa.table( + 'businesspolicyitems', + sa.column('businesspolicyid', sa.Integer), + sa.column('configurationpolicyid', sa.Integer), + sa.column('sortorder', sa.Integer), + sa.column('createddate', sa.DateTime), + sa.column('modifieddate', sa.DateTime), + sa.column('isactive', sa.Boolean), + ) + op.bulk_insert(items, [{ + 'businesspolicyid': business_policy_id, + 'configurationpolicyid': config_policy_id, + 'sortorder': 0, + 'createddate': now, + 'modifieddate': now, + 'isactive': True, + }]) + + +def downgrade(): + bind = op.get_bind() + tables = _tables(bind) + if 'businesspolicies' not in tables or 'configurationpolicies' not in tables: + return + + # Clear children explicitly rather than lean on ON DELETE CASCADE, which a + # test DB (sqlite) does not enforce by default. + business_policy_id = bind.execute( + sa.text('SELECT businesspolicyid FROM businesspolicies ' + 'WHERE name = :name'), + {'name': BUSINESS_POLICY_NAME}).scalar() + config_policy_id = bind.execute( + sa.text('SELECT configurationpolicyid FROM configurationpolicies ' + 'WHERE name = :name'), + {'name': CONFIG_POLICY_NAME}).scalar() + + if business_policy_id is not None: + bind.execute( + sa.text('DELETE FROM businesspolicyitems ' + 'WHERE businesspolicyid = :id'), + {'id': business_policy_id}) + bind.execute( + sa.text('DELETE FROM businesspolicies WHERE businesspolicyid = :id'), + {'id': business_policy_id}) + if config_policy_id is not None: + bind.execute( + sa.text('DELETE FROM configurationpolicyactions ' + 'WHERE configurationpolicyid = :id'), + {'id': config_policy_id}) + bind.execute( + sa.text('DELETE FROM configurationpolicies ' + 'WHERE configurationpolicyid = :id'), + {'id': config_policy_id}) diff --git a/plugins/endpoints/migrations/versions/0008_endpoint_drift.py b/plugins/endpoints/migrations/versions/0008_endpoint_drift.py new file mode 100644 index 00000000..1a1477f5 --- /dev/null +++ b/plugins/endpoints/migrations/versions/0008_endpoint_drift.py @@ -0,0 +1,51 @@ +"""Cache the warn-only reconciler drift on the endpoint state. + +Adds endpointstates.driftjson: a per-asset JSON snapshot of the monitors a site +policy (or the shipped bundle) expects to be running on an endpoint but which the +latest agent report does not show as running. It is computed server-side on every +agent check-in and overwritten in place, the same ledger pattern the other state +snapshots (servicesjson, portsjson) already use - one current row per asset, no +history. Warn-only: nothing here is auto-remediated, it exists so an operator can +see the drift and intervene manually. + +Nullable so an endpoint that has never checked in since the column landed reads +as "never computed" rather than "no drift". + +Idempotent: the column-add checks the live schema first, so a db.create_all() +site (which built the column from the model already) and a re-run both stay clean +on strict MySQL 8. +""" +from alembic import op + +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'endpoints0008drift' +down_revision = 'endpoints0007shippedpolicy' +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 _has_column('endpointstates', 'driftjson'): + return + op.add_column( + 'endpointstates', + sa.Column('driftjson', sa.Text(length=16777215), nullable=True)) + + +def downgrade(): + if not _has_column('endpointstates', 'driftjson'): + return + op.drop_column('endpointstates', 'driftjson') diff --git a/plugins/endpoints/models/__init__.py b/plugins/endpoints/models/__init__.py index 1e7936a9..1e059131 100644 --- a/plugins/endpoints/models/__init__.py +++ b/plugins/endpoints/models/__init__.py @@ -3,7 +3,9 @@ from .endpoint import ( EndpointState, EndpointEvent, EndpointCommand, EndpointApplication, ) +from .monitor import EndpointMonitor __all__ = [ 'EndpointState', 'EndpointEvent', 'EndpointCommand', 'EndpointApplication', + 'EndpointMonitor', ] diff --git a/plugins/endpoints/models/endpoint.py b/plugins/endpoints/models/endpoint.py index e859f125..445a0a7d 100644 --- a/plugins/endpoints/models/endpoint.py +++ b/plugins/endpoints/models/endpoint.py @@ -75,6 +75,13 @@ class EndpointState(db.Model): interfacesjson = db.Column(db.Text(length=16777215), nullable=True) scheduledtasksjson = db.Column(db.Text(length=16777215), nullable=True) + # Warn-only reconciler output, refreshed on every agent check-in in the same + # commit as servicesjson. Each entry names a policy/shipped-attached monitor + # the endpoint is expected to be running but the latest snapshot does not + # report as running, so an operator sees the drift and intervenes manually. + # This is warn-only: nothing here is auto-remediated. Null = never computed. + driftjson = db.Column(db.Text(length=16777215), nullable=True) + # Last-known MTConnect agent run state, cached so the WebUI toggle can show # a real state between command polls. Set by the agent's mtconnect_status # result. Null = never reported (unknown). diff --git a/plugins/endpoints/models/monitor.py b/plugins/endpoints/models/monitor.py new file mode 100644 index 00000000..d5b5a0a5 --- /dev/null +++ b/plugins/endpoints/models/monitor.py @@ -0,0 +1,87 @@ +"""Endpoint monitor definitions (the reusable monitor catalog). + +A monitor definition is the "what": a named process or service the agent should +watch, with a bespoke display label separate from the raw OS name, plus optional +launch and collector paths. Definitions are the noun that a configuration policy +action points at; attaching a monitor to a PC type is a policy concern, not a +property of the definition itself. + +Two sources: + + shipped Internal apps we ship (eDNC, MTConnect, UDC). Seeded by migration. + Read-only in the UI: name and identity are fixed by the product. + local User-defined monitors built on the Settings page. + +A definition is watched on an endpoint when a site policy in effect for that +endpoint's PC type includes a monitor action pointing at it. +""" + +from datetime import datetime + +from shopdb.api import db + + +def _utciso(value): + """Naive-UTC datetime -> ISO string marked UTC, or None.""" + return value.isoformat() + 'Z' if value else None + + +class EndpointMonitor(db.Model): + """One reusable process or service monitor definition.""" + + __tablename__ = 'endpointmonitors' + + endpointmonitorid = db.Column(db.Integer, primary_key=True) + + # 'process' or 'service'. Decides which identity column is authoritative and + # which start/stop/status actions the row exposes. + kind = db.Column(db.String(16), nullable=False) + + # The bespoke label operators see (e.g. 'MTConnect'). Distinct from the raw + # OS name because agent.exe is not obviously MTConnect. + displayname = db.Column(db.String(128), nullable=False) + + # OS identity. For a process monitor, processname is the executable the agent + # matches (e.g. 'agent.exe'); for a service monitor, servicename is the + # Windows service name (e.g. 'MTConnect UDC Agent'). Exactly one is the + # authoritative identity for a given kind; the other stays null. + processname = db.Column(db.String(255), nullable=True) + servicename = db.Column(db.String(255), nullable=True) + + # Optional paths supplied per monitor. launchpath starts the app; the + # collector paths point the agent at where the app writes data and logs. + launchpath = db.Column(db.String(512), nullable=True) + datacollectorpath = db.Column(db.String(512), nullable=True) + logcollectorpath = db.Column(db.String(512), nullable=True) + + # 'shipped' (product-owned, seeded) or 'local' (user-built). Shipped rows are + # not editable or deletable through the CRUD API. + source = db.Column(db.String(16), nullable=False, default='local') + + # Soft on/off so a monitor can be retired without losing policy references. + isactive = db.Column(db.Boolean, nullable=False, default=True) + + createdat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow) + + __table_args__ = ( + db.UniqueConstraint('kind', 'displayname', name='uq_endpointmonitors_kinddisplay'), + db.Index('idx_endpointmonitors_source', 'source'), + ) + + def __repr__(self): + return f"" + + def to_dict(self): + return { + 'endpointmonitorid': self.endpointmonitorid, + 'kind': self.kind, + 'displayname': self.displayname, + 'processname': self.processname, + 'servicename': self.servicename, + 'launchpath': self.launchpath, + 'datacollectorpath': self.datacollectorpath, + 'logcollectorpath': self.logcollectorpath, + 'source': self.source, + 'isactive': self.isactive, + 'createdat': _utciso(self.createdat), + } diff --git a/plugins/endpoints/plugin.py b/plugins/endpoints/plugin.py index aef48945..3dd4d881 100644 --- a/plugins/endpoints/plugin.py +++ b/plugins/endpoints/plugin.py @@ -486,26 +486,6 @@ class EndpointsPlugin(BasePlugin): # the installed-software inventory 'group': 'software', }, - { - # Installed-software (SBOM) inventory. Low-priority reference - # data, not something looked at on every visit, so it starts - # collapsed and sits near the bottom. Raw discovered list for - # now: no managed-application linkage is surfaced yet. - 'id': 'endpoints-applications', - 'title': 'Installed software', - 'assettypes': ['computer'], - 'endpoint': '/api/endpoints/asset/{assetid}/applications', - 'render': 'table', - 'collapsible': True, - 'collapsedDefault': True, - 'columns': [ - {'key': 'displayname', 'label': 'Name'}, - {'key': 'displayversion', 'label': 'Version'}, - {'key': 'publisher', 'label': 'Publisher'}, - ], - 'empty': 'No installed software reported.', - 'position': 30, - }, ] def get_dashboard_widgets(self) -> List[Dict]: @@ -822,6 +802,26 @@ class EndpointsPlugin(BasePlugin): state.lastreportedat = datetime.utcnow() + # Warn-only reconciler: on this check-in, compare the monitors site + # policy expects to be running against what the snapshot reports, and + # cache the drift on the state in THIS transaction (same ledger commit as + # servicesjson). No start/stop is issued - drift is surfaced for an + # operator to act on. Reads the freshly reported services when present, + # else the last-known snapshot, so a report that omits services still + # reconciles against what we last saw. Fail-open: a reconciler error must + # never fail an otherwise-good check-in, so drift is left unchanged. + try: + if 'services' in payload: + reported_services = payload.get('services') or [] + else: + reported_services = json.loads(state.servicesjson or '[]') + from .services.reconciler import compute_drift + state.driftjson = _as_json(compute_drift(assetid, reported_services)) + except Exception: + logger.exception( + 'reconciler failed for asset %s; leaving drift unchanged', + assetid) + # Log any keys the agent sent that the ingest path does not consume yet. # New agent telemetry (disks, bios, memory, ...) lands here first: it is # not persisted until the server grows columns for it, but logging the diff --git a/plugins/endpoints/services/__init__.py b/plugins/endpoints/services/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/plugins/endpoints/services/reconciler.py b/plugins/endpoints/services/reconciler.py new file mode 100644 index 00000000..c5140c75 --- /dev/null +++ b/plugins/endpoints/services/reconciler.py @@ -0,0 +1,153 @@ +"""Warn-only monitor reconciler. + +The reconciler answers one question on every agent check-in: for the monitors a +site policy (or the shipped bundle) attaches to this endpoint, is each one the +endpoint is EXPECTED to be running actually reported as running in the latest +snapshot? Every monitor that is expected-running but not observed-running is +DRIFT, and drift is surfaced to an operator for MANUAL intervention. + +This is warn-only. The reconciler NEVER starts, stops, or enqueues anything - it +compares observed state against expected state and records the difference. Whether +to later automate remediation (a keep-alive that starts a process when it drifts) +is a separate, deliberate opt-in the product does not ship switched on. + +Two seams are kept deliberately narrow so this stays easy to change: + + monitor_expects_running(monitor) + The single place that decides "should this monitor be up?". Today every + attached monitor is treated as expected-running (attachment implies intent + to run). When an observe-only monitor kind arrives - one that only collects + general state and must never be flagged as drift - this is the one function + that branches on it. Nothing else in the reconciler encodes that policy. + + compute_drift(asset_id, reported_services) + Pure given its inputs: it resolves the attached monitors and diffs them + against the reported running set. It reads no request state and writes no + rows, so the ingest path can call it inside its own transaction and tests + can call it directly. +""" + +# The running-state vocabulary a watched process/service can report. Kept in +# lockstep with plugins/endpoints/api/routes.py::_RUNNING_STATES so a service the +# panel paints green is never flagged as drift here. Both read from the same +# agent snapshot; disagreeing would show a running app as out-of-desired-state. +_RUNNING_STATES = ('running', 'started', 'up', 'ok', 'active') + + +def monitor_expects_running(monitor): + """Whether an attached monitor is expected to be running on an endpoint. + + The seam for the "should be up" decision. Today attachment implies intent to + run, so every active monitor returns True. A future observe-only monitor kind + (general state collection, never keep-alive) branches HERE and returns False + so it is never reported as drift - no other part of the reconciler needs to + change. + """ + return bool(getattr(monitor, 'isactive', True)) + + +def _monitor_identity(monitor): + """The (kind, osname) a monitor watches, or None when it has no identity. + + Mirrors how the services panel reads a monitor's OS identity: a process + monitor is keyed by its processname, a service monitor by its servicename. + """ + kind = (getattr(monitor, 'kind', '') or '').lower() or None + if kind == 'process': + osname = getattr(monitor, 'processname', None) + elif kind == 'service': + osname = getattr(monitor, 'servicename', None) + else: + osname = None + osname = (osname or '').strip() + if not kind or not osname: + return None + return kind, osname + + +def _reported_running_keys(reported_services): + """Set of (kind, lowercased-name) the snapshot reports as RUNNING. + + Tolerant of the two shapes the agent snapshot uses: dict entries with + name/service and state/status (and an optional kind), or bare strings. + Only running entries land in the set; a stopped or unknown entry is absent, + which is what makes an expected-running monitor read as drift. + """ + running = set() + for entry in reported_services or (): + if isinstance(entry, dict): + name = entry.get('name') or entry.get('service') or '' + raw = (entry.get('state') or entry.get('status') or '').strip().lower() + kind = (entry.get('kind') or '').lower() or None + else: + name, raw, kind = str(entry), '', None + name = name.strip().lower() + if not name or raw not in _RUNNING_STATES: + continue + running.add((kind, name)) + return running + + +def _is_running(kind, osname, running_keys): + """Whether a monitor's identity appears running in the snapshot. + + A snapshot entry may omit kind (older agents, or a plain string list), so a + match on name alone under an unknown kind still counts as running - the + reconciler must not invent drift just because the agent did not label a row. + """ + needle = osname.strip().lower() + if (kind, needle) in running_keys: + return True + if (None, needle) in running_keys: + return True + return False + + +def compute_drift(asset_id, reported_services): + """Compute the current drift list for an endpoint. Pure, no writes. + + Resolves the monitors attached to the asset's PC type (shipped defaults plus + site policy), keeps only those that are expected-running, and returns one + entry per monitor whose identity is not reported as running. Returns an empty + list when nothing is attached or nothing has drifted, so the caller can store + it as "computed, no drift" distinct from a null "never computed". + + Isolated from Flask request state: it reads the monitor catalog and the + passed-in snapshot only, so the ingest path calls it inside its own + transaction and tests call it directly. + """ + try: + from plugins.endpoints.api.routes import _resolved_monitors_for_asset + except Exception: + return [] + + monitors = _resolved_monitors_for_asset(asset_id) or [] + if not monitors: + return [] + + running_keys = _reported_running_keys(reported_services) + + drift = [] + seen = set() + for monitor in monitors: + if not monitor_expects_running(monitor): + continue + identity = _monitor_identity(monitor) + if identity is None: + continue + kind, osname = identity + key = (kind, osname.lower()) + if key in seen: + continue + seen.add(key) + if _is_running(kind, osname, running_keys): + continue + drift.append({ + 'monitorid': getattr(monitor, 'endpointmonitorid', None), + 'kind': kind, + 'name': osname, + 'displayname': getattr(monitor, 'displayname', None) or osname, + 'expected': 'running', + 'observed': 'notrunning', + }) + return drift diff --git a/shopdb/plugins/alembic_template.py b/shopdb/plugins/alembic_template.py index 3c5590a4..c7ea51a5 100644 --- a/shopdb/plugins/alembic_template.py +++ b/shopdb/plugins/alembic_template.py @@ -53,7 +53,7 @@ PLUGIN_TABLE_OWNERS: dict[str, Iterable[str]] = { 'configurationpolicies', 'configurationpolicyactions', 'businesspolicies', 'businesspolicyitems'), 'employees': ('directoryemployees',), - 'endpoints': ('endpointstates', 'endpointevents', 'endpointcommands', 'endpointapplications'), + 'endpoints': ('endpointstates', 'endpointevents', 'endpointcommands', 'endpointapplications', 'endpointmonitors'), 'geenforce': ('manifestscopes', 'manifestentries', 'manifestentrypctypes', 'manifestentryhostnames', 'manifestentrymachinenumbers', 'manifestinusechecks', 'manifestinusecheckprocesses', diff --git a/tests/test_core/test_public_read_surface.py b/tests/test_core/test_public_read_surface.py index 992c0ce1..e8c55c7f 100644 --- a/tests/test_core/test_public_read_surface.py +++ b/tests/test_core/test_public_read_surface.py @@ -96,6 +96,14 @@ PUBLIC_ENDPOINTS = { 'endpoints.download_app_package', 'endpoints.download_app_package_by_app', 'endpoints.download_script_body', + + # The endpoint agent fetches its server-resolved monitor watch-list (the + # processes/services site policy attaches to its PC type) on each telemetry + # cycle. Authenticated inside the handler by the endpoints collector API key + # - the same shape as claim_next_command - because a bay has no user session. + # It is a read of already-scoped policy output, not an authority: attaching a + # monitor is gated on the policy editor, audited with the requesting user. + 'endpoints.resolved_monitors_for_agent', } # Reads that need a LOGIN but no named permission, because the handler decides diff --git a/tests/test_plugin_migrations.py b/tests/test_plugin_migrations.py index 25159e49..fb57290b 100644 --- a/tests/test_plugin_migrations.py +++ b/tests/test_plugin_migrations.py @@ -92,9 +92,11 @@ EXPECTED_HEAD_REVISION['warranty'] = 'warranty0002proof' # endpoints is a NEW post-cutover plugin: its 0001 genuinely creates the # endpointstates + endpointevents tables the core chain never owned, 0002 # 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'] = 'endpoints0005cmdpin' +# adds the generic command target/targetkind columns, 0004 adds the +# endpointapplications SBOM snapshot table, 0005 pins a command sha256, 0006 +# seeds the reusable monitor catalog, 0007 ships the shopfloor monitor policy +# bundle, and 0008 adds the warn-only reconciler drift cache column. +EXPECTED_HEAD_REVISION['endpoints'] = 'endpoints0008drift' # Plugins built after the cutover: their 0001 baseline really creates tables the # core chain never owned. diff --git a/tests/test_plugins/test_endpoints_policy_monitors.py b/tests/test_plugins/test_endpoints_policy_monitors.py new file mode 100644 index 00000000..e088e8c4 --- /dev/null +++ b/tests/test_plugins/test_endpoints_policy_monitors.py @@ -0,0 +1,199 @@ +"""Policy-attached monitors surface in the Tracked Services/Processes panel. + +A monitor attached to a PC type through an active Business Policy (over a scoped +Configuration Policy) must appear in GET /api/endpoints/asset//services even +before the agent has reported that process/service, so the PC Detail Monitors +tab shows the intended coverage as an Unknown row. +""" + +from shopdb.core.models import Asset, AssetType, Setting + + +SERVICES_URL = '/api/endpoints/asset/{}/services' + + +def _computer(db, hostname, computertype): + """Create a computer Asset + Computer of the given PC type; return the asset.""" + from plugins.computers.models import Computer, ComputerType + + atype = AssetType.query.filter_by(assettype='computer').first() + if not atype: + atype = AssetType(assettype='computer') + db.session.add(atype) + db.session.flush() + + ctype = ComputerType.query.filter_by(computertype=computertype).first() + if not ctype: + ctype = ComputerType(computertype=computertype) + db.session.add(ctype) + db.session.flush() + + asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid, + isactive=True) + db.session.add(asset) + db.session.flush() + comp = Computer(assetid=asset.assetid, hostname=hostname, + computertypeid=ctype.computertypeid) + db.session.add(comp) + db.session.flush() + return asset, ctype + + +def _attach_monitor_to_pctype(db, monitor, computertypeid): + """Create a Configuration Policy scoped to the PC type carrying the monitor, + wrapped in an active Business Policy, with the feature flag enabled.""" + from plugins.computers.models import ( + ConfigurationPolicy, ConfigurationPolicyAction, + BusinessPolicy, BusinessPolicyItem, + ) + + db.session.add(Setting(key='computers_configpolicies_enabled', value='true')) + + policy = ConfigurationPolicy( + name='Edge coverage', + scopejson={'type': 'pctype', 'id': computertypeid}, + ) + db.session.add(policy) + db.session.flush() + db.session.add(ConfigurationPolicyAction( + configurationpolicyid=policy.configurationpolicyid, + action=f'monitor:{monitor.endpointmonitorid}', + mode='gate', + )) + + business = BusinessPolicy(name='Test Policy', priority=100) + db.session.add(business) + db.session.flush() + db.session.add(BusinessPolicyItem( + businesspolicyid=business.businesspolicyid, + configurationpolicyid=policy.configurationpolicyid, + sortorder=0, + )) + db.session.commit() + + +def test_policy_attached_process_monitor_surfaces_as_unknown_row( + client, db, auth_headers): + """A msedge.exe process monitor attached to a PC type by policy appears in + the services panel as an Unknown process row, even with no agent report.""" + from plugins.endpoints.models import EndpointMonitor + + asset, ctype = _computer(db, 'F18BCDB4', 'CMM PC') + monitor = EndpointMonitor( + kind='process', displayname='msedge.exe', processname='msedge.exe', + source='local', isactive=True, + ) + db.session.add(monitor) + db.session.flush() + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + response = client.get(SERVICES_URL.format(asset.assetid), headers=auth_headers) + assert response.status_code == 200 + rows = response.get_json()['data'] + + edge = next((r for r in rows if r['name'] == 'msedge.exe'), None) + assert edge is not None, 'policy-attached monitor missing from panel' + assert edge['kind'] == 'process' + assert edge['statelabel'] == 'Unknown' + assert edge['statecolor'] == 'gray' + # Admin holds endpoints.command, so the row offers a Start control whose + # wire target is the OS process name. + assert edge['action']['target'] == 'msedge.exe' + assert edge['action']['targetkind'] == 'process' + assert edge['action']['running'] is False + + +def test_flag_off_hides_policy_monitor_from_panel(client, db, auth_headers): + """With the feature flag off the panel returns only reported rows (none).""" + from plugins.endpoints.models import EndpointMonitor + + asset, ctype = _computer(db, 'F18BCDB5', 'CMM PC') + monitor = EndpointMonitor( + kind='process', displayname='msedge.exe', processname='msedge.exe', + source='local', isactive=True, + ) + db.session.add(monitor) + db.session.flush() + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + # Flip the flag off after the helper turned it on. + flag = Setting.query.filter_by( + key='computers_configpolicies_enabled').first() + flag.value = 'false' + db.session.commit() + + response = client.get(SERVICES_URL.format(asset.assetid), headers=auth_headers) + assert response.status_code == 200 + rows = response.get_json()['data'] + assert not any(r['name'] == 'msedge.exe' for r in rows) + + +RESOLVED_URL = '/api/endpoints/monitors/resolved' +COLLECTOR_KEY = 'testcollectorkey' + + +def _collector_headers(app): + """Authorize the agent-facing route on the shared collector key.""" + app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY + return {'X-API-Key': COLLECTOR_KEY} + + +def test_resolved_monitors_returns_policy_process_for_agent(client, db, app): + """The agent watch-list route returns a policy-attached process monitor, + shaped as processes[]/services[] of {name, app}, keyed by hostname.""" + from plugins.endpoints.models import EndpointMonitor + + asset, ctype = _computer(db, 'AGENTPC1', 'CMM PC') + monitor = EndpointMonitor( + kind='process', displayname='Edge', processname='msedge.exe', + source='local', isactive=True, + ) + db.session.add(monitor) + db.session.flush() + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + response = client.get(RESOLVED_URL, query_string={'hostname': 'AGENTPC1'}, + headers=_collector_headers(app)) + assert response.status_code == 200 + data = response.get_json()['data'] + assert {'name': 'msedge.exe', 'app': 'Edge'} in data['processes'] + assert data['services'] == [] + + +def test_resolved_monitors_carries_launchpath_for_process(client, db, app): + """A process monitor with a launch path set surfaces launchpath on the wire + so the agent can honor a site-provisioned Start path (after validation).""" + from plugins.endpoints.models import EndpointMonitor + + asset, ctype = _computer(db, 'AGENTPC3', 'CMM PC') + monitor = EndpointMonitor( + kind='process', displayname='Edge', processname='msedge.exe', + launchpath=r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe', + source='local', isactive=True, + ) + db.session.add(monitor) + db.session.flush() + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + response = client.get(RESOLVED_URL, query_string={'hostname': 'AGENTPC3'}, + headers=_collector_headers(app)) + assert response.status_code == 200 + data = response.get_json()['data'] + edge = next((p for p in data['processes'] if p['name'] == 'msedge.exe'), None) + assert edge is not None + assert edge['launchpath'] == \ + r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe' + + +def test_resolved_monitors_requires_collector_auth(client, db, app): + """Without the collector key the agent route refuses (no JWT fallback).""" + _computer(db, 'AGENTPC2', 'CMM PC') + app.config['COLLECTOR_API_KEY'] = COLLECTOR_KEY + response = client.get(RESOLVED_URL, query_string={'hostname': 'AGENTPC2'}) + assert response.status_code in (401, 403) + + +def test_resolved_monitors_unknown_hostname_is_404(client, db, app): + """A hostname with no computer row is a 404, not an empty 200.""" + response = client.get(RESOLVED_URL, query_string={'hostname': 'NOSUCHBOX'}, + headers=_collector_headers(app)) + assert response.status_code == 404 diff --git a/tests/test_plugins/test_endpoints_reconciler.py b/tests/test_plugins/test_endpoints_reconciler.py new file mode 100644 index 00000000..bd3f7164 --- /dev/null +++ b/tests/test_plugins/test_endpoints_reconciler.py @@ -0,0 +1,238 @@ +"""Warn-only reconciler: drift computation and its exposure on the panels. + +The reconciler runs server-side on agent check-in. For every monitor a site +policy attaches to an endpoint's PC type that is EXPECTED to be running, it flags +DRIFT when the latest snapshot does not report it running. It never starts or +stops anything - drift is surfaced for an operator to act on manually. + +These tests cover the pure computer (compute_drift), the check-in write path +(apply_collector_payload persisting driftjson), the panel exposure, and the +observe-only seam (monitor_expects_running) that keeps a non-keepalive monitor +out of the drift list. +""" + +from shopdb.core.models import Asset, AssetType, Setting + + +SERVICES_URL = '/api/endpoints/asset/{}/services' +STATUS_URL = '/api/endpoints/asset/{}/status' + + +def _computer(db, hostname, computertype): + """Create a computer Asset + Computer of the given PC type; return the pair.""" + from plugins.computers.models import Computer, ComputerType + + atype = AssetType.query.filter_by(assettype='computer').first() + if not atype: + atype = AssetType(assettype='computer') + db.session.add(atype) + db.session.flush() + + ctype = ComputerType.query.filter_by(computertype=computertype).first() + if not ctype: + ctype = ComputerType(computertype=computertype) + db.session.add(ctype) + db.session.flush() + + asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid, + isactive=True) + db.session.add(asset) + db.session.flush() + comp = Computer(assetid=asset.assetid, hostname=hostname, + computertypeid=ctype.computertypeid) + db.session.add(comp) + db.session.flush() + return asset, ctype + + +def _attach_monitor_to_pctype(db, monitor, computertypeid): + """Attach a monitor to a PC type via an active Business/Configuration Policy.""" + from plugins.computers.models import ( + ConfigurationPolicy, ConfigurationPolicyAction, + BusinessPolicy, BusinessPolicyItem, + ) + + if not Setting.query.filter_by( + key='computers_configpolicies_enabled').first(): + db.session.add( + Setting(key='computers_configpolicies_enabled', value='true')) + + policy = ConfigurationPolicy( + name='Coverage {}'.format(monitor.endpointmonitorid), + scopejson={'type': 'pctype', 'id': computertypeid}, + ) + db.session.add(policy) + db.session.flush() + db.session.add(ConfigurationPolicyAction( + configurationpolicyid=policy.configurationpolicyid, + action='monitor:{}'.format(monitor.endpointmonitorid), + mode='gate', + )) + business = BusinessPolicy(name='BP {}'.format(monitor.endpointmonitorid), + priority=100) + db.session.add(business) + db.session.flush() + db.session.add(BusinessPolicyItem( + businesspolicyid=business.businesspolicyid, + configurationpolicyid=policy.configurationpolicyid, + sortorder=0, + )) + db.session.commit() + + +def _process_monitor(db, displayname, processname): + from plugins.endpoints.models import EndpointMonitor + + monitor = EndpointMonitor( + kind='process', displayname=displayname, processname=processname, + source='local', isactive=True, + ) + db.session.add(monitor) + db.session.flush() + return monitor + + +# --------------------------------------------------------------------------- +# Pure computer +# --------------------------------------------------------------------------- + +def test_compute_drift_flags_attached_monitor_not_reported_running( + client, db, app): + """An expected-running monitor absent from the snapshot is drift.""" + from plugins.endpoints.services.reconciler import compute_drift + + asset, ctype = _computer(db, 'DRIFT01', 'CMM PC') + monitor = _process_monitor(db, 'eDNC', 'dncMain.exe') + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + drift = compute_drift(asset.assetid, []) + assert len(drift) == 1 + assert drift[0]['name'] == 'dncMain.exe' + assert drift[0]['kind'] == 'process' + assert drift[0]['expected'] == 'running' + assert drift[0]['observed'] == 'notrunning' + + +def test_compute_drift_empty_when_monitor_reported_running(client, db, app): + """A monitor the snapshot reports running is not drift.""" + from plugins.endpoints.services.reconciler import compute_drift + + asset, ctype = _computer(db, 'DRIFT02', 'CMM PC') + monitor = _process_monitor(db, 'eDNC', 'dncMain.exe') + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + reported = [{'name': 'dncMain.exe', 'kind': 'process', 'state': 'running'}] + assert compute_drift(asset.assetid, reported) == [] + + +def test_compute_drift_empty_when_no_monitor_attached(client, db, app): + """No attached monitors means nothing to reconcile.""" + from plugins.endpoints.services.reconciler import compute_drift + + asset, _ = _computer(db, 'DRIFT03', 'CMM PC') + assert compute_drift(asset.assetid, []) == [] + + +def test_monitor_expects_running_is_the_observe_only_seam(): + """The 'should be up' decision lives in one function; inactive is excluded.""" + from plugins.endpoints.services.reconciler import monitor_expects_running + + class _M: + isactive = True + + active = _M() + inactive = _M() + inactive.isactive = False + assert monitor_expects_running(active) is True + assert monitor_expects_running(inactive) is False + + +# --------------------------------------------------------------------------- +# Check-in write path +# --------------------------------------------------------------------------- + +def test_checkin_persists_drift_on_endpoint_state(client, db, app): + """apply_collector_payload caches drift on the state in the same commit.""" + from plugins.endpoints.plugin import EndpointsPlugin + from plugins.endpoints.models import EndpointState + + asset, ctype = _computer(db, 'CHKIN01', 'CMM PC') + monitor = _process_monitor(db, 'eDNC', 'dncMain.exe') + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + plugin = EndpointsPlugin() + plugin.apply_collector_payload({'hostname': 'CHKIN01', 'services': []}) + db.session.commit() + + state = EndpointState.query.filter_by(assetid=asset.assetid).first() + assert state.driftjson is not None + import json + drift = json.loads(state.driftjson) + assert [d['name'] for d in drift] == ['dncMain.exe'] + + +def test_checkin_clears_drift_when_process_returns(client, db, app): + """A later report that shows the process running clears the cached drift.""" + from plugins.endpoints.plugin import EndpointsPlugin + from plugins.endpoints.models import EndpointState + import json + + asset, ctype = _computer(db, 'CHKIN02', 'CMM PC') + monitor = _process_monitor(db, 'eDNC', 'dncMain.exe') + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + plugin = EndpointsPlugin() + plugin.apply_collector_payload({'hostname': 'CHKIN02', 'services': []}) + db.session.commit() + + plugin.apply_collector_payload({ + 'hostname': 'CHKIN02', + 'services': [{'name': 'dncMain.exe', 'kind': 'process', + 'state': 'running'}], + }) + db.session.commit() + + state = EndpointState.query.filter_by(assetid=asset.assetid).first() + assert json.loads(state.driftjson) == [] + + +# --------------------------------------------------------------------------- +# Panel exposure +# --------------------------------------------------------------------------- + +def test_status_panel_surfaces_drift_warning(client, db, app, auth_headers): + """The status panel adds a Needs attention field when drift is cached.""" + from plugins.endpoints.plugin import EndpointsPlugin + + asset, ctype = _computer(db, 'PANEL01', 'CMM PC') + monitor = _process_monitor(db, 'eDNC', 'dncMain.exe') + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + plugin = EndpointsPlugin() + plugin.apply_collector_payload({'hostname': 'PANEL01', 'services': []}) + db.session.commit() + + response = client.get(STATUS_URL.format(asset.assetid), headers=auth_headers) + assert response.status_code == 200 + labels = [f['label'] for f in response.get_json()['data']['fields']] + assert 'Needs attention' in labels + + +def test_services_panel_marks_drifted_row(client, db, app, auth_headers): + """The synthetic Unknown row for an expected-running monitor is drift-flagged.""" + from plugins.endpoints.plugin import EndpointsPlugin + + asset, ctype = _computer(db, 'PANEL02', 'CMM PC') + monitor = _process_monitor(db, 'eDNC', 'dncMain.exe') + _attach_monitor_to_pctype(db, monitor, ctype.computertypeid) + + plugin = EndpointsPlugin() + plugin.apply_collector_payload({'hostname': 'PANEL02', 'services': []}) + db.session.commit() + + response = client.get(SERVICES_URL.format(asset.assetid), headers=auth_headers) + assert response.status_code == 200 + rows = response.get_json()['data'] + edenc = next((r for r in rows if r['name'] == 'eDNC'), None) + assert edenc is not None + assert edenc.get('drift') is True diff --git a/tests/test_plugins/test_endpoints_shipped_monitor_policy.py b/tests/test_plugins/test_endpoints_shipped_monitor_policy.py new file mode 100644 index 00000000..bbdccfeb --- /dev/null +++ b/tests/test_plugins/test_endpoints_shipped_monitor_policy.py @@ -0,0 +1,141 @@ +"""The shipped shopfloor app monitors ride a read-only Site Policy. + +Migration endpoints0007shippedpolicy offloads the eDNC/UDC/MTConnect monitors +from the agent's hardcoded appsettings.json into a product-owned policy bundle: +a shipped ConfigurationPolicy "Shopfloor App Monitors" (global scope, one +desired-state monitor action per shipped monitor) wrapped in a shipped +BusinessPolicy "RPCA Shopfloor App Monitors". + +Run the endpoints chain against a fresh engine, alongside the computers policy +tables the bundle writes into, and assert the bundle exists, is source='shipped' +(read-only), and attaches every shipped monitor as a desired/running:true action. +""" + +import importlib.util +import json +from pathlib import Path + +from sqlalchemy import create_engine, inspect, text +from alembic.migration import MigrationContext +from alembic.operations import Operations + + +_ROOT = Path(__file__).resolve().parent.parent.parent +_ENDPOINTS_VERSIONS = _ROOT / 'plugins' / 'endpoints' / 'migrations' / 'versions' +_COMPUTERS_VERSIONS = _ROOT / 'plugins' / 'computers' / 'migrations' / 'versions' + +CONFIG_POLICY_NAME = 'Shopfloor App Monitors' +BUSINESS_POLICY_NAME = 'RPCA Shopfloor App Monitors' + + +def _load(path, prefix): + spec = importlib.util.spec_from_file_location(prefix + path.stem, str(path)) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _run_bundle_migrations(connection): + """Build the policy tables (computers 0004), the monitor catalog (endpoints + 0006) and the shipped bundle (endpoints 0007) on one engine, in order. + + The core baseline creates every plugin's tables before per-plugin chains run, + so on a real deploy the computers policy tables already exist when endpoints + 0007 runs. Here we stand up exactly the three migrations the bundle depends + on - the earlier endpoints revisions carry FKs to core tables that a bare + engine does not have, and are irrelevant to what 0007 seeds. + """ + operations = Operations(MigrationContext.configure(connection)) + operations._install_proxy() + try: + _load(_COMPUTERS_VERSIONS / '0004_policies.py', + 'computers_mig_').upgrade() + _load(_ENDPOINTS_VERSIONS / '0006_endpoint_monitors.py', + 'endpoints_mig_0006_').upgrade() + _load(_ENDPOINTS_VERSIONS / '0007_shipped_monitor_policy.py', + 'endpoints_mig_0007_').upgrade() + finally: + operations._remove_proxy() + connection.commit() + + +def test_shipped_bundle_seeds_read_only_site_policy(): + engine = create_engine('sqlite://') + with engine.connect() as connection: + _run_bundle_migrations(connection) + + config = connection.execute(text( + 'SELECT configurationpolicyid, source, scopejson ' + 'FROM configurationpolicies WHERE name = :name'), + {'name': CONFIG_POLICY_NAME}).first() + assert config is not None, 'shipped Configuration Policy was not seeded' + config_id, config_source, scopejson = config + assert config_source == 'shipped' + assert json.loads(scopejson) == {'type': 'true'} + + business = connection.execute(text( + 'SELECT businesspolicyid, source FROM businesspolicies ' + 'WHERE name = :name'), {'name': BUSINESS_POLICY_NAME}).first() + assert business is not None, 'shipped Business Policy was not seeded' + business_id, business_source = business + assert business_source == 'shipped' + + item = connection.execute(text( + 'SELECT configurationpolicyid FROM businesspolicyitems ' + 'WHERE businesspolicyid = :id'), {'id': business_id}).first() + assert item is not None and item[0] == config_id + + +def test_shipped_bundle_attaches_every_shipped_monitor(): + engine = create_engine('sqlite://') + with engine.connect() as connection: + _run_bundle_migrations(connection) + + monitor_ids = {row[0] for row in connection.execute(text( + 'SELECT endpointmonitorid FROM endpointmonitors ' + "WHERE source = 'shipped'"))} + assert monitor_ids, 'no shipped monitors to attach' + + config_id = connection.execute(text( + 'SELECT configurationpolicyid FROM configurationpolicies ' + 'WHERE name = :name'), {'name': CONFIG_POLICY_NAME}).scalar() + + actions = connection.execute(text( + 'SELECT action, mode, desiredstate FROM configurationpolicyactions ' + 'WHERE configurationpolicyid = :id'), {'id': config_id}).all() + + attached_ids = set() + for action, mode, desiredstate in actions: + assert action.startswith('monitor:') + # Gate, not desired: the shipped bundle reports state and exposes + # Start/Stop in the UI, but never auto-starts a process. A desired + # running state is a deliberate per-monitor opt-in, not shipped. + assert mode == 'gate' + assert desiredstate is None + attached_ids.add(int(action[len('monitor:'):])) + + assert attached_ids == monitor_ids + + +def test_shipped_bundle_is_idempotent(): + engine = create_engine('sqlite://') + with engine.connect() as connection: + _run_bundle_migrations(connection) + # Re-run only the shipped-bundle migration; it must not duplicate rows. + operations = Operations(MigrationContext.configure(connection)) + operations._install_proxy() + try: + _load(_ENDPOINTS_VERSIONS / '0007_shipped_monitor_policy.py', + 'endpoints_mig_rerun_').upgrade() + finally: + operations._remove_proxy() + connection.commit() + + config_count = connection.execute(text( + 'SELECT COUNT(*) FROM configurationpolicies WHERE name = :name'), + {'name': CONFIG_POLICY_NAME}).scalar() + business_count = connection.execute(text( + 'SELECT COUNT(*) FROM businesspolicies WHERE name = :name'), + {'name': BUSINESS_POLICY_NAME}).scalar() + assert config_count == 1 + assert business_count == 1