*.pyc

def test_status_on_initialised_directory(tmpdir):

    # John is interested in finding out a bit more about what

    # certificates have been already issued in one of the projects he

    # had initialised before.

    tmpdir.chdir()

    run_command('gimmecert', 'init', '-d', '3', '-b', 'My Project')

    run_command('gimmecert', 'server', 'myserver1')

    run_command('gimmecert', 'server', 'myserver2', 'myservice.example.com', 'myotherservice.example.com')

    run_command('gimmecert', 'client', 'myclient1')

    run_command('gimmecert', 'client', 'myclient2')

    # John switches to project directory.

    tmpdir.chdir()

    # John runs the status command.

    stdout, stderr, exit_code = run_command('gimmecert', 'status')

    stdout_lines = stdout.split("\n")

    # Command exits with success without any errors being

    # reported.

    assert exit_code == 0

    assert stderr == ""

    # John notices that information is laid-out in three separate

    # sections - one for CA hierachy, one for server certificates, and

    # one for client certificates.

    assert "CA hierarchy" in stdout

    assert "Server certificates" in stdout

    assert "Client certificates" in stdout

    # John first has a look at information about the CA

    # hierarchy. Hierarchy tree is presented using indentation. Each

    # CA is listed with its full subject DN, as well as not before and

    # not after dates. In addition, the final CA in chain is marked as

    # end entity issuing CA.

    index_ca_1 = stdout_lines.index("CN=My Project Level 1 CA")  # Should not raise

    index_ca_2 = stdout_lines.index("CN=My Project Level 2 CA")  # Should not raise

    index_ca_3 = stdout_lines.index("CN=My Project Level 3 CA [END ENTITY ISSUING CA]")  # Should not raise

    assert index_ca_1 < index_ca_2

    assert index_ca_2 < index_ca_3

    assert stdout_lines[index_ca_1+1].startswith("    Validity: ")

    assert stdout_lines[index_ca_1+2].startswith("    Certificate: ")

    assert stdout_lines[index_ca_2+1].startswith("    Validity: ")

    assert stdout_lines[index_ca_2+2].startswith("    Certificate: ")

    assert stdout_lines[index_ca_3+1].startswith("    Validity: ")

    assert stdout_lines[index_ca_3+2].startswith("    Certificate: ")

    # In addition to CA information, path to full certificate chain is

    # output as well.

    assert "Full certificate chain:" in stdout

    # John then has a look at server certificates. These are presented

    # in a list, and for each certificate is listed with subject DN,

    # not before, not after, and included DNS names. Information for

    # each server is followed by paths to private key and certificate.

    index_myserver1 = stdout_lines.index("CN=myserver1")  # Should not raise

    index_myserver2 = stdout_lines.index("CN=myserver2")  # Should not raise

    assert stdout_lines[index_myserver1+1].startswith("    Validity: ")

    assert stdout_lines[index_myserver1+2] == "    DNS: myserver1"

    assert stdout_lines[index_myserver1+3] == "    Private key: .gimmecert/server/myserver1.key.pem"

    assert stdout_lines[index_myserver1+4] == "    Certificate: .gimmecert/server/myserver1.cert.pem"

    assert stdout_lines[index_myserver2+1].startswith("    Validity: ")

    assert stdout_lines[index_myserver2+2] == "    DNS: myserver2, myservice.example.com, myotherservice.example.com"

    assert stdout_lines[index_myserver2+3] == "    Private key: .gimmecert/server/myserver2.key.pem"

    assert stdout_lines[index_myserver2+4] == "    Certificate: .gimmecert/server/myserver2.cert.pem"

    # For client certificates, John can see that for each certificate

    # he can see its subject DN and validity. Information for each

    # server is followed by paths to private key and certificate.

    index_myclient1 = stdout_lines.index("CN=myclient1")  # Should not raise

    index_myclient2 = stdout_lines.index("CN=myclient2")  # Should not raise

    assert stdout_lines[index_myclient1+1].startswith("    Validity: ")

    assert stdout_lines[index_myclient1+2] == "    Private key: .gimmecert/client/myclient1.key.pem"

    assert stdout_lines[index_myclient1+3] == "    Certificate: .gimmecert/client/myclient1.cert.pem"

    assert stdout_lines[index_myclient2+1].startswith("    Validity: ")

    assert stdout_lines[index_myclient2+2] == "    Private key: .gimmecert/client/myclient2.key.pem"

    assert stdout_lines[index_myclient2+3] == "    Certificate: .gimmecert/client/myclient2.cert.pem"

