"""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 re import time from datetime import datetime, timedelta, timezone from flask import Blueprint, request, current_app, g from flask_jwt_extended import current_user from shopdb.api import ( db, Asset, AssetRelationship, 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, send_alert, ) from ..models import ( EndpointState, EndpointEvent, EndpointCommand, EndpointApplication, EndpointHealthState, OperatorReport, EndpointLogCollection, ) from ..commands import COMMAND_ALLOWLIST, is_allowed_action, ALLOWED_ACTIONS_FOR, row_actions_for, requires_target, action_targetkind, elevated_permission_for, minimum_agent_version_for, action_minimum_versions, parse_agent_version, agent_meets_minimum, PLACE_POLICY_HOLD, ACTIVE_POLICY_HOLDS, HOLDABLE_ACTIONS_FOR, RELEASE_POLICY_HOLD, ACTIVE_LOCK_FOR, PLACE_ENDPOINT_LOCK, RELEASE_ENDPOINT_LOCK from . import agentstore from . import agentauth from ..eventcatalog import CRITICAL_EVENT_IDS, event_severity, event_title from markupsafe import escape endpoints_bp = Blueprint('endpoints', __name__) def _naive_utcnow(): return datetime.now(timezone.utc).replace(tzinfo=None) def _get_site_timezone(): """Get configured site timezone from settings, fallback to UTC.""" try: import pytz setting = Setting.query.filter_by(key='site_timezone').first() if setting and setting.value: return pytz.timezone(setting.value) except Exception: pass return timezone.utc def _localize_timestamp(utc_datetime): """Convert naive UTC datetime to site timezone ISO string for display.""" if not utc_datetime: return None try: import pytz site_tz = _get_site_timezone() # Assume stored datetime is UTC (naive) utc_aware = pytz.UTC.localize(utc_datetime) localized = utc_aware.astimezone(site_tz) return localized.isoformat() except Exception: # Fallback: return as UTC if anything fails return utc_datetime.isoformat() + 'Z' if utc_datetime else 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 from .enrollment import CREDENTIAL_STATES, credential_state_expression 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, ) # Evaluated in the same statement as correlated EXISTS, so the column and # its filter cost no query per row and survive pagination. credentialstate = credential_state_expression( EndpointState.assetid, Computer.hostname, EndpointState.lastreportedat) query = (db.session.query(EndpointState, Computer.computerid, Computer.computertypeid, ComputerType.computertype, Asset.assetnumber, Asset.name, statusrank.label('statusrank'), credentialstate.label('credentialstate')) .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) wantedcredential = (request.args.get('credentialstate') or '').strip().lower() if wantedcredential in CREDENTIAL_STATES: query = query.filter(credentialstate == wantedcredential) 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, credential) 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) row['credentialstate'] = credential 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 def _credential_for_state(state): """Per-PC credential state for one endpoint, redacted below endpoints.fleet. The state itself is as public as the rest of the status panel. The prefix, timestamps, enrollment path and block reason are fleet administration, so a reader holding only endpoints.view (a guest, where a site grants that) gets the state and nothing else. """ from .enrollment import credential_detail hostname = _hostname_for_asset(state.assetid) or state.hostname return credential_detail(state.assetid, hostname, state.lastreportedat, detailed=caller_holds('endpoints.fleet')) @endpoints_bp.route('/', 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) # The currently published agent build, so the page can say "behind" without # hardcoding a version that goes stale the next time one is published. # None when nothing has been staged yet, which reads as "cannot tell". data['agentcurrent'] = agentstore.current_version('agent') # Per-action agent floors. Served rather than duplicated in the browser: # the enqueue guard refuses against these same numbers, so a copy in the # frontend would silently disagree the moment a floor was raised here. data['actionminimums'] = action_minimum_versions() data['credential'] = _credential_for_state(state) # Localize timestamps to site timezone for ts_field in ['lastreportedat', 'lasteventat', 'created_at', 'updated_at']: if ts_field in data and data[ts_field]: data[ts_field] = _localize_timestamp(state.__dict__.get(ts_field)) return success_response(data) @endpoints_bp.route('//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//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 # Annotate the agent version when the endpoint is behind the published # build. Done here rather than as a separate row so the fact sits on the # version it describes. Silent when nothing is published, when the endpoint # has not reported, or when either version is unparseable - claiming # "behind" from a placeholder like '0.0.0-sample' would be a false alarm. agentvalue = state.agentversion published = agentstore.current_version('agent') agentbehind = (agentvalue and published and parse_agent_version(agentvalue) and parse_agent_version(published) and not agent_meets_minimum(agentvalue, published)) if agentbehind: agentvalue = '{} (behind {})'.format(agentvalue, published) 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', agentvalue, mono=True), field('Tray', state.trayversion, mono=True), field('Last reported', state.lastreportedat.isoformat() + 'Z' if state.lastreportedat else None, fmt='datetime'), ] credential = _credential_for_state(state) fields.extend([ field('Credential', credential['statelabel']), field('Credential prefix', credential['credentialprefix'] + '...' if credential['credentialprefix'] else None, mono=True), field('Credential issued', credential['issuedat'], fmt='datetime'), field('Credential last used', credential['lastusedat'], fmt='datetime'), field('Enrolled via', credential['enrolledvia']), field('Re-enroll block reason', credential['reenrollblockedreason']), ]) 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))) # Machine-readable companion to the "(behind X)" text above. The UI needs to # decide whether to offer the upgrade action, and parsing that decision back # out of a display string would tie the button to the wording of a label. # Emitted as a sibling of fields so nothing about the existing rendering # changes; an older UI simply ignores it. return success_response({ 'fields': fields, 'agentbehind': bool(agentbehind), 'agentversion': state.agentversion, 'agentcurrent': published, 'credential': credential, }) @endpoints_bp.route('/asset//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//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//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//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//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//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//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//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//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//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//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//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//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//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//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 _agent_auth_or_error(): """Authorize an agent-facing route: per-PC credential or the shared key. See agentauth. Sets g.agentauth and g.agentassetid. """ return agentauth.authenticate_agent() def _resolve_asset_for_hostname(hostname): """Resolve a reported hostname to its asset id via the computers plugin.""" return agentauth.resolve_hostname(hostname) def _hostname_for_asset(assetid): from plugins.computers.models import Computer computer = Computer.query.filter_by(assetid=assetid).first() return computer.hostname if computer is not None else None def _bind_agent_hostname(rawhostname): """The asset an agent request acts for. Returns (assetid, hostname, error). A credential caller may omit the hostname; a shared-key caller must send one that resolves. hostname comes back filled from the asset when omitted. """ hostname = (rawhostname or '').strip() asset_id, error = agentauth.bind_hostname(hostname) if error is not None: return None, hostname, error if asset_id is None: if not hostname: return None, hostname, error_response( ErrorCodes.VALIDATION_ERROR, 'hostname is required') return None, hostname, error_response( ErrorCodes.NOT_FOUND, f'Unknown hostname: {hostname!r}', http_code=404) if not hostname: hostname = _hostname_for_asset(asset_id) or '' return asset_id, hostname, 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/); 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:" - the app's own uploaded installer (installersha256) - "version:" - a specific AppVersion's pinned package - bare "" - 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:, version:, 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/ 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}' # Extensions collect_log will read. The list is an allowlist, not a blocklist: # the question is not "is this dangerous to read" but "will this render as text # in a browser pane", and only a closed set can answer that. Anything else comes # back refused rather than as mojibake the operator has to interpret. # # The agent enforces this same set independently. Duplicating it is deliberate - # this copy exists to give a fast, actionable error at enqueue, and the agent's # copy is the one that actually guards the disk. # # The Windows ".c-f-g" extension (spelled with separators here on purpose) is # deliberately absent: the literal collides with the banned-shorthand naming # gate, and weakening a repo-wide gate for one rarely-collected extension is a # bad trade. '.conf' and '.ini' cover the same need. LOG_PATH_EXTENSIONS = frozenset({ '.log', '.txt', '.xml', '.json', '.csv', '.ini', '.conf', '.yml', '.yaml', '.trace', '.out', '.err', '.md', }) # Windows path length ceiling for the classic (non-\\?\) API. A longer path # cannot be opened by the agent anyway, so refusing here is honest. LOG_PATH_MAX_LENGTH = 260 # Mirror of the agent's DeniedLogPathReason (CommandExecutor.cs): files SYSTEM # can read that must never be pulled over the wire - deployment answer files, # the agent's own install and state, key material. The server cannot know which # drive Windows lives on, so roots match under any drive letter, and the common # environment variables are expanded textually to those same roots. LOG_PATH_ENVIRONMENT_ROOTS = { '%systemroot%': 'c:\\windows', '%windir%': 'c:\\windows', '%programdata%': 'c:\\programdata', '%allusersprofile%': 'c:\\programdata', '%programfiles%': 'c:\\program files', '%programw6432%': 'c:\\program files', '%programfiles(x86)%': 'c:\\program files (x86)', '%systemdrive%': 'c:', } LOG_PATH_DENIED_ROOTS = ( ('windows\\panther', 'Windows setup answer files (Panther)'), ('windows\\system32\\sysprep', 'Sysprep files'), ('programdata\\shopdb', 'ShopDB agent state (ProgramData\\ShopDB)'), ('program files\\shopdb', 'the ShopDB agent install directory'), ('program files (x86)\\shopdb', 'the ShopDB agent install directory'), ('progra~1\\shopdb', 'the ShopDB agent install directory'), ('progra~2\\shopdb', 'the ShopDB agent install directory'), ('progra~3\\shopdb', 'ShopDB agent state (ProgramData\\ShopDB)'), ) LOG_PATH_DENIED_EXTENSIONS = frozenset({'.pfx', '.p12', '.key'}) def _denied_log_path_reason(path): """Return why a collect_log path is on the sensitive-file denylist, or None. Case-insensitive; slashes, env vars, repeated separators and '.' segments are normalised textually first so trivial respellings do not slip past. """ lowered = (path or '').strip().replace('/', '\\').lower() for variable, root in LOG_PATH_ENVIRONMENT_ROOTS.items(): lowered = lowered.replace(variable, root) parts = [part for part in lowered.split('\\') if part not in ('', '.')] if not parts: return None # Drop a leading drive letter (or \\?\ style prefix) so roots match on any drive. body = list(parts) while body and re.match(r'^([a-z]:|\?)$', body[0]): body.pop(0) joined = '\\'.join(body) for root, label in LOG_PATH_DENIED_ROOTS: if joined == root or joined.startswith(root + '\\'): return f'Collecting files under {label} is not permitted' filename = parts[-1] extension = os.path.splitext(filename)[1] if filename.startswith('unattend') and extension == '.xml': return f'Collecting {filename} (a Windows answer file) is not permitted' if filename == 'autounattend.xml': return f'Collecting {filename} (a Windows answer file) is not permitted' if filename.startswith('appsettings') and extension == '.json': return f'Collecting {filename} (application configuration) is not permitted' if extension in LOG_PATH_DENIED_EXTENSIONS: return f'Collecting {extension} key material is not permitted' return None def _validate_log_path(target): """Reject a collect_log path that cannot be a readable local log file. Returns an error string, or None when the shape is acceptable. This is a SHAPE check for fast operator feedback. It cannot be a security boundary: the server has no view of the endpoint's filesystem, and any caller reaching this code already holds endpoints.updateapp. The agent performs its own independent validation before opening anything. Rejections here are the ones that are wrong no matter what is on disk: a relative path (ambiguous - relative to what, on a service running as SYSTEM), a traversal segment (an attempt to walk out of the directory the operator named, which an honest log path never needs), a UNC path (that would make the endpoint agent read from a file server using the machine account, turning a log collector into a network file reader), and an extension outside the text allowlist. """ path = (target or '').strip() if not path: return 'A log file path is required' if len(path) > LOG_PATH_MAX_LENGTH: return (f'Path is too long ({len(path)} characters); the maximum is ' f'{LOG_PATH_MAX_LENGTH}') normalized = path.replace('/', '\\') # Denylist before the shape checks so an env-var spelling of a sensitive # path is refused with the real reason, not "needs a drive letter". denied_reason = _denied_log_path_reason(normalized) if denied_reason is not None: return denied_reason # A UNC path would have the agent read from a network share as the machine # account. That is a different capability from "read a log off this PC" and # is not what this permission was granted for. if normalized.startswith('\\\\'): return ('UNC/network paths are not supported; collect logs from a ' 'local path on the endpoint') # Require a drive-letter absolute path. A bare or relative path resolves # against the agent service's working directory, which is not somewhere an # operator can reason about. if not re.match(r'^[A-Za-z]:\\', normalized): return ('Enter a full local path including the drive letter ' r'(for example C:\ProgramData\App\app.log)') # '..' would let a typed path escape the directory it appears to name. No # legitimate log path needs it, so it is refused rather than normalised - # normalising would silently read something other than what was reviewed. if any(part == '..' for part in normalized.split('\\')): return 'Path must not contain ".." segments' # Wildcards would make the collection non-deterministic: what it reads would # depend on the directory's contents at run time, so the history could not # say what was actually collected. if '*' in normalized or '?' in normalized: return 'Wildcards are not supported; name a single log file' extension = os.path.splitext(normalized)[1].lower() if extension not in LOG_PATH_EXTENSIONS: allowed = ', '.join(sorted(LOG_PATH_EXTENSIONS)) return (f'Only plaintext log files can be collected. Allowed ' f'extensions: {allowed}') return None def _record_log_collection(command, detail, content): """Persist a collected log for a command and prune that asset's history. Called inside report_command_result, before its commit, so the collection row, the pruning and the command's own finalisation all land together. A separate commit here could leave a collected log with no finalised command if the request failed in between. Pruning is per asset and runs on insert. Doing it here rather than on a schedule means the cap is a real invariant rather than something that is true most of the time: there is no window in which an asset has 40 rows waiting for a cleanup job that may not be running. """ parsed_modified = None raw_modified = (detail or {}).get('modifiedat') if raw_modified: parsed_modified = _parse_agent_datetime(raw_modified) text = content if isinstance(content, str) else None collection = EndpointLogCollection( assetid=command.assetid, endpointcommandid=command.endpointcommandid, # Prefer the path the AGENT resolved; fall back to what was queued so a # row is never left without a path to show. sourcepath=((detail or {}).get('path') or command.target or '')[:512], content=text, filesizebytes=_coerce_int((detail or {}).get('filesizebytes')), contentbytes=(len(text.encode('utf-8', 'replace')) if text is not None else None), linecount=_coerce_int((detail or {}).get('linecount')), istruncated=bool((detail or {}).get('truncated')), filemodifiedat=parsed_modified, # requestedby is the operator who queued the collection. The agent posts # this result with a collector token and has no user identity, so the # attribution has to come off the command row. collectedbyuserid=command.requestedby, collectedbyusername=_username_for(command.requestedby), collectedat=_naive_utcnow(), ) db.session.add(collection) # Make the new row visible to the pruning query below, so the cap counts it. db.session.flush() _prune_log_collections(command.assetid) return collection def _prune_log_collections(assetid): """Keep only the newest RETENTION_PER_ASSET collections for an asset. Ordered by collectedat DESC then id DESC: two collections can share a timestamp at second resolution, and without the id tiebreak the ordering would be non-deterministic, which is how a prune ends up deleting the row it just inserted. """ keep = EndpointLogCollection.RETENTION_PER_ASSET stale = (EndpointLogCollection.query .filter_by(assetid=assetid) .order_by(EndpointLogCollection.collectedat.desc(), EndpointLogCollection.endpointlogcollectionid.desc()) .offset(keep) .all()) for row in stale: db.session.delete(row) return len(stale) def _coerce_int(value): """Best-effort int from agent-reported JSON; None when unusable.""" try: return int(value) except (TypeError, ValueError): return None def _username_for(userid): """Username snapshot for an id, or None. Never raises.""" if not userid: return None user = db.session.get(User, userid) return getattr(user, 'username', None) if user is not None else None def _parse_agent_datetime(value): """Parse an agent ISO-8601 timestamp into a naive UTC datetime. The agent sends an offset-aware string. The column is naive-UTC like every other datetime in this schema, so the offset is applied and then dropped - keeping it would make this the one column that compares wrongly against the rest. """ if not value: return None try: text = str(value).strip().replace('Z', '+00:00') parsed = datetime.fromisoformat(text) except (TypeError, ValueError): return None if parsed.tzinfo is not None: parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None) return parsed def _resolve_run_script(target): """Resolve a run_script target to a script delivery descriptor. Accepts either "script:" 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. captureoutput too: an older agent that never saw the field simply keeps withholding successful output, which is the safe direction to fail. """ 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: 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', 'captureoutput': bool(script.captureoutput), }, None @endpoints_bp.route('//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('//scriptruns', methods=['GET']) @require_permission('endpoints.view') def list_endpoint_script_runs(asset_id: int): """The last few run_script commands for one endpoint, newest first. Separate from list_endpoint_commands because the Run Scripts tab is asking a narrower question - "what happened the last few times I pushed a script HERE" - and filtering a 20-row mixed feed client-side would silently show fewer than five script runs whenever other commands are interleaved. Carries resultdetail, which the plain command list does not. A script that exits zero produces a green badge and nothing else; the exit code, the context it ran as and the directory it ran in are the only evidence the operator gets that the run did anything at all. """ limit = request.args.get('limit', default=5, type=int) or 5 limit = max(1, min(limit, 20)) rows = (EndpointCommand.query .filter_by(assetid=asset_id, action='run_script') .order_by(EndpointCommand.createdat.desc()) .limit(limit) .all()) scriptnames = _script_names_for(rows) payload = [] for row in rows: data = row.to_dict(includedetail=True) data['scriptname'] = scriptnames.get(row.endpointcommandid) # Localize timestamps to site timezone for ts_field in ['createdat', 'claimedat', 'completedat']: if ts_field in data and data[ts_field]: data[ts_field] = _localize_timestamp(row.__dict__.get(ts_field)) payload.append(data) return success_response(payload) @endpoints_bp.route('//logcollections', methods=['GET']) @require_permission('endpoints.view') def list_endpoint_log_collections(asset_id: int): """Collected-log history for one endpoint, newest first. Metadata only - no content. The panel renders up to 20 rows and none of them need the log body, which can be hundreds of kilobytes each; sending them all would make opening the tab far more expensive than reading the one log the operator actually wants. Gated on endpoints.view because this is a listing. Reading a body, which is the part that discloses file contents, requires the collect permission - see get_endpoint_log_collection. """ rows = (EndpointLogCollection.query .filter_by(assetid=asset_id) .order_by(EndpointLogCollection.collectedat.desc(), EndpointLogCollection.endpointlogcollectionid.desc()) .limit(EndpointLogCollection.RETENTION_PER_ASSET) .all()) # Localize timestamps to site timezone payload = [] for row in rows: data = row.to_dict() if 'collectedat' in data and data['collectedat']: data['collectedat'] = _localize_timestamp(row.collectedat) payload.append(data) return success_response(payload) @endpoints_bp.route('/logcollections/', methods=['GET']) @require_permission('endpoints.updateapp') def get_endpoint_log_collection(collection_id: int): """One collected log including its content, for the scrollable pane. Deliberately a heavier permission than the list. The list says a log was collected; this returns what was IN it, which is the disclosure. It rides the same permission that authorised the collection, so nobody gains through this route what they could not have collected themselves. """ row = db.session.get(EndpointLogCollection, collection_id) if row is None: return error_response(ErrorCodes.NOT_FOUND, f'No log collection {collection_id}', http_code=404) return success_response(row.to_dict(includecontent=True)) @endpoints_bp.route('/logcollections//download', methods=['GET']) @require_permission('endpoints.updateapp') def download_endpoint_log_collection(collection_id: int): """Serve a collected log as a file attachment. Served as text/plain with an explicit attachment disposition so the browser saves it instead of rendering it. That matters beyond convenience: a collected log is untrusted endpoint content, and inline rendering of a file that happens to contain markup would execute it in the app's origin. X-Content-Type-Options is set for the same reason - it stops the browser sniffing past the declared text/plain and deciding the bytes are HTML. """ from flask import Response row = db.session.get(EndpointLogCollection, collection_id) if row is None: return error_response(ErrorCodes.NOT_FOUND, f'No log collection {collection_id}', http_code=404) audit_log( action='logcollectiondownloaded', entitytype='Asset', entityid=row.assetid, changes={ 'sourcepath': row.sourcepath, 'endpointlogcollectionid': row.endpointlogcollectionid, 'downloadedby': getattr(current_user, 'username', None), }, details=f'Downloaded collected log {row.sourcepath}', ) db.session.commit() response = Response(row.content or '', mimetype='text/plain; charset=utf-8') # The filename is already sanitised to [A-Za-z0-9._-] by the model, so it # cannot break out of the quoted header value. response.headers['Content-Disposition'] = ( f'attachment; filename="{row.filename}"') response.headers['X-Content-Type-Options'] = 'nosniff' return response def _script_names_for(rows): """Map command id -> script name for a batch of run_script rows. The command stores 'script:', which is not what the operator chose from the picker. Resolved in one query rather than per row. A script deleted since the run leaves the name absent, and the caller falls back to the raw target rather than inventing one. """ wanted = {} for row in rows: kind, _sep, raw = str(row.target or '').partition(':') if kind == 'script' and raw.isdigit(): wanted[row.endpointcommandid] = int(raw) if not wanted: return {} found = { script.scriptid: script.scriptname for script in Script.query.filter( Script.scriptid.in_(set(wanted.values()))).all() } return {cmdid: found[sid] for cmdid, sid in wanted.items() if sid in found} @endpoints_bp.route('//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('//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('//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('//policyholds/', 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('//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 : 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('//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('//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('//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') # The audit row and the hold are written further down, only once every # remaining validation has passed: a refused command must leave neither. policyoverridden = True else: policyoverridden = False # 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') # Capability, not authorization: does the agent on THIS endpoint actually # implement the action? An agent older than the action falls through to its # unknown-action default and answers "Unknown or disallowed action", which # reads as a permission refusal and sends the operator to check a grant they # already hold. Checked after the permission and policy gates so an # unauthorized caller still gets the authorization answer rather than a # version hint about a machine they may not act on. # # Deliberately a hard refusal, not a warning: the command is guaranteed to # fail, and queueing it anyway spends the one in-flight command slot and # writes a failed row that looks like the endpoint misbehaved. minagentversion = minimum_agent_version_for(action) if minagentversion: state = EndpointState.query.filter_by(assetid=asset_id).first() reportedversion = getattr(state, 'agentversion', None) if not agent_meets_minimum(reportedversion, minagentversion): return error_response( ErrorCodes.VALIDATION_ERROR, f'Action {action!r} requires agent {minagentversion} or newer; ' f'this endpoint reports ' f'{reportedversion or "no version"}. Update the agent on this ' f'endpoint to use it.') # 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') # collect_log reads a caller-supplied path. Validate its SHAPE here so an # obviously wrong path fails immediately with a message the operator can act # on, instead of consuming a poll cycle to come back as a generic endpoint # failure. This is a usability gate, not the security boundary: the agent # re-validates independently, because a server-side check cannot know what # is really on that disk and must never be the only thing standing between a # typed string and a file read. if action == 'collect_log': path_error = _validate_log_path(target) if path_error is not None: return error_response(ErrorCodes.VALIDATION_ERROR, path_error) 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) if policyoverridden: # Written after every validation above, so a refused command leaves no # hold and no policyoverridden row. Structured context goes in changes # (JSON-serialized by the model); details is a plain Text column. 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), ) 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('//agentupgrade', methods=['POST']) @require_permission('endpoints.agentupdate') def enqueue_agent_upgrade(asset_id: int): """Push the published agent build to one endpoint. Replacing the agent is self-modifying the control plane: the process that receives this command is the one being swapped. It therefore sits behind endpoints.agentupdate rather than endpoints.command or endpoints.deploy - see the permission's rationale in plugin.py. This does NOT enqueue directly. It prepares the delivery (packages the current build, seeds/refreshes the upgrade script) and then delegates to enqueue_endpoint_command as a run_script push. Writing a second enqueue path here would duplicate - and eventually drift from - the endpoint lock check, the Configuration Policy gate, stale-command reaping and the at-most-one-in-flight rule. Those are exactly the protections a fleet-wide agent swap must not be the one action to skip. Because it lands as run_script, the caller needs endpoints.runscript too (enforced by the delegate as the elevated permission for that action). That is deliberate: this delivers arbitrary SYSTEM PowerShell by the same mechanism as any other script push. """ # Cheap refusals first. The delegate enforces these again, but repackaging # and reseeding rewrite a shared package file and script row, which a caller # who is locked out or lacks endpoints.runscript must not be able to trigger. 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) elevated = elevated_permission_for('run_script') if elevated and not caller_holds(elevated): return error_response( ErrorCodes.FORBIDDEN, f'Action \'run_script\' requires the {elevated} permission', http_code=403) registration, seeded, error = prepare_agent_upgrade() if error is not None: return error payload = request.get_json(silent=True) or {} return queue_prepared_agent_upgrade(asset_id, registration, seeded, payload) def prepare_agent_upgrade(): """Package the current agent build and seed the upgrade script, once. Returns (registration, seeded, error_response). A migration wave calls this once and then queue_prepared_agent_upgrade per PC. """ from . import agentregistry, agentupgradescript # Refresh the package and the script body from the CURRENT staged build # before queueing, so an upgrade cannot ship a stale payload if the row was # written before the latest upload. Both operations are idempotent. try: registration = agentregistry.register_current_agent_package() except Exception as exc: # noqa: BLE001 - surfaced, not swallowed db.session.rollback() return None, None, error_response( ErrorCodes.VALIDATION_ERROR, f'Could not package the agent build: {exc}') if registration is None: return None, None, error_response( ErrorCodes.VALIDATION_ERROR, 'No agent build is published. Upload an agent binary before ' 'pushing an upgrade.') try: seeded = agentupgradescript.seed_upgrade_script() except Exception as exc: # noqa: BLE001 - surfaced, not swallowed db.session.rollback() return None, None, error_response( ErrorCodes.VALIDATION_ERROR, f'Could not prepare the upgrade script: {exc}') if seeded is None: return None, None, error_response( ErrorCodes.VALIDATION_ERROR, 'The agent upgrade script could not be prepared.') return registration, seeded, None def queue_prepared_agent_upgrade(asset_id, registration, seeded, payload): """Queue the prepared upgrade for one endpoint through the real enqueue route. Returns the delegate's response; audits only a push that queued. """ # Rewrite the request body to the equivalent run_script push and hand it to # the real enqueue route. The policyoverride/overridereason fields are # forwarded untouched so break-glass keeps behaving exactly as it does for # a normal script push, including its admin-role and reason requirements. forwarded = { 'action': 'run_script', 'target': 'script:{}'.format(seeded['scriptid']), } if payload.get('policyoverride'): forwarded['policyoverride'] = True forwarded['overridereason'] = payload.get('overridereason') or '' # Swap the parsed JSON for the delegate. Flask caches the parsed body on # the request, so setting the cache is what makes get_json() return this. request._cached_json = (forwarded, forwarded) response = enqueue_endpoint_command(asset_id) # Audit only a push that actually queued. The delegate returns a non-201 # for a lock, a policy denial or an in-flight command, and recording those # as upgrades would put events in the log for pushes that never happened. status = getattr(response, 'status_code', None) if status == 201: AuditLog.log( 'created', 'AgentUpgrade', entityid=asset_id, entityname='agent {}'.format(registration['agentversion']), details='agent upgrade queued for asset {} (version {}, script {})' .format(asset_id, registration['agentversion'], seeded['scriptid'])) db.session.commit() return response @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 = _agent_auth_or_error() if error is not None: return error asset_id, hostname, error = _bind_agent_hostname( request.args.get('hostname')) if error is not None: return error 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 = _agent_auth_or_error() if error is not None: return error asset_id, hostname, error = _bind_agent_hostname( request.args.get('hostname')) if error is not None: return error 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//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 = _agent_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) # Authorization: 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. # # A credential proves which endpoint is calling. The shared key does not, so # its hostname binding is a floor, not a fix. hostname = (request.args.get('hostname') or '').strip() caller_asset_id, error = agentauth.bind_hostname(hostname) if error is not None: return error if agentauth.is_credential_caller(): if caller_asset_id != command.assetid: return agentauth.asset_mismatch_response() elif not hostname: return error_response( ErrorCodes.FORBIDDEN, 'Command result requires an endpoint credential or hostname parameter', http_code=403) 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') # A successful collect_log carries the file body in detail.content. Move it # into its own row and strip it from the detail before that detail is # serialised onto the command: storing it twice would double the cost of the # largest payload this API accepts, and resultjson is read by every command # history list that only wants a status. # # Only on success. A failed collection's detail is small (the reason it # failed) and belongs on the command, where failures are already read. if command.action == 'collect_log' and succeeded and isinstance(detail, dict): detail = dict(detail) content = detail.pop('content', None) _record_log_collection(command, detail, content) 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/', 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 = _agent_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/); # 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/', methods=['GET']) @endpoints_bp.route('/apppackage/app//', 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 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 = _agent_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/', 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 = _agent_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 current_app.logger.info('_send_operator_report_alert called for report %d, asset %d, hostname %s', report.operatorreportid, asset_id, hostname) 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'

