utils.py 文件源码

def decrypt(config, output=False):
    stage = config['stage']
    check_encryption_required_fields(config['stages'][stage])
    region = config['stages'][stage]['keyRegion']
    kms = boto3.client('kms', region_name=region)

    enc_config = get_secret_config(config, stage)
    if isinstance(enc_config, string_types):
        # This is the old-style secretConfig, when everything was encrypted
        # into a single string.
        stage_cfg = base64.b64decode(enc_config)
        resp = kms.decrypt(CiphertextBlob=stage_cfg)
        plain = json.loads(resp['Plaintext'])
        if output:
            print('Decrypted config for stage {}:\n\n{}'.format(
                stage,
                yaml.round_trip_dump(plain)))
        return plain
    elif isinstance(enc_config, dict):
        # This is the new way, where all config items are encrypted separately.
        plain = {}
        for key, value in enc_config.items():
            if is_value_already_encrypted(value):
                ciphertext_blob = base64.b64decode(
                    value[len(ENCRYPTED_PREFIX):])
                resp = kms.decrypt(CiphertextBlob=ciphertext_blob)
                plain[key] = resp['Plaintext']
            else:
                raise Exception('Found unencrypted item in secretConfig: '
                                '{}'.format(key))
        if output:
            print('Decrypted config for stage {}:\n\n{}'.format(
                stage,
                yaml.round_trip_dump(plain)))
        return plain