📝 Description

Currently, the MCP Gateway fails to establish a secure connection (via HTTPS) to any Streamable HTTP server that uses a self-signed TLS/SSL certificate. This is due to the underlying httpx client throwing a CERTIFICATE_VERIFY_FAILED error because the self-signed certificate is not present in the operating system's global trust store.

To support internal or secure development environments where self-signed certificates are common, we must implement a mechanism to explicitly trust a known server certificate during client connection initialization.

Proposed Solution

• Store the self-signed certificate (encrypted) in the database when a new gateway is registered.
• Implement a custom httpx_client_factory function that, when passed to the standard streamablehttp_client and sse_client calls, injects the stored certificate path into the httpx.AsyncClient's verify parameter.
• Cache the decrypted certificate for performance across Gunicorn workers.

📊 Technical Design & Implementation Details

A. Custom HTTPX Client Factory

We will introduce a new, optional parameter, httpx_client_factory, to the streamablehttp_client and sse_client functions. This parameter will receive a callable function that returns an initialized httpx.AsyncClient.

The factory function will dynamically set the verify parameter based on whether the server uses a self-signed certificate:

def make_custom_client_factory(cert_path: str | None) -> Callable[..., httpx.AsyncClient]:
    """
    Returns a client factory function configured to use a specific cert file
    for verification, or default to the system trust store.
    """
    def custom_client_factory(
        headers: dict[str, str] | None = None,
        timeout: httpx.Timeout | None = None,
        auth: httpx.Auth | None = None,
    ) -> httpx.AsyncClient:
        return httpx.AsyncClient(
            # 'cert_path' will be None for standard certificates,
            # or the path to the decrypted self-signed cert file.
            verify=cert_path if cert_path else True,
            follow_redirects=True,
            headers=headers,
            timeout=timeout or httpx.Timeout(30.0),
            auth=auth,
        )
    return custom_client_factory

The client factory will be built in the register_gateway and invoke_tool functions and conditionally passed when calling the server.

B. Certificate Storage & Caching

• Database Storage: The gateways table must be updated to store the self-signed certificate (encrypted using our standard vault/encryption method) and a boolean flag indicating whether the certificate is self-signed.
• Runtime Caching: Upon first retrieval, the decrypted certificate data will be saved to a temporary file (or in-memory cache if file system access is restricted) and the file path will be stored in a TTL-based L1/L2 cache (e.g., using Redis or an equivalent in-memory object) so that all Gunicorn workers can access the path without re-decrypting the file for every request.

C. Self-Signed Verification Logic

To determine if we need to store and trust a specific certificate, we will implement robust logic using the cryptography library to check if the certificate is both Issuer == Subject and is cryptographically self-signed.

from cryptography import x509
# ... import other cryptography modules
# ... implementation of _verify_self_signature(cert: x509.Certificate) -> bool
# ... implementation of is_self_signed_pem_bytes(pem_bytes: bytes) -> bool

def is_self_signed_pem_bytes(pem_bytes: bytes) -> bool:
    cert = x509.load_pem_x509_certificate(pem_bytes, default_backend())
    # 1. Structural Check: Does the issuer match the subject?
    if cert.issuer != cert.subject:
        return False
    # 2. Cryptographic Check: Can the public key verify its own signature?
    return _verify_self_signature(cert)

✅ Required Tasks

1. Database Migration

Build an Alembic script to update the gateways table:

• Add new column: certificate_pem (TEXT, encrypted)
• Add new column: is_self_signed (BOOLEAN, default False)

2. Certificate Management

Implement helper functions for:

• Checking if a PEM certificate is self-signed (is_self_signed_pem_bytes)
• Encrypting and storing the certificate in the database
• Retrieving, decrypting, and caching the certificate (L1/L2) for worker access

3. Gateway Registration Logic (register_gateway)

• During registration, attempt to fetch the server's certificate
• Use the new logic to check if it is self-signed
• Store the certificate (encrypted) and the is_self_signed flag in the database

4. Invocation Logic (invoke_tool / Connection)

• When establishing connections (via streamablehttp_client and sse_client), check the gateway record for is_self_signed
• If True, retrieve the decrypted certificate path from the cache/disk
• Build and pass the custom httpx_client_factory (configured with the certificate path) to the client constructors

5. Refactor streamablehttp_client and sse_client

Update these functions to accept and utilize the optional httpx_client_factory parameter.

🚀 Expected Outcome

Users will be able to register MCP servers that use self-signed certificates without encountering CERTIFICATE_VERIFY_FAILED errors.

This ensures secure, end-to-end communication while maintaining the cryptographic integrity check (by verifying against a known, stored certificate).