def get_ca_and_tls()

in mysqloperator/controller/innodbcluster/cluster_api.py [0:0]

• 47 lines of code
• 16 McCabe index (conditional complexity)

    def get_ca_and_tls(self) -> Dict:
        if self.parsed_spec.tlsUseSelfSigned:
            return {}

        ca_secret = None
        server_tls_secret = None
        same_secret_for_ca_and_tls = False
        ret = {}
        try:
            server_tls_secret = cast(api_client.V1Secret, api_core.read_namespaced_secret(
                                     self.parsed_spec.tlsSecretName, self.namespace))

        except ApiException as e:
            if e.status == 404:
                return {}
            raise

        if "tls.crt" in server_tls_secret.data:
            ret["tls.crt"] = utils.b64decode(server_tls_secret.data["tls.crt"])
        if "tls.key" in server_tls_secret.data:
            ret["tls.key"] = utils.b64decode(server_tls_secret.data["tls.key"])

        if self.parsed_spec.tlsSecretName == self.parsed_spec.tlsCASecretName:
            ca_secret = server_tls_secret
            same_secret_for_ca_and_tls = True
        else:
            try:
                ca_secret = cast(api_client.V1Secret, api_core.read_namespaced_secret(
                                 self.parsed_spec.tlsCASecretName, self.namespace))
            except ApiException as e:
                if e.status == 404:
                    return ret
                raise

        ca_file_name = None
        if "ca.pem" in ca_secret.data:
            ca_file_name = "ca.pem"
        elif "ca.crt" in ca_secret.data:
            ca_file_name = "ca.crt"

        ret["CA"] = ca_file_name
        if ca_file_name:
            ret[ca_file_name] = utils.b64decode(ca_secret.data[ca_file_name])
            ret['same_secret_for_ca_and_tls'] = same_secret_for_ca_and_tls

        # When using HELM a secret should exist, when using bare manifests the secret might
        # not exist (not mentioned directly or using the default name) and so it is not mounted
        # in the router pod, thus not passed to the router.
        try:
            router_tls_secret = cast(api_client.V1Secret, api_core.read_namespaced_secret(
                                     self.parsed_spec.router.tlsSecretName, self.namespace))
            ret["router_tls.crt"] = utils.b64decode(router_tls_secret.data["tls.crt"])
            ret["router_tls.key"] = utils.b64decode(router_tls_secret.data["tls.key"])
        except ApiException as e:
            if e.status != 404:
                raise

        return ret