Coverage for mcpgateway / plugins / framework / external / mcp / tls_utils.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/plugins/framework/external/mcp/tls_utils.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Teryl Taylor 

6 

7TLS/SSL utility functions for external MCP plugin connections. 

8 

9This module provides utilities for creating and configuring SSL contexts for 

10secure communication with external MCP plugin servers. It implements the 

11certificate validation logic that is tested in test_client_certificate_validation.py. 

12 

13Examples: 

14 Create a basic SSL context with default settings: 

15 

16 >>> from mcpgateway.plugins.framework.models import MCPClientTLSConfig 

17 >>> import ssl 

18 >>> config = MCPClientTLSConfig() 

19 >>> ctx = create_ssl_context(config, "ExamplePlugin") 

20 >>> ctx.verify_mode == ssl.CERT_REQUIRED 

21 True 

22 

23 Create an SSL context with hostname verification disabled: 

24 

25 >>> config = MCPClientTLSConfig(verify=True, check_hostname=False) 

26 >>> ctx = create_ssl_context(config, "NoHostnamePlugin") 

27 >>> ctx.verify_mode == ssl.CERT_REQUIRED 

28 True 

29 >>> ctx.check_hostname 

30 False 

31 

32 Verify that TLS version is enforced: 

33 

34 >>> config = MCPClientTLSConfig(verify=True) 

35 >>> ctx = create_ssl_context(config, "VersionTestPlugin") 

36 >>> ctx.minimum_version >= ssl.TLSVersion.TLSv1_2 

37 True 

38 

39 All SSL contexts have TLS 1.2 minimum: 

40 

41 >>> config1 = MCPClientTLSConfig(verify=True) 

42 >>> config2 = MCPClientTLSConfig(verify=False) 

43 >>> ctx1 = create_ssl_context(config1, "Plugin1") 

44 >>> ctx2 = create_ssl_context(config2, "Plugin2") 

45 >>> ctx1.minimum_version == ctx2.minimum_version 

46 True 

47 >>> ctx1.minimum_version.name 

48 'TLSv1_2' 

49 

50 Verify mode differs based on configuration: 

51 

52 >>> config_secure = MCPClientTLSConfig(verify=True) 

53 >>> config_insecure = MCPClientTLSConfig(verify=False) 

54 >>> ctx_secure = create_ssl_context(config_secure, "SecureP") 

55 >>> ctx_insecure = create_ssl_context(config_insecure, "InsecureP") 

56 >>> ctx_secure.verify_mode != ctx_insecure.verify_mode 

57 True 

58 >>> import ssl 

59 >>> ctx_secure.verify_mode == ssl.CERT_REQUIRED 

60 True 

61 >>> ctx_insecure.verify_mode == ssl.CERT_NONE 

62 True 

63""" 

64 

65# Standard 

66import logging 

67import ssl 

68 

69# First-Party 

70from mcpgateway.plugins.framework.errors import PluginError 

71from mcpgateway.plugins.framework.models import MCPClientTLSConfig, PluginErrorModel 

72 

73logger = logging.getLogger(__name__) 

74 

75 

76def create_ssl_context(tls_config: MCPClientTLSConfig, plugin_name: str) -> ssl.SSLContext: 

77 """Create and configure an SSL context for external plugin connections. 

78 

79 This function implements the SSL/TLS security configuration for connecting to 

80 external MCP plugin servers. It supports both standard TLS and mutual TLS (mTLS) 

81 authentication. 

82 

83 Security Features Implemented (per Python ssl docs and OpenSSL): 

84 

85 1. **Invalid Certificate Rejection**: ssl.create_default_context() with CERT_REQUIRED 

86 automatically validates certificate signatures and chains via OpenSSL. 

87 

88 2. **Expired Certificate Handling**: OpenSSL automatically checks notBefore and 

89 notAfter fields per RFC 5280 Section 6. Expired or not-yet-valid certificates 

90 are rejected during the handshake. 

91 

92 3. **Certificate Chain Validation**: Full chain validation up to a trusted CA. 

93 Each certificate in the chain is verified for validity period, signature, etc. 

94 

95 4. **Hostname Verification**: When check_hostname is enabled, the certificate's 

96 Subject Alternative Name (SAN) or Common Name (CN) must match the hostname. 

97 

98 5. **MITM Prevention**: Via mutual authentication when client certificates are 

99 provided (mTLS mode). 

100 

101 Args: 

102 tls_config: TLS configuration containing CA bundle, client certs, and verification settings 

103 plugin_name: Name of the plugin (for error messages) 

104 

105 Returns: 

106 Configured SSLContext ready for use with httpx or other SSL connections 

107 

108 Raises: 

109 PluginError: If SSL context configuration fails 

110 

111 Examples: 

112 Create SSL context with verification enabled (default secure mode): 

113 

114 >>> from mcpgateway.plugins.framework.models import MCPClientTLSConfig 

115 >>> tls_config = MCPClientTLSConfig(verify=True) 

116 >>> ssl_context = create_ssl_context(tls_config, "TestPlugin") 

117 >>> ssl_context.verify_mode == 2 # ssl.CERT_REQUIRED 

118 True 

119 >>> ssl_context.check_hostname 

120 True 

121 

122 Create SSL context with verification disabled (development/testing): 

123 

124 >>> tls_config = MCPClientTLSConfig(verify=False, check_hostname=False) 

125 >>> ssl_context = create_ssl_context(tls_config, "DevPlugin") 

