Move crl fixtures to conftest

This commit is contained in:
Montana 2019-03-12 10:29:59 -04:00
parent 30cd77ff98
commit 2eeb548458
2 changed files with 116 additions and 110 deletions

View File

@ -14,6 +14,15 @@ from atst.queue import queue as atst_queue
import tests.factories as factories import tests.factories as factories
from tests.mocks import PDF_FILENAME, PDF_FILENAME2 from tests.mocks import PDF_FILENAME, PDF_FILENAME2
from datetime import datetime, timezone, timedelta
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import Encoding
from cryptography.x509.oid import NameOID
dictConfig({"version": 1, "handlers": {"wsgi": {"class": "logging.NullHandler"}}}) dictConfig({"version": 1, "handlers": {"wsgi": {"class": "logging.NullHandler"}}})
@ -153,3 +162,105 @@ def extended_financial_verification_data(pdf_upload):
def queue(): def queue():
yield atst_queue yield atst_queue
atst_queue.get_queue().empty() atst_queue.get_queue().empty()
@pytest.fixture
def rsa_key():
def _rsa_key():
return rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend()
)
return _rsa_key
@pytest.fixture
def ca_key(rsa_key):
return rsa_key()
@pytest.fixture
def make_x509():
def _make_x509(private_key, signer_key=None, cn="ATAT", signer_cn="ATAT"):
if signer_key is None:
signer_key = private_key
one_day = timedelta(1, 0, 0)
public_key = private_key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
)
builder = builder.issuer_name(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, signer_cn)])
)
if signer_key == private_key:
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=None), critical=True
)
builder = builder.not_valid_before(datetime.today() - (one_day * 2))
builder = builder.not_valid_after(datetime.today() + (one_day * 30))
builder = builder.serial_number(x509.random_serial_number())
builder = builder.public_key(public_key)
certificate = builder.sign(
private_key=signer_key, algorithm=hashes.SHA256(), backend=default_backend()
)
return certificate
return _make_x509
@pytest.fixture
def make_crl():
def _make_crl(private_key, last_update_days=-1, next_update_days=30, cn="ATAT"):
one_day = timedelta(1, 0, 0)
builder = x509.CertificateRevocationListBuilder()
builder = builder.issuer_name(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
)
builder = builder.last_update(datetime.today() + (one_day * last_update_days))
builder = builder.next_update(datetime.today() + (one_day * next_update_days))
crl = builder.sign(
private_key=private_key,
algorithm=hashes.SHA256(),
backend=default_backend(),
)
return crl
return _make_crl
def serialize_pki_object_to_disk(obj, name, encoding=Encoding.PEM):
with open(name, "wb") as file_:
file_.write(obj.public_bytes(encoding))
return name
@pytest.fixture
def ca_file(make_x509, ca_key, tmpdir):
ca = make_x509(ca_key)
ca_out = tmpdir.join("atat-ca.crt")
serialize_pki_object_to_disk(ca, ca_out)
return ca_out
@pytest.fixture
def expired_crl_file(make_crl, ca_key, tmpdir):
crl = make_crl(ca_key, last_update_days=-7, next_update_days=-1)
crl_out = tmpdir.join("atat-expired.crl")
serialize_pki_object_to_disk(crl, crl_out, encoding=Encoding.DER)
return crl_out
@pytest.fixture
def crl_file(make_crl, ca_key, tmpdir):
crl = make_crl(ca_key)
crl_out = tmpdir.join("atat-valid.crl")
serialize_pki_object_to_disk(crl, crl_out, encoding=Encoding.DER)
return crl_out

View File

