aws_get_certificate.py

#!/usr/bin/env python
'''Retrieve Certificates and Private Keys from AWS

Query Amazon Certificate Manager and Secrets Manager to query for both
the certificate, certificate chain, and the private key.  This results
in a reconstruction of a typical PEM file.
'''

import json
import os
import sys
import boto3
from Crypto.Cipher import AES
from Crypto.Protocol.KDF import PBKDF2
from Crypto.Util import Counter

SALT = 'LTzkm1w/p3ReDm9kmfmnwQ=='
ENCODING = 'utf-8'
AES_KEY_BYTES = 32

AWS_REGION = os.environ.get('AWS_REGION', 'us-east-2')
CERTIFICATE_OUTPUT_DIR = os.environ.get('CERTIFICATE_OUTPUT_DIR',
                                        '/tmp/odyssey')


def __derive_key__(password, salt=None):
    '''Return key using PBKDF2'''
    if not salt:
        salt = SALT
    return PBKDF2(
        password,
        salt.encode(ENCODING),
        AES_KEY_BYTES
    )


def __decrypt_data__(password, salt, data):
    '''Decrypt given data using password and PBKDF2 key stretching'''

    key = __derive_key__(password, salt)
    aes_cipher = AES.new(key, AES.MODE_CTR, counter=Counter.new(128))
    return aes_cipher.decrypt(data).decode(ENCODING)


def main(certificate_path, secret_id, region=None):
    '''Download and decrypt certificate file from EFS

    Use AWS Secrets Manager to download shared key for decryption.

    Arguments:

    - `certificate_path`: Absolute path to encrypted certificate (PEM)
      file.  The basename of this file is used for the output of the
      unencrypted certificate contents.

    - `secret_id`: AWS Secrets Manager secret identifier.  This is
      often referred to as the "secrets path" in conversation.  E.g.,
      `test/certs/rdc/foobar.homecu.io.pem`.  The `secret_id` will
      also be used for the path and name of the unencrypted certificate file.

    - `region`: AWS Region to download secrets from, defaults to
      environment variable `AWS_REGION`.

    '''

    assert os.path.exists(certificate_path)
    if not region:
        region = AWS_REGION
    secrets = get_password_from_sm(secret_id, region)
    salt = secrets['salt']
    password = secrets['password']

    with open(certificate_path, 'rb') as fh:
        encrypted_certificate = fh.read()

    certificate = __decrypt_data__(password, salt, encrypted_certificate)

    certificate_output_name = os.path.join(CERTIFICATE_OUTPUT_DIR,
                                           secret_id)

    make_directory(os.path.dirname(certificate_output_name))

    with open(certificate_output_name, 'w') as fh:
        fh.write(certificate)


def make_directory(directory_name, mode=0o700):
    if not os.path.exists(directory_name):
        os.makedirs(directory_name, mode=mode)


def get_password_from_sm(secret_id, region):
    client = boto3.client('secretsmanager', region_name=region)
    response = client.get_secret_value(SecretId=secret_id)
    return json.loads(response['SecretString'])


if __name__ == '__main__':
    assert len(sys.argv[1:]) >= 2
    main(sys.argv[1], sys.argv[2])