atst/tests/domain/authnid/test_crl.py

# Import installed packages
import pytest
import re
import os
import shutil
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import Encoding
from OpenSSL import crypto

from atst.domain.authnid.crl import (
    CRLCache,
    CRLRevocationException,
    CRLInvalidException,
    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,
    CRLParseError,
    JSON_CACHE,
)

from tests.mocks import FIXTURE_EMAIL_ADDRESS, DOD_CN
from tests.utils import FakeLogger, parse_for_issuer_and_next_update


class MockX509Store:
    def __init__(self):
        self.crls = []
        self.certs = []

    def add_crl(self, crl):
        self.crls.append(crl)

    def add_cert(self, cert):
        self.certs.append(cert)

    def set_flags(self, flag):
        pass


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())
    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


def test_can_build_trusted_root_list(app):
    location = "ssl/server-certs/ca-chain.pem"
    cache = CRLCache(
        location, app.config["CRL_STORAGE_CONTAINER"], store_class=MockX509Store
    )
    with open(location) as f:
        content = f.read()
        assert len(cache.certificate_authorities.keys()) == content.count("BEGIN CERT")


def test_crl_validation_on_login(
    app,
    client,
    ca_key,
    ca_file,
    crl_file,
    rsa_key,
    make_x509,
    make_crl,
    serialize_pki_object_to_disk,
):
    good_cert = make_x509(rsa_key(), signer_key=ca_key, cn="luke")
    bad_cert = make_x509(rsa_key(), signer_key=ca_key, cn="darth")

    crl = make_crl(ca_key, expired_serials=[bad_cert.serial_number])
    serialize_pki_object_to_disk(crl, crl_file, encoding=Encoding.DER)
    crl_dir = os.path.dirname(crl_file)

    cache = CRLCache(ca_file, crl_dir)
    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())


def test_can_dynamically_update_crls(
    ca_key,
    ca_file,
    crl_file,
    rsa_key,
    make_x509,
    make_crl,
    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)
    assert cache.crl_check(client_pem)

    revoked_crl = make_crl(ca_key, expired_serials=[client_cert.serial_number])
    # override the original CRL with one that revokes client_cert
    serialize_pki_object_to_disk(revoked_crl, crl_file, encoding=Encoding.DER)

    with pytest.raises(CRLRevocationException):
        assert cache.crl_check(client_pem)


def test_throws_error_for_missing_issuer(app):
    cache = CRLCache(
        "ssl/server-certs/ca-chain.pem", app.config["CRL_STORAGE_CONTAINER"]
    )
    # this cert is self-signed, and so the application does not have a
    # corresponding CRL for it
    cert = open("tests/fixtures/{}.crt".format(FIXTURE_EMAIL_ADDRESS), "rb").read()
    with pytest.raises(CRLInvalidException) as exc:
        assert cache.crl_check(cert)
    (message,) = exc.value.args
    # objects that the issuer is missing
    assert "issuer" in message
    # names the issuer we were expecting to find a CRL for; same as the
    # certificate subject in this case because the cert is self-signed
    assert DOD_CN in message


def test_multistep_certificate_chain():
    cache = CRLCache("tests/fixtures/chain/ca-chain.pem", "tests/fixtures/chain/")
    cert = open("tests/fixtures/chain/client.crt", "rb").read()
    assert cache.crl_check(cert)


def test_no_op_crl_cache_logs_common_name():
    logger = FakeLogger()
    cert = open("ssl/client-certs/atat.mil.crt", "rb").read()
    cache = NoOpCRLCache(logger=logger)
    assert cache.crl_check(cert)
    assert "ART.GARFUNKEL.1234567890" in logger.messages[-1]


def test_expired_crl_raises_CRLInvalidException_with_failover_config_false(
    app, ca_file, expired_crl_file, ca_key, make_x509, rsa_key
):
    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)
    with pytest.raises(CRLInvalidException):
        crl_cache.crl_check(client_pem)


def test_expired_crl_passes_with_failover_config_true(
    ca_file, expired_crl_file, ca_key, make_x509, rsa_key, crl_failover_open_app
):
    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)

    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):
    dir_ = os.path.dirname(crl_file)
    serialize_crl_locations_cache(dir_)
    cache = load_crl_locations_cache(dir_)
    assert isinstance(cache, dict)