import gimmecert.utils

        return ExitCode.ERROR_NOT_INITIALISED

    def get_section_title(title):

        """

        Small helper function that produces section title surrounded by

        separators for better visibility.

        """

        return "%s\n%s\n%s" % ("-" * len(title), title, "-" * len(title))

    print(get_section_title("CA hierarchy"), file=stdout)

    ca_hierarchy = gimmecert.storage.read_ca_hierarchy(os.path.join(project_directory, '.gimmecert', 'ca'))

    for i, (_, certificate) in enumerate(ca_hierarchy, 1):

        # Separator.

        print("", file=stdout)

        if i == len(ca_hierarchy):

            print(gimmecert.utils.dn_to_str(certificate.subject) + " [END ENTITY ISSUING CA]", file=stdout)

        else:

            print(gimmecert.utils.dn_to_str(certificate.subject), file=stdout)

        print("    Validity: %s" % gimmecert.utils.date_range_to_str(certificate.not_valid_before, certificate.not_valid_after), file=stdout)

        print("    Certificate: .gimmecert/ca/level%d.cert.pem" % i, file=stdout)

    # Separator.

    print("", file=stdout)

    print("Full certificate chain: .gimmecert/ca/chain-full.cert.pem", file=stdout)

    # Section separator.

    print("\n", file=stdout)

    print(get_section_title("Server certificates"), file=stdout)

    certificate_files = sorted([c for c in os.listdir(os.path.join(project_directory, '.gimmecert', 'server')) if c.endswith('.cert.pem')])

    for certificate_file in certificate_files:

        certificate = gimmecert.storage.read_certificate(os.path.join(project_directory, '.gimmecert', 'server', certificate_file))

        # Separator.

        print("", file=stdout)

        print(gimmecert.utils.dn_to_str(certificate.subject), file=stdout)

        print("    Validity: %s" % gimmecert.utils.date_range_to_str(certificate.not_valid_before, certificate.not_valid_after), file=stdout)

        print("    DNS: %s" % ", ".join(gimmecert.utils.get_dns_names(certificate)), file=stdout)

        print("    Private key: .gimmecert/server/%s" % certificate_file.replace('.cert.pem', '.key.pem'), file=stdout)

        print("    Certificate: .gimmecert/server/%s" % certificate_file, file=stdout)

    # Section separator.

    print("\n", file=stdout)

    print(get_section_title("Client certificates"), file=stdout)

    certificate_files = sorted([c for c in os.listdir(os.path.join(project_directory, '.gimmecert', 'client')) if c.endswith('.cert.pem')])

    for certificate_file in certificate_files:

        certificate = gimmecert.storage.read_certificate(os.path.join(project_directory, '.gimmecert', 'client', certificate_file))

        # Separator.

        print("", file=stdout)

        print(gimmecert.utils.dn_to_str(certificate.subject), file=stdout)

        print("    Validity: %s" % gimmecert.utils.date_range_to_str(certificate.not_valid_before, certificate.not_valid_after), file=stdout)

        print("    Private key: .gimmecert/client/%s" % certificate_file.replace('.cert.pem', '.key.pem'), file=stdout)

        print("    Certificate: .gimmecert/client/%s" % certificate_file, file=stdout)

import cryptography.x509

class UnsupportedField(Exception):

    """

    Exception thrown when trying to process an unsupported field in

    subject or issuer DN.

    """

    pass

