ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add endpoints plugin with fleet list view and monitoring drafts

Browse Source 
Server-side RMM telemetry plugin: current-snapshot endpointstates plus append-only curated endpointevents, resolved to existing Computer assets by hostname. Adds ADR-010 asset panels (status, services, recent events), dashboard quiet and stability cards, and a minimal Fleet list view that links into existing PC detail pages. Registers the activity sidebar icon and a dev-only sample seeder. Ingest arrives on the generic collector route; the future C# agent design record lives in docs/DESIGN-DECISIONS-agent.md.
This commit is contained in:
benjamin.baker
2026-08-27 05:41:06 -04:00
parent ce1fbd6d62
commit 8b1ba3c310
18 changed files with 1774 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
plugins/endpoints/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Endpoints plugin package."""
from .plugin import EndpointsPlugin
__all__ = ['EndpointsPlugin']

5
plugins/endpoints/api/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Endpoints plugin API."""
from .routes import endpoints_bp
__all__ = ['endpoints_bp']

350
plugins/endpoints/api/routes.py Normal file
View File

@@ -0,0 +1,350 @@
"""Endpoints plugin API endpoints (read + dashboard).
Ingest is NOT here: agent reports arrive on the generic collector route
POST /api/collector/endpoints (ADR-006), authorized by a collector-scoped
service token, and are handled by EndpointsPlugin.apply_collector_payload.
These routes are the human-facing read surface, gated by user JWT + RBAC.
"""
import json
from datetime import datetime, timedelta, timezone
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from shopdb.api import (
db, Asset, Setting, success_response, error_response, paginated_response,
ErrorCodes, get_pagination_params, paginate_query, require_permission,
)
from ..models import EndpointState, EndpointEvent
from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title
endpoints_bp = Blueprint('endpoints', __name__)
def _naive_utcnow():
return datetime.now(timezone.utc).replace(tzinfo=None)
def _format_uptime(seconds):
"""Whole-seconds uptime -> 'Nd Nh Nm', or None."""
if seconds is None:
return None
try:
seconds = int(seconds)
except (TypeError, ValueError):
return None
days, rem = divmod(seconds, 86400)
hours, rem = divmod(rem, 3600)
minutes = rem // 60
parts = []
if days:
parts.append('{}d'.format(days))
if hours:
parts.append('{}h'.format(hours))
parts.append('{}m'.format(minutes))
return ' '.join(parts)
# =============================================================================
# Endpoint state
# =============================================================================
@endpoints_bp.route('/', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def list_endpoints():
"""List reported endpoint states.
Joins through Computer so each row carries computerid and can deep-link
into the existing PC detail page (/pcs/{computerid}). Endpoint state only
exists for hostnames resolved to a Computer, so the join is not lossy.
"""
from plugins.computers.models import Computer
page, per_page = get_pagination_params(request)
query = (db.session.query(EndpointState, Computer.computerid)
.join(Asset, Asset.assetid == EndpointState.assetid)
.join(Computer, Computer.assetid == EndpointState.assetid)
.filter(Asset.isactive.is_(True)))
if search := request.args.get('search'):
query = query.filter(EndpointState.hostname.ilike(f'%{search}%'))
query = query.order_by(EndpointState.lastreportedat.desc())
items, total = paginate_query(query, page, per_page)
data = []
for state, computerid in items:
row = state.to_dict()
row['computerid'] = computerid
row['uptimedisplay'] = _format_uptime(state.uptimeseconds)
data.append(row)
return paginated_response(data, page, per_page, total)
@endpoints_bp.route('/<int:asset_id>', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def get_endpoint(asset_id: int):
"""Get the current state for one endpoint by asset id."""
state = EndpointState.query.filter_by(assetid=asset_id).first()
if not state:
return error_response(
ErrorCodes.NOT_FOUND,
f'No endpoint state for asset {asset_id}',
http_code=404,
)
return success_response(state.to_dict())
@endpoints_bp.route('/<int:asset_id>/events', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def list_endpoint_events(asset_id: int):
"""List curated Windows events for one endpoint, newest first."""
page, per_page = get_pagination_params(request)
query = EndpointEvent.query.filter_by(assetid=asset_id)
if eventid := request.args.get('eventid'):
try:
query = query.filter(EndpointEvent.eventid == int(eventid))
except (TypeError, ValueError):
return error_response(ErrorCodes.VALIDATION_ERROR,
'eventid must be an integer')
query = query.order_by(EndpointEvent.eventtime.desc())
items, total = paginate_query(query, page, per_page)
data = [e.to_dict() for e in items]
return paginated_response(data, page, per_page, total)
# =============================================================================
# Asset panels (ADR-010) - shaped for the generic PluginAssetPanels renderer
# =============================================================================
@endpoints_bp.route('/asset/<int:asset_id>/status', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def panel_status(asset_id: int):
"""Keyvalue panel: the endpoint's current status snapshot.
Single indexed lookup on endpointstates.assetid. Returns the {fields:[...]}
shape the keyvalue renderer expects; an absent state yields no fields so the
panel hides itself.
"""
state = EndpointState.query.filter_by(assetid=asset_id).first()
if not state:
return success_response({'fields': []})
def field(label, value, mono=False):
return {'label': label, 'value': value, 'mono': mono}
fields = [
field('Logged in user', state.loggedinuser),
field('Uptime', _format_uptime(state.uptimeseconds)),
field('Idle', _format_uptime(state.idleseconds)),
field('Last boot',
state.lastboottime.isoformat() + 'Z' if state.lastboottime else None),
field('CPU %', state.cpupercent),
field('Memory %', state.memorypercent),
field('Agent', state.agentversion, mono=True),
field('Last reported',
state.lastreportedat.isoformat() + 'Z' if state.lastreportedat else None),
]
fields = [f for f in fields if f['value'] not in (None, '')]
return success_response({'fields': fields})
@endpoints_bp.route('/asset/<int:asset_id>/services', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def panel_services(asset_id: int):
"""List panel: monitored services/processes with a run-state badge per row.
Reshapes the latest servicesjson snapshot into rows the list renderer maps
to name + status badge. Tolerant of a missing or unexpected snapshot shape:
an unknown state renders as an Unknown badge rather than failing.
"""
state = EndpointState.query.filter_by(assetid=asset_id).first()
if not state or not state.servicesjson:
return success_response([])
try:
services = json.loads(state.servicesjson)
except (TypeError, ValueError):
return success_response([])
rows = []
for entry in (services or []):
if isinstance(entry, dict):
name = entry.get('name') or entry.get('service') or 'Unknown'
raw = (entry.get('state') or entry.get('status') or '').strip().lower()
else:
name, raw = str(entry), ''
running = raw in ('running', 'started', 'up', 'ok', 'active')
rows.append({
'name': name,
'statelabel': 'Running' if running else (raw.title() or 'Unknown'),
'statecolor': 'green' if running else ('red' if raw else 'gray'),
})
return success_response(rows)
@endpoints_bp.route('/asset/<int:asset_id>/recentevents', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def panel_recent_events(asset_id: int):
"""List panel: the endpoint's most recent curated events, newest first.
Indexed range scan on (assetid, eventtime), capped small - this is the
at-a-glance history card, not the full report (that is the events route
with pagination).
"""
events = (EndpointEvent.query
.filter_by(assetid=asset_id)
.order_by(EndpointEvent.eventtime.desc())
.limit(10)
.all())
rows = [{
'title': event_title(e.eventid),
'severity': event_severity(e.eventid),
'eventtime': e.eventtime.isoformat() + 'Z' if e.eventtime else None,
} for e in events]
return success_response(rows)
# =============================================================================
# Dashboard
# =============================================================================
@endpoints_bp.route('/dashboard/summary', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def dashboard_summary():
"""High-level fleet counts for the endpoints dashboard."""
total = db.session.query(EndpointState).join(
Asset, Asset.assetid == EndpointState.assetid
).filter(Asset.isactive.is_(True)).count()
hours = _quiet_hours()
cutoff = _naive_utcnow() - timedelta(hours=hours)
reporting = db.session.query(EndpointState).join(
Asset, Asset.assetid == EndpointState.assetid
).filter(
Asset.isactive.is_(True),
EndpointState.lastreportedat.isnot(None),
EndpointState.lastreportedat >= cutoff,
).count()
return success_response({
'total': total,
'reporting': reporting,
'quiet': total - reporting,
'quiethours': hours,
})
@endpoints_bp.route('/dashboard/quiet', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def dashboard_quiet():
"""Endpoints that reported before, then went quiet past the window.
Never-reported endpoints are excluded: those are records without an agent,
not agents that stopped. Silence is the only signal - there is no heartbeat
separate from the report.
"""
hours = _quiet_hours()
cutoff = _naive_utcnow() - timedelta(hours=hours)
now = _naive_utcnow()
from plugins.computers.models import Computer
rows = (db.session.query(EndpointState, Computer.computerid)
.join(Asset, Asset.assetid == EndpointState.assetid)
.join(Computer, Computer.assetid == EndpointState.assetid)
.filter(Asset.isactive.is_(True),
EndpointState.lastreportedat.isnot(None),
EndpointState.lastreportedat < cutoff)
.all())
out = [{
'computerid': computerid,
'assetid': s.assetid,
'hostname': s.hostname,
'lastreported': s.lastreportedat.isoformat() + 'Z',
'quietdays': (now - s.lastreportedat).days,
} for s, computerid in rows]
out.sort(key=lambda r: -r['quietdays'])
return success_response(out[:50])
@endpoints_bp.route('/dashboard/stability', methods=['GET'])
@jwt_required()
@require_permission('endpoints.view')
def dashboard_stability():
"""Endpoints with recent crash/shutdown/corruption events.
Bounded to a fixed 48h window and the critical event catalog, capped at 50
rows: cost is a range scan on eventtime filtered to a short id list, not a
scan of all history. Joins through Computer so each row links to the PC
detail page.
"""
from plugins.computers.models import Computer
if not CRITICAL_EVENT_IDS:
return success_response([])
cutoff = _naive_utcnow() - timedelta(hours=48)
rows = (db.session.query(
Computer.computerid,
EndpointState.hostname,
db.func.count(EndpointEvent.endpointeventid),
db.func.max(EndpointEvent.eventtime),
)
.select_from(EndpointEvent)
.join(Asset, Asset.assetid == EndpointEvent.assetid)
.join(Computer, Computer.assetid == EndpointEvent.assetid)
.outerjoin(EndpointState,
EndpointState.assetid == EndpointEvent.assetid)
.filter(Asset.isactive.is_(True),
EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS),
EndpointEvent.eventtime.isnot(None),
EndpointEvent.eventtime >= cutoff)
.group_by(Computer.computerid, EndpointState.hostname)
.all())
out = [{
'computerid': computerid,
'hostname': hostname,
'eventcount': count,
'lastevent': lasttime.isoformat() + 'Z' if lasttime else None,
} for computerid, hostname, count, lasttime in rows]
out.sort(key=lambda r: -r['eventcount'])
return success_response(out[:50])
def _quiet_hours() -> int:
"""Quiet-report window in hours, from a site setting, default 24."""
hours = 24
setting = Setting.query.filter_by(key='endpoints_quietreporthours').first()
if setting and (setting.value or '').strip():
try:
hours = int(setting.value)
except (TypeError, ValueError):
pass
return hours

46
plugins/endpoints/eventcatalog.py Normal file
View File

@@ -0,0 +1,46 @@
"""Curated Windows event catalog: id -> human title and severity.
Single server-side source of truth shared by the recent-events asset panel and
the stability dashboard card, so a 4625 reads as "Failed logon (warning)" in
both places without either restating the mapping. The set of IDs here MUST stay
in step with CURATED_EVENT_IDS in plugin.py; a test asserts they match.
Severity vocabulary matches the dashboard card renderer: 'critical', 'warning',
'info'. The IDs marked critical are the ones an operator/tech should act on
without being asked - unexpected shutdown, crash, filesystem corruption, SCM
timeout. Logon activity is informational history unless it is a failure.
"""
# id -> (title, severity, channel)
EVENT_CATALOG = {
4624: ('Successful logon', 'info', 'Security'),
4625: ('Failed logon', 'warning', 'Security'),
4634: ('Logoff', 'info', 'Security'),
4672: ('Special privileges assigned', 'info', 'Security'),
4697: ('Service installed', 'warning', 'Security'),
4698: ('Scheduled task created', 'warning', 'Security'),
41: ('Unexpected shutdown', 'critical', 'System'),
4618: ('Monitored security event', 'warning', 'Security'),
55: ('File system corruption', 'critical', 'System'),
7011: ('Service timeout', 'critical', 'System'),
10016: ('DCOM permission error', 'warning', 'System'),
1001: ('Application crash', 'critical', 'Application'),
}
# The IDs a tech should see surfaced fleet-wide on the dashboard stability card.
CRITICAL_EVENT_IDS = tuple(
eventid for eventid, (_t, severity, _c) in EVENT_CATALOG.items()
if severity == 'critical'
)
def event_title(eventid):
"""Human title for a curated event id, or a generic fallback."""
entry = EVENT_CATALOG.get(eventid)
return entry[0] if entry else 'Event {}'.format(eventid)
def event_severity(eventid):
"""Severity ('critical'|'warning'|'info') for a curated event id."""
entry = EVENT_CATALOG.get(eventid)
return entry[1] if entry else 'info'

20
plugins/endpoints/frontend/routes.js Normal file
View File

@@ -0,0 +1,20 @@
/**
* Endpoints plugin routes
*
* Groundwork for the fleet view. There is no endpoint-specific detail route
* yet: rows link into the existing PC detail page (/pcs/:id) so telemetry
* stays interoperable with the computers plugin instead of duplicating it.
*/
export default [
{
path: 'endpoints',
name: 'endpoints',
component: () => import('./views/EndpointsList.vue'),
meta: { plugin: 'endpoints' }
}
]
// No full-screen routes: every view in this plugin renders inside
// AppLayout. Declared so routes.gen.js can read `.toplevel` off the module
// without Rollup warning about a missing export.
export const toplevel = []

153
plugins/endpoints/frontend/views/EndpointsList.vue Normal file
View File

@@ -0,0 +1,153 @@
<template>
<div>
<div class="page-header">
<h2>Fleet</h2>
</div>
<!-- Filters -->
<div class="filters">
<input
v-model="search"
type="text"
class="form-control"
placeholder="Search endpoints..."
@input="debouncedSearch"
/>
</div>
<div class="card">
<div v-if="loading" class="loading">Loading...</div>
<template v-else>
<div class="table-container">
<table>
<thead>
<tr>
<th>Hostname</th>
<th>Logged-in User</th>
<th>Uptime</th>
<th>CPU</th>
<th>Memory</th>
<th>Last Reported</th>
</tr>
</thead>
<tbody>
<tr
v-for="item in endpoints"
:key="item.assetid"
class="clickable-row"
:class="{ disabled: !item.computerid }"
@click="openPc(item)"
>
<td>{{ item.hostname || '-' }}</td>
<td>{{ item.loggedinuser || '-' }}</td>
<td>{{ item.uptimedisplay || '-' }}</td>
<td>{{ formatPercent(item.cpupercent) }}</td>
<td>{{ formatPercent(item.memorypercent) }}</td>
<td>{{ formatTime(item.lastreportedat) }}</td>
</tr>
<tr v-if="endpoints.length === 0">
<td colspan="6" style="text-align: center; color: var(--text-light);">
No endpoints found
</td>
</tr>
</tbody>
</table>
</div>
<!-- Pagination -->
<PaginationBar
:page="page"
:totalPages="totalPages"
:perPage="perPage"
@update:page="goToPage"
@update:perPage="changePerPage"
/>
</template>
</div>
</div>
</template>
<script setup>
import { ref, onMounted } from 'vue'
import { useRouter } from 'vue-router'
import { endpointsApi } from '@/api'
import PaginationBar from '@/components/PaginationBar.vue'
import { useListQuery } from '@/composables/listQuery'
const router = useRouter()
const endpoints = ref([])
const loading = ref(true)
const { page, search, setPage, setSearch } = useListQuery({ onChange: loadEndpoints })
const totalPages = ref(1)
const perPage = ref(20)
let searchTimeout = null
onMounted(() => {
loadEndpoints()
})
async function loadEndpoints() {
loading.value = true
try {
const params = {
page: page.value,
perpage: perPage.value
}
if (search.value) params.search = search.value
const response = await endpointsApi.list(params)
endpoints.value = response.data.data || []
totalPages.value = response.data.meta?.pagination?.totalpages || response.data.meta?.pagination?.total_pages || 1
} catch (error) {
console.error('Error loading endpoints:', error)
} finally {
loading.value = false
}
}
// Interoperate with PC Details instead of a bespoke detail page: a row only
// navigates when it resolved to a computerid.
function openPc(item) {
if (item.computerid) {
router.push(`/pcs/${item.computerid}`)
}
}
function formatPercent(value) {
if (value === null || value === undefined) return '-'
return `${Math.round(value)}%`
}
function formatTime(value) {
if (!value) return '-'
return new Date(value).toLocaleString()
}
function debouncedSearch() {
clearTimeout(searchTimeout)
searchTimeout = setTimeout(() => {
setSearch(search.value)
loadEndpoints()
}, 300)
}
function goToPage(p) {
setPage(p)
loadEndpoints()
}
function changePerPage(newPerPage) {
perPage.value = newPerPage
setPage(1)
loadEndpoints()
}
</script>
<style scoped>
.clickable-row.disabled {
cursor: default;
opacity: 0.6;
}
</style>

21
plugins/endpoints/manifest.json Normal file
View File

@@ -0,0 +1,21 @@
{
"name": "endpoints",
"version": "0.1.0",
"description": "Endpoint RMM telemetry: metrics, services, inventory, and curated Windows event records reported by the ShopDB agent",
"author": "ShopDB Team",
"dependencies": ["computers"],
"core_version": ">=0.20.0,<1.0.0",
"api_prefix": "/api/endpoints",
"provides": {
"features": [
"endpoint_metrics",
"service_state",
"inventory_snapshot",
"event_records"
]
},
"settings": {
"report_interval_minutes": 15,
"quiet_report_hours": 24
}
}

15
plugins/endpoints/migrations/env.py Normal file
View File

@@ -0,0 +1,15 @@
"""Alembic environment for the endpoints plugin migration chain.
Delegates to the shared runner in shopdb.plugins.alembic_template, which filters
the metadata to this plugin's tables and drives Alembic against the per-plugin
version table alembic_version_endpoints. See ADR-008 for the ownership model.
This plugin is NEW: its 0001 baseline really CREATES its tables, because the
core chain never built them.
"""
import os
os.environ['PLUGIN_NAME'] = 'endpoints'
from shopdb.plugins.alembic_template import run_migrations # noqa: E402
run_migrations()

24
plugins/endpoints/migrations/script.py.mako Normal file
View File

@@ -0,0 +1,24 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade():
${upgrades if upgrades else "pass"}
def downgrade():
${downgrades if downgrades else "pass"}

93
plugins/endpoints/migrations/versions/0001_endpoints_baseline.py Normal file
View File

@@ -0,0 +1,93 @@
"""endpoints plugin baseline (real create).
A NEW plugin: the core chain never knew about endpointstates or endpointevents,
so this per-plugin chain is their sole authoritative creator.
Emits explicit Alembic ops rather than using the create_plugin_tables helper:
the helper builds a per-plugin MetaData filtered to the plugin's own tables, so
the foreign key to the core assets table cannot resolve at CreateTable-compile
time (NoReferencedTableError). Tables inherit the connection's default charset,
matching how the core chain creates its own.
"""
from alembic import op
from shopdb.plugins.alembic_template import (
create_index_if_missing,
create_table_if_missing,
)
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'endpoints0001baseline'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
create_table_if_missing(
'endpointstates',
sa.Column('endpointstateid', sa.Integer(), nullable=False),
sa.Column('assetid', sa.Integer(), nullable=False),
sa.Column('hostname', sa.String(length=100), nullable=True),
sa.Column('loggedinuser', sa.String(length=100), nullable=True),
sa.Column('agentversion', sa.String(length=40), nullable=True),
sa.Column('lastboottime', sa.DateTime(), nullable=True),
sa.Column('uptimeseconds', sa.BigInteger(), nullable=True),
sa.Column('idleseconds', sa.BigInteger(), nullable=True),
sa.Column('cpupercent', sa.Float(), nullable=True),
sa.Column('memorypercent', sa.Float(), nullable=True),
sa.Column('servicesjson', sa.Text(length=16777215), nullable=True),
sa.Column('portsjson', sa.Text(length=16777215), nullable=True),
sa.Column('interfacesjson', sa.Text(length=16777215), nullable=True),
sa.Column('scheduledtasksjson', sa.Text(length=16777215),
nullable=True),
sa.Column('lastreportedat', sa.DateTime(), nullable=True),
sa.Column('createdat', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['assetid'], ['assets.assetid'],
ondelete='CASCADE'),
sa.PrimaryKeyConstraint('endpointstateid'),
sa.UniqueConstraint('assetid'),
)
create_index_if_missing('idx_endpointstates_lastreportedat',
'endpointstates', ['lastreportedat'])
create_table_if_missing(
'endpointevents',
sa.Column('endpointeventid', sa.Integer(), nullable=False),
sa.Column('assetid', sa.Integer(), nullable=False),
sa.Column('eventid', sa.Integer(), nullable=False),
sa.Column('eventchannel', sa.String(length=50), nullable=True),
sa.Column('eventsource', sa.String(length=255), nullable=True),
sa.Column('eventtime', sa.DateTime(), nullable=True),
sa.Column('eventmessage', sa.Text(), nullable=True),
sa.Column('payloadjson', sa.Text(length=16777215), nullable=True),
sa.Column('recordedat', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['assetid'], ['assets.assetid'],
ondelete='CASCADE'),
sa.PrimaryKeyConstraint('endpointeventid'),
)
create_index_if_missing('ixendpointeventsassetid', 'endpointevents',
['assetid'])
create_index_if_missing('ixendpointeventseventid', 'endpointevents',
['eventid'])
create_index_if_missing('ixendpointeventseventtime', 'endpointevents',
['eventtime'])
# Alerting/history reads by "this endpoint's events over time", so the
# composite is the index that actually gets used.
create_index_if_missing('idx_endpointevents_assetevent', 'endpointevents',
['assetid', 'eventtime'])
def downgrade():
op.drop_index('idx_endpointevents_assetevent',
table_name='endpointevents')
op.drop_index('ixendpointeventseventtime', table_name='endpointevents')
op.drop_index('ixendpointeventseventid', table_name='endpointevents')
op.drop_index('ixendpointeventsassetid', table_name='endpointevents')
op.drop_table('endpointevents')
op.drop_index('idx_endpointstates_lastreportedat',
table_name='endpointstates')
op.drop_table('endpointstates')

5
plugins/endpoints/models/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Endpoints plugin models."""
from .endpoint import EndpointState, EndpointEvent
__all__ = ['EndpointState', 'EndpointEvent']

163
plugins/endpoints/models/endpoint.py Normal file
View File

@@ -0,0 +1,163 @@
"""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)
# 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)
__table_args__ = (
db.Index('idx_endpointstates_lastreportedat', 'lastreportedat'),
)
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,
'lastreportedat': _utciso(self.lastreportedat),
'createdat': _utciso(self.createdat),
}
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),
}

376
plugins/endpoints/plugin.py Normal file
View File

@@ -0,0 +1,376 @@
"""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
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)
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)
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]
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'),
]
def get_navigation_items(self) -> List[Dict]:
"""Sidebar entry for the fleet list view.
Sits just after the PCs item (position 15) because the fleet view is
a telemetry lens over the same computers.
"""
return [
{
'name': 'Fleet',
'icon': 'activity',
'route': '/endpoints',
'position': 16,
},
]
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.
Three panels feeding the generic renderer: the current status snapshot
(keyvalue), monitored services with a run-state badge (list), and the
most recent curated events (list). 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-services',
'title': 'Services and processes',
'assettypes': ['computer'],
'endpoint': '/api/endpoints/asset/{assetid}/services',
'render': 'list',
'map': {
'title': 'name',
'badge': {'label': 'statelabel', 'color': 'statecolor'},
},
'empty': 'No monitored services reported.',
'position': 12,
},
{
'id': 'endpoints-events',
'title': 'Recent events',
'assettypes': ['computer'],
'endpoint': '/api/endpoints/asset/{assetid}/recentevents',
'render': 'list',
'map': {
'title': 'title',
'badge': {
'label': 'severity',
'labelmap': {
'critical': 'Critical',
'warning': 'Warning',
'info': 'Info',
},
},
'meta': [
{'key': 'eventtime', 'label': 'At', 'format': 'date'},
],
},
'empty': 'No recent events.',
'position': 14,
},
]
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',
'endpoint': '/api/endpoints/dashboard/quiet',
'render': 'exceptions',
'severity': 'warning',
'permission': 'endpoints.view',
'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',
'endpoint': '/api/endpoints/dashboard/stability',
'render': 'exceptions',
'severity': 'critical',
'permission': 'endpoints.view',
'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).',
},
'loggedinuser': {'type': 'string'},
'agentversion': {'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.',
},
'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'},
},
},
},
},
},
}
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:
raise ValueError(
'could not resolve hostname to a computer: {!r}'.format(hostname))
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 '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'))
state.lastreportedat = datetime.utcnow()
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,
}

92
plugins/endpoints/tests/test_plugin.py Normal file
View File

@@ -0,0 +1,92 @@
"""Endpoints plugin smoke + collector-contract tests.
Asserts the plugin loads cleanly, satisfies the framework contract, and that
its ADR-006 collector schema is well formed. Extend with DB-backed ingest tests
as the plugin grows.
"""
from flask import Blueprint
from plugins.endpoints.plugin import EndpointsPlugin, CURATED_EVENT_IDS
from plugins.endpoints.eventcatalog import EVENT_CATALOG, CRITICAL_EVENT_IDS
def test_endpoints_plugin_meta_is_valid():
"""meta returns a PluginMeta with the expected name and prefix."""
plugin = EndpointsPlugin()
assert plugin.meta.name == 'endpoints'
assert plugin.meta.api_prefix == '/api/endpoints'
assert 'computers' in plugin.meta.dependencies
def test_endpoints_plugin_get_blueprint_returns_blueprint():
"""get_blueprint returns a Flask Blueprint, not None."""
plugin = EndpointsPlugin()
assert isinstance(plugin.get_blueprint(), Blueprint)
def test_endpoints_plugin_get_models_returns_models():
"""get_models returns the two telemetry tables."""
plugin = EndpointsPlugin()
models = plugin.get_models()
assert len(models) >= 2
for model in models:
assert hasattr(model, '__tablename__')
def test_endpoints_collector_schema_identity_is_hostname():
"""ADR-006 schema declares hostname as the identity field."""
plugin = EndpointsPlugin()
schema = plugin.get_collector_schema()
assert schema is not None
assert schema['identityfield'] == 'hostname'
assert 'hostname' in schema['fields']['required']
def test_endpoints_curated_event_ids_include_core_security_ids():
"""The curated catalog covers the agreed logon/stability IDs."""
for eventid in (4624, 4625, 4634, 4672, 4697, 4698, 41, 4618, 55, 7011,
10016, 1001):
assert eventid in CURATED_EVENT_IDS
def test_curated_ids_match_the_event_catalog():
"""Ingest filter and display catalog cannot drift apart."""
assert CURATED_EVENT_IDS == frozenset(EVENT_CATALOG)
def test_critical_event_ids_are_a_subset_of_the_catalog():
"""Dashboard stability card only queries known critical IDs."""
assert set(CRITICAL_EVENT_IDS)
assert set(CRITICAL_EVENT_IDS).issubset(set(EVENT_CATALOG))
def test_endpoints_asset_panels_are_computer_scoped():
"""Every asset panel targets computer assets only, with an endpoint."""
plugin = EndpointsPlugin()
panels = plugin.get_asset_panels()
assert len(panels) == 3
for panel in panels:
assert panel['assettypes'] == ['computer']
assert panel['endpoint'].startswith('/api/endpoints/asset/{assetid}')
assert panel['render'] in ('keyvalue', 'list')
def test_endpoints_dashboard_widgets_declared():
"""Quiet + stability exception cards are declared and hide when empty."""
plugin = EndpointsPlugin()
widgets = plugin.get_dashboard_widgets()
ids = {w['id'] for w in widgets}
assert {'endpoints-quiet', 'endpoints-stability'} <= ids
for widget in widgets:
assert widget['render'] == 'exceptions'
assert widget['empty'] == 'hide'
assert widget['permission'] == 'endpoints.view'
def test_endpoints_navigation_item_points_at_fleet_view():
"""The fleet list view is exposed as a single /endpoints nav item."""
plugin = EndpointsPlugin()
items = plugin.get_navigation_items()
routes = {item['route'] for item in items}
assert '/endpoints' in routes