dc2f4266bc
The command queue is at-most-one-in-flight, but nothing ever expired a stuck command. When an agent claimed a command (RUNNING) then died before reporting a result - killed, uninstalled, or offline mid-flight - the row stayed in-flight forever and every operator action for that endpoint returned 409 'already in flight'. Reap in-flight rows past a staleness ceiling (15m for RUNNING, 60m for PENDING) to a new STATUS_EXPIRED at enqueue time, inside the same transaction, so the queue self-heals lazily the next time someone queues work. No background job and no schema migration - status is free-text.
339 lines
14 KiB
Python
339 lines
14 KiB
Python
"""Endpoint RMM telemetry models.
|
|
|
|
Two shapes, one per concern:
|
|
|
|
EndpointState The cached CURRENT snapshot for an asset - one row per asset,
|
|
overwritten in place on every agent report. Holds the fast-
|
|
moving metrics (uptime, idle, last boot, logged-in user) and
|
|
the latest inventory snapshots (services/processes, open ports,
|
|
interfaces, scheduled tasks) as serialized JSON. This is the
|
|
cached-current-value half of the ledger pattern: it moves in
|
|
the SAME commit as any EndpointEvent rows the report carries.
|
|
|
|
EndpointEvent Append-only record of a curated Windows Event Log entry. The
|
|
agent only forwards a fixed catalog of IDs (logon/logoff,
|
|
privilege use, service install, unexpected shutdown, DCOM,
|
|
crash) so this stays a readable security/stability history
|
|
rather than a raw event firehose.
|
|
|
|
Both link to the core Asset via assetid (FK, ON DELETE CASCADE), so an endpoint
|
|
is the same physical thing as its Computer/Asset - no duplicate identity. The
|
|
agent reports a hostname; the plugin resolves it to the existing Computer.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
from shopdb.api import db
|
|
|
|
|
|
def _utciso(value):
|
|
"""Naive-UTC datetime -> ISO string marked UTC, or None.
|
|
|
|
Every datetime this plugin stores is naive UTC. Serializing without the Z
|
|
leaves the receiver to guess, and JavaScript guesses browser-local.
|
|
"""
|
|
return value.isoformat() + 'Z' if value else None
|
|
|
|
|
|
class EndpointState(db.Model):
|
|
"""Latest reported state for a single endpoint (one row per asset)."""
|
|
|
|
__tablename__ = 'endpointstates'
|
|
|
|
endpointstateid = db.Column(db.Integer, primary_key=True)
|
|
|
|
# The asset this state belongs to. Unique: exactly one current snapshot per
|
|
# endpoint. Resolved from the reported hostname against the computers plugin.
|
|
assetid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('assets.assetid', ondelete='CASCADE'),
|
|
unique=True,
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
|
|
# Read-only view of the owning asset. No cascade or backref: the asset side
|
|
# must not gain a dependency on this plugin.
|
|
asset = db.relationship('Asset', lazy='joined', viewonly=True)
|
|
|
|
hostname = db.Column(db.String(100), nullable=True)
|
|
loggedinuser = db.Column(db.String(100), nullable=True)
|
|
agentversion = db.Column(db.String(40), nullable=True)
|
|
|
|
# Fast-moving metrics
|
|
lastboottime = db.Column(db.DateTime, nullable=True)
|
|
uptimeseconds = db.Column(db.BigInteger, nullable=True)
|
|
idleseconds = db.Column(db.BigInteger, nullable=True)
|
|
cpupercent = db.Column(db.Float, nullable=True)
|
|
memorypercent = db.Column(db.Float, nullable=True)
|
|
|
|
# Latest inventory snapshots, serialized JSON (MEDIUMTEXT). Kept as the
|
|
# newest snapshot rather than one row per sample: the agent reports the full
|
|
# list each cycle and history of these lists is not what operators ask for.
|
|
servicesjson = db.Column(db.Text(length=16777215), nullable=True)
|
|
portsjson = db.Column(db.Text(length=16777215), nullable=True)
|
|
interfacesjson = db.Column(db.Text(length=16777215), nullable=True)
|
|
scheduledtasksjson = db.Column(db.Text(length=16777215), nullable=True)
|
|
|
|
# Last-known MTConnect agent run state, cached so the WebUI toggle can show
|
|
# a real state between command polls. Set by the agent's mtconnect_status
|
|
# result. Null = never reported (unknown).
|
|
mtconnectrunning = db.Column(db.Boolean, nullable=True)
|
|
mtconnectcheckedat = db.Column(db.DateTime, nullable=True)
|
|
|
|
# When the agent last successfully reported. There is no heartbeat separate
|
|
# from the report, so silence past the quiet-hours window is the only
|
|
# evidence an endpoint has stopped checking in.
|
|
lastreportedat = db.Column(db.DateTime, nullable=True, index=True)
|
|
|
|
createdat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
|
|
|
|
def __repr__(self):
|
|
return f"<EndpointState {self.hostname or self.assetid}>"
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'endpointstateid': self.endpointstateid,
|
|
'assetid': self.assetid,
|
|
'hostname': self.hostname,
|
|
'loggedinuser': self.loggedinuser,
|
|
'agentversion': self.agentversion,
|
|
'lastboottime': _utciso(self.lastboottime),
|
|
'uptimeseconds': self.uptimeseconds,
|
|
'idleseconds': self.idleseconds,
|
|
'cpupercent': self.cpupercent,
|
|
'memorypercent': self.memorypercent,
|
|
'mtconnectrunning': self.mtconnectrunning,
|
|
'mtconnectcheckedat': _utciso(self.mtconnectcheckedat),
|
|
'lastreportedat': _utciso(self.lastreportedat),
|
|
'createdat': _utciso(self.createdat),
|
|
}
|
|
|
|
|
|
class EndpointCommand(db.Model):
|
|
"""A queued operator command targeting one endpoint agent.
|
|
|
|
The WebUI enqueues a named action (server allowlist only, never an arbitrary
|
|
command string) for an asset. The agent claims the next pending command on
|
|
its long-poll, executes it locally, and reports the outcome back, which
|
|
moves the row to a terminal status in the same request. This is a small
|
|
at-most-one-in-flight queue, not a general job system.
|
|
"""
|
|
|
|
__tablename__ = 'endpointcommands'
|
|
|
|
# Lifecycle: pending -> running -> succeeded|failed. A command is claimed
|
|
# (pending -> running) by exactly one agent poll, then finalized by its
|
|
# result post. A command that is claimed but whose agent never reports back
|
|
# (agent killed/uninstalled/offline mid-flight) would otherwise pin the
|
|
# at-most-one guard forever; the enqueue path reaps such rows to STATUS_EXPIRED
|
|
# so a dead in-flight command cannot wedge the endpoint's command queue.
|
|
STATUS_PENDING = 'pending'
|
|
STATUS_RUNNING = 'running'
|
|
STATUS_SUCCEEDED = 'succeeded'
|
|
STATUS_FAILED = 'failed'
|
|
STATUS_EXPIRED = 'expired'
|
|
|
|
# In-flight staleness ceilings. A RUNNING command is expected to finish in
|
|
# seconds to a couple of minutes; the generous ceiling tolerates a slow
|
|
# app install while still freeing the queue when the agent is gone. A
|
|
# PENDING command is claimed within one poll (~25s) by a live agent, so a
|
|
# long-unclaimed row means the agent is offline and the queue should not
|
|
# stay wedged behind it.
|
|
RUNNING_STALE_SECONDS = 15 * 60
|
|
PENDING_STALE_SECONDS = 60 * 60
|
|
|
|
endpointcommandid = db.Column(db.Integer, primary_key=True)
|
|
|
|
assetid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('assets.assetid', ondelete='CASCADE'),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
|
|
asset = db.relationship('Asset', lazy='joined', viewonly=True)
|
|
|
|
# The allowlisted action name (for example 'mtconnect_start'). Validated
|
|
# against the server allowlist at enqueue time, so only known names land.
|
|
action = db.Column(db.String(64), nullable=False)
|
|
|
|
# Generic actions (service_* / process_*) carry their subject over the wire:
|
|
# target is a Windows service name or a process image (for example
|
|
# 'DncMain.exe'); targetkind is 'service' or 'process'. Fixed-target actions
|
|
# (mtconnect_*) leave both null. This is what lets a site build the agent
|
|
# once and drive every future policy/app purely from server data.
|
|
target = db.Column(db.String(255), nullable=True)
|
|
targetkind = db.Column(db.String(16), nullable=True)
|
|
|
|
status = db.Column(db.String(20), nullable=False,
|
|
default=STATUS_PENDING, index=True)
|
|
|
|
# Who queued it (userid) for audit; the agent never sees this.
|
|
requestedby = db.Column(db.Integer, nullable=True)
|
|
|
|
# Agent-reported outcome. resultjson holds structured detail (for example
|
|
# the observed MTConnect running state); resultmessage is a short summary.
|
|
resultmessage = db.Column(db.Text, nullable=True)
|
|
resultjson = db.Column(db.Text(length=16777215), nullable=True)
|
|
|
|
createdat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
|
|
claimedat = db.Column(db.DateTime, nullable=True)
|
|
completedat = db.Column(db.DateTime, nullable=True)
|
|
|
|
__table_args__ = (
|
|
db.Index('idx_endpointcommands_assetstatus', 'assetid', 'status'),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<EndpointCommand {self.action} asset={self.assetid} {self.status}>"
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'endpointcommandid': self.endpointcommandid,
|
|
'assetid': self.assetid,
|
|
'action': self.action,
|
|
'target': self.target,
|
|
'targetkind': self.targetkind,
|
|
'status': self.status,
|
|
'requestedby': self.requestedby,
|
|
'resultmessage': self.resultmessage,
|
|
'createdat': _utciso(self.createdat),
|
|
'claimedat': _utciso(self.claimedat),
|
|
'completedat': _utciso(self.completedat),
|
|
}
|
|
|
|
|
|
class EndpointEvent(db.Model):
|
|
"""A single curated Windows Event Log record forwarded by the agent."""
|
|
|
|
__tablename__ = 'endpointevents'
|
|
|
|
endpointeventid = db.Column(db.Integer, primary_key=True)
|
|
|
|
assetid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('assets.assetid', ondelete='CASCADE'),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
|
|
asset = db.relationship('Asset', lazy='joined', viewonly=True)
|
|
|
|
# The Windows Event ID (for example 4625 failed logon, 41 unexpected
|
|
# shutdown, 1001 application crash). Indexed because alerting queries filter
|
|
# on it.
|
|
eventid = db.Column(db.Integer, nullable=False, index=True)
|
|
|
|
# Event Log channel the record came from: Security, System, Application.
|
|
eventchannel = db.Column(db.String(50), nullable=True)
|
|
eventsource = db.Column(db.String(255), nullable=True)
|
|
|
|
# When the event happened on the endpoint (from the record), not when it
|
|
# reached ShopDB.
|
|
eventtime = db.Column(db.DateTime, nullable=True, index=True)
|
|
|
|
eventmessage = db.Column(db.Text, nullable=True)
|
|
|
|
# Structured fields the agent extracted for this ID (for example the target
|
|
# account on a 4625). Advisory index only, not part of dedup.
|
|
payloadjson = db.Column(db.Text(length=16777215), nullable=True)
|
|
|
|
recordedat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
|
|
|
|
__table_args__ = (
|
|
db.Index('idx_endpointevents_assetevent', 'assetid', 'eventtime'),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<EndpointEvent {self.eventid} asset={self.assetid}>"
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'endpointeventid': self.endpointeventid,
|
|
'assetid': self.assetid,
|
|
'eventid': self.eventid,
|
|
'eventchannel': self.eventchannel,
|
|
'eventsource': self.eventsource,
|
|
'eventtime': _utciso(self.eventtime),
|
|
'eventmessage': self.eventmessage,
|
|
'recordedat': _utciso(self.recordedat),
|
|
}
|
|
|
|
|
|
class EndpointApplication(db.Model):
|
|
"""One installed-software (SBOM) entry reported for an endpoint.
|
|
|
|
The agent enumerates both Windows Uninstall registry hives and emits a
|
|
per-item {displayname, displayversion, publisher}, de-duped on
|
|
name+version. The server stores that as a per-asset SNAPSHOT: on each
|
|
report the existing rows for the asset are replaced with the reported set,
|
|
mirroring how the services/ports inventories are kept as the newest list
|
|
rather than an append-only history. History of installs is not what
|
|
operators ask for; the current inventory is.
|
|
|
|
appid is groundwork for a future link to the core applications catalog
|
|
(the "Managed Applications" concept): a nullable, unenforced-by-match FK
|
|
that a later matcher can populate to associate a discovered install with a
|
|
managed Application row. No matching logic exists yet - the column and the
|
|
relationship are laid down here so the association can be added without a
|
|
second migration.
|
|
"""
|
|
|
|
__tablename__ = 'endpointapplications'
|
|
|
|
endpointapplicationid = db.Column(db.Integer, primary_key=True)
|
|
|
|
# The asset this install belongs to. Not unique: many applications per
|
|
# endpoint. CASCADE so removing the asset clears its SBOM snapshot.
|
|
assetid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('assets.assetid', ondelete='CASCADE'),
|
|
nullable=False,
|
|
index=True,
|
|
)
|
|
|
|
asset = db.relationship('Asset', lazy='joined', viewonly=True)
|
|
|
|
# Wire fields, stored under the same lowercase/no-underscore names the agent
|
|
# emits. displayname is always present (the agent skips entries without it);
|
|
# displayversion and publisher default to '' on the agent side.
|
|
displayname = db.Column(db.String(255), nullable=False, index=True)
|
|
displayversion = db.Column(db.String(128), nullable=True)
|
|
publisher = db.Column(db.String(255), nullable=True)
|
|
|
|
# Groundwork link to the core applications catalog ("Managed Applications").
|
|
# Nullable and unpopulated for now: no discovery-to-catalog matcher is wired
|
|
# yet. SET NULL so retiring a managed Application does not delete discovered
|
|
# SBOM rows - the install is still on the machine, it is just no longer
|
|
# managed.
|
|
appid = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('applications.appid', ondelete='SET NULL'),
|
|
nullable=True,
|
|
index=True,
|
|
)
|
|
|
|
application = db.relationship('Application', lazy='select', viewonly=True)
|
|
|
|
# When this snapshot row was written (the report that produced it).
|
|
recordedat = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
|
|
|
|
__table_args__ = (
|
|
db.Index('idx_endpointapplications_assetname', 'assetid', 'displayname'),
|
|
)
|
|
|
|
def __repr__(self):
|
|
return f"<EndpointApplication {self.displayname!r} asset={self.assetid}>"
|
|
|
|
def to_dict(self):
|
|
return {
|
|
'endpointapplicationid': self.endpointapplicationid,
|
|
'assetid': self.assetid,
|
|
'displayname': self.displayname,
|
|
'displayversion': self.displayversion,
|
|
'publisher': self.publisher,
|
|
'appid': self.appid,
|
|
'recordedat': _utciso(self.recordedat),
|
|
}
|