""" Движок шлюза допустимости действий «mcp-sentinel» (ИБ 3.0). In-line Circuit Breaker для протокола Model Context Protocol (MCP). Перехватывает и верифицирует вызовы инструментов (Tool Calls) до их исполнения, предотвращает деструктивные воздействия и фиксирует доказательную базу в криптографическом журнале по ГОСТ Р 34.11-2012 («Стрибог-256»). """ import json import os import subprocess import sys import threading from typing import Any, Dict, List, Optional, Tuple # Импорт модулей из iss-ledger-117 LEDGER_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "iss-ledger-117")) if LEDGER_DIR not in sys.path: sys.path.insert(0, LEDGER_DIR) from ledger import IssLedger117 from contracts.base import BaseSecurityContract, ContractVerdict from contracts.fs_jail import FSJailContract from contracts.sql_guard import SQLGuardContract from contracts.secret_leak import SecretLeakContract from contracts.shell_barrier import ShellBarrierContract class MCPSentinel: """ Инспектор допустимости и Circuit Breaker для MCP-агентов. """ def __init__( self, ledger_path: Optional[str] = "audit_fstec117.jsonl", allowed_roots: Optional[List[str]] = None, contracts: Optional[List[BaseSecurityContract]] = None, ): self.ledger = IssLedger117(ledger_path) if ledger_path else None if contracts: self.contracts = contracts else: self.contracts = [ FSJailContract(allowed_roots=allowed_roots), SQLGuardContract(allow_ddl=False, strict_where=True), SecretLeakContract(), ShellBarrierContract(), ] def validate_tool_call( self, agent_id: str, tool_name: str, arguments: Dict[str, Any] ) -> Tuple[bool, str, Optional[Dict[str, Any]]]: """ Проверка входящего вызова инструмента агента через набор контрактов. Возвращает (allowed, reason, details). """ target = tool_name # Пытаемся извлечь более точный target (путь, запрос, команду) for k in ("path", "target_file", "file", "query", "sql", "command", "cmd"): if k in arguments and isinstance(arguments[k], str): target = f"{tool_name}:{arguments[k][:50]}" break for contract in self.contracts: verdict: ContractVerdict = contract.check_tool_call(tool_name, arguments) if not verdict.allowed: # Фиксируем инцидент BLOCK в неизменяемом крипто-журнале ГОСТ if self.ledger: self.ledger.append( agent_id=agent_id, action=f"tool_call:{tool_name}", target=target, verdict="BLOCK", reason=verdict.reason, payload=arguments, ) return False, verdict.reason, verdict.details # Все контракты дали согласие (PERMIT) if self.ledger: self.ledger.append( agent_id=agent_id, action=f"tool_call:{tool_name}", target=target, verdict="PERMIT", reason="Аксиоматические инварианты соблюдены", payload=arguments, ) return True, "Approved", None def validate_tool_result(self, agent_id: str, tool_name: str, content: str) -> Tuple[bool, str]: """Проверка исходящего ответа инструмента на утечки секретов.""" for contract in self.contracts: verdict = contract.check_tool_result(tool_name, content) if not verdict.allowed: if self.ledger: self.ledger.append( agent_id=agent_id, action=f"tool_result:{tool_name}", target=tool_name, verdict="BLOCK", reason=verdict.reason, payload=content[:200], ) return False, verdict.reason return True, "Approved" def process_incoming_jsonrpc( self, line: str, agent_id: str = "ai-agent" ) -> Tuple[str, bool]: """ Обработка одной строки JSON-RPC от агента. Возвращает: (response_or_line, should_forward_to_downstream) Если should_forward_to_downstream == False, response_or_line — это готовый синтетический ответ агенту об отказе в доступе. """ line_str = line.strip() if not line_str: return line, True try: msg = json.loads(line_str) except json.JSONDecodeError: # Невалидный JSON, пробрасываем как есть return line, True # Проверяем, является ли сообщение вызовом инструмента if isinstance(msg, dict) and msg.get("method") == "tools/call": params = msg.get("params", {}) tool_name = params.get("name", "unknown") arguments = params.get("arguments", {}) msg_id = msg.get("id") allowed, reason, details = self.validate_tool_call( agent_id=agent_id, tool_name=tool_name, arguments=arguments ) if not allowed: # Генерируем MCP-ответ с ошибкой (Unsat Core) synthetic_error_resp = { "jsonrpc": "2.0", "id": msg_id, "result": { "content": [ { "type": "text", "text": f"[ИБ 3.0 CIRCUIT BREAKER] Действие заблокировано инвариантом безопасности.\n{reason}" } ], "isError": True } } return json.dumps(synthetic_error_resp, ensure_ascii=False) + "\n", False return line, True def run_stdio_proxy( downstream_cmd: List[str], agent_id: str = "ai-agent", ledger_path: str = "audit_fstec117.jsonl", allowed_roots: Optional[List[str]] = None, ) -> None: """ Запускает прозрачный Stdio-шлюз перед дочерним MCP-сервером. """ sentinel = MCPSentinel(ledger_path=ledger_path, allowed_roots=allowed_roots) # Запускаем дочерний процесс (настоящий MCP-сервер) proc = subprocess.Popen( downstream_cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=sys.stderr, bufsize=0, text=True, encoding="utf-8", ) # Поток чтения вывода от дочернего процесса -> наружу def pipe_stdout(): try: for line in proc.stdout: # Можно валидировать на утечки ответов sys.stdout.write(line) sys.stdout.flush() except (BrokenPipeError, OSError): pass t = threading.Thread(target=pipe_stdout, daemon=True) t.start() # Основной поток: чтение от агента -> проверка -> проброс в процесс try: for line in sys.stdin: out_line, should_forward = sentinel.process_incoming_jsonrpc(line, agent_id=agent_id) if should_forward: proc.stdin.write(out_line) proc.stdin.flush() else: # Ответ об аппаратной блокировке отправляем сразу агенту sys.stdout.write(out_line) sys.stdout.flush() except (KeyboardInterrupt, BrokenPipeError): pass finally: proc.terminate()