""" .. module: lemur.plugins.lemur_digicert.digicert :platform: Unix :synopsis: This module is responsible for communicating with the DigiCert ' Advanced API. :license: Apache, see LICENSE for more details. DigiCert CertCentral (v2 API) Documentation https://www.digicert.com/services/v2/documentation Original Implementation: Chris Dorros, github.com/opendns/lemur-digicert .. moduleauthor:: Kevin Glisson """ import json import arrow import requests import pem from retrying import retry from flask import current_app from lemur.extensions import metrics from lemur.plugins.bases import IssuerPlugin, SourcePlugin from lemur.plugins import lemur_digicert as digicert def signature_hash(signing_algorithm): """Converts Lemur's signing algorithm into a format DigiCert understands. :param signing_algorithm: :return: str digicert specific algorithm string """ if not signing_algorithm: return current_app.config.get('DIGICERT_DEFAULT_SIGNING_ALGORITHM', 'sha256') if signing_algorithm == 'sha256WithRSA': return 'sha256' elif signing_algorithm == 'sha384WithRSA': return 'sha384' elif signing_algorithm == 'sha512WithRSA': return 'sha512' raise Exception('Unsupported signing algorithm.') def determine_validity_years(end_date): """Given an end date determine how many years into the future that date is. :param end_date: :return: str validity in years """ now = arrow.utcnow() if end_date < now.replace(years=+1): return 1 elif end_date < now.replace(years=+2): return 2 elif end_date < now.replace(years=+3): return 3 raise Exception("DigiCert issued certificates cannot exceed three" " years in validity") def get_issuance(options): """Get the time range for certificates. :param options: :return: """ if not options['validity_end']: options['validity_end'] = arrow.utcnow().replace(years=current_app.config.get('DIGICERT_DEFAULT_VALIDITY', 1)) validity_years = determine_validity_years(options['validity_end']) return validity_years def process_options(options, csr): """Set the incoming issuer options to DigiCert fields/options. :param options: :param csr: :return: dict or valid DigiCert options """ data = { "certificate": { "common_name": options['common_name'], "csr": csr.decode('utf-8'), "signature_hash": signature_hash(options.get('signing_algorithm')), }, "organization": { "id": current_app.config.get("DIGICERT_ORG_ID") }, } # add SANs if present if options.get('extensions', 'sub_alt_names'): dns_names = [] for san in options['extensions']['sub_alt_names']['names']: dns_names.append(san['value']) data['certificate']['dns_names'] = dns_names validity_years = get_issuance(options) data['custom_expiration_date'] = options['validity_end'].format('YYYY-MM-DD') data['validity_years'] = validity_years return data def handle_response(response): """ Handle the DigiCert API response and any errors it might have experienced. :param response: :return: """ metrics.send('digicert_status_code_{0}'.format(response.status_code), 'counter', 1) if response.status_code not in [200, 201, 302, 301]: raise Exception(response.json()['message']) return response.json() def verify_configuration(): """Verify that needed configuration variables are set before plugin startup.""" if not current_app.config.get('DIGICERT_API_KEY'): raise Exception("No Digicert API key found. Ensure that 'DIGICERT_API_KEY' is set in the Lemur conf.") if not current_app.config.get('DIGICERT_URL'): raise Exception("No Digicert URL found. Ensure that 'DIGICERT_URL' is set in the Lemur conf.") if not current_app.config.get('DIGICERT_ORG_ID'): raise Exception("No Digicert organization ID found. Ensure that 'DIGICERT_ORG_ID' is set in Lemur conf.") if not current_app.config.get('DIGICERT_ROOT'): raise Exception("No Digicert root found. Ensure that 'DIGICERT_ROOT' is set in the Lemur conf.") if not current_app.config.get('DIGICERT_INTERMEDIATE'): raise Exception("No Digicert intermediate found. Ensure that 'DIGICERT_INTERMEDIATE is set in Lemur conf.") @retry(stop_max_attempt_number=10, wait_fixed=100000) def get_certificate_id(session, base_url, order_id): """Retrieve certificate order id from Digicert API.""" order_url = "{0}/services/v2/order/certificate/{1}".format(base_url, order_id) response_data = handle_response(session.get(order_url)) if response_data['status'] == 'issued': raise Exception("Order not in issued state.") return response_data['certificate']['id'] class DigiCertSourcePlugin(SourcePlugin): """Wrap the Digicert Certifcate API.""" title = 'DigiCert' slug = 'digicert-source' description = "Enables the use of Digicert as a source of existing certificates." version = digicert.VERSION author = 'Kevin Glisson' author_url = 'https://github.com/netflix/lemur.git' def __init__(self, *args, **kwargs): """Initialize source with appropriate details.""" verify_configuration() self.session = requests.Session() self.session.headers.update( { 'X-DC-DEVKEY': current_app.config.get('DIGICERT_API_KEY'), 'Content-Type': 'application/json' } ) super(DigiCertSourcePlugin, self).__init__(*args, **kwargs) def get_certificates(self): pass class DigiCertIssuerPlugin(IssuerPlugin): """Wrap the Digicert Issuer API.""" title = 'DigiCert' slug = 'digicert-issuer' description = "Enables the creation of certificates by" "the DigiCert REST API." version = digicert.VERSION author = 'Kevin Glisson' author_url = 'https://github.com/netflix/lemur.git' def __init__(self, *args, **kwargs): """Initialize the issuer with the appropriate details.""" verify_configuration() self.session = requests.Session() self.session.headers.update( { 'X-DC-DEVKEY': current_app.config.get('DIGICERT_API_KEY'), 'Content-Type': 'application/json' } ) super(DigiCertIssuerPlugin, self).__init__(*args, **kwargs) def create_certificate(self, csr, issuer_options): """Create a DigiCert certificate. :param csr: :param issuer_options: :return: :raise Exception: """ base_url = current_app.config.get('DIGICERT_URL') # make certificate request determinator_url = "{0}/services/v2/order/certificate/ssl".format(base_url) data = process_options(issuer_options, csr) response = self.session.post(determinator_url, data=json.dumps(data)) order_id = response.json()['id'] certificate_id = get_certificate_id(self.session, base_url, order_id) # retrieve certificate certificate_url = "{0}/services/v2/certificate/{1}/download/format/pem_all".format(base_url, certificate_id) end_entity, intermediate, root = pem.parse(self.session.get(certificate_url).content) return str(end_entity), str(intermediate) @staticmethod def create_authority(options): """Create an authority. Creates an authority, this authority is then used by Lemur to allow a user to specify which Certificate Authority they want to sign their certificate. :param options: :return: """ role = {'username': '', 'password': '', 'name': 'digicert'} return current_app.config.get('DIGICERT_ROOT'), "", [role]