"""Endpoints plugin main class. Endpoint RMM telemetry reported by the ShopDB agent. Accepts collector input over HTTPS per ADR-006: the agent posts a per-host payload to POST /api/collector/endpoints, the plugin resolves the reported hostname to an existing computer asset, overwrites that asset's cached EndpointState, and appends any curated Windows events the report carried - all in one commit, so the cached snapshot and the events it summarizes never disagree. Identity is the hostname, resolved against the computers plugin. An unresolved hostname is reported as a ValueError (surfaced to the caller as a validation error), never silently dropped: an unknown host is a real gap between the fleet and ShopDB, the same discipline the backups collector uses for machine numbers. """ import json import logging from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Type from flask import Flask, Blueprint from shopdb.plugins.base import BasePlugin, PluginMeta from shopdb.api import db from .api import endpoints_bp from .eventcatalog import EVENT_CATALOG from .models import EndpointState, EndpointEvent logger = logging.getLogger(__name__) # The fixed catalog of Windows Event IDs the agent forwards. Derived from # EVENT_CATALOG (eventcatalog.py) so the ingest filter and the display titles # can never drift. IDs outside this set are ignored on ingest rather than # stored, keeping endpointevents a readable security/stability history instead # of a raw event firehose. CURATED_EVENT_IDS = frozenset(EVENT_CATALOG) def _parse_dt(value): """Best-effort parse of an ISO-8601 string to naive UTC, or None.""" if not value: return None if isinstance(value, datetime): return value.replace(tzinfo=None) text = str(value).strip().replace('Z', '+00:00') try: parsed = datetime.fromisoformat(text) except ValueError: return None if parsed.tzinfo is not None: parsed = parsed.astimezone(tz=None).replace(tzinfo=None) return parsed def _as_json(value): """Serialize a list/dict to JSON text, or None when nothing was sent.""" if value is None: return None return json.dumps(value, separators=(',', ':'), sort_keys=True) class EndpointsPlugin(BasePlugin): """Endpoint RMM telemetry ingest plus read/dashboard surface.""" def __init__(self): self._manifest = self._load_manifest() def _load_manifest(self) -> Dict: manifest_path = Path(__file__).parent / 'manifest.json' if manifest_path.exists(): with open(manifest_path, 'r') as f: return json.load(f) return {} @property def meta(self) -> PluginMeta: return PluginMeta( name=self._manifest.get('name', 'endpoints'), version=self._manifest.get('version', '0.1.0'), description=self._manifest.get( 'description', 'Endpoint RMM telemetry'), author=self._manifest.get('author', 'ShopDB Team'), dependencies=self._manifest.get('dependencies', ['computers']), core_version=self._manifest.get('core_version', '>=0.20.0,<1.0.0'), api_prefix=self._manifest.get('api_prefix', '/api/endpoints'), ) def get_blueprint(self) -> Optional[Blueprint]: return endpoints_bp def get_models(self) -> List[Type]: return [EndpointState, EndpointEvent] def get_permissions(self) -> List: """RBAC for the human-facing read routes. View only. Ingest is not covered here: the collector path authorizes on a collector.ingest-scoped service token, not a user JWT. There is no write/delete permission because operators do not edit telemetry - the agent is the sole author, and history is pruned by policy, not by hand. """ return [ ('endpoints.view', 'View endpoint telemetry', 'endpoints'), ] def get_navigation_items(self) -> List[Dict]: """Sidebar entry for the fleet list view. Sits just after the PCs item (position 15) because the fleet view is a telemetry lens over the same computers. """ return [ { 'name': 'Fleet', 'icon': 'activity', 'route': '/endpoints', 'position': 16, }, ] def init_app(self, app: Flask, db_instance) -> None: logger.info('Endpoints plugin initialized (v%s)', self.meta.version) # ---- ADR-010 UI hooks ----------------------------------------------- def get_asset_panels(self) -> List[Dict]: """Per-PC detail panels, computer assets only. Three panels feeding the generic renderer: the current status snapshot (keyvalue), monitored services with a run-state badge (list), and the most recent curated events (list). Each is a single-asset indexed read, so the cost is the same profile as the warranty/backups panels. Panels hide themselves when their endpoint returns nothing. """ return [ { 'id': 'endpoints-status', 'title': 'Endpoint status', 'assettypes': ['computer'], 'endpoint': '/api/endpoints/asset/{assetid}/status', 'render': 'keyvalue', 'empty': 'No agent report yet.', 'position': 10, }, { 'id': 'endpoints-services', 'title': 'Services and processes', 'assettypes': ['computer'], 'endpoint': '/api/endpoints/asset/{assetid}/services', 'render': 'list', 'map': { 'title': 'name', 'badge': {'label': 'statelabel', 'color': 'statecolor'}, }, 'empty': 'No monitored services reported.', 'position': 12, }, { 'id': 'endpoints-events', 'title': 'Recent events', 'assettypes': ['computer'], 'endpoint': '/api/endpoints/asset/{assetid}/recentevents', 'render': 'list', 'map': { 'title': 'title', 'badge': { 'label': 'severity', 'labelmap': { 'critical': 'Critical', 'warning': 'Warning', 'info': 'Info', }, }, 'meta': [ {'key': 'eventtime', 'label': 'At', 'format': 'date'}, ], }, 'empty': 'No recent events.', 'position': 14, }, ] def get_dashboard_widgets(self) -> List[Dict]: """Fleet-wide exception cards. Run once on dashboard load, not per PC. Both cards hide when empty and cap their rows, so a healthy fleet adds nothing to the dashboard and an unhealthy one adds a bounded list. """ return [ { 'id': 'endpoints-quiet', 'viewall': '/pcs', 'title': 'Endpoints gone quiet', 'endpoint': '/api/endpoints/dashboard/quiet', 'render': 'exceptions', 'severity': 'warning', 'permission': 'endpoints.view', 'empty': 'hide', 'position': 22, 'map': { 'title': 'hostname', 'detail': '', 'meta': [{'key': 'quietdays', 'label': 'silent', 'suffix': ' days'}], 'link': '/pcs/{computerid}', }, }, { 'id': 'endpoints-stability', 'viewall': '/pcs', 'title': 'Endpoints with crashes or shutdowns', 'endpoint': '/api/endpoints/dashboard/stability', 'render': 'exceptions', 'severity': 'critical', 'permission': 'endpoints.view', 'empty': 'hide', 'position': 16, 'map': { 'title': 'hostname', 'detail': '', 'meta': [{'key': 'eventcount', 'suffix': ' events (48h)'}], 'link': '/pcs/{computerid}', }, }, ] # ---- ADR-006 collector contract ------------------------------------- def get_collector_schema(self) -> Optional[Dict]: return { 'identityfield': 'hostname', 'fields': { 'type': 'object', 'required': ['hostname'], 'properties': { 'hostname': { 'type': 'string', 'description': 'Endpoint hostname. Resolved against ' 'the computers plugin (computers.hostname).', }, 'loggedinuser': {'type': 'string'}, 'agentversion': {'type': 'string'}, 'lastboottime': {'type': 'string', 'format': 'date-time'}, 'uptimeseconds': {'type': 'integer'}, 'idleseconds': {'type': 'integer'}, 'cpupercent': {'type': 'number'}, 'memorypercent': {'type': 'number'}, 'services': { 'type': 'array', 'description': 'Monitored services/processes and their ' 'run state (UDC, eDnc, Intune, ...).', }, 'ports': { 'type': 'array', 'description': 'Open listening ports.', }, 'interfaces': { 'type': 'array', 'description': 'Network interface information.', }, 'scheduledtasks': { 'type': 'array', 'description': 'Scheduled task inventory + last run.', }, 'events': { 'type': 'array', 'description': 'Curated Windows Event Log records. Only ' 'IDs in the server catalog are stored.', 'items': { 'type': 'object', 'required': ['eventid'], 'properties': { 'eventid': {'type': 'integer'}, 'eventchannel': {'type': 'string'}, 'eventsource': {'type': 'string'}, 'eventtime': {'type': 'string', 'format': 'date-time'}, 'eventmessage': {'type': 'string'}, 'fields': {'type': 'object'}, }, }, }, }, }, } def apply_collector_payload(self, payload: Dict) -> Dict: """Idempotently upsert one endpoint's state and append its events. The cached EndpointState and any EndpointEvent rows are written in the same transaction (the dispatcher owns the commit), so a reader never sees a snapshot without the events that produced it. """ from plugins.computers.models import Computer warnings: List[str] = [] hostname = (payload.get('hostname') or '').strip() if not hostname: raise ValueError('hostname is required') computer = (Computer.query .filter(db.func.lower(Computer.hostname) == hostname.lower()) .first()) if computer is None: raise ValueError( 'could not resolve hostname to a computer: {!r}'.format(hostname)) assetid = computer.assetid state = EndpointState.query.filter_by(assetid=assetid).first() action = 'updated' if state is None: state = EndpointState(assetid=assetid) db.session.add(state) action = 'created' state.hostname = hostname if 'loggedinuser' in payload: state.loggedinuser = payload.get('loggedinuser') if 'agentversion' in payload: state.agentversion = payload.get('agentversion') if 'lastboottime' in payload: state.lastboottime = _parse_dt(payload.get('lastboottime')) if 'uptimeseconds' in payload: state.uptimeseconds = payload.get('uptimeseconds') if 'idleseconds' in payload: state.idleseconds = payload.get('idleseconds') if 'cpupercent' in payload: state.cpupercent = payload.get('cpupercent') if 'memorypercent' in payload: state.memorypercent = payload.get('memorypercent') if 'services' in payload: state.servicesjson = _as_json(payload.get('services')) if 'ports' in payload: state.portsjson = _as_json(payload.get('ports')) if 'interfaces' in payload: state.interfacesjson = _as_json(payload.get('interfaces')) if 'scheduledtasks' in payload: state.scheduledtasksjson = _as_json(payload.get('scheduledtasks')) state.lastreportedat = datetime.utcnow() stored_events = 0 for record in (payload.get('events') or []): try: eventid = int(record.get('eventid')) except (TypeError, ValueError): warnings.append('skipped event with non-integer eventid') continue if eventid not in CURATED_EVENT_IDS: # Outside the catalog: ignored on purpose, not an error. continue db.session.add(EndpointEvent( assetid=assetid, eventid=eventid, eventchannel=record.get('eventchannel'), eventsource=record.get('eventsource'), eventtime=_parse_dt(record.get('eventtime')), eventmessage=record.get('eventmessage'), payloadjson=_as_json(record.get('fields')), )) stored_events += 1 # flush, not commit: the collector dispatcher owns the transaction. db.session.flush() return { 'action': action, 'assetid': assetid, 'identityvalue': hostname, 'storedevents': stored_events, 'warnings': warnings, }