def dn_to_str(dn):

    """

    Converts passed-in DN to a human-readable OpenSSL-style string

    representation.

    Currently supported fields:

    - Common name.

    :param dn: DN for which to generate string representation.

    :type dn: cryptography.x509.Name

    :returns: OpenSSL-style string representation of passed-in DN.

    :rtype: str

    """

    fields = []

    for field in dn:

        if field.oid == cryptography.x509.oid.NameOID.COMMON_NAME:

            fields.append("CN=%s" % field.value)

        else:

            raise UnsupportedField("Unable to generate string representation for: %s" % field.oid)

    return ",".join(fields)

def date_range_to_str(start, end):

    """

    Converts the provided validity range (with starting and end date),

    into a human-readable string.

    :param begin: Start date in UTC.

    :type begin: datetime.datetime

    :param end: End date in UTC.

    :type end: datetime.datetime

    :returns: String representation of date range, up to a second granularity.

    :rtype: str

    """

    date_format = "%Y-%m-%d %H:%M:%S UTC"

    return "%s - %s" % (start.strftime(date_format), end.strftime(date_format))

def get_dns_names(certificate):

    """

    Retrieves list of DNS subject alternative names from certificate.

    :param certificate: Certificate to process.

    :type certificate: cryptography.x509.Certificate

    :returns: List of DNS subject alternative names extracted from the certificate.

    :rtype: list[str]

    """

    try:

        subject_alternative_name = certificate.extensions.get_extension_for_class(cryptography.x509.SubjectAlternativeName).value

        dns_names = subject_alternative_name.get_values_for_type(cryptography.x509.DNSName)

    except cryptography.x509.extensions.ExtensionNotFound:

        dns_names = []

    return dns_names

from freezegun import freeze_time

    assert status_code == gimmecert.commands.ExitCode.ERROR_NOT_INITIALISED

def test_status_reports_ca_hierarchy_information(tmpdir):

    depth = 3

    stdout_stream = io.StringIO()

    stderr_stream = io.StringIO()

    with freeze_time('2018-01-01 00:15:00'):

        gimmecert.commands.init(io.StringIO(), io.StringIO(), tmpdir.strpath, tmpdir.basename, depth)

    status_code = gimmecert.commands.status(stdout_stream, stderr_stream, tmpdir.strpath)

    stdout = stdout_stream.getvalue()

    stdout_lines = stdout.split("\n")

    stderr = stderr_stream.getvalue()

    assert status_code == gimmecert.commands.ExitCode.SUCCESS

    assert stderr == ""

    assert "CA hierarchy\n------------\n" in stdout

    index_ca_1 = stdout_lines.index("CN=%s Level 1 CA" % tmpdir.basename)  # Should not raise

    index_ca_2 = stdout_lines.index("CN=%s Level 2 CA" % tmpdir.basename)  # Should not raise

    index_ca_3 = stdout_lines.index("CN=%s Level 3 CA [END ENTITY ISSUING CA]" % tmpdir.basename)  # Should not raise

    full_chain = stdout_lines.index("Full certificate chain: .gimmecert/ca/chain-full.cert.pem")  # Shold not raise

    assert full_chain > index_ca_3 > index_ca_2 > index_ca_1, "Output ordering for CA section is wrong:\n%s" % stdout

    ca_1_validity = stdout_lines[index_ca_1 + 1]

    ca_1_certificate_path = stdout_lines[index_ca_1 + 2]

    ca_2_validity = stdout_lines[index_ca_2 + 1]

    ca_2_certificate_path = stdout_lines[index_ca_2 + 2]

    ca_3_validity = stdout_lines[index_ca_3 + 1]

    ca_3_certificate_path = stdout_lines[index_ca_3 + 2]

    assert ca_1_validity == "    Validity: 2018-01-01 00:00:00 UTC - 2019-01-01 00:15:00 UTC"

    assert ca_1_certificate_path == "    Certificate: .gimmecert/ca/level1.cert.pem"

    assert ca_2_validity == "    Validity: 2018-01-01 00:00:00 UTC - 2019-01-01 00:15:00 UTC"

    assert ca_2_certificate_path == "    Certificate: .gimmecert/ca/level2.cert.pem"

    assert ca_3_validity == "    Validity: 2018-01-01 00:00:00 UTC - 2019-01-01 00:15:00 UTC"

    assert ca_3_certificate_path == "    Certificate: .gimmecert/ca/level3.cert.pem"

