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

"""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)
    # The interactive tray app versions independently of the agent service, so
    # it is reported and cached as its own field (AGENT-CONTRACT.md 2). Null
    # when the endpoint has no tray installed or an older agent that predates
    # the field.
    trayversion = 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,
            'trayversion': self.trayversion,
            '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),
        }