ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e8b45bf15d9e391257df23d926a03253bad6957d
shopdb-flask/plugins/endpoints/api/routes.py
Proudlock e8b45bf15d Merge wilm fleet alert scope fixes into wj 
Brings in wilm commits:
- Make alertscope actually target fleet alerts
- Fix fleet alert server surface inherited from wj

Key changes merged:
- Added _alert_targets_endpoint() and _endpoint_scope_facts() for proper
  scope filtering (hostname, assettype, computertype, location, businessunit)
- Alertscope now properly filters which endpoints receive fleet alerts
- Added comprehensive test coverage in test_endpoints_fleet_alerts.py
- Preserved wj's operator report email enhancements (FQDN, machine relationship)

Resolved conflicts:
- plugins/endpoints/api/routes.py: merged scope filtering with operator report changes
- agent csproj: bumped to 1.14.2 combining both branches' changes
2026-09-09 18:42:09 -04:00

3480 lines
147 KiB
Python
Raw Blame History

"""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, caller_is_role, mint_service_token,
list_service_tokens,
revoke_service_tokens,
AuditLog, audit_log, collector_authorized, User,
Application, AppVersion, Script,
)
from ..models import (
EndpointState, EndpointEvent, EndpointCommand, EndpointApplication,
EndpointHealthState, OperatorReport,
)
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for, PLACE_POLICY_HOLD, ACTIVE_POLICY_HOLDS, HOLDABLE_ACTIONS_FOR, RELEASE_POLICY_HOLD, ACTIVE_LOCK_FOR, PLACE_ENDPOINT_LOCK, RELEASE_ENDPOINT_LOCK
from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title
from shopdb.api import send_alert
from markupsafe import escape
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, Asset.name,
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),
Asset.name.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)
states = {state.assetid: state for state, *_ in items}
computertypes = {state.assetid: computertypeid
for state, _computerid, computertypeid, *_ in items}
health = _health_for_assets(states, computertypes)
data = []
for (state, computerid, computertypeid, computertype, assetnumber, alias,
rank) in items:
row = state.to_dict()
row['computerid'] = computerid
row['assetnumber'] = assetnumber
# Named 'alias' rather than 'name' because the fleet row already
# carries a hostname; 'name' next to 'hostname' reads as a duplicate.
row['alias'] = alias
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.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 _health_for_assets(states, computertypes):
"""{assetid: {'level', 'label'}} for a page, from the cached rollup.
ONE query for the page instead of the old per-page critical-event
aggregate, and no ranking work here - the verdict was decided by
services/health.py in the same commit as the telemetry it came from.
Endpoints with no cached row have not checked in since the rollup shipped.
Those are evaluated read-only for display rather than left blank, because a
blank Health column reads as "fine" and not doing that is this page's whole
job. Nothing is persisted on a GET: writing here would invent a firstseenat
no agent ever observed, and would make a page load look like an
observation.
"""
if not states:
return {}
from ..services.health import _verdict
assetids = list(states)
cached = {row.assetid: row for row in
EndpointHealthState.query
.filter(EndpointHealthState.assetid.in_(assetids))
.all()}
now = _naive_utcnow()
out = {}
for assetid, state in states.items():
row = cached.get(assetid)
if row is not None:
out[assetid] = {'level': row.level, 'label': row.label}
continue
verdict = _verdict(state, computertypes.get(assetid), assetid, now)
out[assetid] = {'level': verdict['level'], 'label': verdict['label']}
return out
@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 []
def _dell_updates_summary(state):
"""Parse the cached Dell Command | Update summary off an EndpointState.
dellupdatesjson is written on check-in from the shopdb-dellupdates agent
utility, which reads the log the \\ShopDB-DellCommandScan task produces.
Null means the endpoint has never reported a scan. Tolerant of a malformed
blob so a bad cache never breaks the panel.
"""
raw = getattr(state, 'dellupdatesjson', None) if state else None
if not raw:
return None
try:
parsed = json.loads(raw)
except (TypeError, ValueError):
return None
return parsed if isinstance(parsed, dict) else None
def _dell_updates_value(state):
"""Render the informational 'Pending Dell updates' status value.
Informational only - this row never claims an endpoint needs attention,
because a pending optional driver is not a support problem. Three states are
kept distinct on purpose, and all three are VISIBLE:
never scanned -> 'Never scanned'. Deliberately not 'No': those are
different facts, and claiming 'No' would tell an operator
the machine is current when nobody has ever checked. It
is also not hidden - a missing row reads as a broken
panel, so the unknown state says so out loud.
scanned clean -> 'No'.
pending -> 'Yes (n)', with the count because one optional update and
fourteen urgent ones warrant different urgency.
"""
summary = _dell_updates_summary(state)
if not summary:
return 'Never scanned'
status = summary.get('status')
if status == 'error':
# The scan log exists but could not be read. A different problem from
# never having scanned, and worth saying so rather than implying the
# endpoint was never checked.
return 'Unknown (scan unreadable)'
if status != 'ok':
return 'Never scanned'
try:
count = int(summary.get('count') or 0)
except (TypeError, ValueError):
return 'Unknown (scan unreadable)'
if count <= 0:
return 'No'
return 'Yes ({})'.format(count)
@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, fmt=None):
entry = {'label': label, 'value': value, 'mono': mono}
if fmt:
entry['format'] = fmt
return entry
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,
fmt='datetime'),
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,
fmt='datetime'),
]
fields = [f for f in fields if f['value'] not in (None, '')]
# Informational only, and appended after the null-filter above because it
# always renders - including an explicit "Never scanned", which a hidden row
# would leave looking like a broken panel.
fields.append(field('Pending Dell updates', _dell_updates_value(state)))
# 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)
# Columns for the two cards carved out of the old hand-written Network card on
# the Details tab. That card listed current addresses with a Primary badge and
# nothing else, which undersold what the rows actually hold.
_IPHISTORY_COLUMNS = [
{'key': 'ipaddress', 'label': 'IP address'},
{'key': 'macaddress', 'label': 'MAC'},
{'key': 'status', 'label': 'Status'},
{'key': 'firstseen', 'label': 'First seen'},
{'key': 'lastchanged', 'label': 'Last changed'},
]
def _isostamp(value):
"""ISO-8601 with a Z suffix, matching every other timestamp this file emits.
Formatting is left to the client so it can render in the viewer's locale
and timezone; returning a preformatted string here would pin every reader
to the server's.
"""
return value.isoformat() + 'Z' if value else None
def _communication_rows(asset_id: int):
"""Every Communication row for an asset, newest change first.
These rows are an ACCIDENTAL history rather than a designed one.
_record_interfaces in the computers plugin matches an incoming NIC on MAC
then on IP and inserts when neither matches, and it never deletes - so an
address the machine has not held for a year is still sitting here next to
the current one. That is genuinely useful (it answers "what was this box on
before?"), so it is surfaced as history rather than cleaned away.
"""
from shopdb.api import Communication
# MySQL has no NULLS LAST, so sort on an explicit "is null" key first.
# Rows that never got a modifieddate sink to the bottom on every dialect.
return (Communication.query
.filter_by(assetid=asset_id)
.order_by(Communication.modifieddate.is_(None),
Communication.modifieddate.desc(),
Communication.communicationid.desc())
.all())
def _iphistory_status(comm):
"""What this row IS, in the vocabulary somebody troubleshooting would use.
Primary is the address the collector last picked as this box's main one.
APIPA is called out by name because a 169.254.x.x address is not an address
the machine was given - it is what Windows assigns when DHCP failed, so its
presence in the list is a record of a past network fault, not a past
subnet.
"""
address = (comm.ipaddress or '').strip()
if address.startswith('169.254.'):
return 'APIPA (no DHCP)'
if comm.isprimary:
return 'Primary'
if comm.ismachinenetwork:
return 'Machine network'
return 'Previous'
@endpoints_bp.route('/asset/<int:asset_id>/iphistory', methods=['GET'])
@require_permission('endpoints.view')
def panel_iphistory(asset_id: int):
"""Table panel: every address this asset has held, newest change first.
Replaces the hand-written Network card that used to sit on the Details tab
and show only the current addresses. Dates come from the row's own
lifecycle columns, so they are labelled for what they are: 'Last changed'
is when a field on the row last moved, NOT when the address was last seen.
An address the machine still holds and reports unchanged does not update
modifieddate, so calling that column 'Last seen' would be a lie the data
cannot support.
"""
rows = []
for comm in _communication_rows(asset_id):
address = (comm.ipaddress or '').strip()
if not address:
continue
rows.append({
'ipaddress': address,
'macaddress': (comm.macaddress or '').strip() or '-',
'status': _iphistory_status(comm),
'firstseen': _isostamp(comm.createddate),
'lastchanged': _isostamp(comm.modifieddate),
})
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.
#
# 25s is the value this endpoint is designed around; it was temporarily dropped
# to 5s under the blocking gthread worker pool, where a held connection pinned
# one of a worker's 8 OS threads. Measured then: 14 agents at hold=25 left a
# single worker with 8 of 9 threads parked in timed sleep and UI requests hashed
# to that worker waited 12-24s. Under gevent a held connection is a greenlet and
# the per-worker ceiling is --worker-connections (2000), not 8 threads, so an
# idle hold no longer costs anything a UI request needs.
#
# Restored to 25 because hold length sets the fleet's REQUEST RATE, not its
# concurrency: the agent re-polls immediately on a 204, so hold=5 is 5x the
# requests of hold=25, and TLS handshake CPU was the measured burst bottleneck
# (devnotes/gevent-sizing-simulation.md). 25 also restores the margin against
# the agent's MinIdlePollDuration=5s floor, which CommandWorker.IsIdlePollClose
# uses to tell a normal hold-window close apart from a transient failure; at
# hold=5 that margin was ~20ms.
#
# Constraint: gunicorn --timeout must stay above this or a healthy long-poll is
# killed mid-hold. docker-compose.yml sets 60.
_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, the
per-script run context (system or user), and the optional folder to run in.
The agent fetches the body with its credential and runs it; the body never
crosses the wire.
workingdirectory is sent even when empty so the field is always present in
the claim: an agent that receives it and ignores it is then distinguishable
from one that never saw it, which matters while agents of both vintages are
in the field. errormode rides the same rule, and for the same reason.
"""
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,
'workingdirectory': script.workingdirectory or '',
'errormode': script.errormode or 'default',
}, 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>/policyholds', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_policy_holds(asset_id: int):
"""Active break-glass holds for one endpoint.
Gated by endpoints.view, not by the admin role that places a hold. A hold
suppresses reconciliation, so an operator watching a machine drift from its
policy needs to see that the drift is deliberate, who authorized it and when
it lapses. Hiding that from non-admins turns an audited override back into
unexplained behavior.
"""
return success_response(ACTIVE_POLICY_HOLDS(asset_id))
@endpoints_bp.route('/<int:asset_id>/holdableactions', methods=['GET'])
@require_permission('endpoints.view')
def list_endpoint_holdable_actions(asset_id: int):
"""Desired-mode actions on this endpoint, with current hold state.
Feeds the admin hold picker. Read-gated rather than admin-gated to match the
holds list: knowing WHICH actions this endpoint reconciles is the same class
of provenance as knowing which are currently held.
"""
return success_response(HOLDABLE_ACTIONS_FOR(asset_id))
@endpoints_bp.route('/<int:asset_id>/policyholds', methods=['POST'])
@require_permission('endpoints.command')
def place_endpoint_policy_hold(asset_id: int):
"""Deliberately suspend desired-state reconciliation for one or more actions.
The same authority as break-glass, reached directly instead of as a side
effect of a denied command: an admin who knows they are about to work on a
machine should not have to trip a policy denial first to stop the reconciler
fighting them.
Accepts `actions` (a list) or `action` (a single name, kept so existing
callers keep working). Every named action is validated BEFORE anything is
written, and one bad name rejects the whole request. Partial success on a
break-glass control is the wrong failure mode - "three of your five holds
landed" is worthless to someone about to take a machine apart, and the
missing two are discovered only when the reconciler undoes their work.
One shared window covers every action in the request: holding five actions
means doing one job on one machine, and that job has one duration.
Admin-only via caller_is_role for the same reason as the override branch - a
scoped personal token is a narrower caller than its owner, so a leaked agent
token must not be able to switch off policy enforcement.
"""
if not caller_is_role('admin'):
return error_response(ErrorCodes.FORBIDDEN,
'Placing a policy hold requires the admin role',
http_code=403)
payload = request.get_json(silent=True) or {}
reason = (payload.get('reason') or '').strip()
raw = payload.get('actions')
if raw is None:
raw = [payload.get('action')] if payload.get('action') else []
if not isinstance(raw, list):
return error_response(ErrorCodes.VALIDATION_ERROR,
'actions must be a list of action names')
actions = []
for entry in raw:
name = (entry or '').strip() if isinstance(entry, str) else ''
if name and name not in actions:
actions.append(name)
if not actions:
return error_response(ErrorCodes.VALIDATION_ERROR,
'Select at least one action to hold')
if not reason:
return error_response(ErrorCodes.VALIDATION_ERROR,
'A hold requires a reason explaining why')
for name in actions:
if not is_allowed_action(name):
return error_response(ErrorCodes.VALIDATION_ERROR,
f'Unknown or disallowed action: {name!r}')
# Only desired-mode actions are holdable. Accepting anything else would
# write a row that suppresses nothing, so the UI would report a hold that
# has no effect - worse than refusing. Checked for every action up front so
# the request is rejected whole rather than half-applied.
holdable = {entry['action'] for entry in HOLDABLE_ACTIONS_FOR(asset_id)}
unholdable = [name for name in actions if name not in holdable]
if unholdable:
return error_response(
ErrorCodes.VALIDATION_ERROR,
f'Not under desired-state enforcement on this endpoint, so there '
f'is nothing to hold: {", ".join(sorted(unholdable))}')
minutes = payload.get('minutes')
try:
minutes = int(minutes) if minutes is not None else None
except (TypeError, ValueError):
return error_response(ErrorCodes.VALIDATION_ERROR,
'minutes must be a whole number')
# Zero is legal and means indefinite: the hold stands until released by
# hand. Negatives are not - they would place an already-expired hold, which
# reads as success in the UI while suppressing nothing.
if minutes is not None and minutes < 0:
return error_response(ErrorCodes.VALIDATION_ERROR,
'minutes cannot be negative (use 0 for an '
'indefinite hold)')
indefinite = minutes == 0
window = 'indefinite' if indefinite else minutes
audit_log(
action='policyholdplaced',
entitytype='Asset',
entityid=asset_id,
changes={
'actions': actions,
'reason': reason,
'minutes': window,
'indefinite': indefinite,
'placedby': getattr(current_user, 'username', None),
},
details=(f'Policy hold placed on {", ".join(actions)} '
f'({window}): {reason}'),
)
for name in actions:
placed = PLACE_POLICY_HOLD(
assetid=asset_id,
action=name,
reason=reason,
userid=getattr(current_user, 'userid', None),
username=getattr(current_user, 'username', None),
minutes=minutes,
)
if not placed:
return error_response(ErrorCodes.VALIDATION_ERROR,
'Policy holds are unavailable on this site')
return success_response(ACTIVE_POLICY_HOLDS(asset_id))
@endpoints_bp.route('/<int:asset_id>/policyholds/<action>', methods=['DELETE'])
@require_permission('endpoints.command')
def release_endpoint_policy_hold(asset_id: int, action: str):
"""End a hold early, resuming reconciliation of one action.
Admin-only on the same reasoning as placing one: releasing a hold turns
policy enforcement back on for a machine someone may still be working on,
which is the same authority in the opposite direction.
"""
if not caller_is_role('admin'):
return error_response(ErrorCodes.FORBIDDEN,
'Releasing a policy hold requires the admin role',
http_code=403)
released = RELEASE_POLICY_HOLD(asset_id, action)
if not released:
return error_response(ErrorCodes.NOT_FOUND,
f'No active hold on {action!r} for this endpoint',
http_code=404)
audit_log(
action='policyholdreleased',
entitytype='Asset',
entityid=asset_id,
changes={
'action': action,
'releasedby': getattr(current_user, 'username', None),
},
details=f'Policy hold released early on {action}',
)
return success_response(ACTIVE_POLICY_HOLDS(asset_id))
@endpoints_bp.route('/<int:asset_id>/lock', methods=['GET'])
@require_permission('endpoints.view')
def get_endpoint_lock(asset_id: int):
"""The endpoint lock standing on this endpoint, or null.
Read-gated rather than admin-gated, and that is the point of the feature: an
operator whose command was refused needs to see "locked by <admin>: mid
repair" and self-serve. Hiding the lock from non-admins would leave them
with a bare 403 and a ticket to file about a machine that is working fine.
"""
lock = ACTIVE_LOCK_FOR(asset_id)
return success_response(lock.to_dict() if lock else None)
@endpoints_bp.route('/<int:asset_id>/lock', methods=['POST'])
@require_permission('endpoints.command')
def place_endpoint_lock(asset_id: int):
"""Quarantine this endpoint against non-admin commands.
Admin-only via caller_is_role rather than current_user.hasrole, matching the
override and hold routes: a scoped personal token is a narrower caller than
its owner, so a leaked agent token must not be able to quarantine machines.
A reason is required. The refusal shown to a blocked operator quotes it
verbatim, so an empty one produces a dead end - the operator learns only
that they may not act, not who to ask or when it lifts.
"""
if not caller_is_role('admin'):
return error_response(ErrorCodes.FORBIDDEN,
'Locking an endpoint requires the admin role',
http_code=403)
payload = request.get_json(silent=True) or {}
reason = (payload.get('reason') or '').strip()
if not reason:
return error_response(ErrorCodes.VALIDATION_ERROR,
'A lock requires a reason explaining why')
lock = PLACE_ENDPOINT_LOCK(
assetid=asset_id,
reason=reason,
userid=getattr(current_user, 'userid', None),
username=getattr(current_user, 'username', None),
)
audit_log(
action='endpointlocked',
entitytype='Asset',
entityid=asset_id,
changes={
'reason': reason,
'lockedby': getattr(current_user, 'username', None),
},
details=f'Endpoint locked: {reason}',
)
return success_response(lock.to_dict())
@endpoints_bp.route('/<int:asset_id>/lock', methods=['DELETE'])
@require_permission('endpoints.command')
def release_endpoint_lock(asset_id: int):
"""Lift the quarantine on this endpoint.
Admin-only on the same reasoning as setting one: unlocking returns a machine
someone may still be physically working inside to the whole operator pool,
which is the same authority in the opposite direction.
"""
if not caller_is_role('admin'):
return error_response(ErrorCodes.FORBIDDEN,
'Unlocking an endpoint requires the admin role',
http_code=403)
lock = RELEASE_ENDPOINT_LOCK(
asset_id,
userid=getattr(current_user, 'userid', None),
username=getattr(current_user, 'username', None),
)
if lock is None:
return error_response(ErrorCodes.NOT_FOUND,
'No active lock on this endpoint',
http_code=404)
audit_log(
action='endpointunlocked',
entitytype='Asset',
entityid=asset_id,
changes={
'lockedby': lock.lockedbyusername,
'reason': lock.reason,
'releasedby': getattr(current_user, 'username', None),
},
details=f'Endpoint unlocked (was locked: {lock.reason})',
)
return success_response(None)
@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, refuses non-admin callers while an endpoint lock
stands, 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}')
# Endpoint lock: a quarantine on the CALLER axis, checked before the
# permission and policy gates below so the refusal can name the lock, its
# owner and its reason. Answering with a generic policy denial instead would
# send an operator hunting through Configuration Policy for a cause that is
# not there - the machine is fine, someone has simply reserved it.
#
# Admins pass straight through with no override argument and no reason. The
# lock gates WHO may act, not WHICH actions exist, so there is no policy
# decision here to break past; requiring break-glass would mean an admin had
# to override their own team's quarantine to do the repair it was set for.
lock = ACTIVE_LOCK_FOR(asset_id)
if lock is not None and not caller_is_role('admin'):
return error_response(
ErrorCodes.FORBIDDEN,
f'Endpoint locked by {lock.lockedbyusername or "a deleted user"}: '
f'{lock.reason}',
http_code=403)
# 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)
# Break-glass: an admin may push past a Configuration Policy denial, but
# only by asking for it explicitly and saying why. The override is scoped to
# POLICY gating alone - the elevated permission check above still stands, so
# this cannot conjure a script run for someone who lacks endpoints.script.
# It widens WHERE an action may run, never WHICH actions the caller may use.
override = bool(payload.get('policyoverride'))
overridereason = (payload.get('overridereason') or '').strip()
if action not in ALLOWED_ACTIONS_FOR(assetid=asset_id):
if not override:
return error_response(
ErrorCodes.VALIDATION_ERROR,
f'Action not permitted by policy for this endpoint: {action!r}')
# caller_is_role, not current_user.hasrole: a scoped personal token is a
# narrower caller than its owner, so a leaked agent token must not
# inherit break-glass from the admin who minted it.
if not caller_is_role('admin'):
return error_response(
ErrorCodes.FORBIDDEN,
'Overriding a policy denial requires the admin role',
http_code=403)
if not overridereason:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'An override requires overridereason explaining why')
# Audited before the command is enqueued: if the enqueue fails after
# this point, the record of the attempt still survives. A break-glass
# that is only logged on success hides exactly the cases worth reviewing.
# Structured context goes in changes (JSON-serialized by the model);
# details is a plain Text column, so it carries the readable summary.
audit_log(
action='policyoverridden',
entitytype='Asset',
entityid=asset_id,
changes={
'action': action,
'reason': overridereason,
'overriddenby': getattr(current_user, 'username', None),
},
details=f'Policy denial overridden for {action}: {overridereason}',
)
# A gate-mode override is a one-shot; a desired-mode override is a fight
# with the reconciler, which would re-enqueue the opposite command on the
# next check-in and make the override look broken. Hold it open.
PLACE_POLICY_HOLD(
assetid=asset_id,
action=action,
reason=overridereason,
userid=getattr(current_user, 'userid', None),
username=getattr(current_user, 'username', None),
)
# 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()
def _send_operator_report_alert(report, asset_id, hostname):
"""Send email alert for new operator issue report to default alert recipients."""
from flask import url_for, request
from shopdb.utils.mailer import get_smtp_config, _normalize_recipients
from shopdb.core.models.relationship import AssetRelationship
current_app.logger.info('_send_operator_report_alert called for report %d, asset %d, hostname %s',
report.operatorreportid, asset_id, hostname)
# Check if we have recipients before doing expensive work
config = get_smtp_config()
current_app.logger.info('SMTP config: enabled=%s, host=%s, alert_recipients=%s',
config.get('enabled'), config.get('host'), config.get('alert_recipients'))
recipients = _normalize_recipients(config['alert_recipients'])
if not recipients:
current_app.logger.warning('Skipping operator report alert: no alert recipients configured')
return
current_app.logger.info('Will send alert to %d recipient(s): %s', len(recipients), recipients)
asset = Asset.query.get(asset_id)
if not asset:
current_app.logger.warning('Cannot send operator report alert: asset %d not found', asset_id)
return
# Check if this PC has a relationship to a machine
machine_relationship = (
AssetRelationship.query
.filter_by(sourceassetid=asset_id)
.filter(AssetRelationship.label.like('%machine%'))
.first()
)
machine_info = None
if machine_relationship:
machine_asset = Asset.query.get(machine_relationship.targetassetid)
if machine_asset:
machine_info = {
'assetid': machine_asset.assetid,
'assetnumber': machine_asset.assetnumber,
'name': machine_asset.name
}
# Generate absolute URL to frontend page (not API endpoint)
try:
scheme = request.scheme if request else 'https'
host = request.host if request else 'localhost'
device_url = f'{scheme}://{host}/shopdb/pcs/{asset_id}'
except Exception as e:
current_app.logger.error('Failed to generate device URL: %s', e)
device_url = f'https://localhost/shopdb/pcs/{asset_id}'
reported_at = report.createdat.strftime('%Y-%m-%d %H:%M:%S UTC') if report.createdat else 'Unknown'
screenshot_info = ''
if report.imagefilename:
try:
screenshot_url = url_for('endpoints.serve_operator_report_image',
operatorreportid=report.operatorreportid,
_external=True)
screenshot_info = f'<p><strong>Screenshot:</strong> <a href="{escape(screenshot_url)}">View Screenshot</a></p>'
except Exception as e:
current_app.logger.warning('Failed to generate screenshot URL: %s', e)
# Get PC access domain to build FQDN
from shopdb.core.models import Setting
pc_access_domain_setting = Setting.query.filter_by(key='pc_access_domain').first()
pc_access_domain = pc_access_domain_setting.value if pc_access_domain_setting else ''
# Build full FQDN
if pc_access_domain and not hostname.endswith(pc_access_domain):
fqdn = f'{hostname}.{pc_access_domain}'
else:
fqdn = hostname
# HTML-escape all user-controlled values to prevent XSS
safe_fqdn = escape(fqdn)
safe_hostname = escape(hostname)
safe_name = escape(asset.name or hostname)
safe_assetnumber = escape(asset.assetnumber or 'N/A')
safe_reason = escape(report.reason)
safe_details = escape(report.details or 'No additional details provided.')
# Build machine info for subject and body
machine_subject_part = ''
machine_body_line = ''
if machine_info:
safe_machine_number = escape(machine_info['assetnumber'] or 'N/A')
scheme = request.scheme if request else 'https'
host = request.host if request else 'localhost'
machine_url = f'{scheme}://{host}/shopdb/machines/{machine_info["assetid"]}'
machine_subject_part = f' (Machine #{safe_machine_number})'
machine_body_line = f'<p style="margin: 0;"><strong>Machine#:</strong> <a href="{escape(machine_url)}">{safe_machine_number}</a></p>'
# Build HTML body (don't use render_email to avoid subject duplication)
html_body = f'''
<div style="font-family:Arial,Helvetica,sans-serif;max-width:640px;margin:0 auto;color:#222;">
<h2 style="color:#1a1a1a;margin:0 0 12px;">Operator Issue Report</h2>
<div style="padding: 16px; background: #f9f9f9; border-left: 4px solid #d93f3f; margin-bottom: 16px;">
<p style="margin: 0 0 8px; font-weight: bold; color: #d93f3f;">New Issue Report from Shop Floor</p>
<p style="margin: 0;"><strong>Device:</strong> <a href="{escape(device_url)}">{safe_fqdn}</a></p>
{machine_body_line}
<p style="margin: 0;"><strong>Reason:</strong> {safe_reason}</p>
<p style="margin: 0;"><strong>Reported:</strong> {escape(reported_at)}</p>
</div>
<p><strong>Details:</strong></p>
<p style="white-space: pre-wrap;">{safe_details}</p>
{screenshot_info}
<p><a href="{escape(device_url)}" style="display: inline-block; padding: 10px 16px; background: #1a73e8; color: white; text-decoration: none; border-radius: 4px;">View Device in ShopDB</a></p>
<hr style="border:none;border-top:1px solid #ddd;margin:24px 0 12px;">
<p style="font-size:12px;color:#888;margin:0;">Sent by ShopDB.</p>
</div>
'''
subject = f'Operator Issue Report: {safe_fqdn}{machine_subject_part}'
current_app.logger.info('Sending operator report alert for %s (report %d)',
hostname, report.operatorreportid)
send_alert(subject, html_body)
@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()
# Send email alert to default alert recipients
current_app.logger.info('Operator report %d saved, attempting to send email alert', report.operatorreportid)
try:
_send_operator_report_alert(report, asset_id, hostname)
current_app.logger.info('Email alert function completed for report %d', report.operatorreportid)
except Exception as e:
current_app.logger.error('Failed to send operator report alert email for report %d: %s', report.operatorreportid, e, exc_info=True)
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.operatorreports')
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.operatorreports')
def serve_operator_report_image(operatorreportid: int):
"""Serve one operator report's screenshot.
Requires a user JWT plus endpoints.operatorreports.
This route, not the panel, is what actually protects the image: the panel
only decides whether a tab is drawn, while the bytes are reachable by URL to
anyone who can guess an id. Screenshots are an uncontrolled capture of the
operator's screen, so this must never widen back to 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)
# =============================================================================
# Fleet alerts (agent-facing, collector-token auth)
# =============================================================================
_DEFAULT_ALERT_TYPE_COLOR = '#17a2b8'
# Scope keys an alert may be narrowed by.
# rather than treated as a match-nothing, so a typo cannot silently mute an
# alert - see _alert_targets_endpoint.
_ALERT_SCOPE_KEYS = ('hostname', 'assettype', 'computertype', 'location',
'businessunit')
def _endpoint_scope_facts(hostname):
"""The scope-able attributes of the endpoint asking for its alerts.
Returns a dict of lowercased strings keyed by _ALERT_SCOPE_KEYS, or None if
the hostname resolves to no computer. Values that are unset on the asset are
absent from the dict, which makes them unmatchable rather than empty-string
matchable.
"""
from plugins.computers.models import Computer, ComputerType
from shopdb.api import Location, AssetType
row = (db.session.query(Computer.hostname, AssetType.assettype,
ComputerType.computertype, Location.locationname,
Asset.businessunitid)
.join(Asset, Asset.assetid == Computer.assetid)
.outerjoin(ComputerType,
ComputerType.computertypeid == Computer.computertypeid)
.outerjoin(AssetType, AssetType.assettypeid == Asset.assettypeid)
.outerjoin(Location, Location.locationid == Asset.locationid)
.filter(db.func.lower(Computer.hostname) == hostname.lower())
.first())
if row is None:
return None
hostname_value, assettype, computertype, locationname, businessunitid = row
facts = {'hostname': hostname_value, 'assettype': assettype,
'computertype': computertype, 'location': locationname}
# BusinessUnit is not on the plugin contract surface (shopdb.api), so resolve
# its name by table rather than importing the model and breaking the import
# contract the way the mailer import did.
if businessunitid is not None:
name = db.session.execute(
db.text('SELECT businessunit FROM businessunits '
'WHERE businessunitid = :bid'),
{'bid': businessunitid}).scalar()
facts['businessunit'] = name
return {k: str(v).strip().lower() for k, v in facts.items()
if v is not None and str(v).strip()}
def _alert_targets_endpoint(alertscope, facts):
"""Does an alert's scope include this endpoint?
An absent, empty or unusable scope means the whole fleet - that is the
default and the historical behaviour. A scope with keys narrows the alert:
every recognised key must match, and a key may hold a single value or a list
of values, any of which matches. Matching is case-insensitive because these
are operator-typed names, not identifiers.
Unrecognised keys are IGNORED rather than failing the match. A scope is
authored by hand and a typo like {"hostnam": "BAY-01"} would otherwise mute
the alert everywhere and look like a delivery failure. Ignoring the key
sends the alert wider than intended, which is the safe direction for
something whose purpose is to warn people.
"""
if not isinstance(alertscope, dict) or not alertscope:
return True
for key in _ALERT_SCOPE_KEYS:
if key not in alertscope:
continue
wanted = alertscope[key]
if wanted is None:
continue
if isinstance(wanted, (list, tuple, set)):
candidates = {str(v).strip().lower() for v in wanted if v is not None}
else:
candidates = {str(wanted).strip().lower()}
candidates.discard('')
if not candidates:
continue
if facts.get(key) not in candidates:
return False
return True
@endpoints_bp.route('/alerts', methods=['GET'])
def get_fleet_alerts():
"""Fleet alerts for one endpoint by hostname (collector-token auth).
Called by the endpoint agent to fetch active, unacknowledged fleet alerts
for its hostname. Returns alerts with notification type info (typename and
typecolor from database) formatted for display in the agent tray.
The list is AUTHORITATIVE for this host: it is every alert that should be
live on the endpoint right now, so the agent retires any pending alert it
holds that is absent here. That is how a server-side close or cancel reaches
a machine already showing the alert - an alert deactivated or expired server
side drops out of this list and the agent takes the toast down, rather than
the operator being stuck acknowledging something already withdrawn.
Acknowledgement is terminal, so an acknowledged alert is filtered out here
permanently and a later server-side close is a no-op for that asset.
"""
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)
from plugins.notifications.models import Notification, NotificationType, NotificationAcknowledgement
now = _naive_utcnow()
# Query for active fleet alerts not acknowledged by this asset
query = (db.session.query(Notification, NotificationType)
.join(NotificationType,
NotificationType.notificationtypeid == Notification.notificationtypeid)
.outerjoin(NotificationAcknowledgement,
db.and_(
NotificationAcknowledgement.notificationid == Notification.notificationid,
NotificationAcknowledgement.assetid == asset_id))
.filter(Notification.alertfleet.is_(True),
Notification.isactive.is_(True),
db.or_(
Notification.endtime.is_(None),
Notification.endtime > now),
NotificationAcknowledgement.notificationacknowledgementid.is_(None)))
results = query.all()
# Resolved once per poll. An unknown hostname cannot happen here - the
# asset lookup above already returned - but guard anyway so a missing
# computer row means fleet-wide rather than a crash.
facts = _endpoint_scope_facts(hostname) or {}
alerts = []
for notification, ntype in results:
if not _alert_targets_endpoint(notification.alertscope, facts):
continue
title = notification.notification[:100] if notification.notification else ''
if '\n' in title:
title = title.split('\n')[0][:100] or '(no title)'
alerts.append({
'notificationId': notification.notificationid,
'type': ntype.typename,
'typeColor': ntype.typecolor or _DEFAULT_ALERT_TYPE_COLOR,
'title': title,
'message': notification.notification or '',
'timeout': notification.alerttimeout,
})
return success_response(alerts)
@endpoints_bp.route('/alerts/<int:notification_id>/acknowledge', methods=['POST'])
def acknowledge_fleet_alert(notification_id: int):
"""Agent acknowledges a fleet alert (collector-token auth).
The agent posts the hostname plus assetid when the operator dismisses a
fleet alert from the tray. Records acknowledgement so this alert will not
reappear in future polls. Idempotent - repeated acknowledgements succeed.
Acknowledgement is TERMINAL for this asset. Once recorded, the alert is
filtered out of get_fleet_alerts for this machine permanently, and a later
server-side close or cancel of the same notification is a no-op here: there
is nothing left to withdraw. The agent therefore retries a queued
acknowledgement until it lands without risk of a conflicting later state.
Only a real operator action reaches this route. A toast that merely timed
out on screen is NOT an acknowledgement and the agent does not post it - the
alert stays pending so it can be surfaced again.
"""
error = _collector_auth_or_error()
if error is not None:
return error
payload = request.get_json(silent=True) or {}
hostname = (payload.get('hostname') or '').strip()
assetid = payload.get('assetid')
if not hostname:
return error_response(ErrorCodes.VALIDATION_ERROR,
'hostname is required')
if assetid is None:
return error_response(ErrorCodes.VALIDATION_ERROR,
'assetid is required')
try:
assetid = int(assetid)
except (TypeError, ValueError):
return error_response(ErrorCodes.VALIDATION_ERROR,
'assetid must be an integer')
from plugins.notifications.models import Notification, NotificationAcknowledgement
notification = db.session.get(Notification, notification_id)
if notification is None:
return error_response(ErrorCodes.NOT_FOUND,
f'Notification {notification_id} not found',
http_code=404)
asset = db.session.get(Asset, assetid)
if asset is None:
return error_response(ErrorCodes.NOT_FOUND,
f'Asset {assetid} not found',
http_code=404)
# Validate hostname matches assetid to prevent cross-machine acknowledgements.
# _resolve_asset_for_hostname returns the asset id itself, not an Asset row.
resolved_assetid = _resolve_asset_for_hostname(hostname)
if resolved_assetid is None or resolved_assetid != assetid:
return error_response(ErrorCodes.VALIDATION_ERROR,
'hostname does not match provided assetid')
# Check if already acknowledged (idempotent)
existing = (NotificationAcknowledgement.query
.filter_by(notificationid=notification_id, assetid=assetid)
.first())
if existing:
return success_response({'acknowledged': True},
message='Alert already acknowledged')
# Record acknowledgement
ack = NotificationAcknowledgement(
notificationid=notification_id,
assetid=assetid
)
db.session.add(ack)
try:
db.session.commit()
except Exception as exc:
db.session.rollback()
logger.warning(f'Failed to commit acknowledgement for notification {notification_id}: {exc}')
return error_response(ErrorCodes.INTERNAL_ERROR,
'Failed to record acknowledgement',
http_code=500)
return success_response({'acknowledged': True},
message='Alert acknowledged', http_code=201)
Reference in New Issue View Git Blame Copy Permalink