pk/atst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1177 from dod-ccpo/crls-again

Browse Source 
Maintain static list of CRL URIs and issuers.
This commit is contained in:
dandds
2019-11-14 05:45:51 -05:00
committed by GitHub
parent 92ce3420b6 1b6239893b
commit 9c086e2f85
6 changed files with 307 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
tests/check_crl_parse.py
View File

@@ -1,19 +1,16 @@
import os
import pytest
from atst.domain.authnid.crl.util import scan_for_issuer_and_next_update
from atst.domain.authnid.crl.util import crl_local_path, CRL_LIST
from tests.utils import parse_for_issuer_and_next_update
CRL_DIR = "crls"
_CRLS = [
"{}/{}".format(CRL_DIR, file_) for file_ in os.listdir(CRL_DIR) if ".crl" in file_
]
@pytest.mark.parametrize("crl_path", _CRLS)
def test_crl_scan_against_parse(crl_path):
@pytest.mark.parametrize("crl_uri, issuer", CRL_LIST)
def test_crl_scan_against_parse(crl_uri, issuer):
crl_path = crl_local_path(CRL_DIR, crl_uri)
parsed_der = parse_for_issuer_and_next_update(crl_path)
scanned_der = scan_for_issuer_and_next_update(crl_path)
assert parsed_der == scanned_der
assert issuer == parsed_der.hex()

89
tests/domain/authnid/test_crl.py
View File

