Method kms_request

pymongo/synchronous/encryption.py:166–250

Complete a KMS request. :param kms_context: A :class:`MongoCryptKmsContext`. :return: None

(self, kms_context: MongoCryptKmsContext)

        self._kms_ssl_contexts = opts._kms_ssl_contexts(_IS_SYNC)

    def kms_request(self, kms_context: MongoCryptKmsContext) -> None:
        """Complete a KMS request.

        :param kms_context: A :class:`MongoCryptKmsContext`.

        :return: None
        """
        endpoint = kms_context.endpoint
        message = kms_context.message
        provider = kms_context.kms_provider
        ctx = self._kms_ssl_contexts.get(provider)
        if ctx is None:
            # Enable strict certificate verification, OCSP, match hostname, and
            # SNI using the system default CA certificates.
            ctx = get_ssl_context(
                None,  # certfile
                None,  # passphrase
                None,  # ca_certs
                None,  # crlfile
                False,  # allow_invalid_certificates
                False,  # allow_invalid_hostnames
                False,  # disable_ocsp_endpoint_check
                _IS_SYNC,
            )
        # CSOT: set timeout for socket creation.
        connect_timeout = max(_csot.clamp_remaining(_KMS_CONNECT_TIMEOUT), 0.001)
        opts = PoolOptions(
            connect_timeout=connect_timeout,
            socket_timeout=connect_timeout,
            ssl_context=ctx,
        )
        address = parse_host(endpoint, _HTTPS_PORT)
        sleep_u = kms_context.usleep
        if sleep_u:
            sleep_sec = float(sleep_u) / 1e6
            time.sleep(sleep_sec)
        try:
            conn = _connect_kms(address, opts)
            try:
                sendall(conn, message)
                while kms_context.bytes_needed > 0:
                    # CSOT: update timeout.
                    conn.settimeout(max(_csot.clamp_remaining(_KMS_CONNECT_TIMEOUT), 0))
                    data: memoryview | bytes
                    if _IS_SYNC:
                        data = conn.recv(kms_context.bytes_needed)
                    else:
                        from pymongo.network_layer import (  # type: ignore[attr-defined]
                            receive_data_socket,
                        )

                        data = receive_data_socket(conn, kms_context.bytes_needed)
                    if not data:
                        raise OSError("KMS connection closed")
                    kms_context.feed(data)
            except MongoCryptError:
                raise  # Propagate MongoCryptError errors directly.
            except Exception as exc: