Files
shopdb-flask/plugins/endpoints/models/endpoint.py

368 lines
16 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)
# Warn-only reconciler output, refreshed on every agent check-in in the same
# commit as servicesjson. Each entry names a policy/shipped-attached monitor
# the endpoint is expected to be running but the latest snapshot does not
# report as running, so an operator sees the drift and intervenes manually.
# This is warn-only: nothing here is auto-remediated. Null = never computed.
driftjson = 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)
# The content digest as it stood WHEN AN OPERATOR APPROVED THIS PUSH.
#
# Delivery descriptors are re-resolved at claim time, minutes or hours
# later, so without this the bytes an operator approved and the bytes the
# agent runs are two different lookups against mutable content. Whoever can
# replace a script body or an installer package could swap it in the gap
# and the digest would faithfully describe the new bytes.
#
# Set at enqueue for run_script and update_app; null for actions that
# deliver no content. The claim path refuses a command whose descriptor no
# longer matches, so a swap fails the command instead of shipping.
pinnedsha256 = db.Column(db.String(64), 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'
# BigInteger on MySQL, Integer on SQLite. This table is rewritten wholesale
# on every check-in, so the key never stops climbing and a 32-bit column
# eventually stops accepting inserts - see the migration for the arithmetic.
#
# The variant is not decoration: SQLite only autoincrements a column typed
# exactly INTEGER PRIMARY KEY, so a plain BigInteger makes every insert fail
# with a NOT NULL violation on the test database. Nothing caught that,
# because no test inserts into this table.
endpointapplicationid = db.Column(
db.BigInteger().with_variant(db.Integer, 'sqlite'), 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),
}