def from_object_meta()

in oss2/models.py [0:0]

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

    def from_object_meta(self, headers):
        if not isinstance(headers, CaseInsensitiveDict):
            headers = CaseInsensitiveDict(headers)

        if DEPRECATED_CLIENT_SIDE_ENCRYPTION_KEY in headers:
            self.deprecated = True

        if self.deprecated:
            undecode_encrypted_key = _hget(headers, DEPRECATED_CLIENT_SIDE_ENCRYPTION_KEY)
            undecode_encrypted_iv = _hget(headers, DEPRECATED_CLIENT_SIDE_ENCRYPTION_START)
            cek_alg = _hget(headers, DEPRECATED_CLIENT_SIDE_ENCRYPTION_CEK_ALG)
            wrap_alg = _hget(headers, DEPRECATED_CLIENT_SIDE_ENCRYPTION_WRAP_ALG)
            mat_desc = _hget(headers, DEPRECATED_CLIENT_SIDE_ENCRYTPION_MATDESC)

            if wrap_alg == "kms":
                self.encrypted_key = undecode_encrypted_key
                self.encrypted_iv = undecode_encrypted_iv
                wrap_alg = KMS_ALI_WRAP_ALGORITHM
            else:
                if undecode_encrypted_key:
                    self.encrypted_key = b64decode_from_string(undecode_encrypted_key)
                if undecode_encrypted_iv:
                    self.encrypted_iv = b64decode_from_string(undecode_encrypted_iv)
                wrap_alg = RSA_NONE_OAEPWithSHA1AndMGF1Padding
            if cek_alg == utils.AES_GCM:
                cek_alg = utils.AES_CTR
        else:
            undecode_encrypted_key = _hget(headers, OSS_CLIENT_SIDE_ENCRYPTION_KEY)
            undecode_encrypted_iv = _hget(headers, OSS_CLIENT_SIDE_ENCRYPTION_START)
            if undecode_encrypted_key:
                self.encrypted_key = b64decode_from_string(undecode_encrypted_key)
            if undecode_encrypted_iv:
                self.encrypted_iv = b64decode_from_string(undecode_encrypted_iv)
            cek_alg = _hget(headers, OSS_CLIENT_SIDE_ENCRYPTION_CEK_ALG)
            wrap_alg = _hget(headers, OSS_CLIENT_SIDE_ENCRYPTION_WRAP_ALG)
            mat_desc = _hget(headers, OSS_CLIENT_SIDE_ENCRYTPION_MATDESC)

        if mat_desc:
            self.mat_desc = json.loads(mat_desc)

        if cek_alg and cek_alg != self.cek_alg:
            logger.error("CEK algorithm or is inconsistent, object meta: cek_alg:{0}, material: cek_alg:{1}".
                         format(cek_alg, self.cek_alg))
            err_msg = 'Data encryption/decryption algorithm is inconsistent'
            raise InconsistentError(err_msg, self)

        if wrap_alg and wrap_alg != self.wrap_alg:
            logger.error("WRAP algorithm or is inconsistent, object meta: wrap_alg:{0}, material: wrap_alg:{1}".
                         format(wrap_alg, self.wrap_alg))
            err_msg = 'Envelope encryption/decryption algorithm is inconsistent'
            raise InconsistentError(err_msg, self)

        self.cek_alg = cek_alg
        self.wrap_alg = wrap_alg