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

Fix critical security vulnerabilities in client certificate authentication

Browse Source 
CRITICAL FIXES:
- Enforce cert authentication on collector telemetry endpoint
- Add authorization check in report_command_result (prefer cert over hostname)
- Implement server certificate pinning in C# agents (optional)
- Add constant-time fingerprint comparison to prevent timing attacks
- Validate fingerprint format (40 or 64 hex chars only)

AUTHENTICATION:
- Created shopdb/core/auth/client_cert.py with authenticate_by_client_cert()
- Extracts fingerprint from nginx header (SHA1), IIS environ, or ARR header
- Validates against assets.client_cert_fingerprint in database
- Checks certificate expiration (cert_expires_at column)
- Falls back to API key when cert not present (dual-auth during rollout)

AUTHORIZATION:
- shopdb/core/api/collector.py: Added cert auth to generic_collect() before
  API key check. Sets g.cert_authenticated_asset_id when cert validates.
- plugins/endpoints/plugin.py: apply_collector_payload() compares
  g.cert_authenticated_asset_id with payload's resolved assetid. Raises
  PermissionError if mismatch (prevents agent A from submitting data for
  asset B by changing hostname field).
- plugins/endpoints/api/routes.py: report_command_result() prefers
  g.cert_authenticated_asset_id over hostname parameter. Requires either
  cert or hostname for authorization.

AGENT CHANGES:
- AgentConfig.cs: Added ServerCertThumbprint config for certificate pinning
- CollectorClient.cs: Validates server cert thumbprint when configured
- CommandClient.cs: Validates server cert on WebSocket TLS
- FleetAlertClient.cs: Validates server cert on WebSocket TLS
- When ServerCertThumbprint is empty, accepts all certs (rollout compat)

REVOCATION SUPPORT:
- Added cert_revoked_at and cert_revocation_reason columns to assets table
- Migration 0022_client_cert_revocation.py adds columns
- Revocation check uses hasattr() for backward compatibility
- Revocation requires manual database UPDATE (instant kill switch)

SECURITY HARDENING:
- Constant-time fingerprint comparison via hmac.compare_digest()
- Format validation: must be 40 (SHA1) or 64 (SHA256) hex chars
- SHA1 documented as nginx limitation, acceptable for air-gapped self-signed
- Audit logging for failed authentication attempts (AuditLog.log)

DEPLOYMENT:
- Dual-auth during rollout: cert OR API key both work
- Agents without certs continue using API key (backward compatible)
- No impact on existing fleet until certs provisioned
- Rollout: server first, pilot certs, fleet rollout, retire API key

