diff --git a/frontend/src/components/PluginAssetPanels.vue b/frontend/src/components/PluginAssetPanels.vue index b199931e..18f2e193 100644 --- a/frontend/src/components/PluginAssetPanels.vue +++ b/frontend/src/components/PluginAssetPanels.vue @@ -52,7 +52,20 @@
{{ mapTitle(panel, item) }} - + + {{ mapBadge(panel, item).label }} @@ -68,6 +81,9 @@ {{ m.text }}
+
+ {{ mapBody(panel, item) }} +
{{ panel.manage.label || 'Manage' }} @@ -240,7 +256,7 @@ import { ref, computed, watch, onUnmounted, nextTick } from 'vue' import api from '../api' import { colorStyle } from '@/utils/colorStyle' import { - toApiPath, rows, panelVisible, mapTitle, mapBadge, mapMeta, manageLink, + toApiPath, rows, panelVisible, mapTitle, mapBadge, mapMeta, mapBody, manageLink, keyvalueFields, tableColumns, cell, badges, tabSections, rollupStatus, rowAction, facets, facetedRows, formatValue, } from './pluginAssetPanels' @@ -365,6 +381,24 @@ function visibleRows(panel) { return facetedRows(panel, facetSelection.value[panel.id]) } +// Which row bodies are expanded, keyed by panel id and row index. Closed on +// every panel refresh: the row at index 3 after a refresh is not necessarily +// the row that was opened, so keeping it open would expand an unrelated event. +const bodyOpen = ref({}) + +function bodyKey(panel, index) { + return `${panel.id}:${index}` +} + +function isBodyOpen(panel, index) { + return !!bodyOpen.value[bodyKey(panel, index)] +} + +function toggleBody(panel, index) { + const key = bodyKey(panel, index) + bodyOpen.value = { ...bodyOpen.value, [key]: !bodyOpen.value[key] } +} + const props = defineProps({ assetid: { type: [Number, String], default: null }, // Which detail-page tab to render. Null renders every panel, which is what @@ -525,6 +559,7 @@ async function load() { panels.value = [] revokeReportImageUrls() closeReport() + bodyOpen.value = {} if (!props.assetid) return let declared try { @@ -562,6 +597,31 @@ defineExpose({ refresh: load, populatedGroups }) .pap-title { font-weight: 600; color: var(--text); } .status-badge { padding: 0.15rem 0.6rem; border-radius: 12px; font-size: 0.75rem; font-weight: 600; } .pap-meta { margin-top: 0.35rem; display: flex; flex-wrap: wrap; gap: 0.75rem; font-size: 0.82rem; color: var(--text-light); } +/* The badge doubles as the expander, so it needs to look pressable without + becoming a second style of badge: same pill, just a pointer and a hover + lift. margin-left:auto pins it right even on a row that also has an + action button, where space-between alone would centre it. */ +.pap-badge-toggle { + margin-left: auto; + border: none; + cursor: pointer; + font-family: inherit; + line-height: inherit; +} +.pap-badge-toggle:hover { filter: brightness(1.1); } +.pap-badge-toggle:focus-visible { outline: 2px solid var(--primary, #4181ff); outline-offset: 2px; } +/* pre-wrap, not normal: the message keeps the line breaks the event was + written with, which is what makes a multi-line fault readable. */ +.pap-body { + margin-top: 0.5rem; + padding: 0.5rem 0.6rem; + background: var(--surface-2, rgba(127, 127, 127, 0.08)); + border-radius: 4px; + font-size: 0.82rem; + color: var(--text-light); + white-space: pre-wrap; + overflow-wrap: anywhere; +} .pap-empty { display: flex; align-items: center; gap: 0.6rem; } .pap-facets { diff --git a/frontend/src/components/pluginAssetPanels.js b/frontend/src/components/pluginAssetPanels.js index ce303125..80984e5a 100644 --- a/frontend/src/components/pluginAssetPanels.js +++ b/frontend/src/components/pluginAssetPanels.js @@ -138,6 +138,14 @@ export function mapMeta(panel, item) { .filter(Boolean) } +export function mapBody(panel, item) { + const key = panel.map && panel.map.body + if (!key) return null + const value = item[key] + if (value === null || value === undefined || value === '') return null + return String(value) +} + export function manageLink(panel, assetid) { return panel.manage ? subst(panel.manage.to, assetid) : null } diff --git a/frontend/src/components/pluginAssetPanels.spec.js b/frontend/src/components/pluginAssetPanels.spec.js index f6377ced..f19a2040 100644 --- a/frontend/src/components/pluginAssetPanels.spec.js +++ b/frontend/src/components/pluginAssetPanels.spec.js @@ -1,6 +1,6 @@ import { describe, it, expect } from 'vitest' import { - toApiPath, subst, rows, mapTitle, mapBadge, mapMeta, manageLink, + toApiPath, subst, rows, mapTitle, mapBadge, mapMeta, mapBody, manageLink, keyvalueFields, tableColumns, badges, panelVisible, tabSections, rollupStatus, rowAction, facets, facetedRows, } from './pluginAssetPanels' @@ -64,7 +64,6 @@ describe('list render mapping (warranty)', () => { it('builds the manage link with the assetid', () => { expect(manageLink(warrantyPanel, 7)).toBe('/warranties?addfor=7') }) - it('is visible when it has rows or an empty message', () => { expect(panelVisible(warrantyPanel)).toBe(true) expect(panelVisible({ ...warrantyPanel, _data: [] })).toBe(true) // has empty @@ -72,6 +71,24 @@ describe('list render mapping (warranty)', () => { }) }) +describe('expandable row body (activity feed)', () => { + const activityPanel = { + id: 'endpoints-activity', render: 'list', + map: { title: 'title', body: 'body' }, + } + + it('returns the body only for rows that carry one', () => { + expect(mapBody(activityPanel, { title: 'Crash', body: 'Faulting module' })) + .toBe('Faulting module') + expect(mapBody(activityPanel, { title: 'Crash' })).toBeNull() + expect(mapBody(activityPanel, { title: 'Crash', body: '' })).toBeNull() + }) + + it('returns null when the panel declares no body key', () => { + expect(mapBody(warrantyPanel, { body: 'ignored' })).toBeNull() + }) +}) + describe('list row filtering (tracked services vs processes)', () => { const feedPanel = { id: 'endpoints-tracked-services', render: 'list', diff --git a/plugins/endpoints/api/routes.py b/plugins/endpoints/api/routes.py index 01e4a4ca..11171aeb 100644 --- a/plugins/endpoints/api/routes.py +++ b/plugins/endpoints/api/routes.py @@ -87,6 +87,81 @@ def _format_uptime(seconds): return ' '.join(parts) +def _state_json(state, attr): + """Parse one of the EndpointState inventory JSON blobs. + + These columns are written straight from the agent snapshot, so a partial + or malformed blob from one bad check-in must not take the whole panel + down. Anything unparseable reads as absent. + """ + raw = getattr(state, attr, None) if state else None + if not raw: + return None + try: + return json.loads(raw) + except (TypeError, ValueError): + return None + + +def _format_bytes(value): + """Byte count -> 'N.N GB'. Sizes here are disks and RAM, never sub-MB.""" + if value is None: + return None + try: + value = float(value) + except (TypeError, ValueError): + return None + for unit in ('B', 'KB', 'MB', 'GB', 'TB'): + if abs(value) < 1024.0 or unit == 'TB': + return '{:.1f} {}'.format(value, unit) if unit != 'B' else '{:.0f} B'.format(value) + value /= 1024.0 + return None + + +def _format_agent_date(value): + """Agent timestamps are ISO strings, not datetimes; show the date only. + + A BIOS release date or an OS install date is only ever read at day + resolution, and the agent sends a full 7-digit-fraction UTC stamp. + """ + if not value: + return None + text = str(value) + return text[:10] if len(text) >= 10 else text + + +# A Windows event message runs to several hundred words. It is shown in an +# expander that is closed by default, so the cap here only exists to keep one +# pathological message from bloating the payload - not to fit a line. Blank +# runs are collapsed but real line breaks are kept, because the body is read +# as a block once opened. +_EVENT_BODY_LIMIT = 4000 + + +def _event_body(event): + """Readable message body for one curated event, bounded and normalized. + + Windows messages arrive padded with long runs of blank lines that render + as ragged gaps. Runs of blank lines collapse to one and trailing spaces + go, but single newlines survive so the text keeps its shape when opened. + """ + text = (getattr(event, 'eventmessage', None) or '').strip() + if not text: + return None + lines = [line.rstrip() for line in text.splitlines()] + cleaned = [] + for line in lines: + if not line and cleaned and not cleaned[-1]: + continue + cleaned.append(line) + text = '\n'.join(cleaned).strip() + if not text: + return None + if len(text) <= _EVENT_BODY_LIMIT: + return text + return text[:_EVENT_BODY_LIMIT].rstrip() + '...' + + # ============================================================================= # Endpoint state # ============================================================================= @@ -465,6 +540,180 @@ def panel_status(asset_id: int): return success_response({'fields': fields}) +@endpoints_bp.route('/asset//hardware', methods=['GET']) +@require_permission('endpoints.view') +def panel_hardware(asset_id: int): + """Keyvalue panel: processor, memory and system chassis inventory. + + Only reported by agent 1.7.0 and later. An older agent leaves these + columns null, so the panel yields no fields and hides itself rather than + rendering a card full of blanks. + """ + state = EndpointState.query.filter_by(assetid=asset_id).first() + if not state: + return success_response({'fields': []}) + + processor = _state_json(state, 'processorjson') or {} + memory = _state_json(state, 'memoryjson') or {} + + fields = [ + {'label': 'Processor', 'value': processor.get('name')}, + {'label': 'Cores', 'value': processor.get('cores')}, + {'label': 'Logical processors', 'value': processor.get('logicalprocessors')}, + {'label': 'Max clock', + 'value': '{} MHz'.format(processor['maxclockmhz']) + if processor.get('maxclockmhz') else None}, + {'label': 'Total memory', 'value': _format_bytes(memory.get('totalbytes'))}, + {'label': 'Free memory', 'value': _format_bytes(memory.get('freebytes'))}, + ] + fields = [f for f in fields if f['value'] not in (None, '')] + return success_response({'fields': fields}) + + +@endpoints_bp.route('/asset//osinfo', methods=['GET']) +@require_permission('endpoints.view') +def panel_osinfo(asset_id: int): + """Keyvalue panel: operating system, firmware and domain membership. + + OS and BIOS are one card on purpose: both answer "what is this box + running", and split across two cards each would be three short rows. + """ + state = EndpointState.query.filter_by(assetid=asset_id).first() + if not state: + return success_response({'fields': []}) + + osinfo = _state_json(state, 'osinfojson') or {} + bios = _state_json(state, 'biosjson') or {} + + pendingreboot = None + if state.pendingreboot is not None: + pendingreboot = 'Yes' if state.pendingreboot else 'No' + + fields = [ + {'label': 'OS version', 'value': osinfo.get('version'), 'mono': True}, + {'label': 'Build', 'value': osinfo.get('build'), 'mono': True}, + {'label': 'Architecture', 'value': osinfo.get('architecture')}, + {'label': 'Installed', 'value': _format_agent_date(osinfo.get('installdate'))}, + {'label': 'Domain', 'value': state.domain}, + {'label': 'BIOS vendor', 'value': bios.get('vendor')}, + {'label': 'BIOS version', 'value': bios.get('version'), 'mono': True}, + {'label': 'BIOS date', 'value': _format_agent_date(bios.get('releasedate'))}, + {'label': 'Pending reboot', 'value': pendingreboot}, + ] + fields = [f for f in fields + if f['value'] not in (None, '')] + return success_response({'fields': fields}) + + +@endpoints_bp.route('/asset//storage', methods=['GET']) +@require_permission('endpoints.view') +def panel_storage(asset_id: int): + """Table panel: fixed disks with capacity and free space. + + freepercent is computed by the agent from 1.7.0; for a snapshot written + by an older agent it is derived here from the byte counts so the column + is not empty for the one field an operator actually scans. + """ + state = EndpointState.query.filter_by(assetid=asset_id).first() + disks = _state_json(state, 'disksjson') if state else None + if not isinstance(disks, list): + return success_response([]) + + rows = [] + for disk in disks: + if not isinstance(disk, dict): + continue + freepercent = disk.get('freepercent') + if freepercent is None: + size = disk.get('sizebytes') or 0 + free = disk.get('freebytes') + if size and free is not None: + freepercent = round(free * 100.0 / size, 1) + rows.append({ + 'drive': disk.get('drive'), + 'label': disk.get('label') or '', + 'filesystem': disk.get('filesystem'), + 'size': _format_bytes(disk.get('sizebytes')), + 'free': _format_bytes(disk.get('freebytes')), + 'freepercent': '{}%'.format(freepercent) if freepercent is not None else None, + }) + return success_response(rows) + + +@endpoints_bp.route('/asset//listeners', methods=['GET']) +@require_permission('endpoints.view') +def panel_listeners(asset_id: int): + """Table panel: TCP sockets in LISTEN, with the owning process. + + The listening set is the box's exposed surface, so it is sorted by port + to make an unexpected one easy to spot. The full connection table is a + deliberately on-demand command, not standard telemetry. + """ + state = EndpointState.query.filter_by(assetid=asset_id).first() + ports = _state_json(state, 'portsjson') if state else None + if not isinstance(ports, list): + return success_response([]) + + rows = [] + for entry in ports: + if not isinstance(entry, dict): + continue + # Sample-data rows predate the agent listener collector and carry a + # flat port/process shape; read both rather than showing them blank. + port = entry.get('localport', entry.get('port')) + rows.append({ + 'localport': port, + 'localaddress': entry.get('localaddress'), + 'protocol': (entry.get('protocol') or 'tcp').upper(), + 'processname': entry.get('processname') or entry.get('process'), + 'processid': entry.get('processid'), + }) + rows.sort(key=lambda r: (r['localport'] is None, r['localport'])) + return success_response(rows) + + +@endpoints_bp.route('/asset//usercontext', methods=['GET']) +@require_permission('endpoints.view') +def panel_usercontext(asset_id: int): + """Keyvalue panel: session-scoped data from the interactive user probe. + + Mapped drives and user-owned listeners do not exist in the agent's + session 0, so they are collected by a probe run in the logged-in user's + session. Nothing is reported when no one is signed in, which is a valid + state and not a failure. + """ + state = EndpointState.query.filter_by(assetid=asset_id).first() + if not state: + return success_response({'fields': []}) + + context = _state_json(state, 'usercontextjson') + if not isinstance(context, dict) or not context: + return success_response({'fields': []}) + + netuse = context.get('netuse') + drives = [] + if isinstance(netuse, list): + for entry in netuse: + if isinstance(entry, dict): + local = entry.get('localname') or entry.get('local') or '' + remote = entry.get('remotename') or entry.get('remote') or '' + drives.append('{} {}'.format(local, remote).strip()) + + listeners = context.get('userlisteners') + listenercount = len(listeners) if isinstance(listeners, list) else None + + fields = [ + {'label': 'Probed user', 'value': context.get('username')}, + {'label': 'Collected', 'value': context.get('timestamp')}, + {'label': 'Idle', 'value': _format_uptime(context.get('idleseconds'))}, + {'label': 'Mapped drives', + 'value': ', '.join(drives) if drives else None, 'mono': True}, + {'label': 'User listeners', 'value': listenercount}, + ] + fields = [f for f in fields if f['value'] not in (None, '')] + return success_response({'fields': fields}) + + @endpoints_bp.route('/asset//telemetrymeta', methods=['GET']) @require_permission('endpoints.view') def panel_telemetry_meta(asset_id: int): @@ -723,10 +972,18 @@ def panel_activity(asset_id: int): rows = [] for event in events: + # The provider name is what distinguishes two events sharing an ID, + # and the message body is the line an operator actually reads. Both + # are already stored; the feed just was not showing them. + source = event.eventsource or event.eventchannel + detail = 'Windows event {}'.format(event.eventid) + if source: + detail = '{} - {}'.format(detail, source) rows.append({ 'title': event_title(event.eventid), 'severity': event_severity(event.eventid), - 'detail': 'Windows event {}'.format(event.eventid), + 'detail': detail, + 'body': _event_body(event), 'eventtime': event.eventtime, }) @@ -742,7 +999,9 @@ def panel_activity(asset_id: int): 'title': '{}{}'.format(label, target), # A failed command is the one an operator needs to notice. 'severity': 'critical' if command.status == EndpointCommand.STATUS_FAILED else 'info', - 'detail': '{}{}'.format((command.status or 'queued').title(), who), + 'detail': '{}{}{}'.format('Command - ', (command.status or 'queued').title(), who), + # A failed command's reason is the whole point of noticing it. + 'body': (command.resultmessage or '').strip() or None, 'eventtime': command.createdat, }) @@ -768,6 +1027,7 @@ def panel_activity(asset_id: int): 'severity': r['severity'], 'severitycolor': colours.get(r['severity'], 'gray'), 'detail': r['detail'], + 'body': r['body'], 'eventtime': r['eventtime'].isoformat() + 'Z' if r['eventtime'] else None, } for r in rows[:15]]) diff --git a/plugins/endpoints/plugin.py b/plugins/endpoints/plugin.py index 11c4c38a..8d55bc49 100644 --- a/plugins/endpoints/plugin.py +++ b/plugins/endpoints/plugin.py @@ -52,6 +52,8 @@ KNOWN_PAYLOAD_KEYS = frozenset({ 'agentversion', 'trayversion', 'lastboottime', 'uptimeseconds', 'idleseconds', 'cpupercent', 'memorypercent', 'services', 'ports', 'interfaces', 'scheduledtasks', 'applications', 'events', + 'disks', 'bios', 'memory', 'processor', 'osinfo', 'domain', + 'pendingreboot', 'usercontext', }) @@ -472,6 +474,11 @@ class EndpointsPlugin(BasePlugin): {'key': 'detail'}, {'key': 'eventtime', 'label': 'At', 'format': 'date'}, ], + # Expands in place on click. The message body is several + # hundred words of Windows boilerplate around the one or + # two lines that matter, so it stays collapsed until asked + # for rather than pushing every other row off the card. + 'body': 'body', }, 'empty': 'Nothing recorded for this endpoint yet.', 'position': 20, @@ -524,6 +531,85 @@ class EndpointsPlugin(BasePlugin): # the installed-software inventory 'group': 'software', }, + { + 'id': 'endpoints-hardware', + 'title': 'Hardware', + 'assettypes': ['computer'], + 'endpoint': '/api/endpoints/asset/{assetid}/hardware', + 'render': 'keyvalue', + 'collapsible': True, + 'empty': 'No hardware inventory reported.', + 'position': 14, + 'group': 'monitors', + }, + { + 'id': 'endpoints-osinfo', + 'title': 'Operating system', + 'assettypes': ['computer'], + 'endpoint': '/api/endpoints/asset/{assetid}/osinfo', + 'render': 'keyvalue', + 'collapsible': True, + 'empty': 'No OS inventory reported.', + 'position': 15, + 'group': 'monitors', + }, + { + 'id': 'endpoints-storage', + 'title': 'Storage', + 'assettypes': ['computer'], + 'endpoint': '/api/endpoints/asset/{assetid}/storage', + 'render': 'table', + 'collapsible': True, + 'columns': [ + {'key': 'drive', 'label': 'Drive'}, + {'key': 'label', 'label': 'Label'}, + {'key': 'filesystem', 'label': 'Filesystem'}, + {'key': 'size', 'label': 'Size'}, + {'key': 'free', 'label': 'Free'}, + {'key': 'freepercent', 'label': 'Free %'}, + ], + 'empty': 'No disks reported.', + 'position': 17, + 'group': 'monitors', + }, + { + # The listening set is the box's exposed network surface. The + # full connection table is deliberately not here: it is an + # on-demand command, because a per-cycle capture of every + # connection is a lot of rows nobody reads. + 'id': 'endpoints-listeners', + 'title': 'Listening ports', + 'assettypes': ['computer'], + 'endpoint': '/api/endpoints/asset/{assetid}/listeners', + 'render': 'table', + 'collapsible': True, + 'columns': [ + {'key': 'localport', 'label': 'Port'}, + {'key': 'protocol', 'label': 'Protocol'}, + {'key': 'localaddress', 'label': 'Address'}, + {'key': 'processname', 'label': 'Process'}, + {'key': 'processid', 'label': 'PID'}, + ], + 'empty': 'No listening ports reported.', + 'position': 19, + 'group': 'monitors', + }, + { + # Session-scoped data collected by a probe in the interactive + # user's session, so it is absent whenever nobody is signed in. + # Last on the page: it is the least reliably present card, and + # an empty one at the bottom reads as "not reported" rather + # than pushing the populated cards down. + 'id': 'endpoints-usercontext', + 'title': 'User context', + 'assettypes': ['computer'], + 'endpoint': '/api/endpoints/asset/{assetid}/usercontext', + 'render': 'keyvalue', + 'collapsible': True, + 'empty': 'No user session data reported.', + 'position': 40, + 'group': 'monitors', + }, { # Operator-submitted issue reports: a free-text note and an # optional screenshot the operator captured from the tray. Sits @@ -636,6 +722,49 @@ class EndpointsPlugin(BasePlugin): 'type': 'array', 'description': 'Scheduled task inventory + last run.', }, + 'disks': { + 'type': 'array', + 'description': 'Fixed local disks: {drive, label, ' + 'filesystem, sizebytes, freebytes, ' + 'usedbytes, freepercent}. Newest ' + 'snapshot, replaced on every report.', + }, + 'bios': { + 'type': 'object', + 'description': 'Firmware identity: vendor, version, ' + 'release date.', + }, + 'memory': { + 'type': 'object', + 'description': 'Physical memory totals in bytes.', + }, + 'processor': { + 'type': 'object', + 'description': 'CPU identity, core and logical ' + 'processor counts, max clock.', + }, + 'osinfo': { + 'type': 'object', + 'description': 'OS build, version, and install date.', + }, + 'domain': { + 'type': 'string', + 'description': 'Windows domain or workgroup membership.', + }, + 'pendingreboot': { + 'type': 'boolean', + 'description': 'True when the endpoint is holding a ' + 'reboot. Absent means never reported, ' + 'which is not the same as false.', + }, + 'usercontext': { + 'type': 'object', + 'description': 'Session-scoped data collected by a probe ' + 'running as the signed-in user: mapped ' + 'drives (netuse), user-owned listeners, ' + 'and idle seconds. Absent when nobody is ' + 'signed in.', + }, 'applications': { 'type': 'array', 'description': 'Installed-software SBOM from both ' @@ -663,6 +792,12 @@ class EndpointsPlugin(BasePlugin): 'required': ['eventid'], 'properties': { 'eventid': {'type': 'integer'}, + 'eventrecordid': { + 'type': 'integer', + 'description': 'Windows event record id; ' + 'unique per channel per ' + 'machine, used to dedup.', + }, 'eventchannel': {'type': 'string'}, 'eventsource': {'type': 'string'}, 'eventtime': {'type': 'string', @@ -846,6 +981,33 @@ class EndpointsPlugin(BasePlugin): if 'scheduledtasks' in payload: state.scheduledtasksjson = _as_json(payload.get('scheduledtasks')) + # Hardware/OS inventory. Same presence guard as above: a report that + # omits a key must not wipe the last-known value. These change rarely + # (a RAM upgrade, a BIOS flash), so a transient collection failure on + # one cycle should leave the previous snapshot standing rather than + # blank the asset record. + if 'disks' in payload: + state.disksjson = _as_json(payload.get('disks')) + if 'bios' in payload: + state.biosjson = _as_json(payload.get('bios')) + if 'memory' in payload: + state.memoryjson = _as_json(payload.get('memory')) + if 'processor' in payload: + state.processorjson = _as_json(payload.get('processor')) + if 'osinfo' in payload: + state.osinfojson = _as_json(payload.get('osinfo')) + if 'domain' in payload: + state.domain = payload.get('domain') + if 'pendingreboot' in payload: + state.pendingreboot = payload.get('pendingreboot') + + # Session-scoped probe output. The agent repeats its last result between + # probe runs and omits the key entirely when nobody is signed in, so the + # presence guard preserves the last known mapped drives across an + # unattended period instead of flapping the field to null. + if 'usercontext' in payload: + state.usercontextjson = _as_json(payload.get('usercontext')) + # Installed-software SBOM: stored as a per-asset SNAPSHOT in its own # table (endpointapplications), not a JSON blob on the state, so it can # later join the applications catalog ("Managed Applications"). Only @@ -897,6 +1059,19 @@ class EndpointsPlugin(BasePlugin): hostname, len(unknown_keys), sorted(unknown_keys), preview_text) stored_events = 0 + skipped_events = 0 + + # Records the agent already sent must not be appended a second time. + # The agent carries a per-channel watermark so the steady state is + # "only what is new", but a reinstalled agent, a restored image, or two + # overlapping reports can all replay records. Resolve that here rather + # than relying on the unique index to raise, because an IntegrityError + # would abort the whole collector transaction and lose the state upsert + # along with it. + # + # One query for the whole batch, not one per record: a seeding agent + # sends up to the per-channel cap at once. + candidates = [] for record in (payload.get('events') or []): try: eventid = int(record.get('eventid')) @@ -906,10 +1081,45 @@ class EndpointsPlugin(BasePlugin): if eventid not in CURATED_EVENT_IDS: # Outside the catalog: ignored on purpose, not an error. continue + + try: + eventrecordid = record.get('eventrecordid') + eventrecordid = (int(eventrecordid) + if eventrecordid is not None else None) + except (TypeError, ValueError): + eventrecordid = None + + candidates.append((eventid, eventrecordid, record)) + + keyed = [(record.get('eventchannel'), eventrecordid) + for _e, eventrecordid, record in candidates + if eventrecordid is not None] + already_stored = set() + if keyed: + rows = (db.session.query(EndpointEvent.eventchannel, + EndpointEvent.eventrecordid) + .filter(EndpointEvent.assetid == assetid, + EndpointEvent.eventrecordid.in_( + [recordid for _c, recordid in keyed])) + .all()) + already_stored = {(channel, recordid) for channel, recordid in rows} + + for eventid, eventrecordid, record in candidates: + eventchannel = record.get('eventchannel') + if eventrecordid is not None: + key = (eventchannel, eventrecordid) + if key in already_stored: + skipped_events += 1 + continue + # Guard against the same record appearing twice inside one + # payload as well as against rows already in the table. + already_stored.add(key) + db.session.add(EndpointEvent( assetid=assetid, eventid=eventid, - eventchannel=record.get('eventchannel'), + eventrecordid=eventrecordid, + eventchannel=eventchannel, eventsource=record.get('eventsource'), eventtime=_parse_dt(record.get('eventtime')), eventmessage=record.get('eventmessage'), @@ -917,6 +1127,11 @@ class EndpointsPlugin(BasePlugin): )) stored_events += 1 + if skipped_events: + logger.debug( + 'endpoint %s replayed %d already-stored event record(s)', + hostname, skipped_events) + # flush, not commit: the collector dispatcher owns the transaction. db.session.flush() diff --git a/tests/test_plugins/test_endpoints_plugin.py b/tests/test_plugins/test_endpoints_plugin.py index a779cd03..979bb70b 100644 --- a/tests/test_plugins/test_endpoints_plugin.py +++ b/tests/test_plugins/test_endpoints_plugin.py @@ -69,7 +69,7 @@ def test_endpoints_asset_panels_are_computer_scoped(): """Every asset panel targets computer assets only, with an endpoint.""" plugin = EndpointsPlugin() panels = plugin.get_asset_panels() - assert len(panels) == 7 + assert len(panels) == 12 for panel in panels: assert panel['assettypes'] == ['computer'] assert panel['endpoint'].startswith('/api/endpoints/') @@ -220,18 +220,21 @@ def test_known_payload_keys_cover_persisted_and_relayed_fields(): assert persisted_or_relayed <= KNOWN_PAYLOAD_KEYS -def test_new_hardware_telemetry_keys_are_logged_not_silently_known(): - """Disk/bios/memory inventory is collected at the agent before the server - persists it. Those keys must be OUTSIDE the known set so ingest logs them - rather than dropping them silently.""" +def test_hardware_telemetry_keys_are_persisted_not_merely_logged(): + """Disk/bios/memory/processor/osinfo inventory was collected at the agent and + logged as unknown keys for several versions while the shapes were proven + against real endpoint traffic. They have columns now (endpoints migration + 0013), so they must be INSIDE the known set - leaving them out would log + every report as carrying unpersisted keys while the ingest quietly stores + them, which is the most misleading combination available.""" from plugins.endpoints.plugin import KNOWN_PAYLOAD_KEYS - for key in ('disks', 'bios', 'memory', 'processor', 'osinfo', 'domain'): - assert key not in KNOWN_PAYLOAD_KEYS + for key in ('disks', 'bios', 'memory', 'processor', 'osinfo', 'domain', + 'pendingreboot', 'usercontext'): + assert key in KNOWN_PAYLOAD_KEYS - # 'applications' USED to be in this list. It is persisted now - that is what - # the installed-software work added - so it belongs in the known set, and - # asserting its absence would fail the moment the feature it tests works. + # 'applications' followed this same staging path earlier and is likewise + # persisted, to its own endpointapplications table rather than a state blob. assert 'applications' in KNOWN_PAYLOAD_KEYS