ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
13be0ef90f24578330ffb16a7123969d22defeed
shopdb-flask/plugins/endpoints/plugin.py
cproudlock 13be0ef90f Import the wilm agent store and WINC adoption work, with four fixes 
Eleven commits from the sister site: a reversible Maximo/WINC machine-adoption
import, a versioned agent binary store with upload and rollback, remote-BuildKit
CI, tray version reporting, and UI fixes. Applied as a patch again - the
published mirror rewrites hashes, so the branches have unrelated histories and
git refuses to merge them.

Reviewed by three agents before merging. Four things fixed here.

THE PUBLISH WORKFLOW WAS DEAD. The new remote-BuildKit step in
publish-images.yml sits at column 0 instead of indented under steps:, so the
whole file is invalid YAML and GitHub would refuse it. Their entire GHCR
distribution channel - which Start-ShopDBDev.ps1 and the compose files consume -
was broken at the source and clearly never run. Indented; all five workflows now
parse.

THE IMPORT'S APPROVAL CHECK COULD NOT FAIL. commit() built the plan and the
approval packet from the same upload and then asked the writer whether they
matched, so packet_matches_plan compared a hash against itself. The documented
contract - "a human approves a specific, frozen plan, and the writer can prove
nothing drifted" - did not exist in the web path: an admin could preview file A
and commit file B, gated only by the needs-create group NAMES they ticked. The
preview's planhash is now required at commit and checked against a freshly
rebuilt plan. Their tests committed without previewing, which is what let this
through; they go through the real flow now, and three new tests pin it - all
confirmed to fail with the check stubbed out.

endpoints.agentpackage QUIETLY GREW. It used to mean "wrap the staged exe in a
ZIP"; with the upload routes it also means "choose the executable every future
package installs and runs as SYSTEM". Before this, staging needed filesystem
access to the server. The permission description still said only "generate and
download a package", so an admin granting it could not see what they were
granting. Reworded.

NOTHING RECORDED WHAT WAS STAGED, OR BY WHOM. The manifest held version, size
and timestamp; re-uploading a version replaces the bytes in place under the same
label. A sha256 of the stored file is now recorded, and the upload is
audit-logged with it - after an incident there was previously no answer to who
put that binary on the fleet.

Also documented the nine new routes in docs/api-inventory.json (that gate cannot
run on the mirror, so it falls to us at promotion time again), regenerated the
spec and the project map, and kept our own docs where they conflicted: docs/ is
scrubbed from the mirror, so their DEV-BUILD-TEST.md still had the dev-be
channel mapping we corrected, and their PLUGIN-HOOKS.md was a new 45-line file
rather than our 737-line canonical one. Their journal section is appended to
ours.

Contract version comes across as 0.30.0 for a genuinely additive surface.

Suite 2369, frontend 324, gate green.

Scope: product
2026-09-03 07:54:14 -04:00

912 lines
42 KiB
Python
Raw Blame History

