479 lines
20 KiB
Python
479 lines
20 KiB
Python
"""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())
|