Architecting Air-Gapped Local Model Context Protocol (MCP) Server Daemons with Ollama, Unix Domain Sockets, and Hardware Security Modules (HSM)

import asyncio
import os
import socket
import struct
import json
import logging
from typing import Dict, Any

SOCKET_PATH = "/run/aiconnect/mcp_daemon.sock"
AUTHORIZED_UID = os.getuid()

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def verify_peer_credentials(sock: socket.socket) -> bool:
    """Extract PID, UID, GID from Linux kernel SO_PEERCRED structure."""
    try:
        # SO_PEERCRED returns struct ucred: pid_t (int), uid_t (uint), gid_t (uint)
        ucred = sock.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("3i"))
        pid, uid, gid = struct.unpack("3i", ucred)
        logging.info(f"Incoming IPC connection from PID: {pid}, UID: {uid}, GID: {gid}")
        return uid == AUTHORIZED_UID
    except Exception as e:
        logging.error(f"Kernel SO_PEERCRED verification failed: {e}")
        return False

class AirGappedMCPServer:
    def __init__(self, socket_path: str):
        self.socket_path = socket_path

    async def handle_mcp_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        sock = writer.get_extra_info('socket')
        if not verify_peer_credentials(sock):
            logging.warning("Unauthorized UID detected! Terminating socket.")
            writer.close()
            await writer.wait_closed()
            return

        while not reader.at_eof():
            data = await reader.readline()
            if not data:
                break
            try:
                request = json.loads(data.decode('utf-8'))
                response = await self.dispatch_mcp_request(request)
                writer.write((json.dumps(response) + "
").encode('utf-8'))
                await writer.drain()
            except Exception as err:
                error_resp = {"jsonrpc": "2.0", "error": {"code": -32603, "message": str(err)}, "id": None}
                writer.write((json.dumps(error_resp) + "
").encode('utf-8'))
                await writer.drain()

        writer.close()
        await writer.wait_closed()

    async def dispatch_mcp_request(self, req: Dict[str, Any]) -> Dict[str, Any]:
        method = req.get("method")
        msg_id = req.get("id")

        if method == "tools/list":
            return {
                "jsonrpc": "2.0",
                "result": {
                    "tools": [
                        {
                            "name": "read_secure_vault_file",
                            "description": "Reads local encrypted vault configuration without network egress.",
                            "inputSchema": {
                                "type": "object",
                                "properties": {"filepath": {"type": "string"}},
                                "required": ["filepath"]
                            }
                        }
                    ]
                },
                "id": msg_id
            }
        elif method == "tools/call":
            tool_name = req.get("params", {}).get("name")
            args = req.get("params", {}).get("arguments", {})
            return await self.execute_tool_with_hsm(tool_name, args, msg_id)

        return {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": msg_id}

    async def execute_tool_with_hsm(self, tool_name: str, args: Dict[str, Any], msg_id: Any) -> Dict[str, Any]:
        # Tool execution logic wrapped in local security boundaries
        logging.info(f"Executing authorized local tool: {tool_name}")
        return {
            "jsonrpc": "2.0",
            "result": {
                "content": [{"type": "text", "text": f"Successfully processed {tool_name} air-gapped."}],
                "isError": False
            },
            "id": msg_id
        }

async def run_server():
    if os.path.exists(SOCKET_PATH):
        os.remove(SOCKET_PATH)

    server_instance = AirGappedMCPServer(SOCKET_PATH)
    server = await asyncio.start_unix_server(server_instance.handle_mcp_client, path=SOCKET_PATH)
    os.chmod(SOCKET_PATH, 0o600)  # Restrict permissions to owner only
    logging.info(f"Air-gapped MCP Daemon active on {SOCKET_PATH} (chmod 0600)")
    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    asyncio.run(run_server())

import hashlib
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
# Pseudocode binding to local PKCS#11 hardware token slot
def sign_tool_execution_attestation(tool_name: str, args_json: str, timestamp_ms: int) -> str:
    """Generates a hardware-attested cryptographic signature for tool parameters."""
    payload = f"{tool_name}:{args_json}:{timestamp_ms}".encode('utf-8')
    payload_hash = hashlib.sha256(payload).hexdigest()

    # In production, this call interfaces with libsofthsm2.so or libykcs11.so
    # Private keys never leave the secure boundary of the silicon chip
    logging.info(f"Hardware HSM signing tool invocation hash: {payload_hash}")
    return f"HSM_SIG_ED25519_{payload_hash[:32]}"