"""Endpoints plugin main class.
Endpoint RMM telemetry reported by the ShopDB agent. Accepts collector input
over HTTPS per ADR-006: the agent posts a per-host payload to
POST /api/collector/endpoints, the plugin resolves the reported hostname to an
existing computer asset, overwrites that asset's cached EndpointState, and
appends any curated Windows events the report carried - all in one commit, so
the cached snapshot and the events it summarizes never disagree.
Identity is the hostname, resolved against the computers plugin. An unresolved
hostname is reported as a ValueError (surfaced to the caller as a validation
error), never silently dropped: an unknown host is a real gap between the fleet
and ShopDB, the same discipline the backups collector uses for machine numbers.
"""
import json
import logging
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Type
from flask import Flask, Blueprint
from shopdb.plugins.base import BasePlugin, PluginMeta
from shopdb.api import db
from .api import endpoints_bp
from .eventcatalog import EVENT_CATALOG
from .models import (
EndpointState, EndpointEvent, 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)
# 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',
})
def _parse_dt(value):
"""Best-effort parse of an ISO-8601 string to naive UTC, or None."""
if not value:
return None
if isinstance(value, datetime):
return value.replace(tzinfo=None)
text = str(value).strip().replace('Z', '+00:00')
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is not None:
parsed = parsed.astimezone(tz=None).replace(tzinfo=None)
return parsed
def _as_json(value):
"""Serialize a list/dict to JSON text, or None when nothing was sent."""
if value is None:
return None
return json.dumps(value, separators=(',', ':'), sort_keys=True)
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'),
# 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 grant for software delivery (run_script, update_app).
# Required IN ADDITION to endpoints.command by the enqueue route, so
# running a staged script or pushing an installer is a strictly
# heavier grant than a routine service/process toggle.
('endpoints.deploy',
'Run staged scripts and push application installers to endpoints', '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.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.',
},
]
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': 'Manage the endpoint agent key, build the site '
'agent, and configure what it reports.',
'position': 27,
},
]
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',
},
# TEMPORARY / DEV ONLY. A manual bench to exercise the update_app and
# run_script delivery path while Cameron builds the production UI.
# Gated on endpoints.deploy so only delivery operators see it. Remove
# this entry (and the DeliveryTestBench view/route) once the real
# delivery UI ships.
{
'name': 'Delivery test bench',
'icon': 'send',
'route': '/endpoints/delivery-test',
'position': 17,
'permission': 'endpoints.deploy',
'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.
'group': 'monitors',
},
{
'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': 'monitors',
},
{
# 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'},
{'key': 'eventtime', 'label': 'At', 'format': 'date'},
],
},
'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'},
],
'empty': 'No network interfaces reported.',
'position': 16,
# adapter state is watched, not something you act on
'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',
},
{
# Operator-submitted issue reports: a free-text note and an
# optional screenshot the operator captured from the tray. Sits
# in the Monitors tab beside the other watch-and-observe panels.
# 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': 'monitors',
},
]
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}',
},
},
]
# ---- ADR-006 collector contract -------------------------------------
def get_collector_schema(self) -> Optional[Dict]:
return {
'identityfield': 'hostname',
'fields': {
'type': 'object',
'required': ['hostname'],
'properties': {
'hostname': {
'type': 'string',
'description': 'Endpoint hostname. Resolved against '
'the computers plugin (computers.hostname).',
},
'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.',
},
'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'},
'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))
def _replace_applications(self, assetid: int, applications,
warnings: List[str]) -> int:
"""Replace an asset's SBOM snapshot with the reported set.
The agent reports the full installed-software list each cycle, so the
current inventory is the newest report, not an accumulation. Delete the
asset's existing rows and insert the reported ones in the same
transaction the caller flushes. appid is left null: no
discovery-to-catalog ("Managed Applications") matcher is wired yet.
Malformed items (no displayname) are skipped, not fatal.
"""
EndpointApplication.query.filter_by(assetid=assetid).delete(
synchronize_session=False)
if not applications:
return 0
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
# TRUNCATE EVERY STRING, not just the name. On MySQL in strict
# mode an over-long value raises Data too long and takes down the
# WHOLE check-in transaction, so one machine with a malformed
# Uninstall registry entry stops reporting entirely - telemetry,
# events and all - until somebody uninstalls the offender. SQLite
# does not enforce declared lengths, so no test here would ever see
# it. displayname was already clamped; these two were not.
db.session.add(EndpointApplication(
assetid=assetid,
displayname=displayname[:255],
displayversion=_clamp(item.get('displayversion'), 128),
publisher=_clamp(item.get('publisher'), 255),
))
stored += 1
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')
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'))
# 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
for record in (payload.get('events') or []):
try:
eventid = int(record.get('eventid'))
except (TypeError, ValueError):
warnings.append('skipped event with non-integer eventid')
continue
if eventid not in CURATED_EVENT_IDS:
# Outside the catalog: ignored on purpose, not an error.
continue
db.session.add(EndpointEvent(
assetid=assetid,
eventid=eventid,
eventchannel=record.get('eventchannel'),
eventsource=record.get('eventsource'),
eventtime=_parse_dt(record.get('eventtime')),
eventmessage=record.get('eventmessage'),
payloadjson=_as_json(record.get('fields')),
))
stored_events += 1
# flush, not commit: the collector dispatcher owns the transaction.
db.session.flush()
return {
'action': action,
'assetid': assetid,
'identityvalue': hostname,
'storedevents': stored_events,
'warnings': warnings,
}
Reference in New Issue View Git Blame Copy Permalink