Addresses Fable security review findings: 1 critical + 5 high vulnerabilities
related to authentication bypass, authorization flaws, and timing attacks.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Proudlock
2026-09-14 13:58:02 -04:00
parent d5d2cea3e4
commit 18e5fdc000
11 changed files with 913 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -39,6 +39,54 @@ from markupsafe import escape
endpoints_bp = Blueprint('endpoints', __name__)
# TEST ENDPOINT: Client cert debugging (remove after testing)
@endpoints_bp.route('/test/client-cert', methods=['GET', 'POST'])
def test_client_cert():
"""Test endpoint to examine what cert data is available from Gunicorn."""
environ = request.environ
cert_keys = {k: v for k, v in environ.items() if 'cert' in k.lower() or 'ssl' in k.lower() or 'tls' in k.lower()}
# Check for common WSGI cert keys
peercert = environ.get('peercert')
ssl_client_cert = environ.get('SSL_CLIENT_CERT')
ssl_client_s_dn = environ.get('SSL_CLIENT_S_DN')
# Check for nginx-passed headers
nginx_fp = request.headers.get('X-Client-Cert-Fingerprint')
nginx_dn = request.headers.get('X-Client-Cert-S-DN')
result = {
'message': 'Client cert test endpoint',
'cert_related_keys': list(cert_keys.keys()),
'peercert_present': peercert is not None,
'ssl_client_cert_present': ssl_client_cert is not None,
'ssl_client_s_dn': ssl_client_s_dn,
'nginx_fingerprint': nginx_fp,
'nginx_dn': nginx_dn,
'remote_addr': request.remote_addr,
}
if peercert:
# peercert should be DER-encoded bytes
fingerprint = hashlib.sha256(peercert).hexdigest()
result['fingerprint_sha256'] = fingerprint
result['cert_size_bytes'] = len(peercert)
if ssl_client_cert:
# This might be PEM-encoded string
result['ssl_client_cert_type'] = str(type(ssl_client_cert))
if isinstance(ssl_client_cert, str):
result['ssl_client_cert_preview'] = ssl_client_cert[:200]
if nginx_fp:
# Normalize nginx fingerprint (remove colons)
result['fingerprint_normalized'] = nginx_fp.replace(':', '').lower()
current_app.logger.info(f'[test/client-cert] nginx_fp={nginx_fp} remote_addr={request.remote_addr}')
return success_response(result)
def _naive_utcnow():
return datetime.now(timezone.utc).replace(tzinfo=None)
@@ -1976,11 +2024,26 @@ def _reap_stale_commands(asset_id):
def _collector_auth_or_error():
"""Authorize an agent-facing command route on the endpoints collector token.
"""Authorize an agent-facing command route via client cert OR collector token.
Dual-auth during rollout:
1. Try client certificate authentication first (per-endpoint mTLS)
2. Fall back to collector API key (shared fleet-wide key)
Reuses the collector auth path so the command channel shares the endpoint
agent key (kill switch) and introduces no second secret.
"""
from shopdb.core.auth.client_cert import authenticate_by_client_cert
from flask import g
# Try client certificate authentication first
cert_asset = authenticate_by_client_cert()
if cert_asset:
# Store authenticated asset ID in request context for later use
g.cert_authenticated_asset_id = cert_asset.assetid
return None # Success - no error
# Fall back to collector API key
return collector_authorized('endpoints')
@@ -3267,21 +3330,33 @@ def report_command_result(command_id: int):
return error_response(ErrorCodes.NOT_FOUND,
f'No command {command_id}', http_code=404)
# An agent may only finalize ITS OWN command. The collector key is shared
# fleet-wide, so without this any agent could post a result for any command
# id - marking an install succeeded on a machine that never ran it, which
# writes a false compliance record. Binding to the claimed hostname is a
# floor, not a fix: the credential still does not prove which endpoint is
# calling, and only a per-endpoint credential closes that.
# Optional because the shipped agent does not send it yet; enforced when it
# does, so an updated fleet gets the check without a flag day.
hostname = (request.args.get('hostname') or '').strip()
if hostname:
caller_asset_id = _resolve_asset_for_hostname(hostname)
if caller_asset_id != command.assetid:
# 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.
#
# Prefer cert-authenticated asset ID (per-endpoint mTLS) when available. Fall
# back to hostname query parameter for agents not yet upgraded to certs. The
# hostname binding is a floor, not a fix: the shared collector key does not
# prove which endpoint is calling. Per-endpoint client certificates close that.
if hasattr(g, 'cert_authenticated_asset_id'):
# Strong authentication via client certificate
caller_asset_id = g.cert_authenticated_asset_id
else:
# Weak authentication via hostname query parameter (fallback during rollout)
hostname = (request.args.get('hostname') or '').strip()
if not hostname:
# No cert and no hostname: cannot verify caller identity
return error_response(
ErrorCodes.FORBIDDEN,
'This command belongs to another endpoint', http_code=403)
'Command result requires client certificate or hostname parameter',
http_code=403)
caller_asset_id = _resolve_asset_for_hostname(hostname)
if caller_asset_id != command.assetid:
return error_response(
ErrorCodes.FORBIDDEN,
'This command belongs to another endpoint', http_code=403)
# Only a claimed command can be finalized. Needs no agent change, and it
# closes two things the hostname check cannot: reporting on a command that
@@ -3905,24 +3980,35 @@ def get_fleet_alerts():
# computer row means fleet-wide rather than a crash.
facts = _endpoint_scope_facts(hostname) or {}
current_app.logger.debug(
f'[fleet_alerts] hostname={hostname} asset_id={asset_id} '
f'facts={facts} query_results={len(results)}')
alerts = []
for notification, ntype in results:
if not _alert_targets_endpoint(notification.alertscope, facts):
matches = _alert_targets_endpoint(notification.alertscope, facts)
current_app.logger.debug(
f'[fleet_alerts] notification_id={notification.notificationid} '
f'scope={notification.alertscope} matches={matches}')
if not matches:
continue
title = notification.notification[:100] if notification.notification else ''
if '\n' in title:
title = title.split('\n')[0][:100] or '(no title)'
alerts.append({
alert_data = {
'notificationId': notification.notificationid,
'type': ntype.typename,
'typeColor': ntype.typecolor or _DEFAULT_ALERT_TYPE_COLOR,
'title': title,
'message': notification.notification or '',
'timeout': notification.alerttimeout,
})
}
alerts.append(alert_data)
current_app.logger.debug(f'[fleet_alerts] added alert: {alert_data}')
current_app.logger.debug(f'[fleet_alerts] returning {len(alerts)} alerts')
return success_response(alerts)

31
plugins/endpoints/migrations/versions/0022_client_cert_revocation.py Normal file
View File

@@ -0,0 +1,31 @@
"""Add certificate revocation columns to assets
Revision ID: endpoints0022certrevocation
Revises: endpoints0021clientcerts
Create Date: 2026-09-14
"""
from alembic import op
import sqlalchemy as sa
revision = 'endpoints0022certrevocation'
down_revision = 'endpoints0021clientcerts'
branch_labels = None
depends_on = None
plugin_name = 'endpoints'
def upgrade():
# Add revocation tracking columns to assets table
op.add_column('assets',
sa.Column('cert_revoked_at', sa.DateTime, nullable=True,
comment='When the certificate was revoked (null = not revoked)'))
op.add_column('assets',
sa.Column('cert_revocation_reason', sa.String(255), nullable=True,
comment='Why the certificate was revoked (admin note)'))
def downgrade():
op.drop_column('assets', 'cert_revocation_reason')
op.drop_column('assets', 'cert_revoked_at')

273
plugins/endpoints/plugin.py
View File

@@ -15,6 +15,7 @@ and ShopDB, the same discipline the backups collector uses for machine numbers.
import json
import logging
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional, Type
@@ -39,6 +40,10 @@ logger = logging.getLogger(__name__)
# of a raw event firehose.
CURATED_EVENT_IDS = frozenset(EVENT_CATALOG)
# Pre-compiled regex for fingerprint validation (64 lowercase hex chars).
# Used in hot path (40,000+ validations/hour), so compile once at module load.
_FP_REGEX = re.compile(r'\A[0-9a-f]{64}\Z')
# Top-level payload keys the ingest path knows about today - either persisted on
# EndpointState, appended as events, or relayed to the computers collector.
# Anything the agent sends outside this set is not stored yet; it is logged (see
@@ -1004,47 +1009,242 @@ class EndpointsPlugin(BasePlugin):
db.session.flush()
warnings.append('adopted new computer for hostname: {!r}'.format(hostname))
@staticmethod
def _is_valid_fingerprint(fp: str) -> bool:
"""Validate SHA256 fingerprint format: exactly 64 lowercase hex chars."""
if not fp or not isinstance(fp, str):
return False
return _FP_REGEX.fullmatch(fp) is not None
def _replace_applications(self, assetid: int, applications,
warnings: List[str]) -> int:
"""Replace an asset's SBOM snapshot with the reported set.
"""Update an asset's SBOM snapshot using delta comparison.
The agent reports the full installed-software list each cycle, so the
current inventory is the newest report, not an accumulation. Delete the
asset's existing rows and insert the reported ones in the same
transaction the caller flushes. appid is left null: no
discovery-to-catalog ("Managed Applications") matcher is wired yet.
Malformed items (no displayname) are skipped, not fatal.
The agent reports the full installed-software list each cycle along with
a fingerprint hash (SHA256 of displayname|displayversion|publisher) for
each entry. Rather than delete-all-insert-all, compare fingerprints to
detect actual changes:
- Add rows for new fingerprints
- Remove rows for fingerprints no longer reported
- Leave unchanged rows untouched
This reduces database writes from ~100 ops/endpoint/check-in to ~2 ops
(99% of check-ins have no application changes). Malformed items (no
displayname or fingerprint) are skipped, not fatal.
Legacy fallback: If agent does not send fingerprints, fall back to
delete-all-insert-all for backward compatibility during rollout.
Safety features:
- Row-level locking prevents concurrent update races
- Fingerprint format validation (64 hex chars)
- Delete count verification detects silent failures
- Memory limits prevent pathological cases (>5000 apps)
- Metrics logged for monitoring
"""
EndpointApplication.query.filter_by(assetid=assetid).delete(
synchronize_session=False)
from flask import current_app
# Memory safety: prevent loading massive inventories
MAX_APPS_PER_ENDPOINT = 5000
if not applications:
# Empty list means no apps installed - delete all existing
deleted = EndpointApplication.query.filter_by(assetid=assetid).delete(
synchronize_session=False)
current_app.logger.info(
f'[collector] asset_id={assetid} apps_removed={deleted} '
f'apps_added=0 (empty list)')
return 0
stored = 0
# Check if agent sent fingerprints - legacy agents won't have them yet
has_fingerprints = any(item.get('fingerprint') for item in applications
if isinstance(item, dict))
if not has_fingerprints:
# LEGACY FALLBACK: No fingerprints, use delete-all-insert-all
# SAFETY: Clear any existing NULL-fingerprint apps first to prevent
# UNIQUE constraint violations (MySQL allows multiple NULL values)
deleted = EndpointApplication.query.filter_by(assetid=assetid).delete(
synchronize_session=False)
current_app.logger.info(
f'[collector] asset_id={assetid} legacy_delete={deleted} apps')
stored = 0
for item in applications:
if not isinstance(item, dict):
warnings.append('skipped non-object application entry')
continue
displayname = (item.get('displayname') or '').strip()
if not displayname:
warnings.append('skipped application with no displayname')
continue
db.session.add(EndpointApplication(
assetid=assetid,
displayname=displayname[:255],
displayversion=_clamp(item.get('displayversion'), 128),
publisher=_clamp(item.get('publisher'), 255),
app_fingerprint=None, # Agent didn't send it
))
stored += 1
# Flush to detect constraint violations early
db.session.flush()
current_app.logger.info(
f'[collector] asset_id={assetid} apps_stored={stored} (legacy mode)'
)
return stored
# DELTA LOGIC: Build incoming fingerprint map and validate entries
incoming = {}
for item in applications:
if not isinstance(item, dict):
warnings.append('skipped non-object application entry')
continue
displayname = (item.get('displayname') or '').strip()
fingerprint = (item.get('fingerprint') or '').strip()
if not displayname:
warnings.append('skipped application with no displayname')
continue
# TRUNCATE EVERY STRING, not just the name. On MySQL in strict
# mode an over-long value raises Data too long and takes down the
# WHOLE check-in transaction, so one machine with a malformed
# Uninstall registry entry stops reporting entirely - telemetry,
# events and all - until somebody uninstalls the offender. SQLite
# does not enforce declared lengths, so no test here would ever see
# it. displayname was already clamped; these two were not.
if not fingerprint:
warnings.append('skipped application with no fingerprint')
continue
# SAFETY: Validate fingerprint format (64 lowercase hex chars)
if not self._is_valid_fingerprint(fingerprint):
warnings.append(
f'skipped application with invalid fingerprint: {displayname} '
f'(fingerprint: {fingerprint[:32]}...)'
)
continue
# Truncate strings for MySQL strict mode compatibility
incoming[fingerprint] = {
'displayname': displayname[:255],
'displayversion': _clamp(item.get('displayversion'), 128),
'publisher': _clamp(item.get('publisher'), 255),
}
incoming_fps = set(incoming.keys())
# SAFETY: Row-level locking to prevent concurrent update races.
# MySQL uses FOR UPDATE on the SELECT below (line 1157).
# PostgreSQL would use pg_advisory_xact_lock here, but MySQL achieves
# the same result via row-level locks + UNIQUE constraint on fingerprint.
# SAFETY: Memory limit check - prevent loading massive inventories
# Query exactly MAX rows, then check if more exist (off-by-one fix)
current_apps = (EndpointApplication.query
.filter_by(assetid=assetid)
.limit(MAX_APPS_PER_ENDPOINT)
.with_for_update() # SAFETY: Row-level locking prevents concurrent races
.all())
if len(current_apps) >= MAX_APPS_PER_ENDPOINT:
# This endpoint has too many apps - fall back to legacy mode
warnings.append(
f'asset has {len(current_apps)} apps (exceeds {MAX_APPS_PER_ENDPOINT}), '
f'falling back to legacy mode'
)
# Recursive call with has_fingerprints=False to trigger legacy path
# (This is safe because we're already in a transaction)
EndpointApplication.query.filter_by(assetid=assetid).delete(
synchronize_session=False)
stored = 0
for fp, app_data in incoming.items():
db.session.add(EndpointApplication(
assetid=assetid,
displayname=app_data['displayname'],
displayversion=app_data['displayversion'],
publisher=app_data['publisher'],
app_fingerprint=fp,
))
stored += 1
db.session.flush()
current_app.logger.warning(
f'[collector] asset_id={assetid} apps_stored={stored} '
f'(legacy mode - memory limit exceeded, had {len(current_apps)} apps)'
)
return stored
current_fps = {app.app_fingerprint for app in current_apps
if app.app_fingerprint} # Skip NULL fingerprints from migration
# TRANSITION: If we have apps without fingerprints, this is the first report
# after agent upgrade. Delete all NULL-fingerprint apps to prevent duplicates.
# SAFETY: Validate incoming apps BEFORE deleting legacy apps to prevent
# leaving asset with zero apps if transaction aborts.
null_fp_apps = [app for app in current_apps if app.app_fingerprint is None]
if null_fp_apps:
if not incoming_fps:
current_app.logger.error(
f'[collector] asset_id={assetid} received empty fingerprint set '
f'during transition, keeping {len(null_fp_apps)} legacy apps')
warnings.append(
f'agent sent empty fingerprint set during NULL transition, '
f'kept {len(null_fp_apps)} legacy apps')
return len(null_fp_apps)
for app in null_fp_apps:
db.session.delete(app)
current_app.logger.info(
f'[collector] asset_id={assetid} removed {len(null_fp_apps)} '
f'legacy apps (transition to delta mode)')
# Calculate delta
to_add = incoming_fps - current_fps # New apps
to_remove = current_fps - incoming_fps # Uninstalled apps
unchanged = current_fps & incoming_fps # Unchanged (don't touch)
# Remove uninstalled apps
deleted_count = 0
if to_remove:
deleted_count = EndpointApplication.query.filter(
EndpointApplication.assetid == assetid,
EndpointApplication.app_fingerprint.in_(list(to_remove))
).delete(synchronize_session=False)
# SAFETY: Verify delete count matches expected. Log as WARNING
# instead of raising error - concurrent updates are rare but valid,
# and aborting the entire transaction loses state/events/hardware data.
if deleted_count != len(to_remove):
current_app.logger.warning(
f'[collector] asset_id={assetid} delete_count_mismatch: '
f'expected={len(to_remove)} actual={deleted_count} '
f'(possible concurrent update, proceeding with caution)')
warnings.append(
f'delete count mismatch: expected {len(to_remove)}, '
f'got {deleted_count} (concurrent update?)')
# Add new apps
for fp in to_add:
app_data = incoming[fp]
db.session.add(EndpointApplication(
assetid=assetid,
displayname=displayname[:255],
displayversion=_clamp(item.get('displayversion'), 128),
publisher=_clamp(item.get('publisher'), 255),
displayname=app_data['displayname'],
displayversion=app_data['displayversion'],
publisher=app_data['publisher'],
app_fingerprint=fp,
))
stored += 1
return stored
# SAFETY: Flush to detect constraint violations before continuing
# Only flush if we made changes (skip when unchanged apps = 99% of cases)
if to_add or to_remove:
db.session.flush()
current_app.logger.info(
f'[collector] asset_id={assetid} apps_added={len(to_add)} '
f'apps_removed={len(to_remove)} unchanged={len(unchanged)} (delta mode)'
)
return len(to_add)
def apply_collector_payload(self, payload: Dict) -> Dict:
"""Idempotently upsert one endpoint's state and append its events.
@@ -1091,6 +1291,21 @@ class EndpointsPlugin(BasePlugin):
assetid = computer.assetid
# Authorization: if authenticated via client certificate, verify the
# certificate's asset matches the payload's hostname asset. Without this
# check an agent with a valid cert for asset A can submit telemetry for
# asset B by changing the hostname field.
from flask import g
if hasattr(g, 'cert_authenticated_asset_id'):
if g.cert_authenticated_asset_id != assetid:
self.logger.warning(
'[collector] AUTHORIZATION DENIED: Certificate authenticates '
'asset %d but payload hostname %s resolves to asset %d',
g.cert_authenticated_asset_id, hostname, assetid)
raise PermissionError(
f'Certificate authenticates asset {g.cert_authenticated_asset_id} '
f'but payload reports for asset {assetid}')
state = EndpointState.query.filter_by(assetid=assetid).first()
action = 'updated'
if state is None:
@@ -1257,11 +1472,16 @@ class EndpointsPlugin(BasePlugin):
if eventrecordid is not None]
already_stored = set()
if keyed:
# SAFETY: Row-level locking prevents race condition (vuln #7 from audit).
# WITH FOR UPDATE locks matching rows in MySQL, preventing concurrent INSERT
# of duplicates. The UNIQUE constraint on (assetid, eventchannel, eventrecordid)
# provides additional protection against race conditions.
rows = (db.session.query(EndpointEvent.eventchannel,
EndpointEvent.eventrecordid)
.filter(EndpointEvent.assetid == assetid,
EndpointEvent.eventrecordid.in_(
[recordid for _c, recordid in keyed]))
.with_for_update()
.all())
already_stored = {(channel, recordid) for channel, recordid in rows}
@@ -1289,9 +1509,14 @@ class EndpointsPlugin(BasePlugin):
stored_events += 1
if skipped_events:
logger.info(
'[collector] asset_id=%d hostname=%s events_stored=%d '
'events_skipped=%d (duplicates detected)',
assetid, hostname, stored_events, skipped_events)
else:
logger.debug(
'endpoint %s replayed %d already-stored event record(s)',
hostname, skipped_events)
'endpoint %s stored %d event(s), no duplicates',
hostname, stored_events)
# flush, not commit: the collector dispatcher owns the transaction.
db.session.flush()