066a19052b
Findings F2, F4, F5, F6, F7, F8, F9, F10, F11 and F13 from wj-architectural-review.md. F1/F1a (the non-sargable LOWER pattern) and F3 (the warranty sync holding a pooled connection) are being handled separately and are untouched here. - A collector check-in that raced another for the same PC lost everything: the unique index on the application fingerprint raised, nothing caught it, and the whole check-in rolled back, discarding state, events and hardware, answering the agent with form copy meant for a person. The insert now runs in a savepoint, reconciles what the other transaction already wrote, one row per savepoint so a second collision skips a row rather than the batch, and the check-in commits with a warning. This is the shape the delete side of the same function already had. - Enrollment is deliberately unauthenticated, so it now carries a per-address rate limit and a cap on pending requests, and the cap is a setting an admin can see and change. A flood can no longer bury the approval queue or grow the table without bound. - The fleet alert poll no longer formats debug strings on every request. - BusinessUnit is on the contract surface (0.34.0), so the endpoints plugin stops reaching into a core table with raw SQL. - A test walks every asset type descriptor and proves the endpoint it names really refuses a caller without the declared permission. - The plugin settings routes now need a permission; they were telling anyone which modules a site runs and where its components live. - The connection pool is sized explicitly with a short timeout, and the arithmetic against the server's connection ceiling is written next to the worker count instead of being left to defaults nobody chose. - Fleet alerts carry both the correct lowercase keys and the old camelCase pair. Trays older than 1.7.0 match keys case-sensitively and would have gone quiet on every PC not yet upgraded; the old pair goes when they are. - The dev server proxy points at localhost again, with environment variables for the remote-box workflow, and one developer's address is out of the server certificate page.
1708 lines
81 KiB
Python
1708 lines
81 KiB
Python
"""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
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Type
|
|
|
|
from flask import Flask, Blueprint, g
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
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, EndpointCommand, EndpointApplication,
|
|
)
|
|
|
|
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)
|
|
|
|
# Pre-compiled regex for fingerprint validation (64 lowercase hex chars).
|
|
# Used in hot path (40,000+ validations/hour), so compile once at module load.
|
|
_FP_REGEX = re.compile(r'\A[0-9a-f]{64}\Z')
|
|
|
|
# Top-level payload keys the ingest path knows about today - either persisted on
|
|
# EndpointState, appended as events, or relayed to the computers collector.
|
|
# Anything the agent sends outside this set is not stored yet; it is logged (see
|
|
# apply_collector_payload) so new agent telemetry - disks, bios, memory, etc. -
|
|
# is observable in the logs before the server grows columns for it. Keep this in
|
|
# lockstep with the keys consumed in apply_collector_payload and the computers
|
|
# handoff subset.
|
|
KNOWN_PAYLOAD_KEYS = frozenset({
|
|
'hostname', 'adopt', 'loggedinuser', 'serialnumber', 'pctype',
|
|
'machinenumber', 'vendorname', 'modelnumber', 'osname', 'ipaddress',
|
|
'agentversion', 'trayversion', 'lastboottime', 'uptimeseconds', 'idleseconds',
|
|
'cpupercent', 'memorypercent', 'services', 'ports', 'interfaces',
|
|
'scheduledtasks', 'applications', 'events',
|
|
'disks', 'bios', 'memory', 'processor', 'osinfo', 'domain',
|
|
'pendingreboot', 'usercontext', 'diskhealth',
|
|
'rebootreasons', 'bootinfo', 'defender', 'dellupdates',
|
|
})
|
|
|
|
|
|
def _parse_dt(value):
|
|
"""Best-effort parse of an ISO-8601 string to naive UTC, or None.
|
|
|
|
astimezone(tz=None) converts to the SERVER's local zone, not to UTC, so an
|
|
agent reporting 17:26:16Z was stored as 13:26:16 on an EDT server and then
|
|
re-served labelled 'Z'. The same row's server-written lastreportedat stayed
|
|
true UTC, so one row carried two time bases four hours apart, and the
|
|
dashboard's 48-hour stability window - which compares a real UTC cutoff
|
|
against these values - dropped events four hours early.
|
|
|
|
Invisible on a UTC server, which is why it survived: nothing in the
|
|
Dockerfile, the compose files or the Windows deploy pins TZ, so whether the
|
|
data is right depends on the host's clock settings.
|
|
|
|
AGENT-CONTRACT.md mandates UTC on the wire. A NAIVE value is therefore
|
|
taken as UTC already, rather than guessed at.
|
|
"""
|
|
if not value:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
if value.tzinfo is not None:
|
|
return value.astimezone(timezone.utc).replace(tzinfo=None)
|
|
return value
|
|
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(timezone.utc).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)
|
|
|
|
|
|
def _clamp(value, length):
|
|
"""A string bounded to its column width, or None.
|
|
|
|
Ingest writes whatever an endpoint reports. A registry value longer than its
|
|
column is not a data-quality nuisance on MySQL: in strict mode it raises
|
|
Data too long and fails the whole check-in transaction, so one machine with
|
|
a malformed Uninstall entry stops reporting telemetry and events entirely.
|
|
"""
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text[:length] or None
|
|
|
|
|
|
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, EndpointCommand,
|
|
EndpointApplication]
|
|
|
|
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'),
|
|
# The FLEET-WIDE list and dashboards, separate from endpoints.view,
|
|
# which covers reading ONE endpoint that the caller already reached
|
|
# through its PC page.
|
|
#
|
|
# They were one permission, and that made a reasonable policy
|
|
# impossible to express: a site that wants signed-out visitors to
|
|
# see live status on a PC page had to hand them the whole fleet,
|
|
# including every logged-in username on the floor, because the
|
|
# panels and the list were gated on the same string.
|
|
#
|
|
# The name avoids '.view' for readability, but the NAME no longer
|
|
# decides anything: guest_grantable_permissions is a declared list
|
|
# now, and this plugin blesses endpoints.view alone below. Adding
|
|
# endpoints.fleet there would be what opens the roster, and the
|
|
# docstring on that hook says why not to.
|
|
('endpoints.fleet',
|
|
'View the fleet-wide endpoint list and dashboards', 'endpoints'),
|
|
# Issue an allowlisted command to an endpoint agent (for example the
|
|
# MTConnect toggle). Separate from endpoints.view because acting on a
|
|
# shop-floor PC is a heavier grant than reading its telemetry.
|
|
('endpoints.command',
|
|
'Send allowlisted commands to endpoint agents', 'endpoints'),
|
|
# Operator-submitted issue reports and their screenshots. Separate
|
|
# from endpoints.view because the screenshot is whatever was on the
|
|
# operator's screen when they hit report: it can carry part numbers,
|
|
# customer names or other shopfloor content that reading telemetry
|
|
# does not imply a right to see. Telemetry is machine-collected and
|
|
# uniform; this is an uncontrolled human capture, so it is its own
|
|
# grant rather than a facet of endpoints.view.
|
|
('endpoints.operatorreports',
|
|
'View operator-submitted reports and their screenshots',
|
|
'endpoints'),
|
|
# Elevated grant for destructive endpoint actions (system_restart).
|
|
# Required IN ADDITION to endpoints.command by the enqueue route, so a
|
|
# remote reboot is a strictly heavier grant than a routine toggle.
|
|
('endpoints.restart',
|
|
'Restart endpoint computers (destructive)', 'endpoints'),
|
|
# Elevated grants for the actions gated by SCOPE, never by
|
|
# Configuration Policy. Each is required IN ADDITION to
|
|
# endpoints.command by the enqueue route.
|
|
#
|
|
# These were one permission (endpoints.deploy) covering run_script,
|
|
# update_app and scheduledtask_run. That bundle could not express
|
|
# "may push an application update" without also granting arbitrary
|
|
# SYSTEM code execution, so it is split. The three are NOT equal
|
|
# risk and their descriptions must not pretend otherwise.
|
|
('endpoints.corptools',
|
|
'Start approved scheduled tasks on endpoints (limited to tasks '
|
|
'already registered and allowlisted on the agent)', 'endpoints'),
|
|
('endpoints.updateapp',
|
|
'Push application installers to endpoints (runs a pinned '
|
|
'installer as SYSTEM) and collect plaintext log files from '
|
|
'endpoints by path (read-only, text files only)', 'endpoints'),
|
|
# Blunt on purpose. This is not "system-level changes" - it is
|
|
# arbitrary code execution as SYSTEM unless the script descriptor
|
|
# pins a user context. Whoever grants it must see that here.
|
|
('endpoints.runscript',
|
|
'Run staged scripts on endpoints (ARBITRARY CODE, executes as '
|
|
'SYSTEM unless the script pins a user context; grant only to '
|
|
'fully trusted operators)', 'endpoints'),
|
|
# Packaging the agent is a heavier grant than deploying to one, and
|
|
# it is not the same job either: the generated ZIP carries a live
|
|
# collector key stamped into its install script, so whoever holds
|
|
# this can mint a distributable artefact containing the fleet-wide
|
|
# command-channel credential. It was require_role('admin'), which
|
|
# this codebase otherwise reserves for the few genuinely
|
|
# role-shaped checks - and a role cannot be granted to one person
|
|
# for one job the way a permission can.
|
|
# It now also governs SUPPLYING the binary, not just wrapping it.
|
|
# Before the agent store, the exe could only be staged by someone
|
|
# with filesystem access to instance/siteagent/; the upload routes
|
|
# put that behind this same grant, so a holder chooses the file
|
|
# every future package installs and runs as SYSTEM. Whoever grants
|
|
# this has to be able to see that from the description.
|
|
('endpoints.agentpackage',
|
|
'Upload the agent binary the fleet installs, and generate or '
|
|
'download an agent or adoption-tool package (supplies an '
|
|
'executable that runs as SYSTEM; the package contains a live '
|
|
'collector key)', 'endpoints'),
|
|
# Pushing an agent REPLACEMENT to running endpoints is a distinct,
|
|
# heavier job than either packaging (endpoints.agentpackage) or
|
|
# pushing an ordinary application installer (endpoints.deploy).
|
|
# Kept separate on purpose: the agent runs as SYSTEM and is the
|
|
# command channel itself, so replacing it is self-modifying the
|
|
# control plane. Folding it into endpoints.deploy would let anyone
|
|
# who can push an app silently swap the agent; folding it into
|
|
# endpoints.agentpackage would let a packager who only mints ZIPs
|
|
# also force a live fleet-wide swap. A future Operator role can
|
|
# hold upload without this, or this without upload.
|
|
('endpoints.agentupdate',
|
|
'Push agent/tray binary updates to running endpoints '
|
|
'(replaces the SYSTEM agent)', 'endpoints'),
|
|
# Collector service scope for the endpoint agent. A managed token
|
|
# scoped to ONLY this authorizes the endpoints collector and nothing
|
|
# else, so rotating/revoking it stops endpoint-agent traffic without
|
|
# touching computers/geenforce collectors. Minted from the settings
|
|
# UI (see get_settings_cards). Admins hold it via the catalog.
|
|
('collector.ingest.endpoints',
|
|
'Submit endpoint agent telemetry (fleet reporting)', 'collector'),
|
|
]
|
|
|
|
def get_guest_grantable_permissions(self) -> List[str]:
|
|
"""Declared because a site chose it: live status on a PC page.
|
|
|
|
endpoints.view only. NOT endpoints.fleet - that is the roster, and the
|
|
whole reason the two were split.
|
|
|
|
KNOWN GAP, tracked separately: the per-asset panels this opens are
|
|
keyed on a small integer asset id and the status panel emits the
|
|
logged-in username, so walking ids reconstructs most of the roster the
|
|
split exists to protect. Bounding or rate-limiting the per-asset panels
|
|
is the fix. Until then a site enabling guest access should know the
|
|
split is a speed bump here, not a wall.
|
|
"""
|
|
return ['endpoints.view']
|
|
|
|
def get_settings_defaults(self) -> List[Dict]:
|
|
"""Settings owned by the endpoints agent-key kill switch.
|
|
|
|
`envkeyfallback` gates the bootstrap .env collector key: while true the
|
|
old COLLECTOR_API_KEY[_ENDPOINTS] still authorizes (back-compat, and the
|
|
only path before a UI key exists). The regenerate flow flips it false so
|
|
the .env key stops working and the managed token is the sole credential,
|
|
making revocation an instant kill switch. `agentkeyprovisioned` records
|
|
that a UI-managed key has been minted so the settings page can render
|
|
the rotate/revoke state instead of the first-run generate state.
|
|
"""
|
|
return [
|
|
{
|
|
'key': 'plugin.endpoints.envkeyfallback',
|
|
'value': 'true',
|
|
'valuetype': 'boolean',
|
|
'category': 'plugin.endpoints',
|
|
'description': 'Accept the bootstrap .env collector key for '
|
|
'endpoints. Turned off once a UI-managed agent '
|
|
'key is provisioned so revocation is a real kill '
|
|
'switch.',
|
|
},
|
|
{
|
|
'key': 'plugin.endpoints.sharedkeyallowed',
|
|
'value': 'true',
|
|
'valuetype': 'boolean',
|
|
'category': 'plugin.endpoints',
|
|
'description': 'Accept the shared fleet collector key from PCs '
|
|
'that have not enrolled. Enrolled PCs are always '
|
|
'refused it. Turn off once every PC has a '
|
|
'per-PC credential.',
|
|
},
|
|
{
|
|
'key': 'plugin.endpoints.autoenroll',
|
|
'value': 'true',
|
|
'valuetype': 'boolean',
|
|
'category': 'plugin.endpoints',
|
|
'description': 'Issue a per-PC credential without admin '
|
|
'approval when the hostname and serial number '
|
|
'match exactly one computer.',
|
|
},
|
|
{
|
|
'key': 'plugin.endpoints.agentkeyprovisioned',
|
|
'value': 'false',
|
|
'valuetype': 'boolean',
|
|
'category': 'plugin.endpoints',
|
|
'description': 'True once an operator has minted a UI-managed '
|
|
'endpoint agent key from the settings page.',
|
|
},
|
|
{
|
|
'key': 'plugin.endpoints.maxpendingenrollments',
|
|
'value': '500',
|
|
'valuetype': 'integer',
|
|
'category': 'plugin.endpoints',
|
|
'description': 'How many enrollment requests may wait for an '
|
|
'admin at once. At the cap, a request from a '
|
|
'hostname and serial never seen before is '
|
|
'refused; PCs already in the queue keep asking '
|
|
'and keep being approved. 0 removes the cap. '
|
|
'POST /enroll takes no auth, so this is what '
|
|
'stops the pending queue growing without limit.',
|
|
},
|
|
]
|
|
|
|
def get_settings_cards(self) -> List[Dict]:
|
|
"""The endpoints management card on the global settings page.
|
|
|
|
A settings card rather than a docs page because it is the operator's
|
|
control surface: manage the agent key (the kill switch), build the site
|
|
agent, and configure what it reports. Admin only - key management mints
|
|
a collector service credential.
|
|
"""
|
|
return [
|
|
{
|
|
'group': 'Endpoints',
|
|
'to': '/settings/endpoints/agentkey',
|
|
'icon': 'key',
|
|
'title': 'Endpoints',
|
|
'description': 'The shared agent key PCs use until they '
|
|
'enroll, the site agent installer builder, and '
|
|
'what the agent reports.',
|
|
'position': 27,
|
|
},
|
|
# Its own card rather than a section of the agent key page: the
|
|
# audience differs. Key management is admin-only, while the script
|
|
# library is reachable by any endpoints.runscript holder, and
|
|
# burying it under an admin page would hide it from exactly the
|
|
# operators the grant exists for.
|
|
{
|
|
'group': 'Endpoints',
|
|
'to': '/settings/endpoints/scripts',
|
|
'icon': 'terminal',
|
|
'title': 'System Scripts',
|
|
'description': 'PowerShell the agent can run on an endpoint. '
|
|
'Runs as SYSTEM; used to trigger scheduled '
|
|
'tasks already registered on the endpoint.',
|
|
'position': 28,
|
|
},
|
|
{
|
|
'group': 'Endpoints',
|
|
'to': '/settings/endpoints/server-cert',
|
|
'icon': 'shield-check',
|
|
'title': 'Server Certificate',
|
|
'description': 'The certificate PCs see when they connect, on '
|
|
'IIS or Docker. Source of the server pin and the '
|
|
'trusted root entry for agents.',
|
|
'position': 29,
|
|
},
|
|
{
|
|
'group': 'Endpoints',
|
|
'to': '/settings/endpoints/enrollment',
|
|
'icon': 'badge-check',
|
|
'title': 'Enrollment',
|
|
'description': 'Move agents to per-PC credentials, approve '
|
|
'enrollment requests, and retire the shared key.',
|
|
'position': 30,
|
|
},
|
|
]
|
|
|
|
def get_navigation_items(self) -> List[Dict]:
|
|
"""Sidebar entry for the fleet list view.
|
|
|
|
Filed under Assets, immediately after PCs (position 15), because the
|
|
fleet view is a telemetry lens over the same computers rather than a
|
|
different kind of thing. It shipped declaring a section of its own,
|
|
which rendered a one-item 'Fleet' group at the bottom of the sidebar,
|
|
far from the PCs it describes.
|
|
"""
|
|
return [
|
|
{
|
|
'name': 'Fleet',
|
|
'icon': 'activity',
|
|
'route': '/endpoints',
|
|
'position': 16,
|
|
'permission': 'endpoints.fleet',
|
|
'section': 'assets',
|
|
},
|
|
]
|
|
|
|
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.
|
|
|
|
Panels feeding the generic renderer: the current status snapshot
|
|
(keyvalue), monitored services and processes split into two list cards
|
|
that share one services snapshot (filtered by kind), the most recent
|
|
curated events (list), and the network interfaces (table). 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-tracked-services',
|
|
'title': 'Tracked Services',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/services',
|
|
'render': 'list',
|
|
'collapsible': True,
|
|
'filter': {'key': 'kind', 'equals': 'service'},
|
|
# Facet on the STATE, not the status colour. statecolor is only
|
|
# ever green, red or grey here, so chips derived from it would
|
|
# read "OK" and "Critical" - not how anybody talks about a
|
|
# service. statelabel carries the real vocabulary: Running,
|
|
# Stopped, Paused, and the Windows transitional states. Any
|
|
# value not listed still gets a chip, appended after these.
|
|
'facets': {
|
|
'key': 'statelabel',
|
|
'order': ['Running', 'Stopped', 'Paused', 'Start Pending',
|
|
'Stop Pending', 'Continue Pending', 'Pause Pending',
|
|
'Unknown'],
|
|
},
|
|
'status': {
|
|
'key': 'statecolor',
|
|
'levels': {
|
|
'green': 'ok',
|
|
'yellow': 'warning',
|
|
'red': 'critical',
|
|
},
|
|
},
|
|
'map': {
|
|
'title': 'name',
|
|
'badge': {'label': 'statelabel', 'color': 'statecolor'},
|
|
'action': 'action',
|
|
},
|
|
'commandendpoint': '/api/endpoints/{assetid}/commands',
|
|
'permission': 'endpoints.command',
|
|
'empty': 'No monitored services reported.',
|
|
'position': 12,
|
|
# Service state is what is being watched; the Start/Stop
|
|
# controls belong beside it, not in a tab of their own.
|
|
# Lives in serviceprocess rather than monitors: that tab is the
|
|
# box's own health, this is what runs on it.
|
|
'group': 'serviceprocess',
|
|
},
|
|
{
|
|
'id': 'endpoints-tracked-processes',
|
|
'title': 'Tracked Processes',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/services',
|
|
'render': 'list',
|
|
'collapsible': True,
|
|
'filter': {'key': 'kind', 'equals': 'process'},
|
|
# Facet on the STATE, not the status colour. statecolor is only
|
|
# ever green, red or grey here, so chips derived from it would
|
|
# read "OK" and "Critical" - not how anybody talks about a
|
|
# service. statelabel carries the real vocabulary: Running,
|
|
# Stopped, Paused, and the Windows transitional states. Any
|
|
# value not listed still gets a chip, appended after these.
|
|
'facets': {
|
|
'key': 'statelabel',
|
|
'order': ['Running', 'Stopped', 'Paused', 'Start Pending',
|
|
'Stop Pending', 'Continue Pending', 'Pause Pending',
|
|
'Unknown'],
|
|
},
|
|
'status': {
|
|
'key': 'statecolor',
|
|
'levels': {
|
|
'green': 'ok',
|
|
'yellow': 'warning',
|
|
'red': 'critical',
|
|
},
|
|
},
|
|
'map': {
|
|
'title': 'name',
|
|
'badge': {'label': 'statelabel', 'color': 'statecolor'},
|
|
'action': 'action',
|
|
},
|
|
'commandendpoint': '/api/endpoints/{assetid}/commands',
|
|
'permission': 'endpoints.command',
|
|
'empty': 'No monitored processes reported.',
|
|
'position': 13,
|
|
# observed process state, not something you act on here
|
|
'group': 'serviceprocess',
|
|
},
|
|
{
|
|
# One feed instead of two cards. Events and commands answer the
|
|
# same question - what happened to this box - and reading them
|
|
# separately meant interleaving two lists by eye. Sits high on
|
|
# the page because it is the first thing looked at when
|
|
# something has gone wrong, and NOT collapsed by default for
|
|
# the same reason.
|
|
'id': 'endpoints-activity',
|
|
'title': 'Activity',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/activity',
|
|
'render': 'list',
|
|
'collapsible': True,
|
|
'status': {
|
|
'key': 'severity',
|
|
'levels': {
|
|
'info': 'ok',
|
|
'warning': 'warning',
|
|
'critical': 'critical',
|
|
},
|
|
},
|
|
'map': {
|
|
'title': 'title',
|
|
'badge': {
|
|
'label': 'severity',
|
|
# The renderer reads the colour from a row field; with
|
|
# no colour declared every badge renders grey, which is
|
|
# what the feed spent its first version doing.
|
|
'color': 'severitycolor',
|
|
'labelmap': {
|
|
'critical': 'Critical',
|
|
'warning': 'Warning',
|
|
'info': 'Info',
|
|
},
|
|
},
|
|
'meta': [
|
|
{'key': 'detail'},
|
|
# datetime, not date: 'date' renders via
|
|
# toLocaleDateString and drops the clock time entirely,
|
|
# which made every event on a given day look
|
|
# simultaneous. Ordering a boot against the logon that
|
|
# followed it is the whole point of this feed.
|
|
{'key': 'eventtime', 'label': 'At', 'format': 'datetime'},
|
|
],
|
|
# 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,
|
|
# the event and command feed IS the activity tab
|
|
'group': 'activity',
|
|
},
|
|
{
|
|
'id': 'endpoints-interfaces',
|
|
'title': 'Network interfaces',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/interfaces',
|
|
'render': 'table',
|
|
'collapsible': True,
|
|
'columns': [
|
|
{'key': 'alias', 'label': 'Adapter'},
|
|
{'key': 'ipv4', 'label': 'IPv4'},
|
|
{'key': 'mac', 'label': 'MAC'},
|
|
{'key': 'gateway', 'label': 'Gateway'},
|
|
{'key': 'subnet', 'label': 'Subnet'},
|
|
{'key': 'dhcp', 'label': 'Assignment'},
|
|
{'key': 'role', 'label': 'Role'},
|
|
{'key': 'link', 'label': 'Link'},
|
|
],
|
|
'empty': 'No network interfaces reported.',
|
|
'position': 16,
|
|
# adapter state is watched, not something you act on.
|
|
# Declared in monitors but RENDERED at the top of the Details
|
|
# tab (PCDetail.vue lifts it by id and excludes it here): which
|
|
# adapter is up and what address it holds is asked at the same
|
|
# moment as "what is this box".
|
|
'group': 'monitors',
|
|
},
|
|
{
|
|
# What this box has been on the network, carved out of the
|
|
# hand-written Network card that used to sit on Details and
|
|
# show only current addresses with a Primary badge.
|
|
#
|
|
# The history is ACCIDENTAL: _record_interfaces matches on MAC
|
|
# then IP and never deletes, so old addresses accumulate. That
|
|
# turned out to be worth reading rather than worth cleaning
|
|
# away, so it is presented as what it already is.
|
|
'id': 'endpoints-iphistory',
|
|
'title': 'IP History',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/iphistory',
|
|
'render': 'table',
|
|
'collapsible': True,
|
|
'columns': [
|
|
{'key': 'ipaddress', 'label': 'IP address', 'mono': True},
|
|
{'key': 'macaddress', 'label': 'MAC', 'mono': True},
|
|
{'key': 'status', 'label': 'Status'},
|
|
{'key': 'firstseen', 'label': 'First seen',
|
|
'format': 'date'},
|
|
# NOT "last seen": modifieddate only moves when a field on
|
|
# the row changes, so an address still held and reported
|
|
# unchanged keeps an old stamp here.
|
|
{'key': 'lastchanged', 'label': 'Last changed',
|
|
'format': 'date'},
|
|
],
|
|
'empty': 'No addresses on record for this endpoint.',
|
|
'position': 16.5,
|
|
'group': 'monitors',
|
|
},
|
|
{
|
|
# Installed-software (SBOM) inventory. Raw discovered list for
|
|
# now: no managed-application linkage is surfaced yet.
|
|
#
|
|
# NOT collapsible. It used to be, and started collapsed, because
|
|
# it was low-priority reference data sitting near the bottom of
|
|
# one long page. It is now the Software tab's whole reason to
|
|
# exist, and somebody who clicked that tab has already said what
|
|
# they came for - making them click again to see it is a step
|
|
# that asks a question already answered.
|
|
'id': 'endpoints-applications',
|
|
'title': 'Installed software',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/applications',
|
|
'render': 'table',
|
|
'columns': [
|
|
{'key': 'displayname', 'label': 'Name'},
|
|
{'key': 'displayversion', 'label': 'Version'},
|
|
{'key': 'publisher', 'label': 'Publisher'},
|
|
],
|
|
'empty': 'No installed software reported.',
|
|
'position': 30,
|
|
# 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',
|
|
},
|
|
{
|
|
# Session-scoped, so it sits with the other things watched
|
|
# rather than acted on. The user context card still carries a
|
|
# one-line summary; this is the same data with room to show a
|
|
# remote path and a per-drive badge, and both are built from
|
|
# _mapped_drive_rows so they cannot disagree.
|
|
#
|
|
# Room reserved here for the future one-click reconnect: the
|
|
# list render already supports a row action, so adding it is a
|
|
# map/commandendpoint change rather than a new card.
|
|
'id': 'endpoints-mappeddrives',
|
|
'title': 'Mapped drives',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/mappeddrives',
|
|
'render': 'list',
|
|
'collapsible': True,
|
|
'status': {
|
|
'key': 'statecolor',
|
|
'levels': {
|
|
'green': 'ok',
|
|
'yellow': 'warning',
|
|
'red': 'critical',
|
|
},
|
|
},
|
|
'map': {
|
|
'title': 'drive',
|
|
'meta': [{'key': 'remote', 'mono': True}],
|
|
'badge': {'label': 'statuslabel', 'color': 'statecolor'},
|
|
},
|
|
'empty': 'No mapped drives reported.',
|
|
'position': 21,
|
|
'group': 'monitors',
|
|
},
|
|
{
|
|
# Volumes and physical drive health in one card, as two
|
|
# sections. VISUAL GROUPING ONLY: the two tables share no key
|
|
# (storage is per volume, health is per physical device) and
|
|
# nothing is reconciled across them. They are one card because
|
|
# the operator reading a full disk and the one reading a dying
|
|
# disk are usually the same person in the same minute.
|
|
'id': 'endpoints-storage',
|
|
'title': 'Storage',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/asset/{assetid}/storage',
|
|
'render': 'tabs',
|
|
'collapsible': True,
|
|
'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. Has
|
|
# a tab of its own rather than sitting in System Monitors: that
|
|
# tab is machine-collected health, this is a human telling you
|
|
# something, and the thumbnail grid is far bulkier than the
|
|
# fact cards it used to sit between.
|
|
# A custom render carries the thumbnail grid and the expanding
|
|
# lightbox with left/right cycling the generic modes cannot.
|
|
'id': 'endpoints-operator-reports',
|
|
'title': 'Operator Reports',
|
|
'assettypes': ['computer'],
|
|
'endpoint': '/api/endpoints/panel/{assetid}/operatorreports',
|
|
'render': 'operatorreports',
|
|
'imagebase': '/api/endpoints/operatorreports',
|
|
'collapsible': True,
|
|
'empty': 'No operator reports for this endpoint.',
|
|
'position': 18,
|
|
'group': 'reports',
|
|
# Not endpoints.view: the screenshots can carry shopfloor
|
|
# content that reading telemetry does not imply a right to see.
|
|
# The panel route and the image route re-check this server-side;
|
|
# withholding the declaration only keeps the tab from appearing.
|
|
'permission': 'endpoints.operatorreports',
|
|
},
|
|
]
|
|
|
|
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',
|
|
'shortname': 'Agent quiet',
|
|
'endpoint': '/api/endpoints/dashboard/quiet',
|
|
'render': 'exceptions',
|
|
'severity': 'warning',
|
|
'permission': 'endpoints.fleet',
|
|
'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',
|
|
'shortname': 'Crashes',
|
|
'endpoint': '/api/endpoints/dashboard/stability',
|
|
'render': 'exceptions',
|
|
'severity': 'critical',
|
|
'permission': 'endpoints.fleet',
|
|
'empty': 'hide',
|
|
'position': 16,
|
|
'map': {
|
|
'title': 'hostname',
|
|
'detail': '',
|
|
'meta': [{'key': 'eventcount', 'suffix': ' events (48h)'}],
|
|
'link': '/pcs/{computerid}',
|
|
},
|
|
},
|
|
# One row, not one per PC: the approval itself happens on the
|
|
# Enrollment page, where the asset picker is.
|
|
{
|
|
'id': 'endpoints-enrollment-pending',
|
|
'viewall': '/settings/endpoints/enrollment',
|
|
'title': 'Enrollment requests waiting for approval',
|
|
'shortname': 'Enrollment',
|
|
'endpoint': '/api/endpoints/dashboard/enrollmentpending',
|
|
'render': 'exceptions',
|
|
'severity': 'warning',
|
|
'permission': 'endpoints.fleet',
|
|
'empty': 'hide',
|
|
'position': 24,
|
|
'map': {
|
|
'title': 'summary',
|
|
'detail': '',
|
|
'link': '/settings/endpoints/enrollment',
|
|
},
|
|
},
|
|
# Info, not the queue: a block is deliberate (a rollback), and a
|
|
# week is only the point where it may have been forgotten.
|
|
{
|
|
'id': 'endpoints-reenroll-blocked',
|
|
'viewall': '/settings/endpoints/enrollment',
|
|
'title': 'PCs blocked from re-enrolling over 7 days',
|
|
'shortname': 'Re-enroll blocked',
|
|
'endpoint': '/api/endpoints/dashboard/reenrollblocked',
|
|
'render': 'exceptions',
|
|
'severity': 'info',
|
|
'permission': 'endpoints.fleet',
|
|
'empty': 'hide',
|
|
'position': 60,
|
|
'map': {
|
|
'title': 'hostname',
|
|
'detail': 'reason',
|
|
'meta': [{'key': 'blockeddays', 'label': 'blocked',
|
|
'suffix': ' days'}],
|
|
'link': '/pcs/{computerid}',
|
|
},
|
|
},
|
|
]
|
|
|
|
# ---- ADR-006 collector contract -------------------------------------
|
|
|
|
def resolve_agent_credential(self, secret: str) -> Optional[int]:
|
|
from .api import agentauth
|
|
return agentauth.resolve_credential(secret)
|
|
|
|
def agent_shared_key_refused(self, hostname: str) -> bool:
|
|
from .api import agentauth
|
|
return agentauth.shared_key_refused_for(hostname)
|
|
|
|
def get_collector_schema(self) -> Optional[Dict]:
|
|
return {
|
|
'identityfield': 'hostname',
|
|
'acceptsagentcredential': True,
|
|
'fields': {
|
|
'type': 'object',
|
|
'required': ['hostname'],
|
|
'properties': {
|
|
'hostname': {
|
|
'type': 'string',
|
|
'description': 'Endpoint hostname. Resolved against '
|
|
'the computers plugin (computers.hostname).',
|
|
},
|
|
'adopt': {
|
|
'type': 'boolean',
|
|
'description': 'First-contact only. When true and the '
|
|
'hostname does not resolve, adoption is '
|
|
'delegated to the computers collector '
|
|
'(see DESIGN-DECISIONS-agent.md 1a). This '
|
|
'plugin never creates assets itself.',
|
|
},
|
|
'loggedinuser': {'type': 'string'},
|
|
'agentversion': {'type': 'string'},
|
|
'trayversion': {'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.',
|
|
},
|
|
'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.',
|
|
},
|
|
'diskhealth': {
|
|
'type': 'array',
|
|
'description': 'Per-drive health from two sources: '
|
|
'physicaldisk (the only one reporting '
|
|
'NVMe) and diskdrive (ATA/SATA and '
|
|
'fallback). Each entry carries a '
|
|
'precomputed predictfailure; null there '
|
|
'means the drive did not answer.',
|
|
},
|
|
'rebootreasons': {
|
|
'type': 'array',
|
|
'description': 'Which signal(s) drove pendingreboot: '
|
|
'componentbasedservicing, windowsupdate, '
|
|
'pendingfilerename. Empty array means '
|
|
'checked and nothing pending; omit when '
|
|
'the check failed.',
|
|
},
|
|
'bootinfo': {
|
|
'type': 'object',
|
|
'description': 'hiberbootenabled (fast startup is '
|
|
'configured) and lastboottype (what the '
|
|
'last boot actually did). Uptime is not '
|
|
'trustworthy when fast startup is in '
|
|
'play.',
|
|
},
|
|
'defender': {
|
|
'type': 'object',
|
|
'description': 'Defender service/realtime state and '
|
|
'signature and scan ages in days. '
|
|
'Persisted but not surfaced.',
|
|
},
|
|
'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 '
|
|
'Windows Uninstall registry hives. Each '
|
|
'item is {displayname, displayversion, '
|
|
'publisher}, de-duped on name+version. '
|
|
'Stored as a per-asset snapshot, '
|
|
'replaced on every report.',
|
|
'items': {
|
|
'type': 'object',
|
|
'required': ['displayname'],
|
|
'properties': {
|
|
'displayname': {'type': 'string'},
|
|
'displayversion': {'type': 'string'},
|
|
'publisher': {'type': 'string'},
|
|
},
|
|
},
|
|
},
|
|
'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'},
|
|
'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',
|
|
'format': 'date-time'},
|
|
'eventmessage': {'type': 'string'},
|
|
'fields': {'type': 'object'},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
# PC-identifying and hardware fields forwarded to the computers collector,
|
|
# which owns the Asset/Computer row and renders the Hardware and Network
|
|
# cards. The endpoints plugin only caches telemetry, so these must be
|
|
# relayed or vendor/model/OS/IP never land on the PC (see
|
|
# DESIGN-DECISIONS-agent.md 1a).
|
|
_COMPUTER_HANDOFF_FIELDS = (
|
|
'loggedinuser', 'serialnumber', 'pctype', 'machinenumber',
|
|
'vendorname', 'modelnumber', 'osname', 'ipaddress', 'lastboottime',
|
|
)
|
|
|
|
def _forward_to_computers(self, hostname: str, payload: Dict):
|
|
"""Return the computers plugin and the PC subset to hand off, or None.
|
|
|
|
Both adoption and steady-state reports relay the same hardware subset,
|
|
so the Hardware and Network cards fill in on every report, not just at
|
|
first contact.
|
|
"""
|
|
from flask import current_app
|
|
|
|
manager = current_app.extensions.get('plugin_manager')
|
|
computers = manager.get_plugin('computers') if manager else None
|
|
if computers is None:
|
|
return None, None
|
|
|
|
subset = {'hostname': hostname}
|
|
for field in self._COMPUTER_HANDOFF_FIELDS:
|
|
if payload.get(field):
|
|
subset[field] = payload.get(field)
|
|
return computers, subset
|
|
|
|
def _adopt_computer(self, hostname: str, payload: Dict,
|
|
warnings: List[str]) -> None:
|
|
"""Delegate first-contact adoption to the computers collector.
|
|
|
|
This plugin never creates Asset or Computer rows itself (see
|
|
DESIGN-DECISIONS-agent.md 1a). The computers plugin is the sole
|
|
authority for that, so on an unknown hostname we forward the
|
|
PC-identifying subset of the payload to its collector. It runs inside
|
|
the same transaction, so a later failure rolls the adoption back.
|
|
"""
|
|
computers, adoptpayload = self._forward_to_computers(hostname, payload)
|
|
if computers is None:
|
|
raise ValueError(
|
|
'cannot adopt {!r}: computers plugin is not available'
|
|
.format(hostname))
|
|
|
|
computers.apply_collector_payload(adoptpayload)
|
|
db.session.flush()
|
|
warnings.append('adopted new computer for hostname: {!r}'.format(hostname))
|
|
|
|
@staticmethod
|
|
def _is_valid_fingerprint(fp: str) -> bool:
|
|
"""Validate SHA256 fingerprint format: exactly 64 lowercase hex chars."""
|
|
if not fp or not isinstance(fp, str):
|
|
return False
|
|
return _FP_REGEX.fullmatch(fp) is not None
|
|
|
|
def _replace_applications(self, assetid: int, applications,
|
|
warnings: List[str]) -> int:
|
|
"""Update an asset's SBOM snapshot using delta comparison.
|
|
|
|
The agent reports the full installed-software list each cycle along with
|
|
a fingerprint hash (SHA256 of displayname|displayversion|publisher) for
|
|
each entry. Rather than delete-all-insert-all, compare fingerprints to
|
|
detect actual changes:
|
|
- Add rows for new fingerprints
|
|
- Remove rows for fingerprints no longer reported
|
|
- Leave unchanged rows untouched
|
|
|
|
This reduces database writes from ~100 ops/endpoint/check-in to ~2 ops
|
|
(99% of check-ins have no application changes). Malformed items (no
|
|
displayname or fingerprint) are skipped, not fatal.
|
|
|
|
Legacy fallback: If agent does not send fingerprints, fall back to
|
|
delete-all-insert-all for backward compatibility during rollout.
|
|
|
|
Safety features:
|
|
- Row-level locking prevents concurrent update races
|
|
- Fingerprint format validation (64 hex chars)
|
|
- Delete count verification detects silent failures
|
|
- Memory limits prevent pathological cases (>5000 apps)
|
|
- Metrics logged for monitoring
|
|
"""
|
|
from flask import current_app
|
|
|
|
# Memory safety: prevent loading massive inventories
|
|
MAX_APPS_PER_ENDPOINT = 5000
|
|
if not applications:
|
|
# Empty list means no apps installed - delete all existing
|
|
EndpointApplication.query.filter_by(assetid=assetid).delete(
|
|
synchronize_session=False)
|
|
return 0
|
|
|
|
# Check if agent sent fingerprints - legacy agents won't have them yet
|
|
has_fingerprints = any(item.get('fingerprint') for item in applications
|
|
if isinstance(item, dict))
|
|
|
|
if not has_fingerprints:
|
|
# LEGACY FALLBACK: No fingerprints, use delete-all-insert-all
|
|
EndpointApplication.query.filter_by(assetid=assetid).delete(
|
|
synchronize_session=False)
|
|
|
|
stored = 0
|
|
for item in applications:
|
|
if not isinstance(item, dict):
|
|
warnings.append('skipped non-object application entry')
|
|
continue
|
|
displayname = (item.get('displayname') or '').strip()
|
|
if not displayname:
|
|
warnings.append('skipped application with no displayname')
|
|
continue
|
|
|
|
db.session.add(EndpointApplication(
|
|
assetid=assetid,
|
|
displayname=displayname[:255],
|
|
displayversion=_clamp(item.get('displayversion'), 128),
|
|
publisher=_clamp(item.get('publisher'), 255),
|
|
appfingerprint=None, # Agent didn't send it
|
|
))
|
|
stored += 1
|
|
|
|
return stored
|
|
|
|
# DELTA LOGIC: Build incoming fingerprint map and validate entries
|
|
incoming = {}
|
|
for item in applications:
|
|
if not isinstance(item, dict):
|
|
warnings.append('skipped non-object application entry')
|
|
continue
|
|
|
|
displayname = (item.get('displayname') or '').strip()
|
|
fingerprint = (item.get('fingerprint') or '').strip()
|
|
|
|
if not displayname:
|
|
warnings.append('skipped application with no displayname')
|
|
continue
|
|
if not fingerprint:
|
|
warnings.append('skipped application with no fingerprint')
|
|
continue
|
|
|
|
# SAFETY: Validate fingerprint format (64 lowercase hex chars)
|
|
if not self._is_valid_fingerprint(fingerprint):
|
|
warnings.append(
|
|
f'skipped application with invalid fingerprint: {displayname} '
|
|
f'(fingerprint: {fingerprint[:32]}...)'
|
|
)
|
|
continue
|
|
|
|
# Truncate strings for MySQL strict mode compatibility
|
|
incoming[fingerprint] = {
|
|
'displayname': displayname[:255],
|
|
'displayversion': _clamp(item.get('displayversion'), 128),
|
|
'publisher': _clamp(item.get('publisher'), 255),
|
|
}
|
|
|
|
incoming_fps = set(incoming.keys())
|
|
|
|
# SAFETY: Memory limit check - prevent loading massive inventories
|
|
# Query exactly MAX rows, then check if more exist (off-by-one fix)
|
|
# Concurrency protection: UNIQUE constraint on (assetid, app_fingerprint)
|
|
# handles race conditions without explicit row locks
|
|
current_apps = (EndpointApplication.query
|
|
.filter_by(assetid=assetid)
|
|
.limit(MAX_APPS_PER_ENDPOINT)
|
|
.all())
|
|
|
|
if len(current_apps) >= MAX_APPS_PER_ENDPOINT:
|
|
# This endpoint has too many apps - fall back to legacy mode
|
|
warnings.append(
|
|
f'asset has {len(current_apps)} apps (exceeds {MAX_APPS_PER_ENDPOINT}), '
|
|
f'falling back to legacy mode'
|
|
)
|
|
# Recursive call with has_fingerprints=False to trigger legacy path
|
|
# (This is safe because we're already in a transaction)
|
|
EndpointApplication.query.filter_by(assetid=assetid).delete(
|
|
synchronize_session=False)
|
|
|
|
stored = 0
|
|
for fp, app_data in incoming.items():
|
|
db.session.add(EndpointApplication(
|
|
assetid=assetid,
|
|
displayname=app_data['displayname'],
|
|
displayversion=app_data['displayversion'],
|
|
publisher=app_data['publisher'],
|
|
appfingerprint=fp,
|
|
))
|
|
stored += 1
|
|
|
|
db.session.flush()
|
|
warnings.append(
|
|
f'asset has {len(current_apps)} apps, exceeded memory limit, '
|
|
f'fell back to legacy mode'
|
|
)
|
|
return stored
|
|
|
|
current_fps = {app.appfingerprint for app in current_apps
|
|
if app.appfingerprint} # Skip NULL fingerprints from migration
|
|
|
|
# TRANSITION: If we have apps without fingerprints, this is the first report
|
|
# after agent upgrade. Delete all NULL-fingerprint apps to prevent duplicates.
|
|
# SAFETY: Validate incoming apps BEFORE deleting legacy apps to prevent
|
|
# leaving asset with zero apps if transaction aborts.
|
|
null_fp_apps = [app for app in current_apps if app.appfingerprint is None]
|
|
if null_fp_apps:
|
|
if not incoming_fps:
|
|
warnings.append(
|
|
f'agent sent empty fingerprint set during NULL transition, '
|
|
f'kept {len(null_fp_apps)} legacy apps')
|
|
return len(null_fp_apps)
|
|
|
|
for app in null_fp_apps:
|
|
db.session.delete(app)
|
|
|
|
# Calculate delta
|
|
to_add = incoming_fps - current_fps # New apps
|
|
to_remove = current_fps - incoming_fps # Uninstalled apps
|
|
unchanged = current_fps & incoming_fps # Unchanged (don't touch)
|
|
|
|
# Remove uninstalled apps
|
|
deleted_count = 0
|
|
if to_remove:
|
|
deleted_count = EndpointApplication.query.filter(
|
|
EndpointApplication.assetid == assetid,
|
|
EndpointApplication.appfingerprint.in_(list(to_remove))
|
|
).delete(synchronize_session=False)
|
|
|
|
# SAFETY: Verify delete count matches expected. Log as WARNING
|
|
# instead of raising error - concurrent updates are rare but valid,
|
|
# and aborting the entire transaction loses state/events/hardware data.
|
|
if deleted_count != len(to_remove):
|
|
current_app.logger.warning(
|
|
f'[collector] asset_id={assetid} delete_count_mismatch: '
|
|
f'expected={len(to_remove)} actual={deleted_count} '
|
|
f'(possible concurrent update, proceeding with caution)')
|
|
warnings.append(
|
|
f'delete count mismatch: expected {len(to_remove)}, '
|
|
f'got {deleted_count} (concurrent update?)')
|
|
|
|
# Add new apps.
|
|
#
|
|
# The unique index on (assetid, appfingerprint) DETECTS a concurrent
|
|
# insert of the same row; it does not RESOLVE one. Two check-ins for the
|
|
# same PC can overlap (an agent retry over a slow request, two cloned
|
|
# machines sharing a hostname, the backfill script against a live
|
|
# fleet), and without a savepoint the IntegrityError from the flush
|
|
# below sinks the whole check-in - state, events and hardware - which is
|
|
# exactly the loss the delete-mismatch branch above was written to
|
|
# avoid. So: flush inside a savepoint, and on a collision roll back only
|
|
# the savepoint, re-read what is there now, and insert what is still
|
|
# missing. Same shape as the health block in apply_collector_payload.
|
|
if to_add or to_remove:
|
|
# SAFETY: Flush to detect constraint violations before continuing.
|
|
# Only flush if we made changes (99% of check-ins change nothing).
|
|
try:
|
|
with db.session.begin_nested():
|
|
self._insert_applications(assetid, incoming, to_add)
|
|
db.session.flush()
|
|
except IntegrityError:
|
|
# The context manager has already released the savepoint, so
|
|
# the deletes and everything before them survive.
|
|
current_app.logger.warning(
|
|
'[collector] asset_id=%s application insert collided with a '
|
|
'concurrent check-in (%d new fingerprint(s)); reconciling',
|
|
assetid, len(to_add))
|
|
return self._reconcile_applications(assetid, incoming, to_add,
|
|
warnings)
|
|
|
|
return len(to_add)
|
|
|
|
@staticmethod
|
|
def _insert_applications(assetid: int, incoming: Dict, fingerprints) -> list:
|
|
"""Add one EndpointApplication per fingerprint. Returns the new rows."""
|
|
rows = []
|
|
for fp in fingerprints:
|
|
app_data = incoming[fp]
|
|
row = EndpointApplication(
|
|
assetid=assetid,
|
|
displayname=app_data['displayname'],
|
|
displayversion=app_data['displayversion'],
|
|
publisher=app_data['publisher'],
|
|
appfingerprint=fp,
|
|
)
|
|
rows.append(row)
|
|
db.session.add(row)
|
|
return rows
|
|
|
|
def _reconcile_applications(self, assetid: int, incoming: Dict, to_add,
|
|
warnings: List[str]) -> int:
|
|
"""Re-insert after a unique-index collision, skipping what now exists.
|
|
|
|
Called only when the bulk insert lost a race. Re-reads the asset's
|
|
fingerprints, drops the ones another writer has since committed, and
|
|
inserts the rest ONE ROW PER SAVEPOINT so a second collision skips that
|
|
row instead of failing the batch again. Per-row savepoints are only
|
|
paid on this path; the common case never reaches it.
|
|
|
|
Never raises: a check-in must commit its telemetry even when the
|
|
application inventory cannot be reconciled. Whatever is missed is
|
|
picked up by the next check-in, which reports the full list again.
|
|
"""
|
|
from flask import current_app
|
|
|
|
# The failed flush rolled the savepoint back, which restores the added
|
|
# rows to pending. Expunge them, or the autoflush on the re-read below
|
|
# retries the insert that just collided - this time with no savepoint
|
|
# under it.
|
|
for row in list(db.session.new):
|
|
if isinstance(row, EndpointApplication) and row.assetid == assetid:
|
|
db.session.expunge(row)
|
|
|
|
existing = {app.appfingerprint
|
|
for app in (EndpointApplication.query
|
|
.filter_by(assetid=assetid)
|
|
.all())
|
|
if app.appfingerprint}
|
|
|
|
stored = 0
|
|
skipped = 0
|
|
for fp in to_add:
|
|
if fp in existing:
|
|
skipped += 1
|
|
continue
|
|
try:
|
|
with db.session.begin_nested():
|
|
self._insert_applications(assetid, incoming, [fp])
|
|
db.session.flush()
|
|
stored += 1
|
|
except IntegrityError:
|
|
for row in list(db.session.new):
|
|
if (isinstance(row, EndpointApplication)
|
|
and row.appfingerprint == fp):
|
|
db.session.expunge(row)
|
|
skipped += 1
|
|
|
|
current_app.logger.warning(
|
|
'[collector] asset_id=%s application_insert_conflict: stored=%d '
|
|
'skipped=%d of %d new (concurrent check-in, telemetry kept)',
|
|
assetid, stored, skipped, len(to_add))
|
|
warnings.append(
|
|
'application insert conflict: stored {}, skipped {} of {} new '
|
|
'(concurrent update?)'.format(stored, skipped, len(to_add)))
|
|
return stored
|
|
|
|
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')
|
|
|
|
from .api import agentauth
|
|
if agentauth.is_credential_caller():
|
|
# A credentialed agent reports only for its own asset, and never
|
|
# adopts: its asset already exists.
|
|
if agentauth.resolve_hostname(hostname) != g.agentassetid:
|
|
raise PermissionError(
|
|
'The payload names a different endpoint than the credential')
|
|
|
|
computer = (Computer.query
|
|
.filter(db.func.lower(Computer.hostname) == hostname.lower())
|
|
.first())
|
|
if computer is None:
|
|
# First-contact adoption is delegated to the computers collector,
|
|
# which is the only authority allowed to create Asset + Computer
|
|
# rows (DESIGN-DECISIONS-agent.md 1a). This plugin never invents
|
|
# assets itself. Adoption runs in the dispatcher's transaction, so a
|
|
# later ValueError rolls the whole report back.
|
|
if payload.get('adopt'):
|
|
self._adopt_computer(hostname, payload, warnings)
|
|
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))
|
|
else:
|
|
# Already adopted: relay the hardware/identity subset on every
|
|
# report so the computers plugin keeps vendor/model/OS/IP current.
|
|
# Without this, Hardware and Network stay blank after adoption
|
|
# because this plugin only caches telemetry.
|
|
computers, subset = self._forward_to_computers(hostname, payload)
|
|
if computers is not None and len(subset) > 1:
|
|
computers.apply_collector_payload(subset)
|
|
db.session.flush()
|
|
|
|
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 'trayversion' in payload:
|
|
state.trayversion = payload.get('trayversion')
|
|
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'))
|
|
|
|
# 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:
|
|
# Clamped like every other String column here: an over-long value
|
|
# aborts the whole collector transaction under STRICT_TRANS_TABLES,
|
|
# so one bad agent report would drop a machine's entire telemetry.
|
|
state.domain = _clamp(payload.get('domain'), 255)
|
|
if 'pendingreboot' in payload:
|
|
state.pendingreboot = payload.get('pendingreboot')
|
|
if 'diskhealth' in payload:
|
|
state.diskhealthjson = _as_json(payload.get('diskhealth'))
|
|
if 'rebootreasons' in payload:
|
|
state.rebootreasonsjson = _as_json(payload.get('rebootreasons'))
|
|
if 'bootinfo' in payload:
|
|
state.bootinfojson = _as_json(payload.get('bootinfo'))
|
|
if 'defender' in payload:
|
|
state.defenderjson = _as_json(payload.get('defender'))
|
|
|
|
# Presence-guarded like the probe output below: the agent omits the key
|
|
# entirely when the collector utility is not installed, and overwriting
|
|
# to null there would discard a still-valid scan result from before the
|
|
# utility was removed.
|
|
if 'dellupdates' in payload:
|
|
state.dellupdatesjson = _as_json(payload.get('dellupdates'))
|
|
|
|
# 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
|
|
# replace when the key is present: a report that omits applications must
|
|
# not wipe the last-known inventory. The delete + re-insert happens in
|
|
# this same transaction (the dispatcher owns the commit), so a reader
|
|
# never sees a half-written list. displayname is required; malformed
|
|
# rows are skipped rather than failing the whole report.
|
|
if 'applications' in payload:
|
|
self._replace_applications(assetid, payload.get('applications'),
|
|
warnings)
|
|
|
|
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
|
|
# keys and a compact preview makes incoming telemetry visible in the
|
|
# logs so we can confirm the agent is sending it before wiring the UI.
|
|
unknown_keys = [key for key in payload if key not in KNOWN_PAYLOAD_KEYS]
|
|
if unknown_keys:
|
|
preview = {key: payload.get(key) for key in sorted(unknown_keys)}
|
|
try:
|
|
preview_text = json.dumps(preview, default=str)[:2000]
|
|
except (TypeError, ValueError):
|
|
preview_text = repr(preview)[:2000]
|
|
logger.info(
|
|
'endpoint %s sent %d unpersisted telemetry key(s): %s | preview=%s',
|
|
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'))
|
|
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
|
|
|
|
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:
|
|
# SAFETY: Row-level locking prevents race condition (vuln #7 from audit).
|
|
# WITH FOR UPDATE locks matching rows in MySQL, preventing concurrent INSERT
|
|
# of duplicates. The UNIQUE constraint on (assetid, eventchannel, eventrecordid)
|
|
# provides additional protection against race conditions.
|
|
rows = (db.session.query(EndpointEvent.eventchannel,
|
|
EndpointEvent.eventrecordid)
|
|
.filter(EndpointEvent.assetid == assetid,
|
|
EndpointEvent.eventrecordid.in_(
|
|
[recordid for _c, recordid in keyed]))
|
|
.with_for_update()
|
|
.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,
|
|
eventrecordid=eventrecordid,
|
|
eventchannel=eventchannel,
|
|
eventsource=record.get('eventsource'),
|
|
eventtime=_parse_dt(record.get('eventtime')),
|
|
eventmessage=record.get('eventmessage'),
|
|
payloadjson=_as_json(record.get('fields')),
|
|
))
|
|
stored_events += 1
|
|
|
|
if skipped_events:
|
|
logger.info(
|
|
'[collector] asset_id=%d hostname=%s events_stored=%d '
|
|
'events_skipped=%d (duplicates detected)',
|
|
assetid, hostname, stored_events, skipped_events)
|
|
else:
|
|
logger.debug(
|
|
'endpoint %s stored %d event(s), no duplicates',
|
|
hostname, stored_events)
|
|
|
|
# flush, not commit: the collector dispatcher owns the transaction.
|
|
db.session.flush()
|
|
|
|
# Health rollup, AFTER the flush so the events just written are visible
|
|
# to the evaluator's query in this same transaction. Recomputed here,
|
|
# per asset, rather than by the Fleet page across the whole fleet on
|
|
# every load - the verdict changes when telemetry changes, so this is
|
|
# where it belongs, and it lands in the SAME commit as the telemetry it
|
|
# was derived from.
|
|
#
|
|
# Fail-open, matching the reconciler above: a health error must never
|
|
# fail an otherwise-good check-in. The previous verdict simply stands,
|
|
# which is correct - a failed evaluation is not evidence of health.
|
|
#
|
|
# SAVEPOINT: a failed flush (IntegrityError and friends) otherwise
|
|
# leaves the session in a rolled-back-needed state, and the caught
|
|
# exception would still sink the outer commit of the whole check-in.
|
|
# Rolling back only the savepoint keeps the telemetry flushed above.
|
|
try:
|
|
from .services.health import evaluate_endpoint_health
|
|
with db.session.begin_nested():
|
|
evaluate_endpoint_health(
|
|
assetid, state,
|
|
computertypeid=getattr(computer, 'computertypeid', None))
|
|
db.session.flush()
|
|
except Exception:
|
|
logger.exception(
|
|
'health evaluation failed for asset %s; leaving verdict '
|
|
'unchanged', assetid)
|
|
|
|
return {
|
|
'action': action,
|
|
'assetid': assetid,
|
|
'identityvalue': hostname,
|
|
'storedevents': stored_events,
|
|
'warnings': warnings,
|
|
} |