Screenshot: View Screenshot

' except Exception as e: current_app.logger.warning('Failed to generate screenshot URL: %s', e) # Get PC access domain to build FQDN 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'

Machine#: {safe_machine_number}

' # Build HTML body (don't use render_email to avoid subject duplication) html_body = f'''

Operator Issue Report

New Issue Report from Shop Floor

Device: {safe_fqdn}

{machine_body_line}

Reason: {safe_reason}

Reported: {escape(reported_at)}

Details:

{safe_details}

{screenshot_info}

View Device in ShopDB


Sent by ShopDB.

''' 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 = _agent_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() asset_id, error = agentauth.bind_hostname(hostname) if error is not None: return error if not hostname and asset_id is not None: hostname = _hostname_for_asset(asset_id) or '' if not hostname: return error_response(ErrorCodes.VALIDATION_ERROR, 'hostname is required') 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//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//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 BusinessUnit, 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 joined the contract surface at 0.34.0 precisely so this is a # primary-key get through the model rather than db.text() against a core # table. A raw query here respected the letter of ADR-002 and defeated its # point: hidden coupling to the core schema that no contract bump catches. if businessunitid is not None: unit = db.session.get(BusinessUnit, businessunitid) facts['businessunit'] = unit.businessunit if unit is not None else None 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 (None) or empty scope means the whole fleet - that is the default and the historical behaviour. A scope that is present but not an object (a list, a number, unparseable text) fails CLOSED: nobody can tell what it was meant to target, and create/update now refuse it, so such a row is corrupt rather than a fleet-wide intent. A JSON string holding an object (legacy TEXT column rows) is parsed. 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 isinstance(alertscope, str): if not alertscope.strip(): return True try: alertscope = json.loads(alertscope) except ValueError: return False if alertscope is None: return True if not isinstance(alertscope, dict): return False if 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 = _agent_auth_or_error() if error is not None: return error asset_id, hostname, error = _bind_agent_hostname( request.args.get('hostname')) if error is not None: return error 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.starttime.is_(None), Notification.starttime <= now), 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, # TRANSITION, remove once no tray older than 1.7.0 is left in the # fleet: a deployed 1.6.x tray fetches this feed itself and # deserializes case-sensitively, so dropping these two keys makes # every alert vanish on those PCs with no error anywhere. The # lowercase pair above is the contract; these two are a bridge. # Any NEW field is lowercase only. 'notificationId': notification.notificationid, 'typeColor': ntype.typecolor or _DEFAULT_ALERT_TYPE_COLOR, }) # ONE line, lazily formatted. This runs on every poll from every endpoint, # so an f-string here is formatting work paid at every log level (the cost # HOTFIX 1ed0c4d removed from the collector check-in); per-notification and # per-alert lines multiplied it by the fleet alert count. current_app.logger.debug( '[fleet_alerts] hostname=%s assetid=%s candidates=%d returned=%d', hostname, asset_id, len(results), len(alerts)) return success_response(alerts) @endpoints_bp.route('/alerts//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 = _agent_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') # assetid 0 or absent means "the caller". try: assetid = int(assetid) if assetid not in (None, '') else 0 except (TypeError, ValueError): return error_response(ErrorCodes.VALIDATION_ERROR, 'assetid must be an integer') caller_assetid, error = agentauth.bind_hostname(hostname) if error is not None: return error if agentauth.is_credential_caller(): if assetid and assetid != caller_assetid: return agentauth.asset_mismatch_response() assetid = caller_assetid hostname = hostname or _hostname_for_asset(assetid) or '' else: if not hostname: return error_response(ErrorCodes.VALIDATION_ERROR, 'hostname is required') if not assetid: if caller_assetid is None: return error_response( ErrorCodes.NOT_FOUND, f'Unknown hostname: {hostname!r}', http_code=404) assetid = caller_assetid 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') # Same targeting as the /alerts list. An alert this PC was never shown # answers exactly like an unknown one, so the route cannot probe scopes. facts = _endpoint_scope_facts(hostname) or {} if not _alert_targets_endpoint(notification.alertscope, facts): return error_response(ErrorCodes.NOT_FOUND, f'Notification {notification_id} not found', http_code=404) # 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() current_app.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) # ============================================================================= # Server certificate management # ============================================================================= def _sha1_thumbprint(pembytes): """Uppercase hex SHA-1 of a PEM certificate, as Windows shows a thumbprint. None when the file does not parse. """ from cryptography import x509 from cryptography.hazmat.primitives import hashes try: certificate = x509.load_pem_x509_certificate(pembytes) except ValueError: return None return certificate.fingerprint(hashes.SHA1()).hex().upper() def _spki_sha256(pembytes): """Lowercase hex SHA-256 of a PEM certificate's SubjectPublicKeyInfo. The value agents pin. None when the file does not parse. """ from cryptography import x509 from cryptography.hazmat.primitives import serialization try: certificate = x509.load_pem_x509_certificate(pembytes) except ValueError: return None spki = certificate.public_key().public_bytes( serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo) return hashlib.sha256(spki).hexdigest() @endpoints_bp.route('/settings/server-cert', methods=['GET']) @require_role('admin') def get_server_certificate(): """Get details about the current server certificate. Returns certificate subject, expiry, thumbprint, and validation status. Admin-only - certificate details are sensitive operational data. """ import subprocess from pathlib import Path cert_path = Path('certs/server.crt') if not cert_path.exists(): return error_response(ErrorCodes.NOT_FOUND, 'Server certificate not found at certs/server.crt', http_code=404) try: # Extract certificate details using openssl subject_result = subprocess.run( ['openssl', 'x509', '-in', str(cert_path), '-noout', '-subject'], capture_output=True, text=True, check=True ) subject_cn = subject_result.stdout.strip().split('CN = ')[-1] if 'CN = ' in subject_result.stdout else 'Unknown' issuer_result = subprocess.run( ['openssl', 'x509', '-in', str(cert_path), '-noout', '-issuer'], capture_output=True, text=True, check=True ) issuer = issuer_result.stdout.strip().split('issuer=')[-1] if 'issuer=' in issuer_result.stdout else 'Unknown' dates_result = subprocess.run( ['openssl', 'x509', '-in', str(cert_path), '-noout', '-dates'], capture_output=True, text=True, check=True ) dates_lines = dates_result.stdout.strip().split('\n') not_before = dates_lines[0].split('notBefore=')[-1] if len(dates_lines) > 0 else None not_after = dates_lines[1].split('notAfter=')[-1] if len(dates_lines) > 1 else None thumbprint_result = subprocess.run( ['openssl', 'x509', '-in', str(cert_path), '-noout', '-fingerprint', '-sha256'], capture_output=True, text=True, check=True ) thumbprint = thumbprint_result.stdout.strip().split('=')[-1] if '=' in thumbprint_result.stdout else 'Unknown' # Get SANs san_result = subprocess.run( ['openssl', 'x509', '-in', str(cert_path), '-noout', '-ext', 'subjectAltName'], capture_output=True, text=True ) san_list = [] if san_result.returncode == 0 and san_result.stdout: san_text = san_result.stdout.strip() if 'DNS:' in san_text or 'IP:' in san_text: san_parts = san_text.split('\n') for line in san_parts: if 'DNS:' in line or 'IP:' in line: items = line.strip().split(', ') san_list.extend([item.strip() for item in items if item.strip()]) # Calculate days until expiry if not_after: try: from dateutil import parser expiry_date = parser.parse(not_after) now = datetime.now(timezone.utc) if expiry_date.tzinfo is None: expiry_date = expiry_date.replace(tzinfo=timezone.utc) expires_in_days = (expiry_date - now).days except Exception: expires_in_days = None else: expires_in_days = None # Check if self-signed is_self_signed = (subject_cn == issuer or 'CN = ' + subject_cn in issuer) return success_response({ 'subject_cn': subject_cn, 'issuer': issuer, 'not_before': not_before, 'not_after': not_after, 'thumbprint': thumbprint, 'spkisha256': _spki_sha256(cert_path.read_bytes()), 'sha1thumbprint': _sha1_thumbprint(cert_path.read_bytes()), 'san_list': san_list, 'expires_in_days': expires_in_days, 'is_self_signed': is_self_signed }) except subprocess.CalledProcessError as e: current_app.logger.error(f'Failed to read certificate: {e}') return error_response(ErrorCodes.INTERNAL_ERROR, f'Failed to read certificate details: {e.stderr}', http_code=500) except Exception as e: current_app.logger.error(f'Unexpected error reading certificate: {e}') return error_response(ErrorCodes.INTERNAL_ERROR, f'Failed to read certificate: {str(e)}', http_code=500) @endpoints_bp.route('/settings/server-cert/download', methods=['GET']) @require_permission('endpoints.fleet') def download_server_certificate(): """The PUBLIC server certificate as a DER .cer file, never the key. For PCs on agents older than 1.20, which check the certificate chain when they download the upgrade script: import this into the PC's Trusted Root store before a migration wave reaches it. """ from cryptography import x509 from cryptography.hazmat.primitives import serialization cert_path = _server_cert_dir() / 'server.crt' if not cert_path.exists(): return error_response(ErrorCodes.NOT_FOUND, 'Server certificate not found', http_code=404) try: certificate = x509.load_pem_x509_certificate(cert_path.read_bytes()) except ValueError: return error_response(ErrorCodes.INTERNAL_ERROR, 'The server certificate file does not parse', http_code=500) body = certificate.public_bytes(serialization.Encoding.DER) response = current_app.response_class(body, mimetype='application/pkix-cert') response.headers['Content-Disposition'] = \ 'attachment; filename="shopdb-server.cer"' response.headers['Cache-Control'] = 'no-store' return response _LIVE_CHECK_TIMEOUT_SECONDS = 5 def _presented_certificate(host, port): """DER bytes of the leaf certificate a TLS server presents, never trusted. Verification is off ONLY so a self-signed or internal-CA certificate can be read at all; nothing is sent over the connection and nothing it returns is believed except the certificate bytes, which are hashed and compared. """ import ipaddress import socket import ssl context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) context.check_hostname = False context.verify_mode = ssl.CERT_NONE try: ipaddress.ip_address(host) servername = None except ValueError: servername = host with socket.create_connection((host, port), timeout=_LIVE_CHECK_TIMEOUT_SECONDS) as raw: with context.wrap_socket(raw, server_hostname=servername) as secured: return secured.getpeercert(binary_form=True) @endpoints_bp.route('/settings/server-cert/live', methods=['GET']) @require_permission('endpoints.fleet') def live_server_certificate(): """The certificate PCs actually receive from this site's public address. The file on disk is what this app manages, but a PC pins whatever the site presents: an IIS binding, a reverse proxy or a load balancer in front of the app can serve a different certificate. The host comes from the site_base_url setting and never from the request, and only a TLS handshake is made (no HTTP request), so there is no redirect to follow. """ from urllib.parse import urlsplit from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization baseurl = str(Setting.get('site_base_url') or '').strip() if not baseurl: return success_response({'configured': False}) result = { 'configured': True, 'url': baseurl, 'presentedspkisha256': None, 'presentedsha1thumbprint': None, 'matchesfile': False, 'error': None, } try: parts = urlsplit(baseurl) host = parts.hostname port = parts.port or 443 except ValueError: host = None if not host: result['error'] = 'site_base_url is not a valid URL' return success_response(result) if (parts.scheme or '').lower() != 'https': result['error'] = 'site_base_url does not use https, so no certificate is presented' return success_response(result) try: der = _presented_certificate(host, port) certificate = x509.load_der_x509_certificate(der) except Exception as exc: # noqa: BLE001 - reported to the admin, not raised result['error'] = 'Could not read the certificate from {}:{} ({})'.format( host, port, (str(exc) or exc.__class__.__name__)[:200]) return success_response(result) spki = certificate.public_key().public_bytes( serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo) result['presentedspkisha256'] = hashlib.sha256(spki).hexdigest() result['presentedsha1thumbprint'] = certificate.fingerprint( hashes.SHA1()).hex().upper() certpath = _server_cert_dir() / 'server.crt' if certpath.exists(): filepin = _spki_sha256(certpath.read_bytes()) result['matchesfile'] = bool( filepin and hmac.compare_digest(filepin, result['presentedspkisha256'])) return success_response(result) _CERT_HOSTNAME_PATTERN = re.compile( r'^(\*\.)?(?=.{1,253}$)[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?' r'(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*$') def _server_cert_dir(): """Directory nginx reads server.crt / server.key from.""" from pathlib import Path return Path('certs') def _is_cert_hostname_or_ip(value): """True when value is a plain hostname (optionally *.wildcard) or an IP. Anything else - newlines, '=', brackets, spaces - is refused rather than escaped: these values end up in the certificate and in audit text. """ import ipaddress if not isinstance(value, str) or not value: return False try: ipaddress.ip_address(value) return True except ValueError: pass return bool(_CERT_HOSTNAME_PATTERN.match(value)) def _parse_cert_san(san): """Split 'DNS:a,IP:b' into (dnsnames, ipaddresses) or raise ValueError.""" import ipaddress dnsnames, ipaddresses = [], [] if not san: return dnsnames, ipaddresses for item in (part.strip() for part in san.split(',')): if not item: continue if item[:4].upper() == 'DNS:': name = item[4:].strip() if not _is_cert_hostname_or_ip(name): raise ValueError(f'Invalid DNS name in san: {name!r}') dnsnames.append(name) elif item[:3].upper() == 'IP:': address = item[3:].strip() try: ipaddress.ip_address(address) except ValueError: raise ValueError(f'Invalid IP address in san: {address!r}') ipaddresses.append(address) else: raise ValueError( f'Invalid san entry {item!r}; use DNS:name or IP:address') return dnsnames, ipaddresses def _validate_certificate_pair(cert_bytes, key_bytes): """Parse a PEM certificate and unencrypted PEM key; check they belong together. Returns an error string, or None when the pair is usable. """ from cryptography import x509 from cryptography.hazmat.primitives import serialization try: certificate = x509.load_pem_x509_certificate(cert_bytes) except Exception: return 'Invalid certificate (must be a PEM-encoded X.509 certificate)' try: private_key = serialization.load_pem_private_key(key_bytes, password=None) except TypeError: return 'Private key must be unencrypted (no passphrase)' except Exception: return 'Invalid private key (must be a PEM-encoded private key)' public_format = serialization.PublicFormat.SubjectPublicKeyInfo encoding = serialization.Encoding.DER try: matches = (certificate.public_key().public_bytes(encoding, public_format) == private_key.public_key().public_bytes(encoding, public_format)) except Exception: matches = False if not matches: return 'The private key does not match the certificate public key' return None def _install_server_certificate(cert_bytes, key_bytes): """Back up the current pair, then atomically replace it with the new one. Only called with a pair that has already been validated. The new files are written to unique temp names in the same directory first, so a failure before the swap leaves the existing certificate untouched. """ import shutil import uuid cert_dir = _server_cert_dir() cert_dir.mkdir(parents=True, exist_ok=True) cert_path = cert_dir / 'server.crt' key_path = cert_dir / 'server.key' token = uuid.uuid4().hex cert_temp = cert_dir / f'server.crt.{token}.tmp' key_temp = cert_dir / f'server.key.{token}.tmp' try: cert_temp.write_bytes(cert_bytes) key_temp.write_bytes(key_bytes) key_temp.chmod(0o600) stamp = int(time.time()) if cert_path.exists(): shutil.copy2(cert_path, cert_dir / f'server.crt.backup.{stamp}') if key_path.exists(): shutil.copy2(key_path, cert_dir / f'server.key.backup.{stamp}') os.replace(key_temp, key_path) os.replace(cert_temp, cert_path) finally: for leftover in (cert_temp, key_temp): try: leftover.unlink() except OSError: pass def _build_self_signed_certificate(cn, dnsnames, ipaddresses, days): """Return (cert_pem, key_pem) for a new RSA-2048 self-signed certificate.""" import ipaddress from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID key = rsa.generate_private_key(public_exponent=65537, key_size=2048) name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) now = datetime.now(timezone.utc) builder = (x509.CertificateBuilder() .subject_name(name) .issuer_name(name) .public_key(key.public_key()) .serial_number(x509.random_serial_number()) .not_valid_before(now - timedelta(minutes=5)) .not_valid_after(now + timedelta(days=days))) alternatives = ([x509.DNSName(value) for value in dnsnames] + [x509.IPAddress(ipaddress.ip_address(value)) for value in ipaddresses]) if alternatives: builder = builder.add_extension( x509.SubjectAlternativeName(alternatives), critical=False) certificate = builder.sign(key, hashes.SHA256()) cert_pem = certificate.public_bytes(serialization.Encoding.PEM) key_pem = key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()) return cert_pem, key_pem def _restart_nginx(): """Ask compose to restart nginx; failures are logged, not raised.""" import subprocess try: result = subprocess.run(['docker-compose', 'restart', 'nginx'], capture_output=True, text=True, timeout=30) if result.returncode == 0: current_app.logger.info('Nginx restarted with the new certificate') else: current_app.logger.warning( f'Nginx restart returned {result.returncode}: {result.stderr}') except Exception as restart_error: current_app.logger.error(f'Failed to restart nginx: {restart_error}') @endpoints_bp.route('/settings/server-cert/generate', methods=['POST']) @require_role('admin') def generate_server_certificate(): """Generate a new self-signed server certificate. Expects JSON body: { "cn": "", "san": "DNS:,IP:
", "days": 365 } cn and every san entry must be a plain hostname or IP address; anything else is refused. The pair is built in memory (no openssl config file for a crafted value to rewrite) and validated before the existing certificate is backed up and replaced. Admin-only - generates new TLS certificate and restarts nginx. """ payload = request.get_json(silent=True) or {} cn = payload.get('cn') if isinstance(payload.get('cn'), str) else '' cn = cn.strip() san = payload.get('san') if isinstance(payload.get('san'), str) else '' san = san.strip() days = payload.get('days', 365) if not cn: return error_response(ErrorCodes.VALIDATION_ERROR, 'Common Name (cn) is required', http_code=400) if len(cn) > 64 or not _is_cert_hostname_or_ip(cn): return error_response(ErrorCodes.VALIDATION_ERROR, 'Common Name (cn) must be a hostname or IP address ' 'of at most 64 characters', http_code=400) try: dnsnames, ipaddresses = _parse_cert_san(san) except ValueError as exc: return error_response(ErrorCodes.VALIDATION_ERROR, str(exc), http_code=400) if isinstance(days, bool) or not isinstance(days, int) or days < 1 or days > 3650: return error_response(ErrorCodes.VALIDATION_ERROR, 'days must be an integer between 1 and 3650', http_code=400) try: cert_bytes, key_bytes = _build_self_signed_certificate( cn, dnsnames, ipaddresses, days) except Exception as e: current_app.logger.error(f'Unexpected error generating certificate: {e}') return error_response(ErrorCodes.INTERNAL_ERROR, f'Failed to generate certificate: {str(e)}', http_code=500) pair_error = _validate_certificate_pair(cert_bytes, key_bytes) if pair_error is not None: return error_response(ErrorCodes.INTERNAL_ERROR, f'Generated certificate failed validation: {pair_error}', http_code=500) try: _install_server_certificate(cert_bytes, key_bytes) except Exception as e: current_app.logger.error(f'Failed to install generated certificate: {e}') return error_response(ErrorCodes.INTERNAL_ERROR, f'Failed to write certificate files: {str(e)}', http_code=500) current_app.logger.info(f'Generated new server certificate: CN={cn}, days={days}') _restart_nginx() AuditLog.log('endpoints.server_cert.generated', 'ServerCertificate', entityname=cn, changes={'cn': cn, 'dns': dnsnames, 'ip': ipaddresses, 'days': days}, details=f'Generated self-signed certificate: CN={cn}, ' f'SANs={san}, validity={days} days') db.session.commit() return success_response({ 'message': 'Certificate generated successfully. Nginx is restarting.' }) @endpoints_bp.route('/settings/server-cert/upload', methods=['POST']) @require_role('admin') def upload_server_certificate(): """Upload a new server certificate and private key. Expects multipart/form-data with: - cert: Certificate file (.crt or .pem) - key: Private key file (.key or .pem) Both are parsed and the key is checked against the certificate public key BEFORE anything on disk is touched; only a valid, matching pair causes the existing files to be backed up and replaced. Admin-only - replaces TLS certificate and restarts nginx. """ if 'cert' not in request.files or 'key' not in request.files: return error_response(ErrorCodes.VALIDATION_ERROR, 'Both cert and key files are required', http_code=400) cert_file = request.files['cert'] key_file = request.files['key'] if not cert_file.filename or not key_file.filename: return error_response(ErrorCodes.VALIDATION_ERROR, 'Both cert and key files must be provided', http_code=400) cert_bytes = cert_file.read() key_bytes = key_file.read() pair_error = _validate_certificate_pair(cert_bytes, key_bytes) if pair_error is not None: return error_response(ErrorCodes.VALIDATION_ERROR, pair_error, http_code=400) try: _install_server_certificate(cert_bytes, key_bytes) except Exception as e: current_app.logger.error(f'Failed to write certificate files: {e}') return error_response(ErrorCodes.INTERNAL_ERROR, f'Failed to write certificate files: {str(e)}', http_code=500) current_app.logger.info('Uploaded new server certificate and private key') _restart_nginx() AuditLog.log('endpoints.server_cert.uploaded', 'ServerCertificate', entityname=cert_file.filename, details=f'Uploaded certificate from files: ' f'{cert_file.filename}, {key_file.filename}') db.session.commit() return success_response({ 'message': 'Certificate uploaded successfully. Nginx is restarting.' })