@ -3,14 +3,8 @@ import pytest
import re import re
import os import os
import shutil import shutil
from datetime import datetime, timezone, timedelta
from OpenSSL import crypto, SSL
from cryptography import x509
from cryptography.hazmat.backends import default_backend from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import Encoding from cryptography.hazmat.primitives.serialization import Encoding
from cryptography.x509.oid import NameOID
from atst.domain.authnid.crl import ( from atst.domain.authnid.crl import (
CRLCache, CRLCache,
@ -22,104 +16,6 @@ from atst.domain.authnid.crl import (
from tests.mocks import FIXTURE_EMAIL_ADDRESS, DOD_CN from tests.mocks import FIXTURE_EMAIL_ADDRESS, DOD_CN
def rsa_key():
return rsa.generate_private_key(
public_exponent=65537, key_size=2048, backend=default_backend()
)
@pytest.fixture
def ca_key():
return rsa_key()
@pytest.fixture
def make_x509():
def _make_x509(private_key, signer_key=None, cn="ATAT", signer_cn="ATAT"):
if signer_key is None:
signer_key = private_key
one_day = timedelta(1, 0, 0)
public_key = private_key.public_key()
builder = x509.CertificateBuilder()
builder = builder.subject_name(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
)
builder = builder.issuer_name(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, signer_cn)])
)
if signer_key == private_key:
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=None), critical=True
)
builder = builder.not_valid_before(datetime.today() - (one_day * 2))
builder = builder.not_valid_after(datetime.today() + (one_day * 30))
builder = builder.serial_number(x509.random_serial_number())
builder = builder.public_key(public_key)
certificate = builder.sign(
private_key=signer_key, algorithm=hashes.SHA256(), backend=default_backend()
)
return certificate
return _make_x509
@pytest.fixture
def make_crl():
def _make_crl(private_key, last_update_days=-1, next_update_days=30, cn="ATAT"):
one_day = timedelta(1, 0, 0)
builder = x509.CertificateRevocationListBuilder()
builder = builder.issuer_name(
x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
)
builder = builder.last_update(datetime.today() + (one_day * last_update_days))
builder = builder.next_update(datetime.today() + (one_day * next_update_days))
crl = builder.sign(
private_key=private_key,
algorithm=hashes.SHA256(),
backend=default_backend(),
)
return crl
return _make_crl
def serialize_pki_object_to_disk(obj, name, encoding=Encoding.PEM):
with open(name, "wb") as file_:
file_.write(obj.public_bytes(encoding))
return name
@pytest.fixture
def ca_file(make_x509, ca_key, tmpdir):
ca = make_x509(ca_key)
ca_out = tmpdir.join("atat-ca.crt")
serialize_pki_object_to_disk(ca, ca_out)
return ca_out
@pytest.fixture
def expired_crl_file(make_crl, ca_key, tmpdir):
crl = make_crl(ca_key, last_update_days=-7, next_update_days=-1)
crl_out = tmpdir.join("atat-expired.crl")
serialize_pki_object_to_disk(crl, crl_out, encoding=Encoding.DER)
return crl_out
@pytest.fixture
def crl_file(make_crl, ca_key, tmpdir):
crl = make_crl(ca_key)
crl_out = tmpdir.join("atat-valid.crl")
serialize_pki_object_to_disk(crl, crl_out, encoding=Encoding.DER)
return crl_out
class MockX509Store: class MockX509Store:
def __init__(self): def __init__(self):
self.crls = [] self.crls = []
@ -135,11 +31,8 @@ class MockX509Store:
pass pass
def test_can_build_crl_list(ca_file, ca_key, make_crl, tmpdir): def test_can_build_crl_list(crl_file, ca_key, ca_file, make_crl, tmpdir):
crl = make_crl(ca_key) crl = make_crl(ca_key)
crl_file = tmpdir.join("atat.crl")
serialize_pki_object_to_disk(crl, crl_file, encoding=Encoding.DER)
cache = CRLCache(ca_file, crl_locations=[crl_file], store_class=MockX509Store) cache = CRLCache(ca_file, crl_locations=[crl_file], store_class=MockX509Store)
issuer_der = crl.issuer.public_bytes(default_backend()) issuer_der = crl.issuer.public_bytes(default_backend())
assert len(cache.crl_cache.keys()) == 1 assert len(cache.crl_cache.keys()) == 1
@ -239,7 +132,7 @@ def test_no_op_crl_cache_logs_common_name():
def test_expired_crl_raises_CRLInvalidException( def test_expired_crl_raises_CRLInvalidException(
ca_file, expired_crl_file, ca_key, make_x509 ca_file, expired_crl_file, ca_key, make_x509, rsa_key
): ):
client_cert = make_x509(rsa_key(), signer_key=ca_key, cn="chewbacca") client_cert = make_x509(rsa_key(), signer_key=ca_key, cn="chewbacca")
client_pem = client_cert.public_bytes(Encoding.PEM) client_pem = client_cert.public_bytes(Encoding.PEM)
@ -248,7 +141,9 @@ def test_expired_crl_raises_CRLInvalidException(
crl_cache.crl_check(client_pem) crl_cache.crl_check(client_pem)
def test_updates_expired_certs(ca_file, expired_crl_file, crl_file, ca_key, make_x509): def test_updates_expired_certs(
rsa_key, ca_file, expired_crl_file, crl_file, ca_key, make_x509
):
""" """
Given a CRLCache object with an expired CRL and a function for updating the Given a CRLCache object with an expired CRL and a function for updating the
CRLs, the CRLCache should run the update function before checking a CRLs, the CRLCache should run the update function before checking a