404

D7net
:/ snap / certbot / 5893 / lib / python3.12 / site-packages / certbot / _internal / tests /
Filename : ocsp_test.py
back
"""Tests for ocsp.py"""
# pylint: disable=protected-access
import contextlib
from datetime import datetime
from datetime import timedelta
from datetime import timezone
import sys
import unittest
from unittest import mock

from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.exceptions import UnsupportedAlgorithm
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.x509 import ocsp as ocsp_lib
import pytest

from certbot import crypto_util
from certbot.tests import util as test_util


class OSCPTestCryptography(unittest.TestCase):
    """
    OCSP revokation tests using Cryptography >= 2.4.0
    """

    def setUp(self):
        from certbot._internal import ocsp
        self.checker = ocsp.RevocationChecker()
        self.cert_path = test_util.vector_path('ocsp_certificate.pem')
        self.chain_path = test_util.vector_path('ocsp_issuer_certificate.pem')
        self.cert_obj = mock.MagicMock()
        self.cert_obj.cert_path = self.cert_path
        self.cert_obj.chain_path = self.chain_path
        now = datetime.now(timezone.utc)
        self.mock_notAfter = mock.patch('certbot._internal.ocsp.crypto_util.notAfter',
                                        return_value=now + timedelta(hours=2))
        self.mock_notAfter.start()
        # Ensure the mock.patch is stopped even if test raises an exception
        self.addCleanup(self.mock_notAfter.stop)

    @mock.patch('certbot._internal.ocsp._determine_ocsp_server')
    @mock.patch('certbot._internal.ocsp._check_ocsp_cryptography')
    def test_ensure_cryptography_toggled(self, mock_check, mock_determine):
        mock_determine.return_value = ('http://example.com', 'example.com')
        self.checker.ocsp_revoked(self.cert_obj)

        mock_check.assert_called_once_with(self.cert_path, self.chain_path, 'http://example.com', 10)

    def test_revoke(self):
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED, ocsp_lib.OCSPResponseStatus.SUCCESSFUL):
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked

    def test_responder_is_issuer(self):
        issuer = x509.load_pem_x509_certificate(
            test_util.load_vector('ocsp_issuer_certificate.pem'), default_backend())

        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED,
                        ocsp_lib.OCSPResponseStatus.SUCCESSFUL) as mocks:
            # OCSP response with ResponseID as Name
            mocks['mock_response'].return_value.responder_name = issuer.subject
            mocks['mock_response'].return_value.responder_key_hash = None
            self.checker.ocsp_revoked(self.cert_obj)
            # OCSP response with ResponseID as KeyHash
            key_hash = x509.SubjectKeyIdentifier.from_public_key(issuer.public_key()).digest
            mocks['mock_response'].return_value.responder_name = None
            mocks['mock_response'].return_value.responder_key_hash = key_hash
            self.checker.ocsp_revoked(self.cert_obj)

        # Here responder and issuer are the same. So only the signature of the OCSP
        # response is checked (using the issuer/responder public key).
        assert mocks['mock_check'].call_count == 2
        assert mocks['mock_check'].call_args_list[0][0][0].public_numbers() == \
            issuer.public_key().public_numbers()
        assert mocks['mock_check'].call_args_list[1][0][0].public_numbers() == \
            issuer.public_key().public_numbers()

    def test_responder_is_authorized_delegate(self):
        issuer = x509.load_pem_x509_certificate(
            test_util.load_vector('ocsp_issuer_certificate.pem'), default_backend())
        responder = x509.load_pem_x509_certificate(
            test_util.load_vector('ocsp_responder_certificate.pem'), default_backend())

        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED,
                        ocsp_lib.OCSPResponseStatus.SUCCESSFUL) as mocks:
            # OCSP response with ResponseID as Name
            mocks['mock_response'].return_value.responder_name = responder.subject
            mocks['mock_response'].return_value.responder_key_hash = None
            self.checker.ocsp_revoked(self.cert_obj)
            # OCSP response with ResponseID as KeyHash
            key_hash = x509.SubjectKeyIdentifier.from_public_key(responder.public_key()).digest
            mocks['mock_response'].return_value.responder_name = None
            mocks['mock_response'].return_value.responder_key_hash = key_hash
            self.checker.ocsp_revoked(self.cert_obj)

        # Here responder and issuer are not the same. Two signatures will be checked then,
        # first to verify the responder cert (using the issuer public key), second to
        # to verify the OCSP response itself (using the responder public key).
        assert mocks['mock_check'].call_count == 4
        assert mocks['mock_check'].call_args_list[0][0][0].public_numbers() == \
                         issuer.public_key().public_numbers()
        assert mocks['mock_check'].call_args_list[1][0][0].public_numbers() == \
                         responder.public_key().public_numbers()
        assert mocks['mock_check'].call_args_list[2][0][0].public_numbers() == \
                         issuer.public_key().public_numbers()
        assert mocks['mock_check'].call_args_list[3][0][0].public_numbers() == \
                         responder.public_key().public_numbers()

    def test_revoke_resiliency(self):
        # Server return an invalid HTTP response
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.UNKNOWN, ocsp_lib.OCSPResponseStatus.SUCCESSFUL,
                        http_status_code=400):
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # OCSP response in invalid
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.UNKNOWN, ocsp_lib.OCSPResponseStatus.UNAUTHORIZED):
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # OCSP response is valid, but certificate status is unknown
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.UNKNOWN, ocsp_lib.OCSPResponseStatus.SUCCESSFUL):
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # The OCSP response says that the certificate is revoked, but certificate
        # does not contain the OCSP extension.
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED, ocsp_lib.OCSPResponseStatus.SUCCESSFUL):
            with mock.patch('cryptography.x509.Extensions.get_extension_for_class',
                            side_effect=x509.ExtensionNotFound(
                                'Not found', x509.AuthorityInformationAccessOID.OCSP)):
                revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # OCSP response uses an unsupported signature.
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED, ocsp_lib.OCSPResponseStatus.SUCCESSFUL,
                        check_signature_side_effect=UnsupportedAlgorithm('foo')):
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # OSCP signature response is invalid.
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED, ocsp_lib.OCSPResponseStatus.SUCCESSFUL,
                        check_signature_side_effect=InvalidSignature('foo')):
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # Assertion error on OCSP response validity
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED, ocsp_lib.OCSPResponseStatus.SUCCESSFUL,
                        check_signature_side_effect=AssertionError('foo')):
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # No responder cert in OCSP response
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED,
                        ocsp_lib.OCSPResponseStatus.SUCCESSFUL) as mocks:
            mocks['mock_response'].return_value.certificates = []
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        # Responder cert is not signed by certificate issuer
        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED,
                        ocsp_lib.OCSPResponseStatus.SUCCESSFUL) as mocks:
            cert = mocks['mock_response'].return_value.certificates[0]
            mocks['mock_response'].return_value.certificates[0] = mock.Mock(
                issuer='fake', subject=cert.subject)
            revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False

        with _ocsp_mock(ocsp_lib.OCSPCertStatus.REVOKED, ocsp_lib.OCSPResponseStatus.SUCCESSFUL):
            # This mock is necessary to avoid the first call contained in _determine_ocsp_server
            # of the method cryptography.x509.Extensions.get_extension_for_class.
            with mock.patch('certbot._internal.ocsp._determine_ocsp_server') as mock_server:
                mock_server.return_value = ('https://example.com', 'example.com')
                with mock.patch('cryptography.x509.Extensions.get_extension_for_class',
                                side_effect=x509.ExtensionNotFound(
                                    'Not found', x509.AuthorityInformationAccessOID.OCSP)):
                    revoked = self.checker.ocsp_revoked(self.cert_obj)
        assert revoked is False