def test_status_reports_server_certificate_information(tmpdir):

    depth = 3

    stdout_stream = io.StringIO()

    stderr_stream = io.StringIO()

    with freeze_time('2018-01-01 00:15:00'):

        gimmecert.commands.init(io.StringIO(), io.StringIO(), tmpdir.strpath, tmpdir.basename, depth)

    with freeze_time('2018-02-01 00:15:00'):

        gimmecert.commands.server(io.StringIO(), io.StringIO(), tmpdir.strpath, 'myserver1', None)

    with freeze_time('2018-03-01 00:15:00'):

        gimmecert.commands.server(io.StringIO(), io.StringIO(), tmpdir.strpath, 'myserver2', ['myservice1.example.com', 'myservice2.example.com'])

    status_code = gimmecert.commands.status(stdout_stream, stderr_stream, tmpdir.strpath)

    stdout = stdout_stream.getvalue()

    stdout_lines = stdout.split("\n")

    stderr = stderr_stream.getvalue()

    assert status_code == gimmecert.commands.ExitCode.SUCCESS

    assert stderr == ""

    assert "Server certificates\n-------------------\n" in stdout

    index_myserver1 = stdout_lines.index("CN=myserver1")  # Should not raise

    index_myserver2 = stdout_lines.index("CN=myserver2")  # Should not raise

    myserver1_validity = stdout_lines[index_myserver1 + 1]

    myserver1_dns = stdout_lines[index_myserver1 + 2]

    myserver1_private_key_path = stdout_lines[index_myserver1 + 3]

    myserver1_certificate_path = stdout_lines[index_myserver1 + 4]

    myserver2_validity = stdout_lines[index_myserver2 + 1]

    myserver2_dns = stdout_lines[index_myserver2 + 2]

    myserver2_private_key_path = stdout_lines[index_myserver2 + 3]

    myserver2_certificate_path = stdout_lines[index_myserver2 + 4]

    assert myserver1_validity == "    Validity: 2018-02-01 00:00:00 UTC - 2019-01-01 00:15:00 UTC"

    assert myserver1_dns == "    DNS: myserver1"

    assert myserver1_private_key_path == "    Private key: .gimmecert/server/myserver1.key.pem"

    assert myserver1_certificate_path == "    Certificate: .gimmecert/server/myserver1.cert.pem"

    assert myserver2_validity == "    Validity: 2018-03-01 00:00:00 UTC - 2019-01-01 00:15:00 UTC"

    assert myserver2_dns == "    DNS: myserver2, myservice1.example.com, myservice2.example.com"

    assert myserver2_private_key_path == "    Private key: .gimmecert/server/myserver2.key.pem"

    assert myserver2_certificate_path == "    Certificate: .gimmecert/server/myserver2.cert.pem"

