lemur/lemur/tests/test_certificates.py

755 lines
26 KiB
Python
Raw Normal View History

from __future__ import unicode_literals # at top of module
2015-08-04 06:07:28 +02:00
2016-10-31 19:00:15 +01:00
import datetime
import json
2016-10-31 19:00:15 +01:00
import arrow
import pytest
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from marshmallow import ValidationError
from freezegun import freeze_time
from mock import patch
from lemur.certificates.service import create_csr
2015-07-21 22:06:13 +02:00
from lemur.certificates.views import * # noqa
from lemur.common import utils
from lemur.domains.models import Domain
from lemur.tests.vectors import VALID_ADMIN_API_TOKEN, VALID_ADMIN_HEADER_TOKEN, VALID_USER_HEADER_TOKEN, CSR_STR, \
INTERNAL_VALID_LONG_STR, INTERNAL_VALID_SAN_STR, PRIVATE_KEY_STR
2015-06-25 01:48:40 +02:00
2015-06-27 01:16:13 +02:00
def test_get_or_increase_name(session, certificate):
from lemur.certificates.models import get_or_increase_name
from lemur.tests.factories import CertificateFactory
assert get_or_increase_name(certificate.name, certificate.serial) == '{0}-3E9'.format(certificate.name)
certificate.name = 'test-cert-11111111'
assert get_or_increase_name(certificate.name, certificate.serial) == 'test-cert-11111111-3E9'
certificate.name = 'test-cert-11111111-1'
assert get_or_increase_name('test-cert-11111111-1', certificate.serial) == 'test-cert-11111111-1-3E9'
cert2 = CertificateFactory(name='certificate1-3E9')
session.commit()
assert get_or_increase_name('certificate1', 1001) == 'certificate1-3E9-1'
2016-10-31 19:00:15 +01:00
def test_get_certificate_primitives(certificate):
from lemur.certificates.service import get_certificate_primitives
names = [x509.DNSName(x.name) for x in certificate.domains]
2016-10-31 19:00:15 +01:00
data = {
'common_name': certificate.cn,
'owner': certificate.owner,
'authority': certificate.authority,
'description': certificate.description,
'extensions': {
'sub_alt_names': x509.SubjectAlternativeName(names)
2016-10-31 19:00:15 +01:00
},
'destinations': [],
'roles': [],
'validity_end': arrow.get(2021, 5, 7),
'validity_start': arrow.get(2016, 10, 30),
'country': 'US',
'location': 'A place',
'organization': 'Example',
'organizational_unit': 'Operations',
'state': 'CA'
2016-10-31 19:00:15 +01:00
}
with freeze_time(datetime.date(year=2016, month=10, day=30)):
primitives = get_certificate_primitives(certificate)
2018-05-10 21:57:54 +02:00
assert len(primitives) == 24
2016-10-31 19:00:15 +01:00
def test_certificate_output_schema(session, certificate, issuer_plugin):
from lemur.certificates.schemas import CertificateOutputSchema
# Clear the cached attribute first
if 'parsed_cert' in certificate.__dict__:
del certificate.__dict__['parsed_cert']
# Make sure serialization parses the cert only once (uses cached 'parsed_cert' attribute)
with patch('lemur.common.utils.parse_certificate', side_effect=utils.parse_certificate) as wrapper:
data, errors = CertificateOutputSchema().dump(certificate)
assert data['issuer'] == 'Example'
assert wrapper.call_count == 1
2016-10-18 08:23:14 +02:00
def test_certificate_edit_schema(session):
from lemur.certificates.schemas import CertificateEditInputSchema
input_data = {'owner': 'bob@example.com'}
data, errors = CertificateEditInputSchema().load(input_data)
assert len(data['notifications']) == 3
def test_authority_key_identifier_schema():
from lemur.schemas import AuthorityKeyIdentifierSchema
input_data = {
'useKeyIdentifier': True,
'useAuthorityCert': True
}
data, errors = AuthorityKeyIdentifierSchema().load(input_data)
assert sorted(data) == sorted({
'use_key_identifier': True,
'use_authority_cert': True
})
assert not errors
data, errors = AuthorityKeyIdentifierSchema().dumps(data)
assert sorted(data) == sorted(json.dumps(input_data))
assert not errors
def test_certificate_info_access_schema():
from lemur.schemas import CertificateInfoAccessSchema
input_data = {'includeAIA': True}
data, errors = CertificateInfoAccessSchema().load(input_data)
assert not errors
assert data == {'include_aia': True}
data, errors = CertificateInfoAccessSchema().dump(data)
assert not errors
assert data == input_data
def test_subject_key_identifier_schema():
from lemur.schemas import SubjectKeyIdentifierSchema
input_data = {'includeSKI': True}
data, errors = SubjectKeyIdentifierSchema().load(input_data)
assert not errors
assert data == {'include_ski': True}
data, errors = SubjectKeyIdentifierSchema().dump(data)
assert not errors
assert data == input_data
def test_extension_schema(client):
from lemur.certificates.schemas import ExtensionSchema
input_data = {
'keyUsage': {
'useKeyEncipherment': True,
'useDigitalSignature': True
},
'extendedKeyUsage': {
'useServerAuthentication': True
},
'subjectKeyIdentifier': {
'includeSKI': True
}
}
data, errors = ExtensionSchema().load(input_data)
assert not errors
data, errors = ExtensionSchema().dump(data)
assert not errors
def test_certificate_input_schema(client, authority):
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': 'test.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityEnd': arrow.get(2016, 11, 9).isoformat(),
'validityStart': arrow.get(2015, 11, 9).isoformat(),
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert not errors
assert data['authority'].id == authority.id
# make sure the defaults got set
assert data['common_name'] == 'test.example.com'
assert data['country'] == 'US'
assert data['location'] == 'Los Gatos'
assert len(data.keys()) == 19
def test_certificate_input_with_extensions(client, authority):
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': 'test.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'extensions': {
'keyUsage': {
'digital_signature': True
},
'extendedKeyUsage': {
'useClientAuthentication': True,
'useServerAuthentication': True
},
'subjectKeyIdentifier': {
'includeSKI': True
},
'subAltNames': {
'names': [
{'nameType': 'DNSName', 'value': 'test.example.com'}
]
}
},
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert not errors
def test_certificate_out_of_range_date(client, authority):
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': 'test.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityYears': 100,
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert errors
input_data['validityStart'] = '2017-04-30T00:12:34.513631'
data, errors = CertificateInputSchema().load(input_data)
assert errors
input_data['validityEnd'] = '2018-04-30T00:12:34.513631'
data, errors = CertificateInputSchema().load(input_data)
assert errors
def test_certificate_valid_years(client, authority):
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': 'test.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityYears': 1,
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert not errors
def test_certificate_valid_dates(client, authority):
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': 'test.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityStart': '2020-01-01T00:00:00',
'validityEnd': '2020-01-01T00:00:01',
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert not errors
def test_certificate_cn_admin(client, authority, logged_in_admin):
"""Admin is exempt from CN/SAN domain restrictions."""
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': '*.admin-overrides-whitelist.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityStart': '2020-01-01T00:00:00',
'validityEnd': '2020-01-01T00:00:01',
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert not errors
def test_certificate_allowed_names(client, authority, session, logged_in_user):
"""Test for allowed CN and SAN values."""
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': 'Names with spaces are not checked',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityStart': '2020-01-01T00:00:00',
'validityEnd': '2020-01-01T00:00:01',
'extensions': {
'subAltNames': {
'names': [
{'nameType': 'DNSName', 'value': 'allowed.example.com'},
{'nameType': 'IPAddress', 'value': '127.0.0.1'},
]
}
},
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert not errors
def test_certificate_incative_authority(client, authority, session, logged_in_user):
"""Cannot issue certificates with an inactive authority."""
from lemur.certificates.schemas import CertificateInputSchema
authority.active = False
session.add(authority)
input_data = {
'commonName': 'foo.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityStart': '2020-01-01T00:00:00',
'validityEnd': '2020-01-01T00:00:01',
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert errors['authority'][0] == "The authority is inactive."
def test_certificate_disallowed_names(client, authority, session, logged_in_user):
"""The CN and SAN are disallowed by LEMUR_WHITELISTED_DOMAINS."""
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': '*.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityStart': '2020-01-01T00:00:00',
'validityEnd': '2020-01-01T00:00:01',
'extensions': {
'subAltNames': {
'names': [
{'nameType': 'DNSName', 'value': 'allowed.example.com'},
{'nameType': 'DNSName', 'value': 'evilhacker.org'},
]
}
},
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
data, errors = CertificateInputSchema().load(input_data)
assert errors['common_name'][0].startswith("Domain *.example.com does not match whitelisted domain patterns")
assert (errors['extensions']['sub_alt_names']['names'][0]
.startswith("Domain evilhacker.org does not match whitelisted domain patterns"))
def test_certificate_sensitive_name(client, authority, session, logged_in_user):
"""The CN is disallowed by 'sensitive' flag on Domain model."""
from lemur.certificates.schemas import CertificateInputSchema
input_data = {
'commonName': 'sensitive.example.com',
'owner': 'jim@example.com',
'authority': {'id': authority.id},
'description': 'testtestest',
'validityStart': '2020-01-01T00:00:00',
'validityEnd': '2020-01-01T00:00:01',
2018-04-26 18:04:13 +02:00
'dnsProvider': None,
}
session.add(Domain(name='sensitive.example.com', sensitive=True))
data, errors = CertificateInputSchema().load(input_data)
assert errors['common_name'][0].startswith("Domain sensitive.example.com has been marked as sensitive")
def test_create_basic_csr(client):
csr_config = dict(
common_name='example.com',
2015-08-04 06:07:28 +02:00
organization='Example, Inc.',
organizational_unit='Operations',
2015-08-04 06:07:28 +02:00
country='US',
state='CA',
location='A place',
owner='joe@example.com',
key_type='RSA2048',
extensions=dict(names=dict(sub_alt_names=x509.SubjectAlternativeName([x509.DNSName('test.example.com'), x509.DNSName('test2.example.com')])))
)
csr, pem = create_csr(**csr_config)
2016-11-30 02:15:39 +01:00
csr = x509.load_pem_x509_csr(csr.encode('utf-8'), default_backend())
for name in csr.subject:
assert name.value in csr_config.values()
2015-06-27 01:16:13 +02:00
def test_csr_empty_san(client):
"""Test that an empty "names" list does not produce a CSR with empty SubjectAltNames extension.
The Lemur UI always submits this extension even when no alt names are defined.
"""
csr_text, pkey = create_csr(
common_name='daniel-san.example.com',
owner='daniel-san@example.com',
key_type='RSA2048',
extensions={'sub_alt_names': {'names': x509.SubjectAlternativeName([])}}
)
csr = x509.load_pem_x509_csr(csr_text.encode('utf-8'), default_backend())
with pytest.raises(x509.ExtensionNotFound):
csr.extensions.get_extension_for_class(x509.SubjectAlternativeName)
def test_csr_disallowed_cn(client, logged_in_user):
"""Domain name CN is disallowed via LEMUR_WHITELISTED_DOMAINS."""
from lemur.common import validators
request, pkey = create_csr(
common_name='evilhacker.org',
owner='joe@example.com',
key_type='RSA2048',
)
with pytest.raises(ValidationError) as err:
validators.csr(request)
assert str(err.value).startswith('Domain evilhacker.org does not match whitelisted domain patterns')
def test_csr_disallowed_san(client, logged_in_user):
"""SAN name is disallowed by LEMUR_WHITELISTED_DOMAINS."""
from lemur.common import validators
request, pkey = create_csr(
common_name="CN with spaces isn't a domain and is thus allowed",
owner='joe@example.com',
key_type='RSA2048',
extensions={'sub_alt_names': {'names': x509.SubjectAlternativeName([x509.DNSName('evilhacker.org')])}}
)
with pytest.raises(ValidationError) as err:
validators.csr(request)
assert str(err.value).startswith('Domain evilhacker.org does not match whitelisted domain patterns')
def test_get_name_from_arn(client):
from lemur.certificates.service import get_name_from_arn
2015-06-27 01:16:13 +02:00
arn = 'arn:aws:iam::11111111:server-certificate/mycertificate'
assert get_name_from_arn(arn) == 'mycertificate'
def test_get_account_number(client):
from lemur.certificates.service import get_account_number
2015-06-27 01:16:13 +02:00
arn = 'arn:aws:iam::11111111:server-certificate/mycertificate'
assert get_account_number(arn) == '11111111'
2016-12-08 20:33:40 +01:00
def test_mint_certificate(issuer_plugin, authority):
from lemur.certificates.service import mint
Async Certificate Issuing using Pending Certificates (#1037) * Add PendingCertificate model This change creates a DB table called pending_certificates and associated mapping relationship tables from pending certificate to roles, rotation policy, destination, sources, etc. The table is generated on initialization of Lemur. A pending certificate holds most of the information of a Certificate, while it has not be issued so that it can later backfill the information when the CA has issued the certificate. Change-Id: I277c16b776a71fe5edaf0fa0e76bbedc88924db0 Tickets: PBL-36499 * Create a PendingCertificate if cert is empty IssuePlugins should return empty cert bodies if the request failed to complete immediately (such as Digicert). This way, we can immediately return the certificate, or if not just place into PendingCertificates for later processing. + Fix relation from Certificate to Pending Certificate, as view only. There is no real need for anything more than that since Pending cert only needs to know the cert to replace when it is issued later. + Made PendingCertificate private key be empty: UI does not allow private key on 'Create' but only on 'Import'. For Instart, we require the private key but upstream does not necessarily need it. Thus, if someone at Instart wants to create a CSR / key combo, they should manually issue the cert themselves and import later. Otherwise you should let Lemur generate that. This keeps the workflow transparent for upstream Lemur users. Change-Id: Ib74722a5ed5792d4b10ca702659422739c95ae26 Tickets: PBL-36343 * Fix empty private_key when create Pending Cert On creation of a certificate with a CSR, there is no option for private key. In this case, we actually have a dictionary with private_key as key, but the value is None. This fixes the strip() called on NoneType. Change-Id: I7b265564d8095bfc83d9d4cd14ae13fea3c03199 Tickets: PBL-36499 * Source sync finds and uses pending certificate When a source syncs certificates, it will check for a pending certificate. If that is found via external_id (given by digicert as order_id) then it will use the found Pending Certificate's fields to create a new certificate. Then the pending certificate is deleted. Tickets: PBL-36343 Change-Id: I4f7959da29275ebc47a3996741f7e98d3e2d29d9 * Add Lemur static files and views for pending certs This adds the basic static files to view pending certificates in a table. Tickets: PBL-36343 Change-Id: Ia4362e6664ec730d05d280c5ef5c815a6feda0d9 * Add CLI and plugin based pending fetch This change uses the adds a new function to issuer plugins to fetch certificates like source, but for one order. This way, we can control which pending certificates to try and populate instead of getting all certificates from source. Tickets: PBL-36343 Change-Id: Ifc1747ccdc2cba09a81f298b31ddddebfee1b1d6 * Revert source using Pending Certificate Tickets: PBL-36343 Change-Id: I05121bc951e0530d804070afdb9c9e09baa0bc51 * Fix PendingCertificate init getting authority id Should get authority id from authority.id instead of the authority_id key in kwargs. Change-Id: Ie56df1a5fb0ab2729e91050f3ad1a831853e0623 Tickets: n/a * Add fixtures and basic test for PendingCertificate Change-Id: I4cca34105544d40dac1cc50a87bba93d8af9ab34 Tickets: PBL-36343 * Add User to create_certificate parameters create_certificate now takes a User, which will be used to populate the 'creator' field in certificates.service.upload(). This allows the UI populate with the current user if the owner does not exist in Lemur. + Fix chain being replaced with version from pending certificate, which may be empty (depends on plugin implementation). Change-Id: I516027b36bc643c4978b9c4890060569e03f3049 Tickets: n/a * Fix permalink and filters to pending certs Fixes the permalink button to get a single pending certificate Add argument filter parsing for the pending certificate API Fix comment on API usage Added get_by_name for pending_certificate (currently unused, but useful for CLI, instead of using IDs) Change-Id: Iaa48909c45606bec65dfb193c13d6bd0e816f6db Tickets: PBL-36910 * Update displayed fields for Pending Certificates There are a number of unused / unpopulated fields from Certificate UI that does apply to Pending Certificates. Those ones were removed, and added other useful fields: Owner, number of attempts to fetch and date created Change-Id: I3010a715f0357ba149cf539a19fdb5974c5ce08b Tickets: PBL-36910 * Add common name (cn) to Pending Certificate model Fixes the UI missing the CN for Pending Certificate, as it was originally being parsed from the generated certificate. In the case of pending certificate, the CN from the user generates the request, which means a pending cert can trust the original user putting in the CN instead of having to parse the not-yet-generated certificate. There is no real possibility to return a certificate from a pending certificate where the CN has changed since it was initially ordered. Change-Id: I88a4fa28116d5d8d293e58970d9777ce73fbb2ab Tickets: PBL-36910 * Fix missing imports for service filter + Removed duplicate get_by_name function from old merge Change-Id: I04ae6852533aa42988433338de74390e2868d69b Tickets: PBL-36910 * Add private key viewing to Pending Certificates Add private key API for Pending Certificates, with the same authorization as Certificates (only owner, creator or owner-roles can view private key). Change-Id: Ie5175154a10fe0007cc0e9f35b80c0a01ed48d5b Tickets: PBL-36910 * Add edit capability to pending certificates Like editing certificates, we should be able to modify some parts of a pending certificate so the resulting certificate has the right references, owner, etc. + Added API to update pending certificate + Fix UI to use pending certificate scope instead of reusing Certificate + Change pending_certificate.replaces to non-passive association, so that updates do affect it (similar to roles/notifications/etc) Tickets: PBL-36910 Change-Id: Ibbcb166a33f0337e1b14f426472261222f790ce6 * Add common_name parsing instead using kwargs To fix tests where common name may not be passed in, use the CSR generated to find the official common name. Change-Id: I09f9258fa92c2762d095798676ce210c5d7a3da4 Tickets: PBL-36343 * Add Cancel to pending certificates and plugins This allows pending certificates to be cancelled, which will be handled by the issuer plugin. Change-Id: Ibd6b5627c3977e33aca7860690cfb7f677236ca9 Tickets: PBL-36910 * Add API for Cancelling Pending Certificate Added the DELETE handler for pending_certificates, which will cancel and delete the pending certificate from the pending certs table on successful cancellation via Issuer Plugin. + Add UT for testing cancel API Change-Id: I11b1d87872e4284f6e4f9c366a15da4ddba38bc4 Tickets: PBL-36910 * Remove Export from Pending Certificates Pending Certificates doesn't need an export since it should just be fetched by Lemur via plugins, and the CSR is viewable via the UI. Change-Id: I9a3e65ea11ac5a85316f6428e7f526c3c09178ae Tickets: PBL-36910 * Add cancel button functionality to UI This adds the Cancel option to the dropdown of pending certificates. + Adds modal window for Note (may not be required for all issuers, just Digicert) + Add schema for cancel input + Fix Digitcert plugin for non-existant orders When an order is actually issued, then attempting to cancel will return a 403 from Digicert. This is a case where it should only be done once we know the pending cert has been sitting for too long. Change-Id: I256c81ecd142dd51dcf8e38802d2c202829887b0 Tickets: PBL-36910 * Fix test_pending_cancel UT This change creates and injects a pending cert, which will then be used for the ID so it can be canceled by the unit test. Change-Id: I686e7e0fafd68cdaeb26438fb8504d79de77c346 Tickets: PBL-36343 * Fix test_digicert on non-existent order cancelling a non-existent order is fine since we're cancelling it Change-Id: I70c0e82ba2f4b8723a7f65b113c19e6eeff7e68c Tickets: PBL-36343 * Add migrations for PendingCertificates Added revision for Pending Certificates table and foreign key mapping tables. Change-Id: Ife8202cef1e6b99db377851264639ba540b749db Tickets: n/a * Fix relationship copy from Pending to Certificate When a Pending Certificate is changed to a full Certificate, the relationship fields are not copied via vars() function, as it's not a column but mapped via association table. This adds an explicit copy for these relations. Which will properly copy them to the new Certificate, and thus also update destinations. Change-Id: I322032ce4a9e3e67773f7cf39ee4971054c92685 Tickets: PBL-36343 * Fix renaming of certificates and unit tests The rename flag was not used to rename certificates on creation as expected. Fixed unit test, instead of expunging the session, just copy the pending_certificate so we don't have a weird reference to the object that can't be copied via vars() function. Change-Id: I962943272ed92386ab6eab2af4ed6d074d4cffa0 Tickets: PBL-36343 * Updated developer docs for async certs Added blurb for implementing new issuer functions. Change-Id: I1caed6e914bcd73214eae2d241e4784e1b8a0c4c Tickets: n/a
2018-02-22 17:13:16 +01:00
cert_body, private_key, chain, external_id, csr = mint(authority=authority, csr=CSR_STR)
assert cert_body == INTERNAL_VALID_LONG_STR, INTERNAL_VALID_SAN_STR
def test_create_certificate(issuer_plugin, authority, user):
from lemur.certificates.service import create
cert = create(authority=authority, csr=CSR_STR, owner='joe@example.com', creator=user['user'])
assert str(cert.not_after) == '2040-01-01T20:30:52+00:00'
assert str(cert.not_before) == '2015-06-26T20:30:52+00:00'
assert cert.issuer == 'Example'
assert cert.name == 'long.lived.com-Example-20150626-20400101'
cert = create(authority=authority, csr=CSR_STR, owner='joe@example.com', name='ACustomName1', creator=user['user'])
assert cert.name == 'ACustomName1'
2016-12-08 20:33:40 +01:00
def test_reissue_certificate(issuer_plugin, authority, certificate):
from lemur.certificates.service import reissue_certificate
new_cert = reissue_certificate(certificate)
assert new_cert
def test_create_csr():
csr, private_key = create_csr(owner='joe@example.com', common_name='ACommonName', organization='test', organizational_unit='Meters', country='US',
state='CA', location='Here', key_type='RSA2048')
assert csr
assert private_key
extensions = {'sub_alt_names': {'names': x509.SubjectAlternativeName([x509.DNSName('AnotherCommonName')])}}
csr, private_key = create_csr(owner='joe@example.com', common_name='ACommonName', organization='test', organizational_unit='Meters', country='US',
state='CA', location='Here', extensions=extensions, key_type='RSA2048')
assert csr
assert private_key
def test_import(user):
from lemur.certificates.service import import_certificate
cert = import_certificate(body=INTERNAL_VALID_LONG_STR, chain=INTERNAL_VALID_SAN_STR, private_key=PRIVATE_KEY_STR, creator=user['user'])
assert str(cert.not_after) == '2040-01-01T20:30:52+00:00'
assert str(cert.not_before) == '2015-06-26T20:30:52+00:00'
assert cert.issuer == 'Example'
assert cert.name == 'long.lived.com-Example-20150626-20400101-2'
cert = import_certificate(body=INTERNAL_VALID_LONG_STR, chain=INTERNAL_VALID_SAN_STR, private_key=PRIVATE_KEY_STR, owner='joe@example.com', name='ACustomName2', creator=user['user'])
assert cert.name == 'ACustomName2'
@pytest.mark.skip
def test_upload(user):
from lemur.certificates.service import upload
cert = upload(body=INTERNAL_VALID_LONG_STR, chain=INTERNAL_VALID_SAN_STR, private_key=PRIVATE_KEY_STR, owner='joe@example.com', creator=user['user'])
assert str(cert.not_after) == '2040-01-01T20:30:52+00:00'
assert str(cert.not_before) == '2015-06-26T20:30:52+00:00'
assert cert.issuer == 'Example'
assert cert.name == 'long.lived.com-Example-20150626-20400101-3'
cert = upload(body=INTERNAL_VALID_LONG_STR, chain=INTERNAL_VALID_SAN_STR, private_key=PRIVATE_KEY_STR, owner='joe@example.com', name='ACustomName', creator=user['user'])
assert 'ACustomName' in cert.name
# verify upload with a private key as a str
def test_upload_private_key_str(user):
from lemur.certificates.service import upload
cert = upload(body=INTERNAL_VALID_LONG_STR, chain=INTERNAL_VALID_SAN_STR, private_key=PRIVATE_KEY_STR, owner='joe@example.com', name='ACustomName', creator=user['user'])
assert cert
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 200),
(VALID_ADMIN_HEADER_TOKEN, 200),
(VALID_ADMIN_API_TOKEN, 200),
('', 401)
])
def test_certificate_get_private_key(client, token, status):
assert client.get(api.url_for(Certificates, certificate_id=1), headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 200),
(VALID_ADMIN_HEADER_TOKEN, 200),
(VALID_ADMIN_API_TOKEN, 200),
('', 401)
])
def test_certificate_get(client, token, status):
assert client.get(api.url_for(Certificates, certificate_id=1), headers=token).status_code == status
def test_certificate_get_body(client):
response_body = client.get(api.url_for(Certificates, certificate_id=1), headers=VALID_USER_HEADER_TOKEN).json
assert response_body['serial'] == '1001'
assert response_body['serialHex'] == '3E9'
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificate_post(client, token, status):
assert client.post(api.url_for(Certificates, certificate_id=1), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 400),
(VALID_ADMIN_HEADER_TOKEN, 400),
(VALID_ADMIN_API_TOKEN, 400),
('', 401)
])
def test_certificate_put(client, token, status):
assert client.put(api.url_for(Certificates, certificate_id=1), data={}, headers=token).status_code == status
def test_certificate_put_with_data(client, certificate, issuer_plugin):
resp = client.put(api.url_for(Certificates, certificate_id=certificate.id), data=json.dumps({'owner': 'bob@example.com', 'description': 'test', 'notify': True}), headers=VALID_ADMIN_HEADER_TOKEN)
assert resp.status_code == 200
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificate_delete(client, token, status):
assert client.delete(api.url_for(Certificates, certificate_id=1), headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificate_patch(client, token, status):
assert client.patch(api.url_for(Certificates, certificate_id=1), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 200),
(VALID_ADMIN_HEADER_TOKEN, 200),
(VALID_ADMIN_API_TOKEN, 200),
('', 401)
])
def test_certificates_get(client, token, status):
assert client.get(api.url_for(CertificatesList), headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 400),
(VALID_ADMIN_HEADER_TOKEN, 400),
(VALID_ADMIN_API_TOKEN, 400),
('', 401)
])
def test_certificates_post(client, token, status):
assert client.post(api.url_for(CertificatesList), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificates_put(client, token, status):
assert client.put(api.url_for(CertificatesList), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificates_delete(client, token, status):
assert client.delete(api.url_for(CertificatesList), headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificates_patch(client, token, status):
assert client.patch(api.url_for(CertificatesList), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificate_credentials_post(client, token, status):
assert client.post(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificate_credentials_put(client, token, status):
assert client.put(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificate_credentials_delete(client, token, status):
assert client.delete(api.url_for(CertificatePrivateKey, certificate_id=1), headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificate_credentials_patch(client, token, status):
assert client.patch(api.url_for(CertificatePrivateKey, certificate_id=1), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificates_upload_get(client, token, status):
assert client.get(api.url_for(CertificatesUpload), headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 400),
(VALID_ADMIN_HEADER_TOKEN, 400),
(VALID_ADMIN_API_TOKEN, 400),
('', 401)
])
def test_certificates_upload_post(client, token, status):
assert client.post(api.url_for(CertificatesUpload), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificates_upload_put(client, token, status):
assert client.put(api.url_for(CertificatesUpload), data={}, headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificates_upload_delete(client, token, status):
assert client.delete(api.url_for(CertificatesUpload), headers=token).status_code == status
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
(VALID_ADMIN_API_TOKEN, 405),
('', 405)
])
def test_certificates_upload_patch(client, token, status):
assert client.patch(api.url_for(CertificatesUpload), data={}, headers=token).status_code == status
def test_sensitive_sort(client):
resp = client.get(api.url_for(CertificatesList) + '?sortBy=private_key&sortDir=asc', headers=VALID_ADMIN_HEADER_TOKEN)
assert "'private_key' is not sortable or filterable" in resp.json['message']
def test_boolean_filter(client):
resp = client.get(api.url_for(CertificatesList) + '?filter=notify;true', headers=VALID_ADMIN_HEADER_TOKEN)
assert resp.status_code == 200
# Also don't crash with invalid input (we currently treat that as false)
resp = client.get(api.url_for(CertificatesList) + '?filter=notify;whatisthis', headers=VALID_ADMIN_HEADER_TOKEN)
assert resp.status_code == 200