45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
"""Thin MCP server translating MCP tool calls into calls to the local debug agent.
|
|
No Win32/ctypes dependency of its own -- runs and is tested on any platform, though at
|
|
runtime it must reach the real debug agent server (tools/havoc_debug_agent_server.py)
|
|
running on the same Caspian box."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import socket
|
|
|
|
from tools.havoc_debug_protocol import decode_message, encode_message
|
|
|
|
|
|
class DebugAgentError(Exception):
|
|
"""Raised when the debug agent reports {"ok": false}."""
|
|
|
|
|
|
class DebugAgentClient:
|
|
def __init__(self, host: str, port: int) -> None:
|
|
self._host = host
|
|
self._port = port
|
|
|
|
def call(self, op: str, params: dict) -> dict:
|
|
with socket.create_connection((self._host, self._port), timeout=60) as sock:
|
|
sock.sendall(encode_message(op, params))
|
|
buf = b""
|
|
while not buf.endswith(b"\n"):
|
|
chunk = sock.recv(65536)
|
|
if not chunk:
|
|
break
|
|
buf += chunk
|
|
response = decode_message(buf)
|
|
if not response.get("ok"):
|
|
raise DebugAgentError(response.get("error", "unknown debug agent error"))
|
|
return response.get("result", {})
|
|
|
|
|
|
def check_bearer_token(header_value: str | None, expected: str) -> bool:
|
|
if header_value is None:
|
|
return False
|
|
prefix = "Bearer "
|
|
if not header_value.startswith(prefix):
|
|
return False
|
|
return header_value[len(prefix):] == expected
|