"""Read-only MCP access to curated OfficeCom Sentinel security data.""" import functools import json import logging import os import re import time from datetime import datetime from typing import Any import psycopg from psycopg.rows import dict_row from mcp.server.fastmcp import FastMCP LOGGER = logging.getLogger("ocsentinel.mcp") logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO").upper(), format="%(asctime)s %(levelname)s %(message)s") MAX_RESULT_LIMIT = 100 MAX_LOOKBACK_HOURS = 24 * 90 IDENTIFIER_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") def required_setting(name: str) -> str: value = os.getenv(name, "").strip() if not value or value.startswith("replace-with-"): raise RuntimeError(f"{name} must be configured before starting OCSentinel MCP.") return value AUTH_TOKEN = required_setting("MCP_AUTH_TOKEN") ALLOWED_ORIGINS = {value.strip() for value in os.getenv("MCP_ALLOWED_ORIGINS", "").split(",") if value.strip()} ALLOWED_HOSTS = {value.strip().lower() for value in os.getenv("MCP_ALLOWED_HOSTS", "").split(",") if value.strip()} def db_connection() -> psycopg.Connection: return psycopg.connect( host=required_setting("DB_HOST"), port=os.getenv("DB_PORT", "5432"), dbname=required_setting("DB_NAME"), user=required_setting("DB_USER"), password=required_setting("DB_PASSWORD"), connect_timeout=5, row_factory=dict_row, options="-c default_transaction_read_only=on -c statement_timeout=5000", ) def rows(sql: str, parameters: tuple[Any, ...] = ()) -> list[dict[str, Any]]: with db_connection() as connection, connection.cursor() as cursor: cursor.execute(sql, parameters) return list(cursor.fetchall()) def row(sql: str, parameters: tuple[Any, ...] = ()) -> dict[str, Any] | None: results = rows(sql, parameters) return results[0] if results else None def value(record: dict[str, Any] | None, *names: str, default: Any = None) -> Any: if not isinstance(record, dict): return default for name in names: if name in record: return record[name] return default def json_safe(data: Any) -> Any: return json.loads(json.dumps(data, default=lambda entry: entry.isoformat() if isinstance(entry, datetime) else str(entry))) def bounded_limit(limit: int) -> int: if not isinstance(limit, int) or isinstance(limit, bool): raise ValueError("limit must be a whole number.") return max(1, min(limit, MAX_RESULT_LIMIT)) def bounded_hours(hours: int) -> int: if not isinstance(hours, int) or isinstance(hours, bool): raise ValueError("hours must be a whole number.") return max(1, min(hours, MAX_LOOKBACK_HOURS)) def checked_identifier(identifier: str, field_name: str) -> str: value_to_check = (identifier or "").strip() if value_to_check == "unknown" or IDENTIFIER_PATTERN.fullmatch(value_to_check): return value_to_check raise ValueError(f"{field_name} contains unsupported characters.") def checked_machine_name(machine_name: str) -> str: machine = (machine_name or "").strip() if not machine or len(machine) > 255 or any(character in machine for character in "\r\n\x00"): raise ValueError("machine_name must be a single device name of at most 255 characters.") return machine def organization_context(payload: dict[str, Any] | None) -> tuple[str, str]: ninja = value(payload, "NinjaOne", "ninjaOne", default={}) or {} return ( str(value(ninja, "OrganizationId", "organizationId", default="unknown") or "unknown"), str(value(ninja, "OrganizationName", "organizationName", default="Organisation unbekannt") or "Organisation unbekannt"), ) def sanitized_ransomware(payload: dict[str, Any] | None) -> dict[str, Any]: ransomware = value(payload, "RansomwareBeta", "ransomwareBeta", default={}) or {} sensors = [] for sensor in value(ransomware, "Sensors", "sensors", default=[]) or []: sensors.append( { "name": value(sensor, "Name", "name", default="unknown"), "enabled": bool(value(sensor, "Enabled", "enabled", default=False)), "available": bool(value(sensor, "Available", "available", default=False)), "state": value(sensor, "State", "state", default="unknown"), "eventCount": value(sensor, "EventCount", "eventCount", default=0), } ) signals = [] for signal in value(ransomware, "Signals", "signals", default=[]) or []: # Evidence and raw command lines intentionally never leave the MCP boundary. signals.append( { "timestamp": value(signal, "Timestamp", "timestamp", default=None), "category": value(signal, "Category", "category", default="signal"), "process": value(signal, "Process", "process", default="-"), "source": value(signal, "Source", "source", default="-"), "confidence": value(signal, "Confidence", "confidence", default="low"), } ) return { "enabled": bool(value(ransomware, "Enabled", "enabled", default=False)), "state": value(ransomware, "State", "state", default="disabled"), "reason": value(ransomware, "Reason", "reason", default="Keine Ransomware-Beta-Daten verfuegbar."), "sensors": sensors, "signals": signals[:20], } def summarized_events(payload: dict[str, Any] | None, include_accounts: bool = False) -> list[dict[str, Any]]: groups: dict[tuple[str, str, str], dict[str, Any]] = {} for event in value(payload, "Events", "events", default=[]) or []: event_type = str(value(event, "Target", "target", default="Sicherheitsereignis")) source_ip = str(value(event, "SourceIp", "sourceIp", default="-")) account = str(value(event, "Username", "username", default="-")) if include_accounts else "" key = (event_type, source_ip, account) group = groups.setdefault(key, {"type": event_type, "sourceIp": source_ip, "count": 0, "latest": None}) group["count"] += 1 timestamp = value(event, "Timestamp", "timestamp", default=None) if timestamp and (not group["latest"] or str(timestamp) > str(group["latest"])): group["latest"] = timestamp if include_accounts: group["account"] = account return sorted(groups.values(), key=lambda entry: (str(entry["latest"]), entry["count"]), reverse=True)[:25] def audited(tool_name: str): def decorator(function): @functools.wraps(function) def wrapped(*args, **kwargs): started = time.monotonic() try: result = function(*args, **kwargs) LOGGER.info("mcp_tool=%s outcome=ok duration_ms=%d", tool_name, (time.monotonic() - started) * 1000) return result except Exception: LOGGER.exception("mcp_tool=%s outcome=error duration_ms=%d", tool_name, (time.monotonic() - started) * 1000) raise return wrapped return decorator mcp = FastMCP( "OfficeCom Sentinel", instructions=( "Read-only security context from OfficeCom Sentinel. Use this data to investigate, summarize, and prioritize. " "Do not treat it as authorization to alter devices, NinjaOne, PostgreSQL, or security controls." ), stateless_http=True, json_response=True, streamable_http_path="/mcp", ) @mcp.resource("ocsentinel://read-only-policy") def read_only_policy() -> str: """Explain the data and safety boundary of this server.""" return ( "OfficeCom Sentinel MCP is read-only. It returns curated status, event summaries, ransomware sensor coverage, " "network paths, and weekly-report metadata. Raw event evidence, command lines, SMB sessions, credentials, " "and every write action are deliberately excluded." ) @mcp.prompt() def incident_triage() -> str: """Provide a safe, evidence-oriented workflow for analyzing Sentinel findings.""" return ( "Start with security_overview or get_organization_status. For a flagged device, use get_device_security and " "search_security_events. Separate observed facts from hypotheses, identify the next reversible validation step, " "and recommend escalation to the responsible OfficeCom technician for any containment action." ) @mcp.tool() @audited("security_overview") def security_overview() -> dict[str, Any]: """Return the current cross-organization security posture and the most urgent devices.""" summary = row("SELECT * FROM ocsentinel.organization_summary") or {} coverage = row( """ SELECT count(*) AS known_devices, count(*) FILTER (WHERE received_at >= now() - interval '36 hours') AS reporting_devices, count(*) FILTER (WHERE received_at IS NULL OR received_at < now() - interval '36 hours') AS stale_devices FROM ocsentinel.current_device_status """ ) or {} urgent = rows( """ SELECT machine_name, alert_state, received_at, total_events, unique_ip_count, payload #>> '{NinjaOne,OrganizationId}' AS organization_id, payload #>> '{NinjaOne,OrganizationName}' AS organization_name FROM ocsentinel.current_device_status WHERE alert_state IN ('warning', 'critical') ORDER BY CASE alert_state WHEN 'critical' THEN 0 ELSE 1 END, received_at DESC NULLS LAST LIMIT 20 """ ) return json_safe({"summary": summary, "coverage": coverage, "urgentDevices": urgent}) @mcp.tool() @audited("get_organization_status") def get_organization_status(organization_id: str) -> dict[str, Any]: """Return current coverage and alert state for one NinjaOne organization ID.""" organization_id = checked_identifier(organization_id, "organization_id") devices = rows( """ SELECT machine_name, received_at, alert_state, base_alert_state, total_events, unique_ip_count, cve_total, cve_critical, payload FROM ocsentinel.current_device_status WHERE coalesce(payload #>> '{NinjaOne,OrganizationId}', 'unknown') = %s ORDER BY CASE alert_state WHEN 'critical' THEN 0 WHEN 'warning' THEN 1 ELSE 2 END, machine_name """, (organization_id,), ) if not devices: return {"organizationId": organization_id, "found": False, "devices": []} _, organization_name = organization_context(devices[0]["payload"]) status_counts = {state: sum(device["alert_state"] == state for device in devices) for state in ("ok", "warning", "critical", "unknown")} return json_safe( { "organizationId": organization_id, "organizationName": organization_name, "found": True, "deviceCount": len(devices), "statusCounts": status_counts, "lastReceivedAt": max((device["received_at"] for device in devices if device["received_at"]), default=None), "devices": [ { "machineName": device["machine_name"], "alertState": device["alert_state"], "receivedAt": device["received_at"], "totalEvents": device["total_events"], "uniqueIpCount": device["unique_ip_count"], "criticalCves": device["cve_critical"], } for device in devices ], } ) @mcp.tool() @audited("get_device_security") def get_device_security(machine_name: str, include_accounts: bool = False) -> dict[str, Any]: """Return the latest sanitized security summary for one device. Accounts are omitted by default.""" machine_name = checked_machine_name(machine_name) device = row( """ SELECT machine_name, first_seen_at, last_seen_at, last_client_version, generated_at_utc, received_at, alert_state, base_alert_state, total_events, unique_ip_count, cve_total, cve_critical, payload FROM ocsentinel.current_device_status WHERE machine_name = %s """, (machine_name,), ) if not device: return {"machineName": machine_name, "found": False} organization_id, organization_name = organization_context(device["payload"]) return json_safe( { "found": True, "machineName": device["machine_name"], "organizationId": organization_id, "organizationName": organization_name, "receivedAt": device["received_at"], "generatedAt": device["generated_at_utc"], "alertState": device["alert_state"], "baseAlertState": device["base_alert_state"], "metrics": { "totalEvents": device["total_events"], "uniqueIpCount": device["unique_ip_count"], "cveTotal": device["cve_total"], "criticalCves": device["cve_critical"], }, "events": summarized_events(device["payload"], include_accounts), "ransomwareBeta": sanitized_ransomware(device["payload"]), } ) @mcp.tool() @audited("search_security_events") def search_security_events( hours: int = 168, organization_id: str | None = None, alert_state: str | None = None, limit: int = 25, include_accounts: bool = False, ) -> dict[str, Any]: """Find recent scanned reports with bounded filters. This returns event summaries, not raw evidence.""" hours = bounded_hours(hours) limit = bounded_limit(limit) if organization_id is not None: organization_id = checked_identifier(organization_id, "organization_id") if alert_state is not None and alert_state not in {"ok", "warning", "critical", "unknown"}: raise ValueError("alert_state must be ok, warning, critical, or unknown.") reports = rows( """ SELECT d.machine_name, r.received_at, r.generated_at_utc, r.alert_state, r.base_alert_state, r.total_events, r.unique_ip_count, r.cve_total, r.cve_critical, r.payload FROM ocsentinel.scan_report AS r JOIN ocsentinel.device AS d ON d.id = r.device_id WHERE r.received_at >= now() - (%s * interval '1 hour') AND (%s::text IS NULL OR coalesce(r.payload #>> '{NinjaOne,OrganizationId}', 'unknown') = %s) AND (%s::text IS NULL OR r.alert_state = %s) ORDER BY r.received_at DESC LIMIT %s """, (hours, organization_id, organization_id, alert_state, alert_state, limit), ) results = [] for report in reports: org_id, org_name = organization_context(report["payload"]) results.append( { "machineName": report["machine_name"], "organizationId": org_id, "organizationName": org_name, "receivedAt": report["received_at"], "generatedAt": report["generated_at_utc"], "alertState": report["alert_state"], "baseAlertState": report["base_alert_state"], "totalEvents": report["total_events"], "uniqueIpCount": report["unique_ip_count"], "criticalCves": report["cve_critical"], "events": summarized_events(report["payload"], include_accounts), "ransomwareBeta": sanitized_ransomware(report["payload"]), } ) return json_safe({"lookbackHours": hours, "resultCount": len(results), "reports": results}) @mcp.tool() @audited("get_network_paths") def get_network_paths(days: int = 14, organization_id: str | None = None, limit: int = 50) -> dict[str, Any]: """Return recent observed source-IP to device paths from summarized security events.""" days = max(1, min(bounded_hours(days * 24) // 24, 90)) limit = bounded_limit(limit) if organization_id is not None: organization_id = checked_identifier(organization_id, "organization_id") reports = rows( """ SELECT machine_name, received_at, payload FROM ocsentinel.current_device_status WHERE received_at >= now() - (%s * interval '1 day') AND (%s::text IS NULL OR coalesce(payload #>> '{NinjaOne,OrganizationId}', 'unknown') = %s) """, (days, organization_id, organization_id), ) flows: dict[tuple[str, str, str, str], dict[str, Any]] = {} for report in reports: org_id, org_name = organization_context(report["payload"]) for event in value(report["payload"], "Events", "events", default=[]) or []: source_ip = str(value(event, "SourceIp", "sourceIp", default="")) if not source_ip or source_ip in {"-", "127.0.0.1", "::1"}: continue event_type = str(value(event, "Target", "target", default="Sicherheitsereignis")) key = (source_ip, report["machine_name"], event_type, org_id) flow = flows.setdefault( key, { "sourceIp": source_ip, "machineName": report["machine_name"], "eventType": event_type, "organizationId": org_id, "organizationName": org_name, "count": 0, "lastSeen": report["received_at"], }, ) flow["count"] += 1 timestamp = value(event, "Timestamp", "timestamp", default=None) if timestamp and (not flow["lastSeen"] or str(timestamp) > str(flow["lastSeen"])): flow["lastSeen"] = timestamp paths = sorted(flows.values(), key=lambda entry: (entry["count"], str(entry["lastSeen"])), reverse=True)[:limit] return json_safe({"days": days, "pathCount": len(paths), "paths": paths}) @mcp.tool() @audited("get_weekly_report") def get_weekly_report(organization_id: str) -> dict[str, Any]: """Return the latest weekly report metadata and structured summary for one organization, without report HTML.""" organization_id = checked_identifier(organization_id, "organization_id") report = row( """ SELECT organization_id, organization_name, period_start_utc, period_end_utc, generated_at, device_count, warning_count, critical_count, total_events, unique_ips, cve_total, cve_critical, summary FROM ocsentinel.weekly_organization_report WHERE organization_id = %s ORDER BY period_end_utc DESC, generated_at DESC LIMIT 1 """, (organization_id,), ) if not report: return {"organizationId": organization_id, "found": False} report["found"] = True return json_safe(report) class GuardedMcpApp: """Small ASGI guard without BaseHTTPMiddleware, which can disrupt MCP streaming.""" def __init__(self, wrapped_app): self.wrapped_app = wrapped_app async def __call__(self, scope, receive, send): if scope["type"] != "http": await self.wrapped_app(scope, receive, send) return headers = {key.decode("latin-1").lower(): value.decode("latin-1") for key, value in scope.get("headers", [])} host = headers.get("host", "").lower() origin = headers.get("origin") authorization = headers.get("authorization", "") if ALLOWED_HOSTS and host not in ALLOWED_HOSTS: await self.reject(send, 421, "Untrusted Host header.") return if origin and (not ALLOWED_ORIGINS or origin not in ALLOWED_ORIGINS): await self.reject(send, 403, "Untrusted Origin header.") return if authorization != f"Bearer {AUTH_TOKEN}": await self.reject(send, 401, "Bearer token required.", {b"www-authenticate": b"Bearer"}) return await self.wrapped_app(scope, receive, send) @staticmethod async def reject(send, status: int, message: str, extra_headers: dict[bytes, bytes] | None = None): body = json.dumps({"error": message}).encode("utf-8") headers = [(b"content-type", b"application/json"), (b"content-length", str(len(body)).encode("ascii"))] if extra_headers: headers.extend(extra_headers.items()) await send({"type": "http.response.start", "status": status, "headers": headers}) await send({"type": "http.response.body", "body": body}) app = GuardedMcpApp(mcp.streamable_http_app())