def test_status_reports_client_certificate_information(tmpdir):

    depth = 3

    stdout_stream = io.StringIO()

    stderr_stream = io.StringIO()

    with freeze_time('2018-01-01 00:15:00'):

        gimmecert.commands.init(io.StringIO(), io.StringIO(), tmpdir.strpath, tmpdir.basename, depth)

    with freeze_time('2018-02-01 00:15:00'):

        gimmecert.commands.client(io.StringIO(), io.StringIO(), tmpdir.strpath, 'myclient1')

    with freeze_time('2018-03-01 00:15:00'):

        gimmecert.commands.client(io.StringIO(), io.StringIO(), tmpdir.strpath, 'myclient2')

    status_code = gimmecert.commands.status(stdout_stream, stderr_stream, tmpdir.strpath)

    stdout = stdout_stream.getvalue()

    stdout_lines = stdout.split("\n")

    stderr = stderr_stream.getvalue()

    assert status_code == gimmecert.commands.ExitCode.SUCCESS

    assert stderr == ""

    assert "Client certificates\n-------------------\n" in stdout

    index_myclient1 = stdout_lines.index("CN=myclient1")  # Should not raise

    index_myclient2 = stdout_lines.index("CN=myclient2")  # Should not raise

    myclient1_validity = stdout_lines[index_myclient1 + 1]

    myclient1_private_key_path = stdout_lines[index_myclient1 + 2]

    myclient1_certificate_path = stdout_lines[index_myclient1 + 3]

    myclient2_validity = stdout_lines[index_myclient2 + 1]

    myclient2_private_key_path = stdout_lines[index_myclient2 + 2]

    myclient2_certificate_path = stdout_lines[index_myclient2 + 3]

    assert myclient1_validity == "    Validity: 2018-02-01 00:00:00 UTC - 2019-01-01 00:15:00 UTC"

    assert myclient1_private_key_path == "    Private key: .gimmecert/client/myclient1.key.pem"

    assert myclient1_certificate_path == "    Certificate: .gimmecert/client/myclient1.cert.pem"

    assert myclient2_validity == "    Validity: 2018-03-01 00:00:00 UTC - 2019-01-01 00:15:00 UTC"

    assert myclient2_private_key_path == "    Private key: .gimmecert/client/myclient2.key.pem"

    assert myclient2_certificate_path == "    Certificate: .gimmecert/client/myclient2.cert.pem"

import datetime

import pytest

def test_dn_to_str_with_cn():

    dn = gimmecert.crypto.get_dn('My test 1')

    dn_str = gimmecert.utils.dn_to_str(dn)

    assert isinstance(dn_str, str)

    assert dn_str == "CN=My test 1"

def test_dn_to_str_raises_for_unsupported_field_type():

    dn = cryptography.x509.Name([cryptography.x509.NameAttribute(cryptography.x509.oid.NameOID.COUNTRY_NAME, "RS")])

    with pytest.raises(gimmecert.utils.UnsupportedField):

        gimmecert.utils.dn_to_str(dn)

def test_date_range_to_str():

    begin = datetime.datetime(2017, 1, 2, 3, 4, 5)

    end = datetime.datetime(2018, 6, 7, 8, 9, 10)

    representation = gimmecert.utils.date_range_to_str(begin, end)

    assert isinstance(representation, str)

    assert representation == "2017-01-02 03:04:05 UTC - 2018-06-07 08:09:10 UTC"

def test_get_dns_names_returns_empty_list_if_no_dns_names():

    issuer_private_key, issuer_certificate = gimmecert.crypto.generate_ca_hierarchy('My Test', 1)[0]

    private_key = gimmecert.crypto.generate_private_key()

    certificate = gimmecert.crypto.issue_client_certificate(

        'myclient', private_key.public_key(),

        issuer_private_key, issuer_certificate

    )

    dns_names = gimmecert.utils.get_dns_names(certificate)

    assert isinstance(dns_names, list)

    assert dns_names == []

def test_get_dns_names_returns_list_of_dns_names():

    issuer_private_key, issuer_certificate = gimmecert.crypto.generate_ca_hierarchy('My Test', 1)[0]

    private_key = gimmecert.crypto.generate_private_key()

    certificate = gimmecert.crypto.issue_server_certificate(

        'myserver', private_key.public_key(),

        issuer_private_key, issuer_certificate,

        extra_dns_names=['myservice1.example.com', 'myservice2.example.com']

    )

    dns_names = gimmecert.utils.get_dns_names(certificate)

    assert isinstance(dns_names, list)

    assert dns_names == ['myserver', 'myservice1.example.com', 'myservice2.example.com']