def _update_traffic_key(
        self,
        direction: tls.Direction,
        epoch: tls.Epoch,
        cipher_suite: tls.CipherSuite,
        secret: bytes,
    ) -> None:
        """
        Callback which is invoked by the TLS engine when new traffic keys are
        available.
        """
        # For clients, determine the negotiated protocol version.
        if (
            self._is_client
            and self._crypto_packet_version is not None
            and not self._version_negotiated_compatible
        ):
            self._version = self._crypto_packet_version
            self._version_negotiated_compatible = True
            self._logger.info(
                "Negotiated protocol version %s", pretty_protocol_version(self._version)
            )

        secrets_log_file = self._configuration.secrets_log_file
        if secrets_log_file is not None:
            label_row = self._is_client == (direction == tls.Direction.DECRYPT)
            label = SECRETS_LABELS[label_row][epoch.value]
            secrets_log_file.write(
                "%s %s %s\n" % (label, self.tls.client_random.hex(), secret.hex())
            )
            secrets_log_file.flush()

        crypto = self._cryptos[epoch]
        if direction == tls.Direction.ENCRYPT:
            crypto.send.setup(
                cipher_suite=cipher_suite, secret=secret, version=self._version
            )
        else:
            crypto.recv.setup(
                cipher_suite=cipher_suite, secret=secret, version=self._version
            )

        Callback which is invoked by the TLS engine when new traffic keys are
        available.
        
src/aioquic/quic/connection.py