@@ -14,16 +14,14 @@ from atst.domain.authnid.crl import (
NoOpCRLCache,
)
from atst.domain.authnid.crl.util import (
scan_for_issuer_and_next_update,
build_crl_locations_cache,
serialize_crl_locations_cache,
load_crl_locations_cache,
serialize_crl_locations_cache,
CRLParseError,
JSON_CACHE,
)
from tests.mocks import FIXTURE_EMAIL_ADDRESS, DOD_CN
from tests.utils import FakeLogger, parse_for_issuer_and_next_update
from tests.utils import FakeLogger, parse_for_issuer_and_next_update, make_crl_list
class MockX509Store:
@@ -43,14 +41,13 @@ class MockX509Store:
def test_can_build_crl_list(crl_file, ca_key, ca_file, make_crl, tmpdir):
crl = make_crl(ca_key)
dir_ = os.path.dirname(crl_file)
serialize_crl_locations_cache(dir_)
cache = CRLCache(ca_file, dir_, store_class=MockX509Store)
issuer_der = crl.issuer.public_bytes(default_backend())
dir_ = os.path.dirname(crl_file)
serialize_crl_locations_cache(dir_, crl_list=[(str(crl_file), issuer_der.hex())])
cache = CRLCache(ca_file, dir_, store_class=MockX509Store)
assert len(cache.crl_cache.keys()) == 1
assert issuer_der in cache.crl_cache
assert cache.crl_cache[issuer_der]["location"] == crl_file
assert cache.crl_cache[issuer_der]["expires"] == crl.next_update
assert cache.crl_cache[issuer_der] == crl_file
def test_can_build_trusted_root_list(app):
@@ -81,7 +78,8 @@ def test_crl_validation_on_login(
serialize_pki_object_to_disk(crl, crl_file, encoding=Encoding.DER)
crl_dir = os.path.dirname(crl_file)
cache = CRLCache(ca_file, crl_dir)
crl_list = make_crl_list(good_cert, crl_file)
cache = CRLCache(ca_file, crl_dir, crl_list=crl_list)
assert cache.crl_check(good_cert.public_bytes(Encoding.PEM).decode())
with pytest.raises(CRLRevocationException):
cache.crl_check(bad_cert.public_bytes(Encoding.PEM).decode())
@@ -97,9 +95,10 @@ def test_can_dynamically_update_crls(
serialize_pki_object_to_disk,
):
crl_dir = os.path.dirname(crl_file)
cache = CRLCache(ca_file, crl_dir)
client_cert = make_x509(rsa_key(), signer_key=ca_key, cn="chewbacca")
client_pem = client_cert.public_bytes(Encoding.PEM)
crl_list = make_crl_list(client_cert, crl_file)
cache = CRLCache(ca_file, crl_dir, crl_list=crl_list)
assert cache.crl_check(client_pem)
revoked_crl = make_crl(ca_key, expired_serials=[client_cert.serial_number])
@@ -147,7 +146,8 @@ def test_expired_crl_raises_CRLInvalidException_with_failover_config_false(
client_cert = make_x509(rsa_key(), signer_key=ca_key, cn="chewbacca")
client_pem = client_cert.public_bytes(Encoding.PEM)
crl_dir = os.path.dirname(expired_crl_file)
crl_cache = CRLCache(ca_file, crl_dir)
crl_list = make_crl_list(client_cert, expired_crl_file)
crl_cache = CRLCache(ca_file, crl_dir, crl_list=crl_list)
with pytest.raises(CRLInvalidException):
crl_cache.crl_check(client_pem)
@@ -158,72 +158,13 @@ def test_expired_crl_passes_with_failover_config_true(
client_cert = make_x509(rsa_key(), signer_key=ca_key, cn="chewbacca")
client_pem = client_cert.public_bytes(Encoding.PEM)
crl_dir = os.path.dirname(expired_crl_file)
crl_cache = CRLCache(ca_file, crl_dir)
crl_list = make_crl_list(client_cert, expired_crl_file)
crl_cache = CRLCache(ca_file, crl_dir, crl_list=crl_list)
assert crl_cache.crl_check(client_pem)
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
CRLs, the CRLCache should run the update function before checking a
certificate that requires the expired CRL.
"""
client_cert = make_x509(rsa_key(), signer_key=ca_key, cn="chewbacca")
client_pem = client_cert.public_bytes(Encoding.PEM)
def _crl_update_func():
shutil.copyfile(crl_file, expired_crl_file)
crl_dir = os.path.dirname(expired_crl_file)
crl_cache = CRLCache(ca_file, crl_dir, crl_update_func=_crl_update_func)
crl_cache.crl_check(client_pem)
def test_scan_for_issuer_and_next_update(crl_file):
parsed = parse_for_issuer_and_next_update(crl_file)
scanned = scan_for_issuer_and_next_update(crl_file)
assert parsed == scanned
@pytest.fixture
def bad_crl(tmpdir):
bad_file = tmpdir.join("bad.crl")
with open(bad_file, "wb") as bad:
bad.write(b"definitely not a crl")
return bad_file
def test_scan_for_issuer_and_next_update_with_bad_data(bad_crl):
with pytest.raises(CRLParseError):
scan_for_issuer_and_next_update(bad_crl)
def test_build_crl_locations_cache(crl_file):
issuer_der, next_update = parse_for_issuer_and_next_update(crl_file)
cache = build_crl_locations_cache([crl_file])
assert cache == {issuer_der: {"location": crl_file, "expires": next_update}}
def test_build_crl_locations_cache_with_bad_data(crl_file, bad_crl):
logger = FakeLogger()
issuer_der, next_update = parse_for_issuer_and_next_update(crl_file)
cache = build_crl_locations_cache([crl_file, bad_crl], logger=logger)
assert cache == {issuer_der: {"location": crl_file, "expires": next_update}}
assert logger.messages
assert str(bad_crl) in logger.messages[0]
def test_serialize_crl_locations_cache(crl_file, bad_crl):
dir_ = os.path.dirname(crl_file)
serialize_crl_locations_cache(dir_)
assert os.path.isfile("{}/{}".format(dir_, JSON_CACHE))
def test_load_crl_locations_cache(crl_file, bad_crl):
def test_load_crl_locations_cache(crl_file):
dir_ = os.path.dirname(crl_file)
serialize_crl_locations_cache(dir_)
cache = load_crl_locations_cache(dir_)

10
tests/test_auth.py
View File

@@ -6,14 +6,16 @@ from datetime import datetime
from flask import session, url_for
from cryptography.hazmat.primitives.serialization import Encoding
from .mocks import DOD_SDN_INFO, DOD_SDN, FIXTURE_EMAIL_ADDRESS
from atst.domain.users import Users
from atst.domain.permission_sets import PermissionSets
from atst.domain.exceptions import NotFoundError
from atst.domain.authnid.crl import CRLInvalidException
from atst.domain.auth import UNPROTECTED_ROUTES
from atst.domain.authnid.crl import CRLCache
from .factories import UserFactory
from .mocks import DOD_SDN_INFO, DOD_SDN, FIXTURE_EMAIL_ADDRESS
from .utils import make_crl_list
MOCK_USER = {"id": "438567dd-25fa-4d83-a8cc-8aa8366cb24a"}
@@ -149,7 +151,8 @@ def swap_crl_cache(
crl = make_crl(ca_key)
serialize_pki_object_to_disk(crl, crl_file, encoding=Encoding.DER)
crl_dir = os.path.dirname(crl_file)
app.crl_cache = CRLCache(ca_file, crl_dir)
crl_list = make_crl_list(crl, crl_file)
app.crl_cache = CRLCache(ca_file, crl_dir, crl_list=crl_list)
yield _swap_crl_cache
@@ -175,7 +178,8 @@ def test_crl_validation_on_login(
serialize_pki_object_to_disk(crl, crl_file, encoding=Encoding.DER)
crl_dir = os.path.dirname(crl_file)
cache = CRLCache(ca_file, crl_dir)
crl_list = make_crl_list(good_cert, crl_file)
cache = CRLCache(ca_file, crl_dir, crl_list=crl_list)
swap_crl_cache(cache)
# bad cert is on the test CRL

14
tests/utils.py
View File

@@ -1,7 +1,10 @@
from flask import template_rendered
from contextlib import contextmanager
import os
from unittest.mock import Mock
from OpenSSL import crypto
from cryptography.hazmat.backends import default_backend
from flask import template_rendered
from atst.utils.notification_sender import NotificationSender
@@ -49,5 +52,10 @@ FakeNotificationSender = lambda: Mock(spec=NotificationSender)
def parse_for_issuer_and_next_update(crl):
with open(crl, "rb") as crl_file:
parsed = crypto.load_crl(crypto.FILETYPE_ASN1, crl_file.read())
next_update = parsed.to_cryptography().next_update
return (parsed.get_issuer().der(), next_update)
return parsed.get_issuer().der()
def make_crl_list(x509_obj, x509_path):
issuer = x509_obj.issuer.public_bytes(default_backend())
filename = os.path.basename(x509_path)
return [(filename, issuer.hex())]