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

harden CSV import reconciliation and preserve saved model selections

Browse Source 
Default fuzzy reconcile threshold to 0 and refuse merges when digit runs differ so numbered labels like Bay 311 never bind to Bay 310. Keep edit forms from clearing saved model fields when category filtering hides untyped rows. Includes in-progress computers policy and endpoints command feature work.
This commit is contained in:
benjamin.baker
2026-08-28 09:29:09 -04:00
parent d68c6db9ab
commit 01734e5449
46 changed files with 4241 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

372
plugins/endpoints/api/routes.py
View File

@@ -7,17 +7,21 @@ 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
from flask_jwt_extended import jwt_required
from flask import Blueprint, request, current_app
from flask_jwt_extended import jwt_required, 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
from ..models import EndpointState, EndpointEvent, EndpointCommand
from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind
from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title
endpoints_bp = Blueprint('endpoints', __name__)
@@ -188,14 +192,27 @@ def panel_services(asset_id: int):
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 = str(entry), ''
name, raw, app, kind = str(entry), '', None, None
running = raw in ('running', 'started', 'up', 'ok', 'active')
rows.append({
row = {
'name': name,
'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)
@@ -223,6 +240,48 @@ def panel_recent_events(asset_id: int):
return success_response(rows)
@endpoints_bp.route('/asset/<int:asset_id>/interfaces', methods=['GET'])
@jwt_required()
@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
# =============================================================================
@@ -348,3 +407,304 @@ def _quiet_hours() -> int:
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'])
@jwt_required()
@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'])
@jwt_required()
@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'])
@jwt_required()
@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'])
@jwt_required()
@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'])
@jwt_required()
@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}')
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')