(self)
| 1916 | class TestKmsTLSOptions(EncryptionIntegrationTest): |
| 1917 | @unittest.skipUnless(any(AWS_CREDS.values()), "AWS environment credentials are not set") |
| 1918 | def setUp(self): |
| 1919 | super().setUp() |
| 1920 | # 1, create client with only tlsCAFile. |
| 1921 | providers: dict = copy.deepcopy(ALL_KMS_PROVIDERS) |
| 1922 | providers["azure"]["identityPlatformEndpoint"] = "127.0.0.1:9002" |
| 1923 | providers["gcp"]["endpoint"] = "127.0.0.1:9002" |
| 1924 | kms_tls_opts_ca_only = { |
| 1925 | "aws": {"tlsCAFile": CA_PEM}, |
| 1926 | "azure": {"tlsCAFile": CA_PEM}, |
| 1927 | "gcp": {"tlsCAFile": CA_PEM}, |
| 1928 | "kmip": {"tlsCAFile": CA_PEM}, |
| 1929 | } |
| 1930 | self.client_encryption_no_client_cert = self.create_client_encryption( |
| 1931 | providers, "keyvault.datakeys", self.client, OPTS, kms_tls_options=kms_tls_opts_ca_only |
| 1932 | ) |
| 1933 | # 2, same providers as above but with tlsCertificateKeyFile. |
| 1934 | kms_tls_opts = copy.deepcopy(kms_tls_opts_ca_only) |
| 1935 | for p in kms_tls_opts: |
| 1936 | kms_tls_opts[p]["tlsCertificateKeyFile"] = CLIENT_PEM |
| 1937 | self.client_encryption_with_tls = self.create_client_encryption( |
| 1938 | providers, "keyvault.datakeys", self.client, OPTS, kms_tls_options=kms_tls_opts |
| 1939 | ) |
| 1940 | # 3, update endpoints to expired host. |
| 1941 | providers: dict = copy.deepcopy(providers) |
| 1942 | providers["azure"]["identityPlatformEndpoint"] = "127.0.0.1:9000" |
| 1943 | providers["gcp"]["endpoint"] = "127.0.0.1:9000" |
| 1944 | providers["kmip"]["endpoint"] = "127.0.0.1:9000" |
| 1945 | self.client_encryption_expired = self.create_client_encryption( |
| 1946 | providers, "keyvault.datakeys", self.client, OPTS, kms_tls_options=kms_tls_opts_ca_only |
| 1947 | ) |
| 1948 | # 3, update endpoints to invalid host. |
| 1949 | providers: dict = copy.deepcopy(providers) |
| 1950 | providers["azure"]["identityPlatformEndpoint"] = "127.0.0.1:9001" |
| 1951 | providers["gcp"]["endpoint"] = "127.0.0.1:9001" |
| 1952 | providers["kmip"]["endpoint"] = "127.0.0.1:9001" |
| 1953 | self.client_encryption_invalid_hostname = self.create_client_encryption( |
| 1954 | providers, "keyvault.datakeys", self.client, OPTS, kms_tls_options=kms_tls_opts_ca_only |
| 1955 | ) |
| 1956 | # Errors when client has no cert, some examples: |
| 1957 | # [SSL: TLSV13_ALERT_CERTIFICATE_REQUIRED] tlsv13 alert certificate required (_ssl.c:2623) |
| 1958 | self.cert_error = ( |
| 1959 | "certificate required|SSL handshake failed|" |
| 1960 | "KMS connection closed|Connection reset by peer|ECONNRESET|EPIPE" |
| 1961 | ) |
| 1962 | # On Python 3.10+ this error might be: |
| 1963 | # EOF occurred in violation of protocol (_ssl.c:2384) |
| 1964 | if sys.version_info[:2] >= (3, 10): |
| 1965 | self.cert_error += "|EOF" |
| 1966 | # On Windows this error might be: |
| 1967 | # [WinError 10054] An existing connection was forcibly closed by the remote host |
| 1968 | if sys.platform == "win32": |
| 1969 | self.cert_error += "|forcibly closed" |
| 1970 | # 4, Test named KMS providers. |
| 1971 | providers = { |
| 1972 | "aws:no_client_cert": AWS_CREDS, |
| 1973 | "azure:no_client_cert": {"identityPlatformEndpoint": "127.0.0.1:9002", **AZURE_CREDS}, |
| 1974 | "gcp:no_client_cert": {"endpoint": "127.0.0.1:9002", **GCP_CREDS}, |
| 1975 | "kmip:no_client_cert": KMIP_CREDS, |
nothing calls this directly
no test coverage detected