9725c47814
The endpoint agent already reports an installed-software list (SBOM), but ingress dropped it: the collector schema had no entry for it and there was nowhere to store it, so operators could not see what is installed on a PC. Persist it as a per-asset snapshot in a new endpointapplications table, replaced on every report (current inventory is what operators ask for, not install history). Add applications to the collector schema and KNOWN_PAYLOAD_KEYS so ingress stops discarding it, a read route for the raw collection, and a collapsed-by-default Installed software panel to surface it. appid is nullable groundwork for a future Managed Applications link; no matcher is wired yet.
1078 lines
44 KiB
Python
1078 lines
44 KiB
Python
"""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
|
|
import time
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from flask import Blueprint, request, current_app
|
|
from flask_jwt_extended import current_user
|
|
|
|
from shopdb.api import (
|
|
db, Asset, Setting, success_response, error_response, paginated_response,
|
|
ErrorCodes, get_pagination_params, paginate_query, require_permission,
|
|
require_role, caller_holds, mint_service_token, list_service_tokens,
|
|
revoke_service_tokens,
|
|
AuditLog, collector_authorized, User,
|
|
)
|
|
|
|
from ..models import EndpointState, EndpointEvent, EndpointCommand, EndpointApplication
|
|
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for
|
|
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)
|
|
|
|
|
|
# A watched service or process reports a STATE STRING, not a boolean, and the
|
|
# vocabulary varies by source. Both the services panel and the fleet health
|
|
# rollup have to agree on what counts as running, so the rule lives here once -
|
|
# reading it wrong makes every service look stopped.
|
|
_RUNNING_STATES = ('running', 'started', 'up', 'ok', 'active')
|
|
|
|
|
|
def _service_running(entry):
|
|
raw = (entry.get('state') or entry.get('status') or '').strip().lower()
|
|
return raw in _RUNNING_STATES
|
|
|
|
|
|
def _service_name(entry):
|
|
return entry.get('name') or entry.get('service') or 'Unknown'
|
|
|
|
|
|
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'])
|
|
@require_permission('endpoints.fleet')
|
|
def list_endpoints():
|
|
"""List reported endpoint states.
|
|
|
|
Registered as '' so the rule is '/api/endpoints', with no trailing slash.
|
|
CONTRIBUTING.md's API section spells endpoints that way ('/api/machines'),
|
|
and every other plugin's list route follows it - this was the only route in
|
|
331 paths that did not.
|
|
|
|
It was not cosmetic. A rule ending in a slash makes Flask answer the
|
|
unslashed URL with a 308 to an ABSOLUTE url. On dev that resolves and
|
|
nobody notices. Behind the IIS subpath it does not: with no ProxyFix,
|
|
Werkzeug builds that Location from the scheme and host it believes it has,
|
|
so it emitted http:// and dropped the /shopdb prefix. The browser refuses
|
|
to follow an https page to http, the list request failed, and the page
|
|
rendered "No endpoints found" - indistinguishable from a fleet with nothing
|
|
in it. The agent had been reporting correctly the whole time.
|
|
|
|
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.
|
|
|
|
Sorted WORST FIRST by default: never-reported, then quiet, then reporting.
|
|
It used to sort by lastreportedat descending, which floated the healthiest
|
|
endpoints to the top and buried the ones somebody needs to look at.
|
|
|
|
Each row also carries what the fleet view is actually for: which MACHINE
|
|
depends on this PC, its PC type, and a health rollup. On a shop floor a PC
|
|
matters because a machine stops without it, so that is the column an
|
|
operator reads first.
|
|
"""
|
|
from plugins.computers.models import Computer, ComputerType
|
|
|
|
page, per_page = get_pagination_params(request)
|
|
cutoff = _naive_utcnow() - timedelta(hours=_quiet_hours())
|
|
|
|
# Status as a sortable expression rather than a post-hoc label, so
|
|
# worst-first survives pagination. 0 never reported, 1 quiet, 2 reporting.
|
|
statusrank = db.case(
|
|
(EndpointState.lastreportedat.is_(None), 0),
|
|
(EndpointState.lastreportedat < cutoff, 1),
|
|
else_=2,
|
|
)
|
|
|
|
query = (db.session.query(EndpointState, Computer.computerid,
|
|
Computer.computertypeid, ComputerType.computertype,
|
|
Asset.assetnumber, statusrank.label('statusrank'))
|
|
.join(Asset, Asset.assetid == EndpointState.assetid)
|
|
.join(Computer, Computer.assetid == EndpointState.assetid)
|
|
.outerjoin(ComputerType,
|
|
ComputerType.computertypeid == Computer.computertypeid)
|
|
.filter(Asset.isactive.is_(True)))
|
|
|
|
# Search covers hostname, asset number and the machine number a PC serves.
|
|
# Hostname alone was too narrow: people look a PC up by the bay it runs.
|
|
if search := (request.args.get('search') or '').strip():
|
|
like = f'%{search}%'
|
|
machinehits = _assetids_serving_machine_like(like)
|
|
clauses = [EndpointState.hostname.ilike(like), Asset.assetnumber.ilike(like)]
|
|
if machinehits:
|
|
clauses.append(EndpointState.assetid.in_(machinehits))
|
|
query = query.filter(db.or_(*clauses))
|
|
|
|
if status := request.args.get('status'):
|
|
wanted = {'never': 0, 'quiet': 1, 'reporting': 2}.get(status.lower())
|
|
if wanted is not None:
|
|
query = query.filter(statusrank == wanted)
|
|
|
|
if computertypeid := request.args.get('computertypeid'):
|
|
query = query.filter(Computer.computertypeid == computertypeid)
|
|
if locationid := request.args.get('locationid'):
|
|
query = query.filter(Asset.locationid == locationid)
|
|
if businessunitid := request.args.get('businessunitid'):
|
|
query = query.filter(Asset.businessunitid == businessunitid)
|
|
|
|
sort = (request.args.get('sort') or 'status').lower()
|
|
sorts = {
|
|
'status': (statusrank.asc(), EndpointState.hostname.asc()),
|
|
'hostname': (EndpointState.hostname.asc(),),
|
|
'lastreported': (EndpointState.lastreportedat.desc(),),
|
|
'uptime': (EndpointState.uptimeseconds.desc(),),
|
|
}
|
|
query = query.order_by(*sorts.get(sort, sorts['status']))
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
|
|
# Batch the per-row extras for this PAGE only. Doing them inline would be
|
|
# one query per endpoint for machines and another for events.
|
|
assetids = [state.assetid for state, *_ in items]
|
|
machines = _machines_for_assets(assetids)
|
|
critical = _critical_counts_for_assets(assetids)
|
|
|
|
data = []
|
|
for state, computerid, computertypeid, computertype, assetnumber, rank in items:
|
|
row = state.to_dict()
|
|
row['computerid'] = computerid
|
|
row['assetnumber'] = assetnumber
|
|
row['computertypeid'] = computertypeid
|
|
row['computertype'] = computertype
|
|
row['uptimedisplay'] = _format_uptime(state.uptimeseconds)
|
|
row['status'] = ('never', 'quiet', 'reporting')[rank]
|
|
row['machines'] = machines.get(state.assetid, [])
|
|
row['health'] = _health_for(state, critical.get(state.assetid, 0))
|
|
data.append(row)
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
def _assetids_serving_machine_like(like):
|
|
"""Asset ids of PCs whose controlled machine matches a search term.
|
|
|
|
Checked in BOTH directions on purpose. The canonical edge is PC controls
|
|
machine, but production also holds reciprocal rows, so matching only
|
|
source->target silently drops half the fleet from a machine-number search.
|
|
"""
|
|
from shopdb.api import AssetRelationship, RelationshipType, AssetType
|
|
|
|
controls = (RelationshipType.query
|
|
.filter(db.func.lower(RelationshipType.relationshiptype) == 'controls')
|
|
.first())
|
|
if controls is None:
|
|
return []
|
|
|
|
# Only MACHINE assets. A PC also 'controls' printers and other kit, and
|
|
# matching any controlled asset put office locations in the machine column.
|
|
other = db.aliased(Asset)
|
|
othertype = db.aliased(AssetType)
|
|
forward = (db.session.query(AssetRelationship.sourceassetid)
|
|
.join(other, other.assetid == AssetRelationship.targetassetid)
|
|
.join(othertype, othertype.assettypeid == other.assettypeid)
|
|
.filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
|
|
db.func.lower(othertype.assettype) == 'machine',
|
|
other.assetnumber.ilike(like)))
|
|
backward = (db.session.query(AssetRelationship.targetassetid)
|
|
.join(other, other.assetid == AssetRelationship.sourceassetid)
|
|
.join(othertype, othertype.assettypeid == other.assettypeid)
|
|
.filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
|
|
db.func.lower(othertype.assettype) == 'machine',
|
|
other.assetnumber.ilike(like)))
|
|
return [r[0] for r in forward.union(backward).all()]
|
|
|
|
|
|
def _machines_for_assets(assetids):
|
|
"""{assetid: [{assetid, assetnumber, name}]} for the machines each PC serves.
|
|
|
|
Both edge directions again, deduplicated by machine asset id. One query for
|
|
the whole page rather than one per row.
|
|
"""
|
|
from shopdb.api import AssetRelationship, RelationshipType, AssetType
|
|
|
|
if not assetids:
|
|
return {}
|
|
|
|
controls = (RelationshipType.query
|
|
.filter(db.func.lower(RelationshipType.relationshiptype) == 'controls')
|
|
.first())
|
|
if controls is None:
|
|
return {}
|
|
|
|
other = db.aliased(Asset)
|
|
othertype = db.aliased(AssetType)
|
|
forward = (db.session.query(AssetRelationship.sourceassetid.label('pc'),
|
|
other.assetid, other.assetnumber, other.name)
|
|
.join(other, other.assetid == AssetRelationship.targetassetid)
|
|
.join(othertype, othertype.assettypeid == other.assettypeid)
|
|
.filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
|
|
db.func.lower(othertype.assettype) == 'machine',
|
|
AssetRelationship.sourceassetid.in_(assetids)))
|
|
backward = (db.session.query(AssetRelationship.targetassetid.label('pc'),
|
|
other.assetid, other.assetnumber, other.name)
|
|
.join(other, other.assetid == AssetRelationship.sourceassetid)
|
|
.join(othertype, othertype.assettypeid == other.assettypeid)
|
|
.filter(AssetRelationship.relationshiptypeid == controls.relationshiptypeid,
|
|
db.func.lower(othertype.assettype) == 'machine',
|
|
AssetRelationship.targetassetid.in_(assetids)))
|
|
|
|
out = {}
|
|
for pc, machineid, assetnumber, name in forward.union(backward).all():
|
|
seen = out.setdefault(pc, [])
|
|
if not any(m['assetid'] == machineid for m in seen):
|
|
seen.append({'assetid': machineid, 'assetnumber': assetnumber, 'name': name})
|
|
return out
|
|
|
|
|
|
def _critical_counts_for_assets(assetids):
|
|
"""{assetid: count} of critical events in the stability window."""
|
|
if not assetids:
|
|
return {}
|
|
cutoff = _naive_utcnow() - timedelta(hours=48)
|
|
rows = (db.session.query(EndpointEvent.assetid,
|
|
db.func.count(EndpointEvent.endpointeventid))
|
|
.filter(EndpointEvent.assetid.in_(assetids),
|
|
EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS),
|
|
EndpointEvent.eventtime.isnot(None),
|
|
EndpointEvent.eventtime >= cutoff)
|
|
.group_by(EndpointEvent.assetid)
|
|
.all())
|
|
return dict(rows)
|
|
|
|
|
|
def _health_for(state, criticalcount):
|
|
"""Worst-of rollup: a stopped watched service, or a recent critical event.
|
|
|
|
Deliberately a PHRASE, not a score. "eDNC stopped" tells an operator what
|
|
to do; a percentage does not.
|
|
"""
|
|
if criticalcount:
|
|
return {'level': 'critical',
|
|
'label': f'{criticalcount} critical' if criticalcount > 1 else '1 critical'}
|
|
|
|
stopped = []
|
|
try:
|
|
for entry in json.loads(state.servicesjson or '[]'):
|
|
if not _service_running(entry):
|
|
stopped.append(_service_name(entry))
|
|
except (TypeError, ValueError):
|
|
# Malformed payload is not a health verdict - say nothing rather than
|
|
# claiming the endpoint is unhealthy.
|
|
return {'level': 'unknown', 'label': None}
|
|
|
|
if stopped:
|
|
first = stopped[0]
|
|
extra = f' +{len(stopped) - 1}' if len(stopped) > 1 else ''
|
|
return {'level': 'warning', 'label': f'{first} stopped{extra}'}
|
|
|
|
if state.lastreportedat is None:
|
|
return {'level': 'unknown', 'label': 'no agent'}
|
|
return {'level': 'ok', 'label': 'ok'}
|
|
|
|
|
|
@endpoints_bp.route('/<int:asset_id>', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def get_endpoint(asset_id: int):
|
|
"""Get the current state for one endpoint by asset id.
|
|
|
|
Carries the same reporting/quiet/never verdict the fleet list uses. The
|
|
quiet window is a site setting, so the SERVER decides it: a caller that
|
|
guessed a threshold in the browser would disagree with the fleet list on
|
|
the same endpoint the moment a site changed the setting.
|
|
"""
|
|
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,
|
|
)
|
|
|
|
cutoff = _naive_utcnow() - timedelta(hours=_quiet_hours())
|
|
if state.lastreportedat is None:
|
|
status = 'never'
|
|
elif state.lastreportedat < cutoff:
|
|
status = 'quiet'
|
|
else:
|
|
status = 'reporting'
|
|
|
|
data = state.to_dict()
|
|
data['status'] = status
|
|
data['uptimedisplay'] = _format_uptime(state.uptimeseconds)
|
|
return success_response(data)
|
|
|
|
|
|
@endpoints_bp.route('/<int:asset_id>/events', methods=['GET'])
|
|
@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'])
|
|
@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>/telemetrymeta', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_telemetry_meta(asset_id: int):
|
|
"""Lightweight freshness probe for the panels.
|
|
|
|
Returns just the endpoint's lastreportedat so the WebUI can tell whether a
|
|
genuinely new telemetry snapshot has landed after it forces a collect. The
|
|
UI captures this before enqueuing collect_now (or a row Start/Stop), then
|
|
polls until the stamp advances - a command reaching 'succeeded' only means
|
|
the agent ran it, not that it has yet posted the fresh services snapshot, so
|
|
gating the refresh on this timestamp is what avoids re-rendering stale rows.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
stamp = (state.lastreportedat.isoformat() + 'Z'
|
|
if state and state.lastreportedat else None)
|
|
return success_response({'lastreportedat': stamp})
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/services', methods=['GET'])
|
|
@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([])
|
|
|
|
maycommand = caller_holds('endpoints.command')
|
|
|
|
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()
|
|
app = entry.get('app')
|
|
kind = entry.get('kind')
|
|
else:
|
|
name, raw, app, kind = str(entry), '', None, None
|
|
running = _service_running(entry)
|
|
row = {
|
|
'name': name,
|
|
'kind': (kind or '').lower() or None,
|
|
'statelabel': 'Running' if running else (raw.title() or 'Unknown'),
|
|
'statecolor': 'green' if running else ('red' if raw else 'gray'),
|
|
}
|
|
# Only offer Start/Stop to a caller who could actually run it. The
|
|
# emission never consulted the caller, so a read-only viewer was shown
|
|
# controls on a shop-floor PC that 403 on click - wrong signal, and it
|
|
# gets worse the moment guests can read this panel.
|
|
targets = row_actions_for(app, kind, name) if maycommand else None
|
|
if targets:
|
|
row['action'] = {
|
|
'running': running,
|
|
'start': targets['start'],
|
|
'stop': targets['stop'],
|
|
'status': targets['status'],
|
|
'target': targets['target'],
|
|
'targetkind': targets['targetkind'],
|
|
}
|
|
rows.append(row)
|
|
return success_response(rows)
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/recentevents', methods=['GET'])
|
|
@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)
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/activity', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_activity(asset_id: int):
|
|
"""List panel: one chronological feed of what happened to this endpoint.
|
|
|
|
Curated Windows events and operator commands merged and sorted newest
|
|
first. They answer the same question - what happened to this box - and
|
|
reading them as two separate cards means interleaving them by eye.
|
|
|
|
Both sources are already keyed by assetid, so this is two indexed reads and
|
|
a merge sort, not a join. The core AuditLog is deliberately NOT a third
|
|
source: its rows carry the action as the entity name and the asset only
|
|
inside a details string, so pulling it in would mean parsing prose.
|
|
|
|
The requester's name is withheld from callers who cannot themselves issue
|
|
commands. A read-only viewer - a guest, on a site that grants this panel -
|
|
has no need for the roster of who restarted what, and the feed is just as
|
|
useful without it.
|
|
"""
|
|
naming = caller_holds('endpoints.command')
|
|
|
|
events = (EndpointEvent.query
|
|
.filter_by(assetid=asset_id)
|
|
.order_by(EndpointEvent.eventtime.desc())
|
|
.limit(15)
|
|
.all())
|
|
|
|
commands = (EndpointCommand.query
|
|
.filter_by(assetid=asset_id)
|
|
.order_by(EndpointCommand.createdat.desc())
|
|
.limit(15)
|
|
.all())
|
|
|
|
# requestedby holds a user ID, not a name. One batched lookup rather than a
|
|
# query per row, and only when the caller is allowed to see the names at all.
|
|
names = {}
|
|
if naming:
|
|
userids = {c.requestedby for c in commands if c.requestedby}
|
|
if userids:
|
|
names = {u.userid: (u.username or '')
|
|
for u in User.query.filter(User.userid.in_(userids)).all()}
|
|
|
|
rows = []
|
|
for event in events:
|
|
rows.append({
|
|
'title': event_title(event.eventid),
|
|
'severity': event_severity(event.eventid),
|
|
'detail': 'Windows event {}'.format(event.eventid),
|
|
'eventtime': event.eventtime,
|
|
})
|
|
|
|
for command in commands:
|
|
label = COMMAND_ALLOWLIST.get(command.action, {}).get('label') or command.action
|
|
target = ' on {}'.format(command.target) if command.target else ''
|
|
who = ''
|
|
if naming and command.requestedby:
|
|
username = names.get(command.requestedby)
|
|
if username:
|
|
who = ' by {}'.format(username)
|
|
rows.append({
|
|
'title': '{}{}'.format(label, target),
|
|
# A failed command is the one an operator needs to notice.
|
|
'severity': 'critical' if command.status == EndpointCommand.STATUS_FAILED else 'info',
|
|
'detail': '{}{}'.format((command.status or 'queued').title(), who),
|
|
'eventtime': command.createdat,
|
|
})
|
|
|
|
# Undated rows sort last rather than crashing the comparison.
|
|
rows.sort(key=lambda r: (r['eventtime'] is not None, r['eventtime']), reverse=True)
|
|
|
|
return success_response([{
|
|
'title': r['title'],
|
|
'severity': r['severity'],
|
|
'detail': r['detail'],
|
|
'eventtime': r['eventtime'].isoformat() + 'Z' if r['eventtime'] else None,
|
|
} for r in rows[:15]])
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/interfaces', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_interfaces(asset_id: int):
|
|
"""Table panel: every hardware NIC the agent last reported.
|
|
|
|
Reshapes the latest interfacesjson snapshot into one row per adapter with
|
|
the {alias, ipv4, mac, gateway, subnet, dhcp} shape the collector emits. A
|
|
gatewayless adapter is flagged as the likely machine-IPS link (see
|
|
DESIGN-DECISIONS-agent.md 9.3): those NICs speak to the machine, not the
|
|
corporate LAN. Tolerant of a missing or unexpected snapshot: an unreadable
|
|
payload yields no rows so the panel hides itself.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
if not state or not state.interfacesjson:
|
|
return success_response([])
|
|
|
|
try:
|
|
interfaces = json.loads(state.interfacesjson)
|
|
except (TypeError, ValueError):
|
|
return success_response([])
|
|
|
|
rows = []
|
|
for entry in (interfaces or []):
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
gateway = (entry.get('gateway') or '').strip()
|
|
ipv4 = (entry.get('ipv4') or '').strip()
|
|
rows.append({
|
|
'alias': entry.get('alias') or 'Unknown',
|
|
'ipv4': ipv4 or '-',
|
|
'mac': (entry.get('mac') or '').strip() or '-',
|
|
'gateway': gateway or '-',
|
|
'subnet': (entry.get('subnet') or '').strip() or '-',
|
|
'dhcp': 'DHCP' if entry.get('dhcp') else 'Static',
|
|
# A gatewayless NIC with an address is the machine link; the gateway
|
|
# NIC is the corporate LAN.
|
|
'role': 'Machine link' if (ipv4 and not gateway) else 'LAN',
|
|
})
|
|
return success_response(rows)
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/applications', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_applications(asset_id: int):
|
|
"""Table panel: the installed-software (SBOM) snapshot for this endpoint.
|
|
|
|
Reads the endpointapplications rows the collector last wrote for this asset
|
|
(a per-report snapshot, not history) and returns them sorted by name. This
|
|
is the raw discovered inventory: appid is carried through but not yet linked
|
|
to the applications catalog ("Managed Applications"), so no managed status
|
|
is surfaced here. An endpoint with no reported SBOM yields no rows and the
|
|
panel hides itself.
|
|
"""
|
|
rows = (EndpointApplication.query
|
|
.filter_by(assetid=asset_id)
|
|
.order_by(db.func.lower(EndpointApplication.displayname))
|
|
.all())
|
|
return success_response([
|
|
{
|
|
'displayname': row.displayname,
|
|
'displayversion': row.displayversion or '-',
|
|
'publisher': row.publisher or '-',
|
|
}
|
|
for row in rows
|
|
])
|
|
|
|
|
|
# =============================================================================
|
|
# Dashboard
|
|
# =============================================================================
|
|
|
|
@endpoints_bp.route('/dashboard/summary', methods=['GET'])
|
|
@require_permission('endpoints.fleet')
|
|
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()
|
|
|
|
# Counted separately from quiet. An endpoint that has NEVER reported is a
|
|
# different problem from one that reported and stopped: the first is an
|
|
# enrolment or key failure, the second is a machine that went down. Folding
|
|
# them together (quiet = total - reporting) hid the enrolment failures.
|
|
never = db.session.query(EndpointState).join(
|
|
Asset, Asset.assetid == EndpointState.assetid
|
|
).filter(
|
|
Asset.isactive.is_(True),
|
|
EndpointState.lastreportedat.is_(None),
|
|
).count()
|
|
|
|
return success_response({
|
|
'total': total,
|
|
'reporting': reporting,
|
|
'quiet': total - reporting - never,
|
|
'never': never,
|
|
'quiethours': hours,
|
|
})
|
|
|
|
|
|
@endpoints_bp.route('/dashboard/quiet', methods=['GET'])
|
|
@require_permission('endpoints.fleet')
|
|
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'])
|
|
@require_permission('endpoints.fleet')
|
|
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
|
|
|
|
|
|
# =============================================================================
|
|
# Endpoint agent key (UI-managed collector.ingest.endpoints kill switch)
|
|
# =============================================================================
|
|
|
|
# The single permission the endpoint agent key authorizes. A managed token
|
|
# scoped to ONLY this authorizes the endpoints collector and nothing else, so
|
|
# rotating/revoking it stops endpoint-agent traffic without touching the
|
|
# computers/geenforce collectors.
|
|
AGENT_KEY_SCOPE = 'collector.ingest.endpoints'
|
|
AGENT_KEY_NAME = 'Endpoint agent key'
|
|
|
|
|
|
def _agentkey_status_payload():
|
|
"""Current state of the endpoint agent key for the settings page.
|
|
|
|
Reports whether an active managed key exists (and its non-secret metadata)
|
|
plus whether the bootstrap .env fallback is still accepted. Never exposes a
|
|
secret; the full secret is only ever returned once, at regenerate time.
|
|
"""
|
|
tokens = list_service_tokens(AGENT_KEY_SCOPE)
|
|
active = tokens[0] if tokens else None
|
|
envfallback = Setting.get('plugin.endpoints.envkeyfallback', default=True)
|
|
provisioned = Setting.get('plugin.endpoints.agentkeyprovisioned',
|
|
default=False)
|
|
return {
|
|
'provisioned': bool(provisioned) or active is not None,
|
|
'envkeyfallback': bool(envfallback),
|
|
'key': active.to_dict() if active is not None else None,
|
|
}
|
|
|
|
|
|
@endpoints_bp.route('/agentkey', methods=['GET'])
|
|
@require_role('admin')
|
|
def get_agentkey():
|
|
"""Report the endpoint agent key state (no secret)."""
|
|
return success_response(_agentkey_status_payload())
|
|
|
|
|
|
@endpoints_bp.route('/agentkey/regenerate', methods=['POST'])
|
|
@require_role('admin')
|
|
def regenerate_agentkey():
|
|
"""Mint a fresh endpoint agent key, revoking any previous one.
|
|
|
|
Hard cutover: the old key is deactivated in the same call, and the .env
|
|
fallback is switched off, so the new managed token becomes the sole
|
|
credential and revoking it is a true kill switch. The full secret is
|
|
returned ONCE here and never again.
|
|
"""
|
|
try:
|
|
token, secret = mint_service_token(
|
|
AGENT_KEY_SCOPE, AGENT_KEY_NAME, current_user.userid)
|
|
except PermissionError as error:
|
|
return error_response(ErrorCodes.FORBIDDEN, str(error), http_code=403)
|
|
|
|
Setting.set('plugin.endpoints.agentkeyprovisioned', True,
|
|
valuetype='boolean', category='plugin.endpoints')
|
|
Setting.set('plugin.endpoints.envkeyfallback', False,
|
|
valuetype='boolean', category='plugin.endpoints')
|
|
|
|
result = _agentkey_status_payload()
|
|
# The secret appears here and NOWHERE else, ever. Not stored, not logged.
|
|
result['secret'] = secret
|
|
result['warning'] = ('Save this key now. It will not be shown again. '
|
|
'Configure it as the agent X-API-Key.')
|
|
return success_response(result, message='Endpoint agent key generated',
|
|
http_code=201)
|
|
|
|
|
|
@endpoints_bp.route('/agentkey', methods=['DELETE'])
|
|
@require_role('admin')
|
|
def revoke_agentkey():
|
|
"""Revoke the endpoint agent key: stop all managed-token endpoint traffic.
|
|
|
|
Leaves envkeyfallback off (the kill switch stays armed). To restore
|
|
reporting an operator regenerates a new key.
|
|
"""
|
|
revoked = revoke_service_tokens(AGENT_KEY_SCOPE)
|
|
Setting.set('plugin.endpoints.agentkeyprovisioned', False,
|
|
valuetype='boolean', category='plugin.endpoints')
|
|
result = _agentkey_status_payload()
|
|
result['revoked'] = revoked
|
|
return success_response(result, message='Endpoint agent key revoked')
|
|
|
|
|
|
# =============================================================================
|
|
# Endpoint commands (WebUI -> agent, allowlisted actions only)
|
|
# =============================================================================
|
|
#
|
|
# Operators enqueue a named, server-allowlisted action for one endpoint (JWT +
|
|
# endpoints.command). The agent long-polls /commands/next with its collector
|
|
# token, claims the single next pending command, runs it locally, and reports
|
|
# the outcome to /commands/<id>/result. The wire never carries a command string
|
|
# - only an allowlisted action name - so a compromised WebUI cannot ask an
|
|
# endpoint to run something arbitrary.
|
|
|
|
# How long the agent claim poll is held open server-side before returning empty.
|
|
# Kept below the agent HttpClient timeout and typical proxy idle timeouts so a
|
|
# quiet fleet re-polls cleanly instead of tripping a read timeout.
|
|
_CLAIM_HOLD_SECONDS = 25
|
|
_CLAIM_POLL_INTERVAL = 1.0
|
|
|
|
|
|
def _collector_auth_or_error():
|
|
"""Authorize an agent-facing command route on the endpoints collector token.
|
|
|
|
Reuses the collector auth path so the command channel shares the endpoint
|
|
agent key (kill switch) and introduces no second secret.
|
|
"""
|
|
return collector_authorized('endpoints')
|
|
|
|
|
|
def _resolve_asset_for_hostname(hostname):
|
|
"""Resolve a reported hostname to its asset id via the computers plugin."""
|
|
from plugins.computers.models import Computer
|
|
|
|
computer = (Computer.query
|
|
.filter(db.func.lower(Computer.hostname) == hostname.lower())
|
|
.first())
|
|
return computer.assetid if computer is not None else None
|
|
|
|
|
|
@endpoints_bp.route('/<int:asset_id>/commands', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def list_endpoint_commands(asset_id: int):
|
|
"""Recent commands for one endpoint, newest first (operator view)."""
|
|
rows = (EndpointCommand.query
|
|
.filter_by(assetid=asset_id)
|
|
.order_by(EndpointCommand.createdat.desc())
|
|
.limit(20)
|
|
.all())
|
|
return success_response([row.to_dict() for row in rows])
|
|
|
|
|
|
@endpoints_bp.route('/<int:asset_id>/commands', methods=['POST'])
|
|
@require_permission('endpoints.command')
|
|
def enqueue_endpoint_command(asset_id: int):
|
|
"""Queue an allowlisted command for one endpoint.
|
|
|
|
Rejects unknown actions and refuses to stack a second pending/running
|
|
command on the same endpoint, so the queue stays at-most-one-in-flight and
|
|
the WebUI toggle cannot race itself.
|
|
"""
|
|
payload = request.get_json(silent=True) or {}
|
|
action = (payload.get('action') or '').strip()
|
|
if not is_allowed_action(action):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
f'Unknown or disallowed action: {action!r}')
|
|
|
|
# Destructive actions (for example system_restart) demand a stricter grant
|
|
# ON TOP OF endpoints.command, so a remote reboot cannot ride the routine
|
|
# toggle permission.
|
|
elevated = elevated_permission_for(action)
|
|
if elevated and not current_user.haspermission(elevated):
|
|
return error_response(
|
|
ErrorCodes.FORBIDDEN,
|
|
f'Action {action!r} requires the {elevated} permission',
|
|
http_code=403)
|
|
|
|
if action not in ALLOWED_ACTIONS_FOR(assetid=asset_id):
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
f'Action not permitted by policy for this endpoint: {action!r}')
|
|
|
|
# Generic actions (service_* / process_*) must carry the subject to act on;
|
|
# fixed-target actions (mtconnect_*) must not. Validate both directions so a
|
|
# caller cannot send a targetless generic action (nothing to act on) or
|
|
# smuggle a target onto a pinned action.
|
|
target = (payload.get('target') or '').strip() or None
|
|
targetkind = action_targetkind(action)
|
|
if requires_target(action):
|
|
if not target:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
f'Action {action!r} requires a target')
|
|
elif target is not None:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
f'Action {action!r} does not accept a target')
|
|
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
if state is None:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'No endpoint state for asset {asset_id}',
|
|
http_code=404)
|
|
|
|
pending = (EndpointCommand.query
|
|
.filter(EndpointCommand.assetid == asset_id,
|
|
EndpointCommand.status.in_((
|
|
EndpointCommand.STATUS_PENDING,
|
|
EndpointCommand.STATUS_RUNNING)))
|
|
.first())
|
|
if pending is not None:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'A command is already in flight for this endpoint',
|
|
http_code=409)
|
|
|
|
command = EndpointCommand(
|
|
assetid=asset_id,
|
|
action=action,
|
|
target=target,
|
|
targetkind=targetkind if target else None,
|
|
status=EndpointCommand.STATUS_PENDING,
|
|
requestedby=getattr(current_user, 'userid', None),
|
|
)
|
|
db.session.add(command)
|
|
|
|
AuditLog.log('created', 'EndpointCommand',
|
|
entityname=action,
|
|
details=f'action={action} target={target} asset={asset_id}')
|
|
|
|
db.session.commit()
|
|
return success_response(command.to_dict(),
|
|
message='Command queued', http_code=201)
|
|
|
|
|
|
@endpoints_bp.route('/commands/next', methods=['GET'])
|
|
def claim_next_command():
|
|
"""Agent long-poll: claim the next pending command for a hostname.
|
|
|
|
Collector-token auth (not JWT). Holds the request open briefly and returns
|
|
the single oldest pending command for the caller's endpoint, atomically
|
|
marking it running so no other poll claims it. Returns 204 when nothing is
|
|
queued within the hold window - the agent simply re-polls.
|
|
"""
|
|
error = _collector_auth_or_error()
|
|
if error is not None:
|
|
return error
|
|
|
|
hostname = (request.args.get('hostname') or '').strip()
|
|
if not hostname:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'hostname is required')
|
|
|
|
asset_id = _resolve_asset_for_hostname(hostname)
|
|
if asset_id is None:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Unknown hostname: {hostname!r}', http_code=404)
|
|
|
|
deadline = time.monotonic() + _CLAIM_HOLD_SECONDS
|
|
while True:
|
|
command = (EndpointCommand.query
|
|
.filter_by(assetid=asset_id,
|
|
status=EndpointCommand.STATUS_PENDING)
|
|
.order_by(EndpointCommand.createdat.asc())
|
|
.first())
|
|
if command is not None:
|
|
command.status = EndpointCommand.STATUS_RUNNING
|
|
command.claimedat = _naive_utcnow()
|
|
db.session.commit()
|
|
return success_response({
|
|
'endpointcommandid': command.endpointcommandid,
|
|
'action': command.action,
|
|
'target': command.target,
|
|
'targetkind': command.targetkind,
|
|
})
|
|
|
|
# Nothing queued: release the session so a concurrent enqueue is visible
|
|
# on the next poll, wait a beat, then re-check until the hold expires.
|
|
db.session.rollback()
|
|
if time.monotonic() >= deadline:
|
|
return ('', 204)
|
|
time.sleep(_CLAIM_POLL_INTERVAL)
|
|
|
|
|
|
@endpoints_bp.route('/commands/<int:command_id>/result', methods=['POST'])
|
|
def report_command_result(command_id: int):
|
|
"""Agent reports a claimed command's outcome (collector-token auth).
|
|
|
|
Finalizes the command row and, for MTConnect actions that observed the run
|
|
state, caches it on the EndpointState so the WebUI toggle reflects reality
|
|
between polls. Both writes land in one commit.
|
|
"""
|
|
error = _collector_auth_or_error()
|
|
if error is not None:
|
|
return error
|
|
|
|
command = db.session.get(EndpointCommand, command_id)
|
|
if command is None:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'No command {command_id}', http_code=404)
|
|
|
|
payload = request.get_json(silent=True) or {}
|
|
succeeded = bool(payload.get('succeeded'))
|
|
command.status = (EndpointCommand.STATUS_SUCCEEDED if succeeded
|
|
else EndpointCommand.STATUS_FAILED)
|
|
command.resultmessage = (payload.get('message') or '')[:2000] or None
|
|
detail = payload.get('detail')
|
|
command.resultjson = json.dumps(detail, separators=(',', ':')) \
|
|
if detail is not None else None
|
|
command.completedat = _naive_utcnow()
|
|
|
|
# Cache the observed MTConnect state so the toggle shows a real value. The
|
|
# agent sends mtconnectrunning on any mtconnect_* result that probed it.
|
|
if 'mtconnectrunning' in payload:
|
|
state = EndpointState.query.filter_by(assetid=command.assetid).first()
|
|
if state is not None:
|
|
running = payload.get('mtconnectrunning')
|
|
state.mtconnectrunning = bool(running) if running is not None else None
|
|
state.mtconnectcheckedat = _naive_utcnow()
|
|
|
|
db.session.commit()
|
|
return success_response(command.to_dict(), message='Result recorded')
|
|
|