Bring the wilm branch into main: endpoints plugin, agent, policies, CSV import
Ports the corporate-GitHub wilm branch onto main, keeping the two branches
separate. Thirteen upstream commits arrive with their authorship intact; the
review fixes and the portability work below are on top.
What arrives:
endpoints plugin - a ShopDB agent reporting endpoint state, telemetry,
services and interfaces, with a command channel the agent
long-polls. Ingest stays on the generic collector route
(ADR-006); the agent key is a UI-managed single-scope
token whose revocation is an instant kill switch.
computers policies - Configuration and Business Policies (RMM-style), scope
trees validated at write time, actions checked against
the endpoints command allowlist. Feature-gated: every
route 404s while the setting is off, so enabling it can
never half-expose the surface.
asset CSV import - templates, a collector script carrying the site path, and
a validate-then-commit import. A plugin can own a virtual
column plus a post-write row hook, which is how a PC's
machine number becomes a controls link without a
hand-built relationships sheet.
agent/ - the Windows agent and its installer.
Contract 0.21.0 -> 0.24.0, additive throughout: register_import_hook (0.22),
mint/list/revoke_service_token (0.23), collector_authorized (0.24). The stability
table had no rows for any of them and now does.
MySQL 5.6 portability. The branch stored three policy columns as sa.JSON, the
only JSON columns in the product, and 5.6 has no JSON type: the migration failed
with a bare syntax error near the word JSON. They are TEXT now, through a
JSONText type decorator, so all seventeen call sites still read and write dicts.
The convention elsewhere is already TEXT (guides.blocksjson,
geenforce.manifestjson).
Review fixes carried on top:
- six policy GETs were jwt_required(optional=True), which authenticates a
caller without checking anything; they require computers.view now, and six
redundant jwt_required decorators are gone
- the agent installer took a default BaseUrl and CollectorKey, so a mistyped
run silently pointed an endpoint at the wrong instance; both are mandatory
and a non-https BaseUrl throws unless -AllowInsecureHttp is passed
- a committing CSV import wrote no audit row
Documentation: the 34 new routes are described in docs/api-inventory.json, which
is hand-maintained on purpose, and the generated spec now covers 331 paths and
477 operations across 22 surfaces. DEVELOPMENT-SETUP's 1071 troubleshooting row
said "use MySQL 8 for dev" and now gives the actual fix, including the trap that
5.6 downgrades ROW_FORMAT=DYNAMIC to Compact with a warning instead of an error.
Full suite 2123 passed, vitest 260 passed, all seventeen plugin chains migrate
clean on MySQL 5.6 and the naming, docs, parity and read-surface gates are green.
This commit is contained in:
cproudlock
5
plugins/endpoints/api/__init__.py
Normal file
5
plugins/endpoints/api/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
||||
"""Endpoints plugin API."""
|
||||
|
||||
from .routes import endpoints_bp
|
||||
|
||||
__all__ = ['endpoints_bp']
|
||||
724
plugins/endpoints/api/routes.py
Normal file
724
plugins/endpoints/api/routes.py
Normal file
@@ -0,0 +1,724 @@
|
||||
"""Endpoints plugin API endpoints (read + dashboard).
|
||||
|
||||
Ingest is NOT here: agent reports arrive on the generic collector route
|
||||
POST /api/collector/endpoints (ADR-006), authorized by a collector-scoped
|
||||
service token, and are handled by EndpointsPlugin.apply_collector_payload.
|
||||
These routes are the human-facing read surface, gated by user JWT + RBAC.
|
||||
"""
|
||||
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from flask import Blueprint, request, current_app
|
||||
from flask_jwt_extended import current_user
|
||||
|
||||
from shopdb.api import (
|
||||
db, Asset, Setting, success_response, error_response, paginated_response,
|
||||
ErrorCodes, get_pagination_params, paginate_query, require_permission,
|
||||
require_role, mint_service_token, list_service_tokens, revoke_service_tokens,
|
||||
AuditLog, collector_authorized,
|
||||
)
|
||||
|
||||
from ..models import EndpointState, EndpointEvent, EndpointCommand
|
||||
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for
|
||||
from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title
|
||||
|
||||
endpoints_bp = Blueprint('endpoints', __name__)
|
||||
|
||||
|
||||
def _naive_utcnow():
|
||||
return datetime.now(timezone.utc).replace(tzinfo=None)
|
||||
|
||||
|
||||
def _format_uptime(seconds):
|
||||
"""Whole-seconds uptime -> 'Nd Nh Nm', or None."""
|
||||
if seconds is None:
|
||||
return None
|
||||
try:
|
||||
seconds = int(seconds)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
days, rem = divmod(seconds, 86400)
|
||||
hours, rem = divmod(rem, 3600)
|
||||
minutes = rem // 60
|
||||
parts = []
|
||||
if days:
|
||||
parts.append('{}d'.format(days))
|
||||
if hours:
|
||||
parts.append('{}h'.format(hours))
|
||||
parts.append('{}m'.format(minutes))
|
||||
return ' '.join(parts)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Endpoint state
|
||||
# =============================================================================
|
||||
|
||||
@endpoints_bp.route('/', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def list_endpoints():
|
||||
"""List reported endpoint states.
|
||||
|
||||
Joins through Computer so each row carries computerid and can deep-link
|
||||
into the existing PC detail page (/pcs/{computerid}). Endpoint state only
|
||||
exists for hostnames resolved to a Computer, so the join is not lossy.
|
||||
"""
|
||||
from plugins.computers.models import Computer
|
||||
|
||||
page, per_page = get_pagination_params(request)
|
||||
|
||||
query = (db.session.query(EndpointState, Computer.computerid)
|
||||
.join(Asset, Asset.assetid == EndpointState.assetid)
|
||||
.join(Computer, Computer.assetid == EndpointState.assetid)
|
||||
.filter(Asset.isactive.is_(True)))
|
||||
|
||||
if search := request.args.get('search'):
|
||||
query = query.filter(EndpointState.hostname.ilike(f'%{search}%'))
|
||||
|
||||
query = query.order_by(EndpointState.lastreportedat.desc())
|
||||
|
||||
items, total = paginate_query(query, page, per_page)
|
||||
data = []
|
||||
for state, computerid in items:
|
||||
row = state.to_dict()
|
||||
row['computerid'] = computerid
|
||||
row['uptimedisplay'] = _format_uptime(state.uptimeseconds)
|
||||
data.append(row)
|
||||
|
||||
return paginated_response(data, page, per_page, total)
|
||||
|
||||
|
||||
@endpoints_bp.route('/<int:asset_id>', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def get_endpoint(asset_id: int):
|
||||
"""Get the current state for one endpoint by asset id."""
|
||||
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
||||
|
||||
if not state:
|
||||
return error_response(
|
||||
ErrorCodes.NOT_FOUND,
|
||||
f'No endpoint state for asset {asset_id}',
|
||||
http_code=404,
|
||||
)
|
||||
|
||||
return success_response(state.to_dict())
|
||||
|
||||
|
||||
@endpoints_bp.route('/<int:asset_id>/events', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def list_endpoint_events(asset_id: int):
|
||||
"""List curated Windows events for one endpoint, newest first."""
|
||||
page, per_page = get_pagination_params(request)
|
||||
|
||||
query = EndpointEvent.query.filter_by(assetid=asset_id)
|
||||
|
||||
if eventid := request.args.get('eventid'):
|
||||
try:
|
||||
query = query.filter(EndpointEvent.eventid == int(eventid))
|
||||
except (TypeError, ValueError):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'eventid must be an integer')
|
||||
|
||||
query = query.order_by(EndpointEvent.eventtime.desc())
|
||||
|
||||
items, total = paginate_query(query, page, per_page)
|
||||
data = [e.to_dict() for e in items]
|
||||
|
||||
return paginated_response(data, page, per_page, total)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Asset panels (ADR-010) - shaped for the generic PluginAssetPanels renderer
|
||||
# =============================================================================
|
||||
|
||||
@endpoints_bp.route('/asset/<int:asset_id>/status', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def panel_status(asset_id: int):
|
||||
"""Keyvalue panel: the endpoint's current status snapshot.
|
||||
|
||||
Single indexed lookup on endpointstates.assetid. Returns the {fields:[...]}
|
||||
shape the keyvalue renderer expects; an absent state yields no fields so the
|
||||
panel hides itself.
|
||||
"""
|
||||
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
||||
if not state:
|
||||
return success_response({'fields': []})
|
||||
|
||||
def field(label, value, mono=False):
|
||||
return {'label': label, 'value': value, 'mono': mono}
|
||||
|
||||
fields = [
|
||||
field('Logged in user', state.loggedinuser),
|
||||
field('Uptime', _format_uptime(state.uptimeseconds)),
|
||||
field('Idle', _format_uptime(state.idleseconds)),
|
||||
field('Last boot',
|
||||
state.lastboottime.isoformat() + 'Z' if state.lastboottime else None),
|
||||
field('CPU %', state.cpupercent),
|
||||
field('Memory %', state.memorypercent),
|
||||
field('Agent', state.agentversion, mono=True),
|
||||
field('Last reported',
|
||||
state.lastreportedat.isoformat() + 'Z' if state.lastreportedat else None),
|
||||
]
|
||||
fields = [f for f in fields if f['value'] not in (None, '')]
|
||||
return success_response({'fields': fields})
|
||||
|
||||
|
||||
@endpoints_bp.route('/asset/<int:asset_id>/telemetrymeta', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def panel_telemetry_meta(asset_id: int):
|
||||
"""Lightweight freshness probe for the panels.
|
||||
|
||||
Returns just the endpoint's lastreportedat so the WebUI can tell whether a
|
||||
genuinely new telemetry snapshot has landed after it forces a collect. The
|
||||
UI captures this before enqueuing collect_now (or a row Start/Stop), then
|
||||
polls until the stamp advances - a command reaching 'succeeded' only means
|
||||
the agent ran it, not that it has yet posted the fresh services snapshot, so
|
||||
gating the refresh on this timestamp is what avoids re-rendering stale rows.
|
||||
"""
|
||||
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
||||
stamp = (state.lastreportedat.isoformat() + 'Z'
|
||||
if state and state.lastreportedat else None)
|
||||
return success_response({'lastreportedat': stamp})
|
||||
|
||||
|
||||
@endpoints_bp.route('/asset/<int:asset_id>/services', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def panel_services(asset_id: int):
|
||||
"""List panel: monitored services/processes with a run-state badge per row.
|
||||
|
||||
Reshapes the latest servicesjson snapshot into rows the list renderer maps
|
||||
to name + status badge. Tolerant of a missing or unexpected snapshot shape:
|
||||
an unknown state renders as an Unknown badge rather than failing.
|
||||
"""
|
||||
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
||||
if not state or not state.servicesjson:
|
||||
return success_response([])
|
||||
|
||||
try:
|
||||
services = json.loads(state.servicesjson)
|
||||
except (TypeError, ValueError):
|
||||
return success_response([])
|
||||
|
||||
rows = []
|
||||
for entry in (services or []):
|
||||
if isinstance(entry, dict):
|
||||
name = entry.get('name') or entry.get('service') or 'Unknown'
|
||||
raw = (entry.get('state') or entry.get('status') or '').strip().lower()
|
||||
app = entry.get('app')
|
||||
kind = entry.get('kind')
|
||||
else:
|
||||
name, raw, app, kind = str(entry), '', None, None
|
||||
running = raw in ('running', 'started', 'up', 'ok', 'active')
|
||||
row = {
|
||||
'name': name,
|
||||
'kind': (kind or '').lower() or None,
|
||||
'statelabel': 'Running' if running else (raw.title() or 'Unknown'),
|
||||
'statecolor': 'green' if running else ('red' if raw else 'gray'),
|
||||
}
|
||||
targets = row_actions_for(app, kind, name)
|
||||
if targets:
|
||||
row['action'] = {
|
||||
'running': running,
|
||||
'start': targets['start'],
|
||||
'stop': targets['stop'],
|
||||
'status': targets['status'],
|
||||
'target': targets['target'],
|
||||
'targetkind': targets['targetkind'],
|
||||
}
|
||||
rows.append(row)
|
||||
return success_response(rows)
|
||||
|
||||
|
||||
@endpoints_bp.route('/asset/<int:asset_id>/recentevents', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def panel_recent_events(asset_id: int):
|
||||
"""List panel: the endpoint's most recent curated events, newest first.
|
||||
|
||||
Indexed range scan on (assetid, eventtime), capped small - this is the
|
||||
at-a-glance history card, not the full report (that is the events route
|
||||
with pagination).
|
||||
"""
|
||||
events = (EndpointEvent.query
|
||||
.filter_by(assetid=asset_id)
|
||||
.order_by(EndpointEvent.eventtime.desc())
|
||||
.limit(10)
|
||||
.all())
|
||||
|
||||
rows = [{
|
||||
'title': event_title(e.eventid),
|
||||
'severity': event_severity(e.eventid),
|
||||
'eventtime': e.eventtime.isoformat() + 'Z' if e.eventtime else None,
|
||||
} for e in events]
|
||||
return success_response(rows)
|
||||
|
||||
|
||||
@endpoints_bp.route('/asset/<int:asset_id>/interfaces', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def panel_interfaces(asset_id: int):
|
||||
"""Table panel: every hardware NIC the agent last reported.
|
||||
|
||||
Reshapes the latest interfacesjson snapshot into one row per adapter with
|
||||
the {alias, ipv4, mac, gateway, subnet, dhcp} shape the collector emits. A
|
||||
gatewayless adapter is flagged as the likely machine-IPS link (see
|
||||
DESIGN-DECISIONS-agent.md 9.3): those NICs speak to the machine, not the
|
||||
corporate LAN. Tolerant of a missing or unexpected snapshot: an unreadable
|
||||
payload yields no rows so the panel hides itself.
|
||||
"""
|
||||
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
||||
if not state or not state.interfacesjson:
|
||||
return success_response([])
|
||||
|
||||
try:
|
||||
interfaces = json.loads(state.interfacesjson)
|
||||
except (TypeError, ValueError):
|
||||
return success_response([])
|
||||
|
||||
rows = []
|
||||
for entry in (interfaces or []):
|
||||
if not isinstance(entry, dict):
|
||||
continue
|
||||
gateway = (entry.get('gateway') or '').strip()
|
||||
ipv4 = (entry.get('ipv4') or '').strip()
|
||||
rows.append({
|
||||
'alias': entry.get('alias') or 'Unknown',
|
||||
'ipv4': ipv4 or '-',
|
||||
'mac': (entry.get('mac') or '').strip() or '-',
|
||||
'gateway': gateway or '-',
|
||||
'subnet': (entry.get('subnet') or '').strip() or '-',
|
||||
'dhcp': 'DHCP' if entry.get('dhcp') else 'Static',
|
||||
# A gatewayless NIC with an address is the machine link; the gateway
|
||||
# NIC is the corporate LAN.
|
||||
'role': 'Machine link' if (ipv4 and not gateway) else 'LAN',
|
||||
})
|
||||
return success_response(rows)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Dashboard
|
||||
# =============================================================================
|
||||
|
||||
@endpoints_bp.route('/dashboard/summary', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def dashboard_summary():
|
||||
"""High-level fleet counts for the endpoints dashboard."""
|
||||
total = db.session.query(EndpointState).join(
|
||||
Asset, Asset.assetid == EndpointState.assetid
|
||||
).filter(Asset.isactive.is_(True)).count()
|
||||
|
||||
hours = _quiet_hours()
|
||||
cutoff = _naive_utcnow() - timedelta(hours=hours)
|
||||
|
||||
reporting = db.session.query(EndpointState).join(
|
||||
Asset, Asset.assetid == EndpointState.assetid
|
||||
).filter(
|
||||
Asset.isactive.is_(True),
|
||||
EndpointState.lastreportedat.isnot(None),
|
||||
EndpointState.lastreportedat >= cutoff,
|
||||
).count()
|
||||
|
||||
return success_response({
|
||||
'total': total,
|
||||
'reporting': reporting,
|
||||
'quiet': total - reporting,
|
||||
'quiethours': hours,
|
||||
})
|
||||
|
||||
|
||||
@endpoints_bp.route('/dashboard/quiet', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def dashboard_quiet():
|
||||
"""Endpoints that reported before, then went quiet past the window.
|
||||
|
||||
Never-reported endpoints are excluded: those are records without an agent,
|
||||
not agents that stopped. Silence is the only signal - there is no heartbeat
|
||||
separate from the report.
|
||||
"""
|
||||
hours = _quiet_hours()
|
||||
cutoff = _naive_utcnow() - timedelta(hours=hours)
|
||||
now = _naive_utcnow()
|
||||
|
||||
from plugins.computers.models import Computer
|
||||
|
||||
rows = (db.session.query(EndpointState, Computer.computerid)
|
||||
.join(Asset, Asset.assetid == EndpointState.assetid)
|
||||
.join(Computer, Computer.assetid == EndpointState.assetid)
|
||||
.filter(Asset.isactive.is_(True),
|
||||
EndpointState.lastreportedat.isnot(None),
|
||||
EndpointState.lastreportedat < cutoff)
|
||||
.all())
|
||||
|
||||
out = [{
|
||||
'computerid': computerid,
|
||||
'assetid': s.assetid,
|
||||
'hostname': s.hostname,
|
||||
'lastreported': s.lastreportedat.isoformat() + 'Z',
|
||||
'quietdays': (now - s.lastreportedat).days,
|
||||
} for s, computerid in rows]
|
||||
|
||||
out.sort(key=lambda r: -r['quietdays'])
|
||||
return success_response(out[:50])
|
||||
|
||||
|
||||
@endpoints_bp.route('/dashboard/stability', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def dashboard_stability():
|
||||
"""Endpoints with recent crash/shutdown/corruption events.
|
||||
|
||||
Bounded to a fixed 48h window and the critical event catalog, capped at 50
|
||||
rows: cost is a range scan on eventtime filtered to a short id list, not a
|
||||
scan of all history. Joins through Computer so each row links to the PC
|
||||
detail page.
|
||||
"""
|
||||
from plugins.computers.models import Computer
|
||||
|
||||
if not CRITICAL_EVENT_IDS:
|
||||
return success_response([])
|
||||
|
||||
cutoff = _naive_utcnow() - timedelta(hours=48)
|
||||
|
||||
rows = (db.session.query(
|
||||
Computer.computerid,
|
||||
EndpointState.hostname,
|
||||
db.func.count(EndpointEvent.endpointeventid),
|
||||
db.func.max(EndpointEvent.eventtime),
|
||||
)
|
||||
.select_from(EndpointEvent)
|
||||
.join(Asset, Asset.assetid == EndpointEvent.assetid)
|
||||
.join(Computer, Computer.assetid == EndpointEvent.assetid)
|
||||
.outerjoin(EndpointState,
|
||||
EndpointState.assetid == EndpointEvent.assetid)
|
||||
.filter(Asset.isactive.is_(True),
|
||||
EndpointEvent.eventid.in_(CRITICAL_EVENT_IDS),
|
||||
EndpointEvent.eventtime.isnot(None),
|
||||
EndpointEvent.eventtime >= cutoff)
|
||||
.group_by(Computer.computerid, EndpointState.hostname)
|
||||
.all())
|
||||
|
||||
out = [{
|
||||
'computerid': computerid,
|
||||
'hostname': hostname,
|
||||
'eventcount': count,
|
||||
'lastevent': lasttime.isoformat() + 'Z' if lasttime else None,
|
||||
} for computerid, hostname, count, lasttime in rows]
|
||||
|
||||
out.sort(key=lambda r: -r['eventcount'])
|
||||
return success_response(out[:50])
|
||||
|
||||
|
||||
def _quiet_hours() -> int:
|
||||
"""Quiet-report window in hours, from a site setting, default 24."""
|
||||
hours = 24
|
||||
setting = Setting.query.filter_by(key='endpoints_quietreporthours').first()
|
||||
if setting and (setting.value or '').strip():
|
||||
try:
|
||||
hours = int(setting.value)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return hours
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Endpoint agent key (UI-managed collector.ingest.endpoints kill switch)
|
||||
# =============================================================================
|
||||
|
||||
# The single permission the endpoint agent key authorizes. A managed token
|
||||
# scoped to ONLY this authorizes the endpoints collector and nothing else, so
|
||||
# rotating/revoking it stops endpoint-agent traffic without touching the
|
||||
# computers/geenforce collectors.
|
||||
AGENT_KEY_SCOPE = 'collector.ingest.endpoints'
|
||||
AGENT_KEY_NAME = 'Endpoint agent key'
|
||||
|
||||
|
||||
def _agentkey_status_payload():
|
||||
"""Current state of the endpoint agent key for the settings page.
|
||||
|
||||
Reports whether an active managed key exists (and its non-secret metadata)
|
||||
plus whether the bootstrap .env fallback is still accepted. Never exposes a
|
||||
secret; the full secret is only ever returned once, at regenerate time.
|
||||
"""
|
||||
tokens = list_service_tokens(AGENT_KEY_SCOPE)
|
||||
active = tokens[0] if tokens else None
|
||||
envfallback = Setting.get('plugin.endpoints.envkeyfallback', default=True)
|
||||
provisioned = Setting.get('plugin.endpoints.agentkeyprovisioned',
|
||||
default=False)
|
||||
return {
|
||||
'provisioned': bool(provisioned) or active is not None,
|
||||
'envkeyfallback': bool(envfallback),
|
||||
'key': active.to_dict() if active is not None else None,
|
||||
}
|
||||
|
||||
|
||||
@endpoints_bp.route('/agentkey', methods=['GET'])
|
||||
@require_role('admin')
|
||||
def get_agentkey():
|
||||
"""Report the endpoint agent key state (no secret)."""
|
||||
return success_response(_agentkey_status_payload())
|
||||
|
||||
|
||||
@endpoints_bp.route('/agentkey/regenerate', methods=['POST'])
|
||||
@require_role('admin')
|
||||
def regenerate_agentkey():
|
||||
"""Mint a fresh endpoint agent key, revoking any previous one.
|
||||
|
||||
Hard cutover: the old key is deactivated in the same call, and the .env
|
||||
fallback is switched off, so the new managed token becomes the sole
|
||||
credential and revoking it is a true kill switch. The full secret is
|
||||
returned ONCE here and never again.
|
||||
"""
|
||||
try:
|
||||
token, secret = mint_service_token(
|
||||
AGENT_KEY_SCOPE, AGENT_KEY_NAME, current_user.userid)
|
||||
except PermissionError as error:
|
||||
return error_response(ErrorCodes.FORBIDDEN, str(error), http_code=403)
|
||||
|
||||
Setting.set('plugin.endpoints.agentkeyprovisioned', True,
|
||||
valuetype='boolean', category='plugin.endpoints')
|
||||
Setting.set('plugin.endpoints.envkeyfallback', False,
|
||||
valuetype='boolean', category='plugin.endpoints')
|
||||
|
||||
result = _agentkey_status_payload()
|
||||
# The secret appears here and NOWHERE else, ever. Not stored, not logged.
|
||||
result['secret'] = secret
|
||||
result['warning'] = ('Save this key now. It will not be shown again. '
|
||||
'Configure it as the agent X-API-Key.')
|
||||
return success_response(result, message='Endpoint agent key generated',
|
||||
http_code=201)
|
||||
|
||||
|
||||
@endpoints_bp.route('/agentkey', methods=['DELETE'])
|
||||
@require_role('admin')
|
||||
def revoke_agentkey():
|
||||
"""Revoke the endpoint agent key: stop all managed-token endpoint traffic.
|
||||
|
||||
Leaves envkeyfallback off (the kill switch stays armed). To restore
|
||||
reporting an operator regenerates a new key.
|
||||
"""
|
||||
revoked = revoke_service_tokens(AGENT_KEY_SCOPE)
|
||||
Setting.set('plugin.endpoints.agentkeyprovisioned', False,
|
||||
valuetype='boolean', category='plugin.endpoints')
|
||||
result = _agentkey_status_payload()
|
||||
result['revoked'] = revoked
|
||||
return success_response(result, message='Endpoint agent key revoked')
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Endpoint commands (WebUI -> agent, allowlisted actions only)
|
||||
# =============================================================================
|
||||
#
|
||||
# Operators enqueue a named, server-allowlisted action for one endpoint (JWT +
|
||||
# endpoints.command). The agent long-polls /commands/next with its collector
|
||||
# token, claims the single next pending command, runs it locally, and reports
|
||||
# the outcome to /commands/<id>/result. The wire never carries a command string
|
||||
# - only an allowlisted action name - so a compromised WebUI cannot ask an
|
||||
# endpoint to run something arbitrary.
|
||||
|
||||
# How long the agent claim poll is held open server-side before returning empty.
|
||||
# Kept below the agent HttpClient timeout and typical proxy idle timeouts so a
|
||||
# quiet fleet re-polls cleanly instead of tripping a read timeout.
|
||||
_CLAIM_HOLD_SECONDS = 25
|
||||
_CLAIM_POLL_INTERVAL = 1.0
|
||||
|
||||
|
||||
def _collector_auth_or_error():
|
||||
"""Authorize an agent-facing command route on the endpoints collector token.
|
||||
|
||||
Reuses the collector auth path so the command channel shares the endpoint
|
||||
agent key (kill switch) and introduces no second secret.
|
||||
"""
|
||||
return collector_authorized('endpoints')
|
||||
|
||||
|
||||
def _resolve_asset_for_hostname(hostname):
|
||||
"""Resolve a reported hostname to its asset id via the computers plugin."""
|
||||
from plugins.computers.models import Computer
|
||||
|
||||
computer = (Computer.query
|
||||
.filter(db.func.lower(Computer.hostname) == hostname.lower())
|
||||
.first())
|
||||
return computer.assetid if computer is not None else None
|
||||
|
||||
|
||||
@endpoints_bp.route('/<int:asset_id>/commands', methods=['GET'])
|
||||
@require_permission('endpoints.view')
|
||||
def list_endpoint_commands(asset_id: int):
|
||||
"""Recent commands for one endpoint, newest first (operator view)."""
|
||||
rows = (EndpointCommand.query
|
||||
.filter_by(assetid=asset_id)
|
||||
.order_by(EndpointCommand.createdat.desc())
|
||||
.limit(20)
|
||||
.all())
|
||||
return success_response([row.to_dict() for row in rows])
|
||||
|
||||
|
||||
@endpoints_bp.route('/<int:asset_id>/commands', methods=['POST'])
|
||||
@require_permission('endpoints.command')
|
||||
def enqueue_endpoint_command(asset_id: int):
|
||||
"""Queue an allowlisted command for one endpoint.
|
||||
|
||||
Rejects unknown actions and refuses to stack a second pending/running
|
||||
command on the same endpoint, so the queue stays at-most-one-in-flight and
|
||||
the WebUI toggle cannot race itself.
|
||||
"""
|
||||
payload = request.get_json(silent=True) or {}
|
||||
action = (payload.get('action') or '').strip()
|
||||
if not is_allowed_action(action):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
f'Unknown or disallowed action: {action!r}')
|
||||
|
||||
# Destructive actions (for example system_restart) demand a stricter grant
|
||||
# ON TOP OF endpoints.command, so a remote reboot cannot ride the routine
|
||||
# toggle permission.
|
||||
elevated = elevated_permission_for(action)
|
||||
if elevated and not current_user.haspermission(elevated):
|
||||
return error_response(
|
||||
ErrorCodes.FORBIDDEN,
|
||||
f'Action {action!r} requires the {elevated} permission',
|
||||
http_code=403)
|
||||
|
||||
if action not in ALLOWED_ACTIONS_FOR(assetid=asset_id):
|
||||
return error_response(
|
||||
ErrorCodes.VALIDATION_ERROR,
|
||||
f'Action not permitted by policy for this endpoint: {action!r}')
|
||||
|
||||
# Generic actions (service_* / process_*) must carry the subject to act on;
|
||||
# fixed-target actions (mtconnect_*) must not. Validate both directions so a
|
||||
# caller cannot send a targetless generic action (nothing to act on) or
|
||||
# smuggle a target onto a pinned action.
|
||||
target = (payload.get('target') or '').strip() or None
|
||||
targetkind = action_targetkind(action)
|
||||
if requires_target(action):
|
||||
if not target:
|
||||
return error_response(
|
||||
ErrorCodes.VALIDATION_ERROR,
|
||||
f'Action {action!r} requires a target')
|
||||
elif target is not None:
|
||||
return error_response(
|
||||
ErrorCodes.VALIDATION_ERROR,
|
||||
f'Action {action!r} does not accept a target')
|
||||
|
||||
state = EndpointState.query.filter_by(assetid=asset_id).first()
|
||||
if state is None:
|
||||
return error_response(ErrorCodes.NOT_FOUND,
|
||||
f'No endpoint state for asset {asset_id}',
|
||||
http_code=404)
|
||||
|
||||
pending = (EndpointCommand.query
|
||||
.filter(EndpointCommand.assetid == asset_id,
|
||||
EndpointCommand.status.in_((
|
||||
EndpointCommand.STATUS_PENDING,
|
||||
EndpointCommand.STATUS_RUNNING)))
|
||||
.first())
|
||||
if pending is not None:
|
||||
return error_response(
|
||||
ErrorCodes.VALIDATION_ERROR,
|
||||
'A command is already in flight for this endpoint',
|
||||
http_code=409)
|
||||
|
||||
command = EndpointCommand(
|
||||
assetid=asset_id,
|
||||
action=action,
|
||||
target=target,
|
||||
targetkind=targetkind if target else None,
|
||||
status=EndpointCommand.STATUS_PENDING,
|
||||
requestedby=getattr(current_user, 'userid', None),
|
||||
)
|
||||
db.session.add(command)
|
||||
|
||||
AuditLog.log('created', 'EndpointCommand',
|
||||
entityname=action,
|
||||
details=f'action={action} target={target} asset={asset_id}')
|
||||
|
||||
db.session.commit()
|
||||
return success_response(command.to_dict(),
|
||||
message='Command queued', http_code=201)
|
||||
|
||||
|
||||
@endpoints_bp.route('/commands/next', methods=['GET'])
|
||||
def claim_next_command():
|
||||
"""Agent long-poll: claim the next pending command for a hostname.
|
||||
|
||||
Collector-token auth (not JWT). Holds the request open briefly and returns
|
||||
the single oldest pending command for the caller's endpoint, atomically
|
||||
marking it running so no other poll claims it. Returns 204 when nothing is
|
||||
queued within the hold window - the agent simply re-polls.
|
||||
"""
|
||||
error = _collector_auth_or_error()
|
||||
if error is not None:
|
||||
return error
|
||||
|
||||
hostname = (request.args.get('hostname') or '').strip()
|
||||
if not hostname:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'hostname is required')
|
||||
|
||||
asset_id = _resolve_asset_for_hostname(hostname)
|
||||
if asset_id is None:
|
||||
return error_response(ErrorCodes.NOT_FOUND,
|
||||
f'Unknown hostname: {hostname!r}', http_code=404)
|
||||
|
||||
deadline = time.monotonic() + _CLAIM_HOLD_SECONDS
|
||||
while True:
|
||||
command = (EndpointCommand.query
|
||||
.filter_by(assetid=asset_id,
|
||||
status=EndpointCommand.STATUS_PENDING)
|
||||
.order_by(EndpointCommand.createdat.asc())
|
||||
.first())
|
||||
if command is not None:
|
||||
command.status = EndpointCommand.STATUS_RUNNING
|
||||
command.claimedat = _naive_utcnow()
|
||||
db.session.commit()
|
||||
return success_response({
|
||||
'endpointcommandid': command.endpointcommandid,
|
||||
'action': command.action,
|
||||
'target': command.target,
|
||||
'targetkind': command.targetkind,
|
||||
})
|
||||
|
||||
# Nothing queued: release the session so a concurrent enqueue is visible
|
||||
# on the next poll, wait a beat, then re-check until the hold expires.
|
||||
db.session.rollback()
|
||||
if time.monotonic() >= deadline:
|
||||
return ('', 204)
|
||||
time.sleep(_CLAIM_POLL_INTERVAL)
|
||||
|
||||
|
||||
@endpoints_bp.route('/commands/<int:command_id>/result', methods=['POST'])
|
||||
def report_command_result(command_id: int):
|
||||
"""Agent reports a claimed command's outcome (collector-token auth).
|
||||
|
||||
Finalizes the command row and, for MTConnect actions that observed the run
|
||||
state, caches it on the EndpointState so the WebUI toggle reflects reality
|
||||
between polls. Both writes land in one commit.
|
||||
"""
|
||||
error = _collector_auth_or_error()
|
||||
if error is not None:
|
||||
return error
|
||||
|
||||
command = db.session.get(EndpointCommand, command_id)
|
||||
if command is None:
|
||||
return error_response(ErrorCodes.NOT_FOUND,
|
||||
f'No command {command_id}', http_code=404)
|
||||
|
||||
payload = request.get_json(silent=True) or {}
|
||||
succeeded = bool(payload.get('succeeded'))
|
||||
command.status = (EndpointCommand.STATUS_SUCCEEDED if succeeded
|
||||
else EndpointCommand.STATUS_FAILED)
|
||||
command.resultmessage = (payload.get('message') or '')[:2000] or None
|
||||
detail = payload.get('detail')
|
||||
command.resultjson = json.dumps(detail, separators=(',', ':')) \
|
||||
if detail is not None else None
|
||||
command.completedat = _naive_utcnow()
|
||||
|
||||
# Cache the observed MTConnect state so the toggle shows a real value. The
|
||||
# agent sends mtconnectrunning on any mtconnect_* result that probed it.
|
||||
if 'mtconnectrunning' in payload:
|
||||
state = EndpointState.query.filter_by(assetid=command.assetid).first()
|
||||
if state is not None:
|
||||
running = payload.get('mtconnectrunning')
|
||||
state.mtconnectrunning = bool(running) if running is not None else None
|
||||
state.mtconnectcheckedat = _naive_utcnow()
|
||||
|
||||
db.session.commit()
|
||||
return success_response(command.to_dict(), message='Result recorded')
|
||||
|
||||
Reference in New Issue
Block a user