126 >>> ssl_context.verify_mode == 0 # ssl.CERT_NONE 

127 True 

128 >>> ssl_context.check_hostname 

129 False 

130 

131 Verify TLS 1.2 minimum version enforcement: 

132 

133 >>> tls_config = MCPClientTLSConfig(verify=True) 

134 >>> ssl_context = create_ssl_context(tls_config, "SecurePlugin") 

135 >>> ssl_context.minimum_version.name 

136 'TLSv1_2' 

137 

138 Mixed security settings (verify enabled, hostname check disabled): 

139 

140 >>> tls_config = MCPClientTLSConfig(verify=True, check_hostname=False) 

141 >>> ssl_context = create_ssl_context(tls_config, "MixedPlugin") 

142 >>> ssl_context.verify_mode == 2 # ssl.CERT_REQUIRED 

143 True 

144 >>> ssl_context.check_hostname 

145 False 

146 

147 Default configuration is secure: 

148 

149 >>> tls_config = MCPClientTLSConfig() 

150 >>> ssl_context = create_ssl_context(tls_config, "DefaultPlugin") 

151 >>> ssl_context.verify_mode == 2 # ssl.CERT_REQUIRED 

152 True 

153 >>> ssl_context.check_hostname 

154 True 

155 >>> ssl_context.minimum_version.name 

156 'TLSv1_2' 

157 

158 Test error handling with invalid certificate file: 

159 

160 >>> import tempfile 

161 >>> import os 

162 >>> tmp_dir = tempfile.mkdtemp() 

163 >>> bad_cert = os.path.join(tmp_dir, "bad.pem") 

164 >>> with open(bad_cert, 'w') as f: 

165 ... _ = f.write("INVALID CERT") 

166 >>> tls_config = MCPClientTLSConfig(certfile=bad_cert, keyfile=bad_cert, verify=False) 

167 >>> try: 

168 ... ssl_context = create_ssl_context(tls_config, "BadCertPlugin") 

169 ... except PluginError as e: 

170 ... "Failed to configure SSL context" in e.error.message 

171 True 

172 

173 Verify logging occurs for different configurations: 

174 

175 >>> import logging 

176 >>> tls_config = MCPClientTLSConfig(verify=False) 

177 >>> ssl_context = create_ssl_context(tls_config, "LogTestPlugin") 

178 >>> ssl_context is not None 

179 True 

180 """ 

181 try: 

182 # Create SSL context with secure defaults 

183 # Per Python docs: "The settings are chosen by the ssl module, and usually 

184 # represent a higher security level than when calling the SSLContext 

185 # constructor directly." 

186 # This sets verify_mode to CERT_REQUIRED by default, which enables: 

187 # - Certificate signature validation 

188 # - Certificate chain validation up to trusted CA 

189 # - Automatic expiration checking (notBefore/notAfter per RFC 5280) 

190 ssl_context = ssl.create_default_context() # NOSONAR as this will fail check_hostname from NOT tls_config.verify 

191 

192 # Enforce TLS 1.2 or higher for security 

193 ssl_context.minimum_version = ssl.TLSVersion.TLSv1_2 

194 

195 if not tls_config.verify: 

196 # Disable certificate verification (not recommended for production) 

197 logger.warning(f"Certificate verification disabled for plugin '{plugin_name}'. This is not recommended for production use.") 

198 ssl_context.check_hostname = False # NOSONAR as this is specifically NOT tls_config.verify 

199 ssl_context.verify_mode = ssl.CERT_NONE # nosec B502 # noqa: DUO122 #NOSONAR as this is specifically NOT tls_config.verify 

200 else: 

201 # Enable strict certificate verification (production mode) 

202 # Load CA certificate bundle for server certificate validation 

203 if tls_config.ca_bundle: 

204 # This CA bundle will be used to validate the server's certificate 

205 # OpenSSL will check: 

206 # - Certificate is signed by a trusted CA in this bundle 

207 # - Certificate hasn't expired (notAfter > now) 

208 # - Certificate is already valid (notBefore < now) 

209 # - Certificate chain is complete and valid 

210 ssl_context.load_verify_locations(cafile=tls_config.ca_bundle) 

211 

212 # Hostname verification 

213 # When enabled, certificate's SAN or CN must match the server hostname 

214 if not tls_config.check_hostname: 

215 logger.warning(f"Hostname verification disabled for plugin '{plugin_name}'. This increases risk of MITM attacks.") 

216 ssl_context.check_hostname = False 

217 

218 # Load client certificate for mTLS (mutual authentication) 

219 # If provided, the client will authenticate itself to the server 

220 if tls_config.certfile: 

221 ssl_context.load_cert_chain( 

222 certfile=tls_config.certfile, 

223 keyfile=tls_config.keyfile, 

224 password=tls_config.keyfile_password, 

225 ) 

226 logger.debug(f"mTLS enabled for plugin '{plugin_name}' with client certificate: {tls_config.certfile}") 

227 

228 # Log security configuration 

229 logger.debug( 

230 f"SSL context created for plugin '{plugin_name}': verify_mode={ssl_context.verify_mode}, check_hostname={ssl_context.check_hostname}, minimum_version={ssl_context.minimum_version}" 

231 ) 

232 

233 return ssl_context 

234 

235 except Exception as exc: 

236 error_msg = f"Failed to configure SSL context for plugin '{plugin_name}': {exc}" 

237 logger.error(error_msg) 

238 raise PluginError(error=PluginErrorModel(message=error_msg, plugin_name=plugin_name)) from exc