0c9bd1ed32
Mapped drives existed only as a comma-joined line in the user context card, which is fine for three drives and unreadable for a dozen. They now get their own list card in Monitors: one row per letter, remote path as meta, health as a badge. Reuses the statecolor/levels vocabulary the tracked-services cards already declare rather than new CSS. Only unhealthy drives are coloured. Green on every working drive makes the broken one just another badge in a row of badges, and the card exists to make the exception findable at a glance. Unhealthy rows sort first for the same reason - a user with a dozen mappings should not hunt for the broken one - and the sort is on health alone, so drive-letter order survives within each group and the list does not reshuffle between refreshes. The merge logic is untouched. _mapped_drive_rows now returns structured rows and the user-context summary line is built from them, so the card and the line cannot drift apart about a drive's status. The two silent-failure paths are mutation-tested: forcing every badge green failed 7 tests, and the first sort test passed a mutation that added a secondary key because every broken drive in it said the same word - it now uses statuses that would reorder under a text sort. Also corrects a stale warning claiming the merge is untested; it has been covered since Phase 1.
2577 lines
108 KiB
Python
2577 lines
108 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 hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
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,
|
|
Application, AppVersion, Script,
|
|
)
|
|
|
|
from ..models import (
|
|
EndpointState, EndpointEvent, EndpointCommand, EndpointApplication,
|
|
OperatorReport,
|
|
)
|
|
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)
|
|
|
|
|
|
# What counts as RUNNING is defined once, by the reconciler, and imported here.
|
|
# The panel and the reconciler read the same agent snapshot, so a disagreement
|
|
# would paint a service green and flag it as drift on the same page. Importing
|
|
# api -> services is the direction the plugin layout already runs in; the
|
|
# reconciler's own import of this module is deferred, so there is no cycle.
|
|
from ..services.reconciler import RUNNING_STATES as _RUNNING_STATES
|
|
|
|
|
|
def _service_running(entry):
|
|
raw = (entry.get('state') or entry.get('status') or '').strip().lower()
|
|
return raw in _RUNNING_STATES
|
|
|
|
|
|
# Paused, plus the four transitional states Windows reports while a service is
|
|
# on its way somewhere.
|
|
_TRANSITIONAL_STATES = ('paused', 'start pending', 'stop pending',
|
|
'continue pending', 'pause pending', 'starting',
|
|
'stopping', 'pausing')
|
|
|
|
|
|
def _service_transitional(raw):
|
|
return (raw or '').strip().lower() in _TRANSITIONAL_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)
|
|
|
|
|
|
def _state_json(state, attr):
|
|
"""Parse one of the EndpointState inventory JSON blobs.
|
|
|
|
These columns are written straight from the agent snapshot, so a partial
|
|
or malformed blob from one bad check-in must not take the whole panel
|
|
down. Anything unparseable reads as absent.
|
|
"""
|
|
raw = getattr(state, attr, None) if state else None
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _format_bytes(value):
|
|
"""Byte count -> 'N.N GB'. Sizes here are disks and RAM, never sub-MB."""
|
|
if value is None:
|
|
return None
|
|
try:
|
|
value = float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
for unit in ('B', 'KB', 'MB', 'GB', 'TB'):
|
|
if abs(value) < 1024.0 or unit == 'TB':
|
|
return '{:.1f} {}'.format(value, unit) if unit != 'B' else '{:.0f} B'.format(value)
|
|
value /= 1024.0
|
|
return None
|
|
|
|
|
|
def _format_agent_date(value):
|
|
"""Agent timestamps are ISO strings, not datetimes; show the date only.
|
|
|
|
A BIOS release date or an OS install date is only ever read at day
|
|
resolution, and the agent sends a full 7-digit-fraction UTC stamp.
|
|
"""
|
|
if not value:
|
|
return None
|
|
text = str(value)
|
|
return text[:10] if len(text) >= 10 else text
|
|
|
|
|
|
# A Windows event message runs to several hundred words. It is shown in an
|
|
# expander that is closed by default, so the cap here only exists to keep one
|
|
# pathological message from bloating the payload - not to fit a line. Blank
|
|
# runs are collapsed but real line breaks are kept, because the body is read
|
|
# as a block once opened.
|
|
_EVENT_BODY_LIMIT = 4000
|
|
|
|
|
|
def _event_body(event):
|
|
"""Readable message body for one curated event, bounded and normalized.
|
|
|
|
Windows messages arrive padded with long runs of blank lines that render
|
|
as ragged gaps. Runs of blank lines collapse to one and trailing spaces
|
|
go, but single newlines survive so the text keeps its shape when opened.
|
|
"""
|
|
text = (getattr(event, 'eventmessage', None) or '').strip()
|
|
if not text:
|
|
return None
|
|
lines = [line.rstrip() for line in text.splitlines()]
|
|
cleaned = []
|
|
for line in lines:
|
|
if not line and cleaned and not cleaned[-1]:
|
|
continue
|
|
cleaned.append(line)
|
|
text = '\n'.join(cleaned).strip()
|
|
if not text:
|
|
return None
|
|
if len(text) <= _EVENT_BODY_LIMIT:
|
|
return text
|
|
return text[:_EVENT_BODY_LIMIT].rstrip() + '...'
|
|
|
|
|
|
# =============================================================================
|
|
# Endpoint state
|
|
# =============================================================================
|
|
|
|
@endpoints_bp.route('', methods=['GET'], strict_slashes=False)
|
|
@endpoints_bp.route('/', methods=['GET'], strict_slashes=False)
|
|
@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))
|
|
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: (total, dominant_eventid)} of critical events in the window.
|
|
|
|
Broken down BY EVENT ID, not just totalled. A bare count says "418
|
|
critical", which tells an operator nothing about what to do; naming the
|
|
event that dominates - service timeout, application crash - is the
|
|
difference between a number and a lead.
|
|
"""
|
|
if not assetids:
|
|
return {}
|
|
cutoff = _naive_utcnow() - timedelta(hours=48)
|
|
rows = (db.session.query(EndpointEvent.assetid, EndpointEvent.eventid,
|
|
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, EndpointEvent.eventid)
|
|
.all())
|
|
|
|
perasset = {}
|
|
for assetid, eventid, count in rows:
|
|
total, dominant, best = perasset.get(assetid, (0, None, 0))
|
|
total += count
|
|
if count > best:
|
|
dominant, best = eventid, count
|
|
perasset[assetid] = (total, dominant, best)
|
|
return {assetid: (total, dominant)
|
|
for assetid, (total, dominant, _best) in perasset.items()}
|
|
|
|
|
|
def _health_for(state, critical):
|
|
"""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 - and neither does "418 critical", which is
|
|
why the dominant event is named rather than merely counted.
|
|
"""
|
|
if critical:
|
|
total, dominant = critical
|
|
title = event_title(dominant) if dominant else 'critical'
|
|
label = title if total == 1 else '{} x{}'.format(title, total)
|
|
return {'level': 'critical', 'label': label}
|
|
|
|
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
|
|
# =============================================================================
|
|
|
|
def _drift_for_state(state):
|
|
"""Parse the cached warn-only reconciler drift list off an EndpointState.
|
|
|
|
driftjson is written on every check-in by the endpoints ingest: the monitors
|
|
site policy expects running that the latest snapshot did not report running.
|
|
Null means never computed; an empty list means computed with no drift.
|
|
Tolerant of a malformed blob so a bad cache never breaks the panel.
|
|
"""
|
|
raw = getattr(state, 'driftjson', None) if state else None
|
|
if not raw:
|
|
return []
|
|
try:
|
|
parsed = json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
return []
|
|
return parsed if isinstance(parsed, list) else []
|
|
|
|
|
|
@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('Tray', state.trayversion, 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, '')]
|
|
|
|
# Warn-only reconciler summary: monitors site policy expects to be running
|
|
# that the last check-in did not report as running. Surfaced as a status
|
|
# field so an operator sees the drift on the panel; nothing is auto-started.
|
|
drift = _drift_for_state(state)
|
|
if drift:
|
|
names = ', '.join(entry.get('displayname') or entry.get('name')
|
|
for entry in drift)
|
|
fields.append(field(
|
|
'Needs attention',
|
|
'{} monitor(s) expected running but not reported: {}'.format(
|
|
len(drift), names)))
|
|
|
|
return success_response({'fields': fields})
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/hardware', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_hardware(asset_id: int):
|
|
"""Keyvalue panel: processor, memory and system chassis inventory.
|
|
|
|
Only reported by agent 1.7.0 and later. An older agent leaves these
|
|
columns null, so the panel yields no fields and hides itself rather than
|
|
rendering a card full of blanks.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
if not state:
|
|
return success_response({'fields': []})
|
|
|
|
processor = _state_json(state, 'processorjson') or {}
|
|
memory = _state_json(state, 'memoryjson') or {}
|
|
|
|
fields = [
|
|
{'label': 'Processor', 'value': processor.get('name')},
|
|
{'label': 'Cores', 'value': processor.get('cores')},
|
|
{'label': 'Logical processors', 'value': processor.get('logicalprocessors')},
|
|
{'label': 'Max clock',
|
|
'value': '{} MHz'.format(processor['maxclockmhz'])
|
|
if processor.get('maxclockmhz') else None},
|
|
{'label': 'Total memory', 'value': _format_bytes(memory.get('totalbytes'))},
|
|
{'label': 'Free memory', 'value': _format_bytes(memory.get('freebytes'))},
|
|
]
|
|
fields = [f for f in fields if f['value'] not in (None, '')]
|
|
return success_response({'fields': fields})
|
|
|
|
|
|
# Wire values to operator-readable text. The wire keys are stable contract;
|
|
# these labels are not, so they live here rather than in the agent.
|
|
_REBOOT_REASON_LABELS = {
|
|
'componentbasedservicing': 'servicing',
|
|
'windowsupdate': 'Windows Update',
|
|
'pendingfilerename': 'files staged for replacement',
|
|
}
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/osinfo', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_osinfo(asset_id: int):
|
|
"""Keyvalue panel: operating system, firmware and domain membership.
|
|
|
|
OS and BIOS are one card on purpose: both answer "what is this box
|
|
running", and split across two cards each would be three short rows.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
if not state:
|
|
return success_response({'fields': []})
|
|
|
|
osinfo = _state_json(state, 'osinfojson') or {}
|
|
bios = _state_json(state, 'biosjson') or {}
|
|
bootinfo = _state_json(state, 'bootinfojson') or {}
|
|
|
|
pendingreboot = None
|
|
if state.pendingreboot is not None:
|
|
pendingreboot = 'Yes' if state.pendingreboot else 'No'
|
|
# Say WHICH signal fired when we know. A staged file rename means the
|
|
# box is mid-servicing; a lone servicing key can sit set for weeks on a
|
|
# machine nobody rebooted. Same "Yes", very different urgency.
|
|
reasons = _state_json(state, 'rebootreasonsjson')
|
|
if state.pendingreboot and isinstance(reasons, list) and reasons:
|
|
pendingreboot = '{} ({})'.format(
|
|
pendingreboot, ', '.join(_REBOOT_REASON_LABELS.get(r, r)
|
|
for r in reasons if isinstance(r, str)))
|
|
|
|
# Only worth a row when fast startup actually applies. On a box with it
|
|
# disabled and a cold last boot there is nothing to warn about, and a
|
|
# permanent "Cold" row is noise on a card that is already dense.
|
|
boottype = bootinfo.get('lastboottype')
|
|
if boottype in (None, '', 'Cold', 'Unknown'):
|
|
boottype = None
|
|
elif boottype == 'Fast startup':
|
|
# Uptime is measured from the resume, not the last true boot, so it
|
|
# understates how long the kernel has actually been up. Said plainly
|
|
# because this is the field that makes uptime read wrong.
|
|
boottype = 'Fast startup (uptime not since cold boot)'
|
|
|
|
fields = [
|
|
{'label': 'OS version', 'value': osinfo.get('version'), 'mono': True},
|
|
{'label': 'Build', 'value': osinfo.get('build'), 'mono': True},
|
|
{'label': 'Architecture', 'value': osinfo.get('architecture')},
|
|
{'label': 'Installed', 'value': _format_agent_date(osinfo.get('installdate'))},
|
|
{'label': 'Domain', 'value': state.domain},
|
|
{'label': 'BIOS vendor', 'value': bios.get('vendor')},
|
|
{'label': 'BIOS version', 'value': bios.get('version'), 'mono': True},
|
|
{'label': 'BIOS date', 'value': _format_agent_date(bios.get('releasedate'))},
|
|
{'label': 'Last boot', 'value': boottype},
|
|
{'label': 'Pending reboot', 'value': pendingreboot},
|
|
]
|
|
fields = [f for f in fields
|
|
if f['value'] not in (None, '')]
|
|
return success_response({'fields': fields})
|
|
|
|
|
|
def _storage_rows(disks):
|
|
"""Turn the agent's volume list into display rows.
|
|
|
|
freepercent is computed by the agent from 1.7.0; for a snapshot written
|
|
by an older agent it is derived here from the byte counts so the column
|
|
is not empty for the one field an operator actually scans.
|
|
"""
|
|
if not isinstance(disks, list):
|
|
return []
|
|
|
|
rows = []
|
|
for disk in disks:
|
|
if not isinstance(disk, dict):
|
|
continue
|
|
freepercent = disk.get('freepercent')
|
|
if freepercent is None:
|
|
size = disk.get('sizebytes') or 0
|
|
free = disk.get('freebytes')
|
|
if size and free is not None:
|
|
freepercent = round(free * 100.0 / size, 1)
|
|
rows.append({
|
|
'drive': disk.get('drive'),
|
|
'label': disk.get('label') or '',
|
|
'filesystem': disk.get('filesystem'),
|
|
'size': _format_bytes(disk.get('sizebytes')),
|
|
'free': _format_bytes(disk.get('freebytes')),
|
|
'freepercent': '{}%'.format(freepercent) if freepercent is not None else None,
|
|
})
|
|
return rows
|
|
|
|
|
|
_STORAGE_COLUMNS = [
|
|
{'key': 'drive', 'label': 'Drive'},
|
|
{'key': 'label', 'label': 'Label'},
|
|
{'key': 'filesystem', 'label': 'Filesystem'},
|
|
{'key': 'size', 'label': 'Size'},
|
|
{'key': 'free', 'label': 'Free'},
|
|
{'key': 'freepercent', 'label': 'Free %'},
|
|
]
|
|
|
|
_DISKHEALTH_COLUMNS = [
|
|
{'key': 'model', 'label': 'Drive'},
|
|
{'key': 'mediatype', 'label': 'Media'},
|
|
{'key': 'bustype', 'label': 'Bus'},
|
|
{'key': 'healthstatus', 'label': 'Health'},
|
|
{'key': 'verdict', 'label': 'Verdict'},
|
|
{'key': 'source', 'label': 'Source'},
|
|
]
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/storage', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_storage(asset_id: int):
|
|
"""Tabs panel: volume capacity and physical drive health as two sections.
|
|
|
|
One card because an operator looking at a full disk and an operator looking
|
|
at a dying disk are usually the same person in the same minute. TWO
|
|
SECTIONS because the two do not share a key: storage is per volume (C:, D:)
|
|
and health is per physical device, and one drive can back several volumes
|
|
or none the operator sees. Nothing is reconciled across the sections -
|
|
this is not the mapped-drive merge, where two sources described the same
|
|
drive and the worse status won.
|
|
|
|
Both sections are emitted even when empty so the section keeps its own
|
|
empty message; a dropped section would take its tab with it and read as
|
|
"no disks" rather than "nothing reported".
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
disks = _state_json(state, 'disksjson') if state else None
|
|
drives = _state_json(state, 'diskhealthjson') if state else None
|
|
return success_response({'sections': [
|
|
{
|
|
'label': 'Volumes',
|
|
'columns': _STORAGE_COLUMNS,
|
|
'rows': _storage_rows(disks),
|
|
'empty': 'No disks reported.',
|
|
},
|
|
{
|
|
'label': 'Disk health',
|
|
'columns': _DISKHEALTH_COLUMNS,
|
|
'rows': _diskhealth_rows(drives),
|
|
'empty': 'No disk health reported.',
|
|
},
|
|
]})
|
|
|
|
|
|
def _diskhealth_rows(drives):
|
|
"""Turn the agent's disk health list into display rows, worst first.
|
|
|
|
Both agent sources are kept rather than merged. MSFT_PhysicalDisk is the
|
|
only source that reports NVMe at all, and Win32_DiskDrive covers hardware
|
|
and hosts where the Storage namespace is unavailable. When both answer for
|
|
the same hardware, two rows that disagree is the useful signal; collapsing
|
|
them would hide exactly the case worth looking at.
|
|
"""
|
|
if not isinstance(drives, list):
|
|
return []
|
|
|
|
rows = []
|
|
for drive in drives:
|
|
if not isinstance(drive, dict):
|
|
continue
|
|
predictfailure = drive.get('predictfailure')
|
|
if predictfailure is None:
|
|
verdict = 'Unknown'
|
|
elif predictfailure:
|
|
verdict = 'Failure predicted'
|
|
else:
|
|
verdict = 'OK'
|
|
rows.append({
|
|
'deviceid': drive.get('deviceid'),
|
|
'model': drive.get('model') or '',
|
|
'mediatype': drive.get('mediatype') or '',
|
|
'bustype': drive.get('bustype') or '',
|
|
'healthstatus': drive.get('healthstatus') or 'Unknown',
|
|
'verdict': verdict,
|
|
'source': drive.get('source') or '',
|
|
})
|
|
|
|
# Failure first, then unknown, then OK. Unknown outranks OK because a drive
|
|
# that stopped answering is a weaker claim of health than one that said OK.
|
|
# An unrecognised verdict sorts as unknown rather than last, so a future
|
|
# verdict string cannot bury itself under the healthy drives.
|
|
order = {'Failure predicted': 0, 'Unknown': 1, 'OK': 2}
|
|
rows.sort(key=lambda row: (order.get(row['verdict'], 1),
|
|
row['deviceid'] or ''))
|
|
return rows
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/diskhealth', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_diskhealth(asset_id: int):
|
|
"""Physical drive health rows.
|
|
|
|
No longer backs a panel of its own - the Storage card renders it as a
|
|
section. Kept as a route because it is a stable read for callers outside
|
|
the panel renderer and removing it would break them for a UI-only change.
|
|
|
|
Rows are ordered so anything predicting failure sorts first: a failing
|
|
drive must not be below the fold on a machine with several disks.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
drives = _state_json(state, 'diskhealthjson') if state else None
|
|
return success_response(_diskhealth_rows(drives))
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/listeners', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_listeners(asset_id: int):
|
|
"""Table panel: TCP sockets in LISTEN, with the owning process.
|
|
|
|
The listening set is the box's exposed surface, so it is sorted by port
|
|
to make an unexpected one easy to spot. The full connection table is a
|
|
deliberately on-demand command, not standard telemetry.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
ports = _state_json(state, 'portsjson') if state else None
|
|
if not isinstance(ports, list):
|
|
return success_response([])
|
|
|
|
rows = []
|
|
for entry in ports:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
# Sample-data rows predate the agent listener collector and carry a
|
|
# flat port/process shape; read both rather than showing them blank.
|
|
port = entry.get('localport', entry.get('port'))
|
|
rows.append({
|
|
'localport': port,
|
|
'localaddress': entry.get('localaddress'),
|
|
'protocol': (entry.get('protocol') or 'tcp').upper(),
|
|
'processname': entry.get('processname') or entry.get('process'),
|
|
'processid': entry.get('processid'),
|
|
})
|
|
rows.sort(key=lambda r: (r['localport'] is None, r['localport']))
|
|
return success_response(rows)
|
|
|
|
|
|
def _drive_letter(value):
|
|
"""Normalise a drive to its bare letter so the two sources can be paired.
|
|
|
|
net use reports 'Z:' and Get-SmbMapping reports 'Z:' or occasionally a
|
|
trailing separator, so compare on the letter alone.
|
|
"""
|
|
text = (value or '').strip().rstrip('\\/')
|
|
return text.upper() or None
|
|
|
|
|
|
# Worst-first. The panel shows the worse of the two sources for a drive, so a
|
|
# share that one source calls OK and the other calls Unavailable is reported as
|
|
# Unavailable: a drive that is broken in either view is broken for the operator.
|
|
#
|
|
# WARNING - ranks must stay DISTINCT across statuses that can genuinely differ.
|
|
# _worse_drive_status resolves by min(), which breaks a tie by argument order,
|
|
# so two statuses sharing a rank make the verdict depend on which SOURCE
|
|
# happened to report which - the same drive reads healthy or broken based on
|
|
# enumeration order. Deliberate synonyms are fine (disconnected/unavailable,
|
|
# ok/connected) because either answer is the same answer. Adding a NEW status at
|
|
# an existing rank is not. test_which_source_reported_a_status_does_not_change
|
|
# _the_verdict in tests/test_plugins/test_endpoints_mapped_drives.py asserts
|
|
# both directions of each pair and is what catches this; the single-direction
|
|
# tests stay green through it.
|
|
_DRIVE_STATUS_RANK = {
|
|
'unavailable': 0,
|
|
'disconnected': 0,
|
|
'paused': 1,
|
|
'reconnecting': 2,
|
|
'ok': 3,
|
|
'connected': 3,
|
|
}
|
|
|
|
|
|
def _worse_drive_status(left, right):
|
|
"""Return whichever status is worse, preferring a known one over None.
|
|
|
|
An unrecognised status ranks as Reconnecting, deliberately below OK: a
|
|
status we cannot interpret must not be assumed healthy.
|
|
"""
|
|
candidates = [s for s in (left, right) if s]
|
|
if not candidates:
|
|
return None
|
|
unknown_rank = _DRIVE_STATUS_RANK['reconnecting']
|
|
return min(
|
|
candidates,
|
|
key=lambda s: _DRIVE_STATUS_RANK.get(str(s).strip().lower(), unknown_rank)
|
|
)
|
|
|
|
|
|
def _mapped_drive_rows(context):
|
|
"""Merge the two mapped-drive views into one structured row per letter.
|
|
|
|
Windows exposes mapped drives through Win32_NetworkConnection (the 'net use'
|
|
table) and through Get-SmbMapping, and they do not always agree - a share can
|
|
be Unavailable in one and OK or absent in the other. The probe collects both
|
|
without reconciling them, so the disagreement survives to here.
|
|
|
|
A drive present in only one source is still a drive: it is reported with the
|
|
status that source gave it, rather than dropped for lack of a second opinion.
|
|
|
|
WARNING - three behaviours here are load-bearing and easy to break while
|
|
"simplifying". They are covered by tests/test_plugins/test_endpoints_mapped
|
|
_drives.py; if you touch this, read those first:
|
|
|
|
1. Worse-status-wins. Never report the healthier of the two sources. A
|
|
drive that either source calls Unavailable IS unavailable to the
|
|
operator, and rendering it OK sends someone to a share that is down.
|
|
2. Single-source drives are kept, not dropped. Get-SmbMapping is absent on
|
|
some builds and session types; requiring agreement would silently hide
|
|
every mapped drive on those machines.
|
|
3. Unknown statuses rank as Reconnecting, not as OK. An unrecognised
|
|
string must never be optimistically treated as healthy.
|
|
|
|
Changing any of these makes a broken drive look fine, which is a silent
|
|
failure - the card stays green and nobody is told.
|
|
"""
|
|
drives = {}
|
|
|
|
netuse = context.get('netuse')
|
|
if isinstance(netuse, list):
|
|
for entry in netuse:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
letter = _drive_letter(entry.get('localname') or entry.get('local'))
|
|
if not letter:
|
|
continue
|
|
drives[letter] = {
|
|
'remote': entry.get('remotename') or entry.get('remote') or '',
|
|
'status': entry.get('status'),
|
|
}
|
|
|
|
smbmapping = context.get('smbmapping')
|
|
if isinstance(smbmapping, list):
|
|
for entry in smbmapping:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
letter = _drive_letter(entry.get('localpath'))
|
|
if not letter:
|
|
continue
|
|
existing = drives.get(letter)
|
|
if existing:
|
|
existing['status'] = _worse_drive_status(
|
|
existing.get('status'), entry.get('status'))
|
|
existing['remote'] = existing['remote'] or entry.get('remotepath') or ''
|
|
else:
|
|
drives[letter] = {
|
|
'remote': entry.get('remotepath') or '',
|
|
'status': entry.get('status'),
|
|
}
|
|
|
|
rows = []
|
|
for letter in sorted(drives):
|
|
drive = drives[letter]
|
|
# A drive neither source gave a status for is Unknown, never blank. An
|
|
# unannotated row reads as healthy, so a failed check would render as a
|
|
# working drive - the one outcome this card must never produce.
|
|
status = (drive.get('status') or '').strip() or 'Unknown'
|
|
healthy = status.lower() in ('ok', 'connected')
|
|
rows.append({
|
|
'drive': letter,
|
|
'remote': (drive.get('remote') or '').strip(),
|
|
'statuslabel': status,
|
|
# Only unhealthy drives are coloured. Green on every working drive
|
|
# makes the broken one just another badge in a row of badges; the
|
|
# card's job is to make the exception findable at a glance.
|
|
'statecolor': 'green' if healthy else 'red',
|
|
})
|
|
return rows
|
|
|
|
|
|
def _mapped_drives(context):
|
|
"""The mapped drives as one-line labels, for the user context card.
|
|
|
|
Built on _mapped_drive_rows so the summary line and the card can never
|
|
disagree about a drive's status.
|
|
"""
|
|
labels = []
|
|
for row in _mapped_drive_rows(context):
|
|
label = '{} {}'.format(row['drive'], row['remote']).strip()
|
|
# Only the problem states are annotated. Tagging every healthy drive
|
|
# with "(OK)" would bury the one that is not.
|
|
if row['statecolor'] != 'green':
|
|
label = '{} ({})'.format(label, row['statuslabel'])
|
|
labels.append(label)
|
|
return labels
|
|
|
|
|
|
def _mapped_drives_unhealthy_first(rows):
|
|
"""Order drive rows so anything not healthy sorts first.
|
|
|
|
A user with a dozen mappings should not have to hunt for the broken one.
|
|
Sorting only on health keeps it stable: drive-letter order survives within
|
|
each group, so the list does not reshuffle between refreshes.
|
|
"""
|
|
return sorted(rows, key=lambda row: row['statecolor'] == 'green')
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/mappeddrives', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_mappeddrives(asset_id: int):
|
|
"""List panel: mapped network drives with a health badge each.
|
|
|
|
Session-scoped: mapped drives do not exist in the agent's session 0, so
|
|
these come from the interactive user probe. Nothing reported when nobody
|
|
is signed in, which is a valid state and not a failure.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
context = _state_json(state, 'usercontextjson') if state else None
|
|
if not isinstance(context, dict) or not context:
|
|
return success_response([])
|
|
return success_response(_mapped_drives_unhealthy_first(_mapped_drive_rows(context)))
|
|
|
|
|
|
@endpoints_bp.route('/asset/<int:asset_id>/usercontext', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_usercontext(asset_id: int):
|
|
"""Keyvalue panel: session-scoped data from the interactive user probe.
|
|
|
|
Mapped drives and user-owned listeners do not exist in the agent's
|
|
session 0, so they are collected by a probe run in the logged-in user's
|
|
session. Nothing is reported when no one is signed in, which is a valid
|
|
state and not a failure.
|
|
|
|
Mapped drives arrive from two sources that can disagree; the worse of the
|
|
two statuses is shown so a half-broken share is never rendered as healthy.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
if not state:
|
|
return success_response({'fields': []})
|
|
|
|
context = _state_json(state, 'usercontextjson')
|
|
if not isinstance(context, dict) or not context:
|
|
return success_response({'fields': []})
|
|
|
|
drives = _mapped_drives(context)
|
|
|
|
listeners = context.get('userlisteners')
|
|
listenercount = len(listeners) if isinstance(listeners, list) else None
|
|
|
|
fields = [
|
|
{'label': 'Probed user', 'value': context.get('username')},
|
|
{'label': 'Collected', 'value': context.get('collectedat')},
|
|
{'label': 'Idle', 'value': _format_uptime(context.get('idleseconds'))},
|
|
{'label': 'Mapped drives',
|
|
'value': ', '.join(drives) if drives else None, 'mono': True},
|
|
{'label': 'User listeners', 'value': listenercount},
|
|
]
|
|
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})
|
|
|
|
|
|
def _pctype_for_asset(asset_id):
|
|
"""Resolve an asset's PC type name, or None.
|
|
|
|
Endpoints stays optional-on-computers: a missing plugin, a non-computer
|
|
asset, or an unset type all resolve to None so the caller falls back to
|
|
reported-only behaviour.
|
|
"""
|
|
try:
|
|
from plugins.computers.models import Computer, ComputerType
|
|
except Exception:
|
|
return None
|
|
row = (db.session.query(ComputerType.computertype)
|
|
.join(Computer,
|
|
Computer.computertypeid == ComputerType.computertypeid)
|
|
.filter(Computer.assetid == asset_id)
|
|
.first())
|
|
return row[0] if row else None
|
|
|
|
|
|
def _resolved_monitors_for_asset(asset_id):
|
|
"""Monitor definitions attached to this asset's PC type by policy/shipped.
|
|
|
|
Reuses the endpoints monitor resolver (shipped defaults plus policy-attached
|
|
monitors). Returns a list of EndpointMonitor rows, or an empty list when the
|
|
PC type is unknown or nothing resolves.
|
|
"""
|
|
pctype = _pctype_for_asset(asset_id)
|
|
if not pctype:
|
|
return []
|
|
try:
|
|
from .monitors import _resolve_pctype_monitors
|
|
return _resolve_pctype_monitors(pctype) or []
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
def _row_action_dict(app, kind, name, running):
|
|
"""Build the inline Start/Stop action map for a services/processes row."""
|
|
targets = row_actions_for(app, kind, name)
|
|
if not targets:
|
|
return None
|
|
return {
|
|
'running': running,
|
|
'start': targets['start'],
|
|
'stop': targets['stop'],
|
|
'status': targets['status'],
|
|
'target': targets['target'],
|
|
'targetkind': targets['targetkind'],
|
|
}
|
|
|
|
|
|
@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, then MERGES the monitors that site policy (and the
|
|
shipped defaults) attach to this asset's PC type. A policy-attached monitor
|
|
the agent has not yet reported surfaces as an Unknown row with a Start
|
|
control, so an operator sees the intent immediately rather than waiting for
|
|
the next agent snapshot. Tolerant of a missing snapshot or plugin: an
|
|
unknown state renders as an Unknown badge rather than failing.
|
|
"""
|
|
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
|
maycommand = caller_holds('endpoints.command')
|
|
|
|
services = []
|
|
if state and state.servicesjson:
|
|
try:
|
|
services = json.loads(state.servicesjson) or []
|
|
except (TypeError, ValueError):
|
|
services = []
|
|
|
|
rows = []
|
|
# Reported identities, keyed (kind, lowercased name), so a policy monitor
|
|
# the agent already reports is not duplicated as a synthetic row.
|
|
reported = set()
|
|
# Warn-only reconciler drift, keyed (kind, lowercased name), so a row that is
|
|
# expected running but not reported running can be flagged for the operator.
|
|
# kind may be None on an older snapshot, so match on name too.
|
|
drift_keys = set()
|
|
for entry in _drift_for_state(state):
|
|
name = (entry.get('name') or '').strip().lower()
|
|
if not name:
|
|
continue
|
|
drift_keys.add(((entry.get('kind') or '').lower() or None, name))
|
|
drift_keys.add((None, name))
|
|
|
|
def _is_drift(kind, name):
|
|
key = (name or '').strip().lower()
|
|
return (kind, key) in drift_keys or (None, key) in drift_keys
|
|
|
|
for entry in services:
|
|
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)
|
|
normkind = (kind or '').lower() or None
|
|
row = {
|
|
'name': name,
|
|
'kind': normkind,
|
|
'statelabel': 'Running' if running else (raw.title() or 'Unknown'),
|
|
# Amber for paused and the four Windows transitional states. A
|
|
# service stuck in Start Pending is failing to start on a retry
|
|
# cycle - which is what raises event 7011 - and a paused one was
|
|
# paused deliberately. Painting either the same red as Stopped
|
|
# loses the distinction that tells you which to chase.
|
|
'statecolor': ('green' if running
|
|
else 'yellow' if _service_transitional(raw)
|
|
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.
|
|
action = _row_action_dict(app, kind, name, running) if maycommand else None
|
|
if action:
|
|
row['action'] = action
|
|
if not running and _is_drift(normkind, name):
|
|
row['drift'] = True
|
|
rows.append(row)
|
|
reported.add((normkind, name.strip().lower()))
|
|
|
|
# Merge policy/shipped monitors this PC type should watch. A monitor already
|
|
# reported by the agent stays as its live row; one not yet seen is appended
|
|
# as an Unknown row so the operator sees the intended coverage and can Start
|
|
# it. The monitor's displayname is the operator label (e.g. 'msedge.exe'
|
|
# -> 'Edge'); its process/service name is the wire target for actions.
|
|
for monitor in _resolved_monitors_for_asset(asset_id):
|
|
kind = (monitor.kind or '').lower() or None
|
|
osname = (monitor.processname if kind == 'process'
|
|
else monitor.servicename)
|
|
osname = (osname or '').strip()
|
|
if not osname:
|
|
continue
|
|
if (kind, osname.lower()) in reported:
|
|
continue
|
|
row = {
|
|
'name': monitor.displayname or osname,
|
|
'kind': kind,
|
|
'statelabel': 'Unknown',
|
|
'statecolor': 'gray',
|
|
}
|
|
action = _row_action_dict(None, kind, osname, False) if maycommand else None
|
|
if action:
|
|
row['action'] = action
|
|
if _is_drift(kind, osname):
|
|
row['drift'] = True
|
|
rows.append(row)
|
|
reported.add((kind, osname.lower()))
|
|
|
|
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())
|
|
|
|
# The newest 15 of ANY severity can be entirely routine logons while the
|
|
# health rollup is shouting about a burst of crashes underneath them - the
|
|
# page then contradicts itself for no visible reason. Pull the newest
|
|
# criticals separately and merge, so what health is reporting is always
|
|
# present in the feed that is supposed to explain it.
|
|
criticals = (EndpointEvent.query
|
|
.filter(EndpointEvent.assetid == asset_id,
|
|
EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS))
|
|
.order_by(EndpointEvent.eventtime.desc())
|
|
.limit(5)
|
|
.all())
|
|
seen = {e.endpointeventid for e in events}
|
|
events.extend(e for e in criticals if e.endpointeventid not in seen)
|
|
|
|
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:
|
|
# The provider name is what distinguishes two events sharing an ID,
|
|
# and the message body is the line an operator actually reads. Both
|
|
# are already stored; the feed just was not showing them.
|
|
source = event.eventsource or event.eventchannel
|
|
detail = 'Windows event {}'.format(event.eventid)
|
|
if source:
|
|
detail = '{} - {}'.format(detail, source)
|
|
rows.append({
|
|
'title': event_title(event.eventid),
|
|
'severity': event_severity(event.eventid),
|
|
'detail': detail,
|
|
'body': _event_body(event),
|
|
'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 - ', (command.status or 'queued').title(), who),
|
|
# A failed command's reason is the whole point of noticing it.
|
|
'body': (command.resultmessage or '').strip() or None,
|
|
'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)
|
|
|
|
# Trim to the cap without dropping the criticals that were fetched on
|
|
# purpose: they keep their place in time, but they survive the cut.
|
|
if len(rows) > 15:
|
|
keep = [r for r in rows if r['severity'] == 'critical'][:5]
|
|
rest = [r for r in rows if r not in keep]
|
|
rows = sorted(keep + rest[:15 - len(keep)],
|
|
key=lambda r: (r['eventtime'] is not None, r['eventtime']),
|
|
reverse=True)
|
|
|
|
# The list renderer reads a badge colour from a ROW FIELD, so without this
|
|
# every badge falls back to grey and the feed loses the one signal it is
|
|
# there to carry.
|
|
colours = {'critical': 'red', 'warning': 'yellow', 'info': 'gray'}
|
|
|
|
return success_response([{
|
|
'title': r['title'],
|
|
'severity': r['severity'],
|
|
'severitycolor': colours.get(r['severity'], 'gray'),
|
|
'detail': r['detail'],
|
|
'body': r['body'],
|
|
'eventtime': r['eventtime'].isoformat() + 'Z' if r['eventtime'] else None,
|
|
} for r in rows[:15]])
|
|
|
|
|
|
def _link_label(speedbps, fullduplex):
|
|
"""Render NIC link speed and duplex as one cell.
|
|
|
|
Null speed means the agent could not read link state OR the adapter is
|
|
disconnected - verified on a live box that a disconnected NIC returns empty
|
|
values. Either way we do not know, so it reads '-' rather than 0 Mb.
|
|
|
|
Half duplex is called out explicitly because that is the actionable case: a
|
|
NIC that negotiated half duplex on a switch port is a cable or port fault
|
|
that otherwise presents as "the machine is slow sometimes". Full duplex is
|
|
the normal case and gets no annotation, so the only thing standing out in
|
|
the column is the thing worth looking at.
|
|
"""
|
|
if not speedbps:
|
|
return '-'
|
|
|
|
if speedbps >= 1000000000:
|
|
speed = '{:g} Gb'.format(speedbps / 1000000000)
|
|
else:
|
|
speed = '{:g} Mb'.format(speedbps / 1000000)
|
|
|
|
if fullduplex is False:
|
|
return '{} half duplex'.format(speed)
|
|
return speed
|
|
|
|
|
|
@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',
|
|
'link': _link_label(entry.get('linkspeedbps'),
|
|
entry.get('fullduplex')),
|
|
})
|
|
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 _pin_holds(command, descriptor):
|
|
"""Does the descriptor still carry the digest this command was queued for?
|
|
|
|
True when nothing was pinned (actions that deliver no content), or when the
|
|
descriptor is missing entirely - a vanished script or version already sends
|
|
the command through with no descriptor so the agent fails closed, and that
|
|
path is not a content swap.
|
|
"""
|
|
if not command.pinnedsha256:
|
|
return True
|
|
if descriptor is None:
|
|
return True
|
|
return hmac.compare_digest(
|
|
str(descriptor.get('sha256') or ''), command.pinnedsha256)
|
|
|
|
|
|
def _fail_on_pin_drift(command, action):
|
|
"""Refuse a claim whose content changed after the operator approved it."""
|
|
command.status = EndpointCommand.STATUS_FAILED
|
|
command.resultmessage = (
|
|
'Refused: the {} content changed after this command was queued'
|
|
.format(action))
|
|
command.completedat = _naive_utcnow()
|
|
db.session.commit()
|
|
current_app.logger.warning(
|
|
'Endpoint command %s (%s, asset %s) refused: pinned sha256 %s no '
|
|
'longer matches the resolved descriptor',
|
|
command.endpointcommandid, action, command.assetid,
|
|
command.pinnedsha256)
|
|
AuditLog.log('refused', 'EndpointCommand',
|
|
entityid=command.endpointcommandid,
|
|
entityname=action,
|
|
details='content digest changed after enqueue')
|
|
db.session.commit()
|
|
return ('', 204)
|
|
|
|
|
|
def _reap_stale_commands(asset_id):
|
|
"""Expire in-flight commands an agent will never finish for this endpoint.
|
|
|
|
The queue is at-most-one-in-flight: a single PENDING or RUNNING row blocks
|
|
every new command. If the agent is killed, uninstalled, or goes offline
|
|
after a command is claimed (RUNNING) or before it claims one (PENDING), that
|
|
row would pin the queue forever and every operator action returns 409. Reap
|
|
rows past their staleness ceiling to STATUS_EXPIRED so the queue self-heals.
|
|
Caller commits (this runs inside the enqueue transaction).
|
|
"""
|
|
now = _naive_utcnow()
|
|
running_cutoff = now - timedelta(seconds=EndpointCommand.RUNNING_STALE_SECONDS)
|
|
pending_cutoff = now - timedelta(seconds=EndpointCommand.PENDING_STALE_SECONDS)
|
|
|
|
stale = (EndpointCommand.query
|
|
.filter(EndpointCommand.assetid == asset_id,
|
|
db.or_(
|
|
db.and_(
|
|
EndpointCommand.status == EndpointCommand.STATUS_RUNNING,
|
|
EndpointCommand.claimedat.isnot(None),
|
|
EndpointCommand.claimedat < running_cutoff),
|
|
db.and_(
|
|
EndpointCommand.status == EndpointCommand.STATUS_PENDING,
|
|
EndpointCommand.createdat < pending_cutoff)))
|
|
.all())
|
|
for command in stale:
|
|
command.status = EndpointCommand.STATUS_EXPIRED
|
|
command.completedat = now
|
|
if not command.resultmessage:
|
|
command.resultmessage = (
|
|
'Expired: agent did not report a result before the staleness '
|
|
'ceiling (agent offline, killed, or reinstalled mid-command).')
|
|
return stale
|
|
|
|
|
|
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
|
|
|
|
|
|
_HOSTED_PACKAGE_PREFIX = '/api/applications/package/'
|
|
|
|
|
|
def _hosted_package_filename(application):
|
|
"""Return the stored filename for a ShopDB-hosted installer, or None.
|
|
|
|
installpath is the core package URL (/api/applications/package/<filename>);
|
|
a site-typed share path (not our upload) has no servable file here.
|
|
"""
|
|
installpath = application.installpath or ''
|
|
if not installpath.startswith(_HOSTED_PACKAGE_PREFIX):
|
|
return None
|
|
return installpath[len(_HOSTED_PACKAGE_PREFIX):]
|
|
|
|
|
|
def _backfill_app_installer_sha256(application):
|
|
"""Compute and persist installersha256 for a pre-migration package.
|
|
|
|
Packages uploaded before the 7d36 migration have a null installersha256,
|
|
which used to force a re-upload (Bug B). When the hosted package file is
|
|
present on disk, hash it once, stamp the app, and commit so future resolves
|
|
hit the stored value. Returns the digest, or None when no servable file.
|
|
"""
|
|
filename = _hosted_package_filename(application)
|
|
if not filename:
|
|
return None
|
|
package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
|
|
package_path = os.path.join(package_dir, filename)
|
|
if not os.path.isfile(package_path):
|
|
return None
|
|
digest = hashlib.sha256()
|
|
with open(package_path, 'rb') as package_file:
|
|
for chunk in iter(lambda: package_file.read(65536), b''):
|
|
digest.update(chunk)
|
|
application.installersha256 = digest.hexdigest()
|
|
db.session.commit()
|
|
return application.installersha256
|
|
|
|
|
|
def _resolve_update_app(target):
|
|
"""Resolve an update_app target to a delivery descriptor.
|
|
|
|
Accepts either an app-level or a version-level source, so a push works
|
|
whether or not the site tracks AppVersion rows (the applications UI stores a
|
|
single app-level installer; versioning is optional):
|
|
|
|
- "app:<appid>" - the app's own uploaded installer (installersha256)
|
|
- "version:<id>" - a specific AppVersion's pinned package
|
|
- bare "<int>" - legacy: treated as an appversionid
|
|
|
|
Returns (descriptor, error_message). The descriptor carries what the agent
|
|
needs to fetch and run the pinned installer: a collector-authenticated
|
|
download URL, its SHA-256, and the silent-install switches. The agent
|
|
fetches the bytes with its existing collector credential; no bytes here.
|
|
"""
|
|
kind, _sep, raw = str(target or '').partition(':')
|
|
if not raw:
|
|
# No prefix: legacy bare appversionid.
|
|
kind, raw = 'version', kind
|
|
|
|
try:
|
|
ident = int(raw)
|
|
except (TypeError, ValueError):
|
|
return None, 'update_app target must be app:<id>, version:<id>, or an appversionid'
|
|
|
|
if kind == 'app':
|
|
application = db.session.get(Application, ident)
|
|
if application is None:
|
|
return None, f'Unknown appid: {ident}'
|
|
if not application.installpath:
|
|
return None, (f'Application {ident} has no uploaded installer package')
|
|
# Backfill installersha256 for packages uploaded before the 7d36 migration
|
|
# so app-level delivery resolves without a re-upload (Bug B).
|
|
sha256 = application.installersha256 or _backfill_app_installer_sha256(application)
|
|
if not sha256:
|
|
return None, (f'Application {ident} has no uploaded installer package')
|
|
# Carry the real installer filename (with extension) on the delivery URL so
|
|
# the agent's Path.GetExtension routes MSI through msiexec (Bug A rc2). The
|
|
# trailing filename is cosmetic for the agent; the app_id drives the lookup.
|
|
filename = _hosted_package_filename(application)
|
|
installpath = '/api/endpoints/apppackage/app/{}'.format(application.appid)
|
|
if filename:
|
|
installpath = '{}/{}'.format(installpath, filename)
|
|
return {
|
|
'source': 'app',
|
|
'appid': application.appid,
|
|
'appversionid': None,
|
|
'appname': application.appname,
|
|
'version': None,
|
|
'installpath': installpath,
|
|
'sha256': sha256,
|
|
'silentargs': None,
|
|
}, None
|
|
|
|
if kind == 'version':
|
|
version = db.session.get(AppVersion, ident)
|
|
if version is None:
|
|
return None, f'Unknown appversionid: {ident}'
|
|
application = db.session.get(Application, version.appid)
|
|
if application is None or not application.installpath:
|
|
return None, ('Application for appversionid {} has no installer package'
|
|
.format(ident))
|
|
# PIN THE FILE THAT IS ACTUALLY SERVED.
|
|
#
|
|
# /api/endpoints/apppackage/<appversionid> resolves the version to its
|
|
# APPLICATION and serves application.installpath - there is no
|
|
# version-specific artifact behind that route. So version.sha256 pinned
|
|
# a digest that need not describe the bytes delivered, and where it was
|
|
# null the descriptor carried no digest at all and the agent had nothing
|
|
# to verify against. Deriving it from the application, exactly as the
|
|
# app branch does, makes the pin describe the delivery.
|
|
sha256 = (application.installersha256
|
|
or _backfill_app_installer_sha256(application))
|
|
if not sha256:
|
|
return None, ('Application for appversionid {} has an installer '
|
|
'package with no recorded SHA-256, so the delivery '
|
|
'cannot be verified'.format(ident))
|
|
|
|
# Carry the real filename so the agent's Path.GetExtension routes an MSI
|
|
# through msiexec. Without it every MSI push failed, because this branch
|
|
# ended the URL at the id and the extension was simply absent.
|
|
filename = _hosted_package_filename(application)
|
|
installpath = '/api/endpoints/apppackage/{}'.format(version.appversionid)
|
|
if filename:
|
|
installpath = '/api/endpoints/apppackage/app/{}/{}'.format(
|
|
application.appid, filename)
|
|
|
|
return {
|
|
'source': 'version',
|
|
'appversionid': version.appversionid,
|
|
'appid': version.appid,
|
|
'appname': application.appname,
|
|
'version': version.version,
|
|
'installpath': installpath,
|
|
'sha256': sha256,
|
|
'silentargs': version.silentargs,
|
|
}, None
|
|
|
|
return None, f'Unknown update_app target kind: {kind!r}'
|
|
|
|
|
|
def _resolve_run_script(target):
|
|
"""Resolve a run_script target to a script delivery descriptor.
|
|
|
|
Accepts either "script:<scriptid>" or a bare script name. The descriptor
|
|
carries a collector-authenticated download URL, the body's SHA-256, and the
|
|
per-script run context (system or user). The agent fetches the body with
|
|
its collector credential and runs it; the body never crosses the wire.
|
|
"""
|
|
kind, _sep, raw = str(target or '').partition(':')
|
|
script = None
|
|
if kind == 'script' and raw:
|
|
try:
|
|
script = db.session.get(Script, int(raw))
|
|
except (TypeError, ValueError):
|
|
return None, 'run_script target script:<id> must be numeric'
|
|
else:
|
|
# Bare name (the whole target, prefix included when there was no colon).
|
|
name = (target or '').strip()
|
|
if name:
|
|
script = Script.query.filter_by(scriptname=name).first()
|
|
|
|
if script is None:
|
|
return None, f'Unknown script: {target!r}'
|
|
if not script.filename or not script.sha256:
|
|
return None, f'Script {script.scriptname!r} has no uploaded body'
|
|
if not script.isactive:
|
|
return None, f'Script {script.scriptname!r} is inactive'
|
|
|
|
return {
|
|
'scriptid': script.scriptid,
|
|
'scriptname': script.scriptname,
|
|
'installpath': '/api/endpoints/scriptbody/{}'.format(script.scriptid),
|
|
'sha256': script.sha256,
|
|
'runcontext': script.runcontext,
|
|
}, 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.
|
|
# caller_holds, not current_user.haspermission: the question is what THIS
|
|
# caller may do, and a scoped personal token is a narrower caller than its
|
|
# owner. Asking the owner let a token scoped to endpoints.command queue a
|
|
# remote restart, a script run and an app deploy - the exact bypass
|
|
# require_permission exists to prevent, reintroduced by checking by hand.
|
|
elevated = elevated_permission_for(action)
|
|
if elevated and not caller_holds(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')
|
|
|
|
# update_app pins a specific application version; verify it resolves to a
|
|
# real installer now so a broken push is rejected at enqueue rather than
|
|
# failing opaquely on the endpoint minutes later.
|
|
# Digest of the content this push was APPROVED against. Kept on the row so
|
|
# the claim path can refuse a descriptor that has changed since; see
|
|
# EndpointCommand.pinnedsha256.
|
|
pinnedsha256 = None
|
|
|
|
if action == 'update_app':
|
|
descriptor, resolve_error = _resolve_update_app(target)
|
|
if resolve_error is not None:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, resolve_error)
|
|
pinnedsha256 = (descriptor or {}).get('sha256')
|
|
|
|
# run_script pins a ShopDB-hosted script; verify it resolves to a real
|
|
# uploaded body now so a broken push is rejected at enqueue rather than
|
|
# failing opaquely on the endpoint later.
|
|
if action == 'run_script':
|
|
script, script_error = _resolve_run_script(target)
|
|
if script_error is not None:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, script_error)
|
|
pinnedsha256 = (script or {}).get('sha256')
|
|
|
|
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)
|
|
|
|
# Self-heal a queue wedged by a command the agent will never finish (killed,
|
|
# uninstalled, or offline mid-flight) before enforcing at-most-one-in-flight,
|
|
# so a dead in-flight row cannot permanently 409 every operator action.
|
|
_reap_stale_commands(asset_id)
|
|
|
|
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),
|
|
pinnedsha256=pinnedsha256,
|
|
)
|
|
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('/monitors/resolved', methods=['GET'])
|
|
def resolved_monitors_for_agent():
|
|
"""Agent watch list: the monitors this endpoint should report on.
|
|
|
|
Collector-token auth (not JWT), same as the command channel. The agent
|
|
calls this each telemetry cycle with its hostname and receives the process
|
|
and service monitors that site policy (and the shipped defaults) attach to
|
|
this endpoint's PC type, so a policy-attached monitor is enumerated and
|
|
reported without editing appsettings.json on the box. The shape mirrors the
|
|
agent's MonitorConfig (processes[]/services[] of {name, app}) so the agent
|
|
can bind it directly: name is the wire identity (process image or service
|
|
key), app is the operator label. An unknown hostname or a PC type with no
|
|
monitors returns empty lists, which the agent merges as "watch nothing extra".
|
|
"""
|
|
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)
|
|
|
|
processes = []
|
|
services = []
|
|
for monitor in _resolved_monitors_for_asset(asset_id):
|
|
kind = (monitor.kind or '').lower()
|
|
if kind == 'process':
|
|
wirename = (monitor.processname or '').strip()
|
|
bucket = processes
|
|
elif kind == 'service':
|
|
wirename = (monitor.servicename or '').strip()
|
|
bucket = services
|
|
else:
|
|
continue
|
|
if not wirename:
|
|
continue
|
|
entry = {
|
|
'name': wirename,
|
|
'app': (monitor.displayname or wirename),
|
|
}
|
|
# Only process monitors carry a launch path, and only when set. The
|
|
# agent never trusts it blindly: it validates the path against an
|
|
# approved-root allowlist before it becomes launchable (fail closed).
|
|
if kind == 'process':
|
|
launchpath = (monitor.launchpath or '').strip()
|
|
if launchpath:
|
|
entry['launchpath'] = launchpath
|
|
bucket.append(entry)
|
|
|
|
return success_response({'processes': processes, 'services': services})
|
|
|
|
|
|
@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()
|
|
claim = {
|
|
'endpointcommandid': command.endpointcommandid,
|
|
'action': command.action,
|
|
'target': command.target,
|
|
'targetkind': command.targetkind,
|
|
}
|
|
# update_app rides a resolved delivery descriptor: the agent gets
|
|
# the package URL, its pinned SHA-256, and the version's silent-
|
|
# install switches, and fetches the bytes with its collector
|
|
# credential. If the version was deleted between enqueue and claim,
|
|
# send the command through with no descriptor and let the agent
|
|
# fail closed rather than 500 the poll.
|
|
if command.action == 'update_app':
|
|
descriptor, _resolve_error = _resolve_update_app(command.target)
|
|
if not _pin_holds(command, descriptor):
|
|
return _fail_on_pin_drift(command, 'update_app')
|
|
if descriptor is not None:
|
|
claim['app'] = descriptor
|
|
# run_script rides a resolved script descriptor: the agent gets a
|
|
# collector-authenticated body URL, the pinned SHA-256, and the
|
|
# per-script run context. Fetch happens agent-side with the
|
|
# collector credential. If the script vanished between enqueue and
|
|
# claim, send the command through with no descriptor and let the
|
|
# agent fail closed rather than 500 the poll.
|
|
if command.action == 'run_script':
|
|
script_descriptor, _script_error = _resolve_run_script(command.target)
|
|
if not _pin_holds(command, script_descriptor):
|
|
return _fail_on_pin_drift(command, 'run_script')
|
|
if script_descriptor is not None:
|
|
claim['script'] = script_descriptor
|
|
return success_response(claim)
|
|
|
|
# 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)
|
|
|
|
# An agent may only finalize ITS OWN command. The collector key is shared
|
|
# fleet-wide, so without this any agent could post a result for any command
|
|
# id - marking an install succeeded on a machine that never ran it, which
|
|
# writes a false compliance record. Binding to the claimed hostname is a
|
|
# floor, not a fix: the credential still does not prove which endpoint is
|
|
# calling, and only a per-endpoint credential closes that.
|
|
# Optional because the shipped agent does not send it yet; enforced when it
|
|
# does, so an updated fleet gets the check without a flag day.
|
|
hostname = (request.args.get('hostname') or '').strip()
|
|
if hostname:
|
|
caller_asset_id = _resolve_asset_for_hostname(hostname)
|
|
if caller_asset_id != command.assetid:
|
|
return error_response(
|
|
ErrorCodes.FORBIDDEN,
|
|
'This command belongs to another endpoint', http_code=403)
|
|
|
|
# Only a claimed command can be finalized. Needs no agent change, and it
|
|
# closes two things the hostname check cannot: reporting on a command that
|
|
# was never claimed, and overwriting a terminal result - which would let a
|
|
# recorded failure be replaced by a success.
|
|
if command.status != EndpointCommand.STATUS_RUNNING:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'Command {} is {}, not running'.format(
|
|
command_id, command.status),
|
|
http_code=409)
|
|
|
|
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')
|
|
|
|
|
|
@endpoints_bp.route('/apppackage/<int:appversionid>', methods=['GET'])
|
|
def download_app_package(appversionid: int):
|
|
"""Agent-facing installer download (collector-token auth).
|
|
|
|
The core /api/applications/package route is gated on applications.view, a
|
|
permission the endpoint agent's collector token does not hold. This route
|
|
authorizes on the same endpoints collector token the agent already uses for
|
|
telemetry and command polling, so the whole app-delivery path rides one key.
|
|
Serves the exact bytes the update_app descriptor pinned by SHA-256.
|
|
"""
|
|
from shopdb.api import send_upload
|
|
|
|
error = _collector_auth_or_error()
|
|
if error is not None:
|
|
return error
|
|
|
|
version = db.session.get(AppVersion, appversionid)
|
|
if version is None:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
f'Unknown appversionid: {appversionid}', http_code=404)
|
|
application = db.session.get(Application, version.appid)
|
|
if application is None or not application.installpath:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No installer package for this application',
|
|
http_code=404)
|
|
|
|
# installpath is the core package URL (/api/applications/package/<filename>);
|
|
# take the stored filename and serve it from the core package directory. A
|
|
# site-typed share path (not our upload) has no servable file here.
|
|
prefix = '/api/applications/package/'
|
|
if not application.installpath.startswith(prefix):
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'Application installer is not a ShopDB-hosted package',
|
|
http_code=404)
|
|
filename = application.installpath[len(prefix):]
|
|
package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
|
|
return send_upload(package_dir, filename, as_attachment=True)
|
|
|
|
|
|
@endpoints_bp.route('/apppackage/app/<int:app_id>', methods=['GET'])
|
|
@endpoints_bp.route('/apppackage/app/<int:app_id>/<path:filename>', methods=['GET'])
|
|
def download_app_package_by_app(app_id: int, filename: str = None):
|
|
"""Agent-facing installer download by APP (collector-token auth).
|
|
|
|
The app-level counterpart to download_app_package: serves the application's
|
|
own uploaded installer (installpath + installersha256) so update_app can
|
|
resolve an install source by app alone, with no AppVersion row. Same
|
|
collector-token auth and same package directory as the per-version route.
|
|
|
|
The optional trailing <filename> segment lets the delivery URL carry the
|
|
real installer extension (so the agent routes MSI through msiexec); it is
|
|
validated against the stored filename and does not drive the lookup.
|
|
"""
|
|
from shopdb.api import send_upload
|
|
|
|
error = _collector_auth_or_error()
|
|
if error is not None:
|
|
return error
|
|
|
|
application = db.session.get(Application, app_id)
|
|
if application is None or not application.installpath:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No installer package for this application',
|
|
http_code=404)
|
|
prefix = '/api/applications/package/'
|
|
if not application.installpath.startswith(prefix):
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'Application installer is not a ShopDB-hosted package',
|
|
http_code=404)
|
|
stored_filename = application.installpath[len(prefix):]
|
|
if filename is not None and filename != stored_filename:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'Requested installer filename does not match this application',
|
|
http_code=404)
|
|
package_dir = os.path.join(current_app.instance_path, 'applicationpackages')
|
|
return send_upload(package_dir, stored_filename, as_attachment=True)
|
|
|
|
|
|
@endpoints_bp.route('/scriptbody/<int:script_id>', methods=['GET'])
|
|
def download_script_body(script_id: int):
|
|
"""Agent-facing script body download (collector-token auth).
|
|
|
|
The core /api/scripts/body route is gated on applications.view, which the
|
|
endpoint agent's collector token does not hold. This route authorizes on
|
|
the same endpoints collector token the agent already uses, keeping the
|
|
whole script-delivery path on one key. Serves the exact bytes the
|
|
run_script descriptor pinned by SHA-256.
|
|
"""
|
|
from shopdb.api import send_upload
|
|
|
|
error = _collector_auth_or_error()
|
|
if error is not None:
|
|
return error
|
|
|
|
script = db.session.get(Script, script_id)
|
|
if script is None or not script.filename:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No body for this script', http_code=404)
|
|
script_dir = os.path.join(current_app.instance_path, 'scripts')
|
|
return send_upload(script_dir, script.filename, as_attachment=True)
|
|
|
|
|
|
# Operator issue reports (tray screenshot + free-text notes)
|
|
# ---------------------------------------------------------------------------
|
|
# Retention groundwork: how many days an operator report (and its screenshot
|
|
# file) is kept before the purge sweep removes it. Read from the Setting store
|
|
# so Agent Configuration can expose a toggle later without a code change; the
|
|
# default is 7 days. A value of 0 or less means "keep forever" (purge disabled).
|
|
OPERATOR_REPORT_RETENTION_SETTING = 'endpoints.operatorreport.retentiondays'
|
|
OPERATOR_REPORT_RETENTION_DEFAULT_DAYS = 7
|
|
|
|
# Screenshots are small PNGs from a virtual-screen grab. Cap the upload so a
|
|
# malformed or hostile agent cannot fill the disk with one request.
|
|
OPERATOR_REPORT_MAX_IMAGE_BYTES = 10 * 1024 * 1024
|
|
# The multipart envelope carries the other form fields and the part boundaries
|
|
# too, so content_length is legitimately larger than the image. Allow for that
|
|
# rather than rejecting a screenshot that is just under the cap.
|
|
_OPERATOR_REPORT_FORM_SLACK = 64 * 1024
|
|
# PNG signature. The route stores and serves the file as image/png, so this is
|
|
# the one thing it accepts.
|
|
_PNG_MAGIC = b'\x89PNG\r\n\x1a\n'
|
|
|
|
|
|
def _operator_report_dir():
|
|
return os.path.join(current_app.instance_path, 'operatorreports')
|
|
|
|
|
|
def _operator_report_retention_days():
|
|
"""Configured retention window in whole days (default 7, <=0 disables)."""
|
|
raw = Setting.get(OPERATOR_REPORT_RETENTION_SETTING)
|
|
if raw is None:
|
|
return OPERATOR_REPORT_RETENTION_DEFAULT_DAYS
|
|
try:
|
|
return int(raw)
|
|
except (TypeError, ValueError):
|
|
return OPERATOR_REPORT_RETENTION_DEFAULT_DAYS
|
|
|
|
|
|
def _purge_expired_operator_reports():
|
|
"""Delete operator reports older than the retention window, files and all.
|
|
|
|
Best-effort and idempotent: called opportunistically on upload so the feed
|
|
stays bounded without a scheduler. A retention of <=0 disables the sweep.
|
|
"""
|
|
days = _operator_report_retention_days()
|
|
if days <= 0:
|
|
return
|
|
|
|
cutoff = _naive_utcnow() - timedelta(days=days)
|
|
stale = OperatorReport.query.filter(OperatorReport.createdat < cutoff).all()
|
|
if not stale:
|
|
return
|
|
|
|
report_dir = _operator_report_dir()
|
|
for report in stale:
|
|
if report.imagefilename:
|
|
path = os.path.join(report_dir, report.imagefilename)
|
|
try:
|
|
os.remove(path)
|
|
except OSError:
|
|
pass
|
|
db.session.delete(report)
|
|
db.session.commit()
|
|
|
|
|
|
@endpoints_bp.route('/operatorreports', methods=['POST'])
|
|
def upload_operator_report():
|
|
"""Agent uploads one operator issue report (collector-token auth).
|
|
|
|
The tray captures the operator's free-text note and an optional screenshot,
|
|
then hands them to the service, which posts here as multipart/form-data on
|
|
the same endpoints collector token used for telemetry. The reported hostname
|
|
resolves to an asset; an unresolved hostname is a validation error. The
|
|
screenshot bytes are written to the operator-reports uploads dir and the row
|
|
holds only metadata plus the stored filename. One retention sweep runs
|
|
opportunistically so the feed stays bounded.
|
|
"""
|
|
import uuid
|
|
|
|
error = _collector_auth_or_error()
|
|
if error is not None:
|
|
return error
|
|
|
|
# Refuse an oversized body BEFORE anything parses it. content_length is a
|
|
# header, so it is readable without touching the body; request.form is not
|
|
# - the first access to it makes werkzeug parse the whole multipart and
|
|
# spool the file part to a temp file. An earlier version of this check sat
|
|
# below that line and so ran after the thing it exists to prevent. The
|
|
# value is caller-supplied, which is why this is an early out and not the
|
|
# enforcement; the bounded read below is.
|
|
declared = request.content_length or 0
|
|
if declared > OPERATOR_REPORT_MAX_IMAGE_BYTES + _OPERATOR_REPORT_FORM_SLACK:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'screenshot exceeds the {} byte limit'.format(
|
|
OPERATOR_REPORT_MAX_IMAGE_BYTES))
|
|
|
|
hostname = (request.form.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.VALIDATION_ERROR,
|
|
'could not resolve hostname to a computer: {!r}'.format(hostname))
|
|
|
|
reason = (request.form.get('reason') or 'Other (free text)').strip()[:128]
|
|
details = (request.form.get('details') or '').strip() or None
|
|
|
|
imagefilename = None
|
|
imagemimetype = None
|
|
imagebytesize = None
|
|
|
|
upload = request.files.get('screenshot')
|
|
if upload is not None and upload.filename:
|
|
# Read one byte past the cap: enough to know it was exceeded, without
|
|
# taking the whole of a body that lied about its length.
|
|
data = upload.read(OPERATOR_REPORT_MAX_IMAGE_BYTES + 1)
|
|
if len(data) > OPERATOR_REPORT_MAX_IMAGE_BYTES:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'screenshot exceeds the {} byte limit'.format(
|
|
OPERATOR_REPORT_MAX_IMAGE_BYTES))
|
|
if data and not data.startswith(_PNG_MAGIC):
|
|
# The stored file is served back as image/png. Storing whatever
|
|
# arrived under that name meant the extension and the bytes could
|
|
# disagree; the upload headers make the document case inert, but
|
|
# nothing is gained by keeping a file we already know is not what it
|
|
# claims.
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'screenshot must be a PNG')
|
|
if data:
|
|
imagefilename = 'report-{}.png'.format(uuid.uuid4().hex)
|
|
imagemimetype = 'image/png'
|
|
imagebytesize = len(data)
|
|
report_dir = _operator_report_dir()
|
|
os.makedirs(report_dir, exist_ok=True)
|
|
with open(os.path.join(report_dir, imagefilename), 'wb') as handle:
|
|
handle.write(data)
|
|
|
|
report = OperatorReport(
|
|
assetid=asset_id,
|
|
reason=reason,
|
|
details=details,
|
|
imagefilename=imagefilename,
|
|
imagemimetype=imagemimetype,
|
|
imagebytesize=imagebytesize,
|
|
reportedhostname=hostname,
|
|
)
|
|
db.session.add(report)
|
|
db.session.commit()
|
|
|
|
# Keep the feed bounded without a scheduler. Best-effort: a purge failure
|
|
# must not fail the upload the operator already completed.
|
|
try:
|
|
_purge_expired_operator_reports()
|
|
except Exception:
|
|
db.session.rollback()
|
|
|
|
return success_response(report.to_dict(),
|
|
message='Operator report received', http_code=201)
|
|
|
|
|
|
@endpoints_bp.route('/panel/<int:asset_id>/operatorreports', methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def panel_operator_reports(asset_id: int):
|
|
"""Operator Reports panel: this PC's recent operator issue reports.
|
|
|
|
Newest first. Each row carries a hasimage flag so the viewer knows whether
|
|
to request the screenshot from the serve route below. Runs a retention
|
|
sweep opportunistically so a viewed PC does not accumulate stale rows.
|
|
"""
|
|
try:
|
|
_purge_expired_operator_reports()
|
|
except Exception:
|
|
db.session.rollback()
|
|
|
|
reports = (OperatorReport.query
|
|
.filter_by(assetid=asset_id)
|
|
.order_by(OperatorReport.createdat.desc())
|
|
.all())
|
|
return success_response({'reports': [r.to_dict() for r in reports]})
|
|
|
|
|
|
@endpoints_bp.route('/operatorreports/<int:operatorreportid>/image',
|
|
methods=['GET'])
|
|
@require_permission('endpoints.view')
|
|
def serve_operator_report_image(operatorreportid: int):
|
|
"""Serve one operator report's screenshot (user JWT + endpoints.view)."""
|
|
from shopdb.api import send_upload
|
|
|
|
report = db.session.get(OperatorReport, operatorreportid)
|
|
if report is None or not report.imagefilename:
|
|
return error_response(ErrorCodes.NOT_FOUND,
|
|
'No screenshot for this report', http_code=404)
|
|
return send_upload(_operator_report_dir(), report.imagefilename)
|
|
|
|
|