class TestDeprecation:
    """Tests related to the deprecation of certbot.ocsp.

    These tests can be deleted after this module is removed from Certbot.

    """
    def test_deprecation_warning(self):
        with pytest.warns(DeprecationWarning, match='certbot.ocsp is deprecated'):
            import certbot.ocsp  # noqa: F401

    def test_no_changes(self):
        from certbot._internal import ocsp
        expected_hash = '4f595b3c6e63749af1f71b5b4890b94e04734bb75f8bff95cf7d7a7e4752d5c1'
        failure_message = ('Despite being prefixed by _internal, certbot._internal.ocsp is still '
            'part of our public API while certbot.ocsp exists. You are free to make changes to '
            'this file and update the hash in this test however, please be sure your changes do '
            'not affect the API of the certbot.ocsp module.')
        assert crypto_util.sha256sum(ocsp.__file__) == expected_hash, failure_message


@contextlib.contextmanager
def _ocsp_mock(certificate_status, response_status,
               http_status_code=200, check_signature_side_effect=None):
    with mock.patch('certbot._internal.ocsp.ocsp.load_der_ocsp_response') as mock_response:
        mock_response.return_value = _construct_mock_ocsp_response(
            certificate_status, response_status)
        with mock.patch('certbot._internal.ocsp.requests.post') as mock_post:
            mock_post.return_value = mock.Mock(status_code=http_status_code)
            with mock.patch('certbot._internal.ocsp.crypto_util.verify_signed_payload') \
                as mock_check:
                if check_signature_side_effect:
                    mock_check.side_effect = check_signature_side_effect
                yield {
                    'mock_response': mock_response,
                    'mock_post': mock_post,
                    'mock_check': mock_check,
                }


def _construct_mock_ocsp_response(certificate_status, response_status):
    cert = x509.load_pem_x509_certificate(
        test_util.load_vector('ocsp_certificate.pem'), default_backend())
    issuer = x509.load_pem_x509_certificate(
        test_util.load_vector('ocsp_issuer_certificate.pem'), default_backend())
    responder = x509.load_pem_x509_certificate(
        test_util.load_vector('ocsp_responder_certificate.pem'), default_backend())
    builder = ocsp_lib.OCSPRequestBuilder()
    builder = builder.add_certificate(cert, issuer, hashes.SHA1())
    request = builder.build()

    return mock.Mock(
        response_status=response_status,
        certificate_status=certificate_status,
        serial_number=request.serial_number,
        issuer_key_hash=request.issuer_key_hash,
        issuer_name_hash=request.issuer_name_hash,
        responder_name=responder.subject,
        certificates=[responder],
        hash_algorithm=hashes.SHA1(),
        next_update_utc=datetime.now(timezone.utc) + timedelta(days=1),
        this_update_utc=datetime.now(timezone.utc) - timedelta(days=1),
        signature_algorithm_oid=x509.oid.SignatureAlgorithmOID.RSA_WITH_SHA1,
    )


if __name__ == '__main__':
    sys.exit(pytest.main(sys.argv[1:] + [__file__]))  # pragma: no cover