<commit_before>"""Tests for letsencrypt.client."""
import os
import shutil
import tempfile
import unittest

import OpenSSL
import mock

from acme import jose

from letsencrypt import account
from letsencrypt import errors
from letsencrypt import le_util

from letsencrypt.tests import test_util


KEY = test_util.load_vector("rsa512_key.pem")
CSR_SAN = test_util.load_vector("csr-san.der")


class ConfigHelper(object):
    """Creates a dummy object to imitate a namespace object

        Example: cfg = ConfigHelper(redirect=True, hsts=False, uir=False)
        will result in: cfg.redirect=True, cfg.hsts=False, etc.
    """
    def __init__(self, **kwds):
        self.__dict__.update(kwds)

class RegisterTest(unittest.TestCase):
    """Tests for letsencrypt.client.register."""

    def setUp(self):
        self.config = mock.MagicMock(rsa_key_size=1024, register_unsafely_without_email=False)
        self.account_storage = account.AccountMemoryStorage()
        self.tos_cb = mock.MagicMock()

    def _call(self):
        from letsencrypt.client import register
        return register(self.config, self.account_storage, self.tos_cb)

    def test_no_tos(self):
        with mock.patch("letsencrypt.client.acme_client.Client") as mock_client:
            mock_client.register().terms_of_service = "http://tos"
            with mock.patch("letsencrypt.account.report_new_account"):
                self.tos_cb.return_value = False
                self.assertRaises(errors.Error, self._call)

                self.tos_cb.return_value = True
                self._call()

                self.tos_cb = None
                self._call()

    def test_it(self):
        with mock.patch("letsencrypt.client.acme_client.Client"):
            with mock.patch("letsencrypt.account.report_new_account"):
                self._call()

    @mock.patch("letsencrypt.account.report_new_account")
    @mock.patch("letsencrypt.client.display_ops.get_email")
    def test_email_retry(self, _rep, mock_get_email):
        from acme import messages
        msg = "Validation of contact mailto:<EMAIL> failed"
        mx_err = messages.Error(detail=msg, typ="malformed", title="title")
        with mock.patch("letsencrypt.client.acme_client.Client") as mock_client:
            mock_client().register.side_effect = [mx_err, mock.MagicMock()]
            self._call()
            self.assertEqual(mock_get_email.call_count, 1)

    def test_needs_email(self):
        self.config.email = None
        self.assertRaises(errors.Error, self._call)

    @mock.patch("letsencrypt.client.logger")
    def test_without_email(self, mock_logger):
        with mock.patch("letsencrypt.client.acme_client.Client"):
            with mock.patch("letsencrypt.account.report_new_account"):
                self.config.email = None
                self.config.register_unsafely_without_email = True
                self._call()
                mock_logger.warn.assert_called_once_with(mock.ANY)

    def test_unsupported_error(self):
        from acme import messages
        msg = "Test"
        mx_err = messages.Error(detail=msg, typ="malformed", title="title")
        with mock.patch("letsencrypt.client.acme_client.Client") as mock_client:
            mock_client().register.side_effect = [mx_err, mock.MagicMock()]
            self.assertRaises(messages.Error, self._call)

class ClientTest(unittest.TestCase):
    """Tests for letsencrypt.client.Client."""

    def setUp(self):
        self.config = mock.MagicMock(
            no_verify_ssl=False, config_dir="/etc/letsencrypt")
        # pylint: disable=star-args
        self.account = mock.MagicMock(**{"key.pem": KEY})
        self.eg_domains = ["example.com", "www.example.com"]

        from letsencrypt.client import Client
        with mock.patch("letsencrypt.client.acme_client.Client") as acme:
            self.acme_client = acme
            self.acme = acme.return_value = mock.MagicMock()
            self.client = Client(
                config=self.config, account_=self.account,
                dv_auth=None, installer=None)

    def test_init_acme_verify_ssl(self):
        net = self.acme_client.call_args[1]["net"]
        self.assertTrue(net.verify_ssl)

    def _mock_obtain_certificate(self):
        self.client.auth_handler = mock.MagicMock()
        self.acme.request_issuance.return_value = mock.sentinel.certr
        self.acme.fetch_chain.return_value = mock.sentinel.chain

    def _check_obtain_certificate(self):
        self.client.auth_handler.get_authorizations.assert_called_once_with(self.eg_domains)
        self.acme.request_issuance.assert_called_once_with(
            jose.ComparableX509(OpenSSL.crypto.load_certificate_request(
                OpenSSL.crypto.FILETYPE_ASN1, CSR_SAN)),
            self.client.auth_handler.get_authorizations())
        self.acme.fetch_chain.assert_called_once_with(mock.sentinel.certr)

    # FIXME move parts of this to test_cli.py...
    @mock.patch("letsencrypt.cli._process_domain")
    def test_obtain_certificate_from_csr(self, mock_process_domain):
        self._mock_obtain_certificate()
        from letsencrypt import cli
        test_csr = le_util.CSR(form="der", file=None, data=CSR_SAN)
        mock_parsed_args = mock.MagicMock()
        with mock.patch("letsencrypt.client.le_util.CSR") as mock_CSR:
            mock_CSR.return_value = test_csr
            mock_parsed_args.domains = self.eg_domains[:]
            mock_parser = mock.MagicMock(cli.HelpfulArgumentParser)
            cli.HelpfulArgumentParser.handle_csr(mock_parser, mock_parsed_args)

            # make sure cli processing occurred
            cli_processed = (call[0][1] for call in mock_process_domain.call_args_list)
            self.assertEqual(set(cli_processed), set(("example.com", "www.example.com")))
            # Now provoke an inconsistent domains error...
            mock_parsed_args.domains.append("hippopotamus.io")
            self.assertRaises(errors.ConfigurationError,
                cli.HelpfulArgumentParser.handle_csr, mock_parser, mock_parsed_args)

            self.assertEqual(
                (mock.sentinel.certr, mock.sentinel.chain),
                self.client.obtain_certificate_from_csr(self.eg_domains, test_csr))
            # and that the cert was obtained correctly
            self._check_obtain_certificate()


    @mock.patch("letsencrypt.client.crypto_util")
    def test_obtain_certificate(self, mock_crypto_util):
        self._mock_obtain_certificate()

        csr = le_util.CSR(form="der", file=None, data=CSR_SAN)
        mock_crypto_util.init_save_csr.return_value = csr
        mock_crypto_util.init_save_key.return_value = mock.sentinel.key
        domains = ["example.com", "www.example.com"]

        self.assertEqual(
            self.client.obtain_certificate(domains),
            (mock.sentinel.certr, mock.sentinel.chain, mock.sentinel.key, csr))

        mock_crypto_util.init_save_key.assert_called_once_with(
            self.config.rsa_key_size, self.config.key_dir)
        mock_crypto_util.init_save_csr.assert_called_once_with(
            mock.sentinel.key, domains, self.config.csr_dir)
        self._check_obtain_certificate()

    def test_save_certificate(self):
        certs = ["matching_cert.pem", "cert.pem", "cert-san.pem"]
        tmp_path = tempfile.mkdtemp()<commit_msg>Adding test for obtain_certificate_from_csr with auth_handler set to None<commit_after>"""Tests for letsencrypt.client."""
import os
import shutil
import tempfile
import unittest

import OpenSSL
import mock

from acme import jose

from letsencrypt import account
from letsencrypt import errors
from letsencrypt import le_util

from letsencrypt.tests import test_util


KEY = test_util.load_vector("rsa512_key.pem")
CSR_SAN = test_util.load_vector("csr-san.der")


class ConfigHelper(object):
    """Creates a dummy object to imitate a namespace object

        Example: cfg = ConfigHelper(redirect=True, hsts=False, uir=False)
        will result in: cfg.redirect=True, cfg.hsts=False, etc.
    """
    def __init__(self, **kwds):
        self.__dict__.update(kwds)

class RegisterTest(unittest.TestCase):
    """Tests for letsencrypt.client.register."""

    def setUp(self):
        self.config = mock.MagicMock(rsa_key_size=1024, register_unsafely_without_email=False)
        self.account_storage = account.AccountMemoryStorage()
        self.tos_cb = mock.MagicMock()

    def _call(self):
        from letsencrypt.client import register
        return register(self.config, self.account_storage, self.tos_cb)

    def test_no_tos(self):
        with mock.patch("letsencrypt.client.acme_client.Client") as mock_client:
            mock_client.register().terms_of_service = "http://tos"
            with mock.patch("letsencrypt.account.report_new_account"):
                self.tos_cb.return_value = False
                self.assertRaises(errors.Error, self._call)

                self.tos_cb.return_value = True
                self._call()

                self.tos_cb = None
                self._call()

    def test_it(self):
        with mock.patch("letsencrypt.client.acme_client.Client"):
            with mock.patch("letsencrypt.account.report_new_account"):
                self._call()

    @mock.patch("letsencrypt.account.report_new_account")
    @mock.patch("letsencrypt.client.display_ops.get_email")
    def test_email_retry(self, _rep, mock_get_email):
        from acme import messages
        msg = "Validation of contact mailto:<EMAIL> failed"
        mx_err = messages.Error(detail=msg, typ="malformed", title="title")
        with mock.patch("letsencrypt.client.acme_client.Client") as mock_client:
            mock_client().register.side_effect = [mx_err, mock.MagicMock()]
            self._call()
            self.assertEqual(mock_get_email.call_count, 1)

    def test_needs_email(self):
        self.config.email = None
        self.assertRaises(errors.Error, self._call)

    @mock.patch("letsencrypt.client.logger")
    def test_without_email(self, mock_logger):
        with mock.patch("letsencrypt.client.acme_client.Client"):
            with mock.patch("letsencrypt.account.report_new_account"):
                self.config.email = None
                self.config.register_unsafely_without_email = True
                self._call()
                mock_logger.warn.assert_called_once_with(mock.ANY)

    def test_unsupported_error(self):
        from acme import messages
        msg = "Test"
        mx_err = messages.Error(detail=msg, typ="malformed", title="title")
        with mock.patch("letsencrypt.client.acme_client.Client") as mock_client:
            mock_client().register.side_effect = [mx_err, mock.MagicMock()]
            self.assertRaises(messages.Error, self._call)

class ClientTest(unittest.TestCase):
    """Tests for letsencrypt.client.Client."""

    def setUp(self):
        self.config = mock.MagicMock(
            no_verify_ssl=False, config_dir="/etc/letsencrypt")
        # pylint: disable=star-args
        self.account = mock.MagicMock(**{"key.pem": KEY})
        self.eg_domains = ["example.com", "www.example.com"]

        from letsencrypt.client import Client
        with mock.patch("letsencrypt.client.acme_client.Client") as acme:
            self.acme_client = acme
            self.acme = acme.return_value = mock.MagicMock()
            self.client = Client(
                config=self.config, account_=self.account,
                dv_auth=None, installer=None)

    def test_init_acme_verify_ssl(self):
        net = self.acme_client.call_args[1]["net"]
        self.assertTrue(net.verify_ssl)

    def _mock_obtain_certificate(self):
        self.client.auth_handler = mock.MagicMock()
        self.acme.request_issuance.return_value = mock.sentinel.certr
        self.acme.fetch_chain.return_value = mock.sentinel.chain

    def _check_obtain_certificate(self):
        self.client.auth_handler.get_authorizations.assert_called_once_with(self.eg_domains)
        self.acme.request_issuance.assert_called_once_with(
            jose.ComparableX509(OpenSSL.crypto.load_certificate_request(
                OpenSSL.crypto.FILETYPE_ASN1, CSR_SAN)),
            self.client.auth_handler.get_authorizations())
        self.acme.fetch_chain.assert_called_once_with(mock.sentinel.certr)

    # FIXME move parts of this to test_cli.py...
    @mock.patch("letsencrypt.client.logger")
    @mock.patch("letsencrypt.cli._process_domain")
    def test_obtain_certificate_from_csr(self, mock_process_domain, mock_logger):
        self._mock_obtain_certificate()
        from letsencrypt import cli
        test_csr = le_util.CSR(form="der", file=None, data=CSR_SAN)
        mock_parsed_args = mock.MagicMock()
        with mock.patch("letsencrypt.client.le_util.CSR") as mock_CSR:
            mock_CSR.return_value = test_csr
            mock_parsed_args.domains = self.eg_domains[:]
            mock_parser = mock.MagicMock(cli.HelpfulArgumentParser)
            cli.HelpfulArgumentParser.handle_csr(mock_parser, mock_parsed_args)

            # make sure cli processing occurred
            cli_processed = (call[0][1] for call in mock_process_domain.call_args_list)
            self.assertEqual(set(cli_processed), set(("example.com", "www.example.com")))
            # Now provoke an inconsistent domains error...
            mock_parsed_args.domains.append("hippopotamus.io")
            self.assertRaises(errors.ConfigurationError,
                cli.HelpfulArgumentParser.handle_csr, mock_parser, mock_parsed_args)

            self.assertEqual(
                (mock.sentinel.certr, mock.sentinel.chain),
                self.client.obtain_certificate_from_csr(self.eg_domains, test_csr))
            # and that the cert was obtained correctly
            self._check_obtain_certificate()

            # Test for no auth_handler
            self.client.auth_handler = None
            self.assertRaises(
                errors.Error,
                self.client.obtain_certificate_from_csr,
                self.eg_domains,
                test_csr)
            mock_logger.warning.assert_called_once_with(mock.ANY)

    @mock.patch("letsencrypt.client.crypto_util")
    def test_obtain_certificate(self, mock_crypto_util):
        self._mock_obtain_certificate()

        csr = le_util.CSR(form="der", file=None, data=CSR_SAN)
        mock_crypto_util.init_save_csr.return_value = csr
        mock_crypto_util.init_save_key.return_value = mock.sentinel.key
        domains = ["example.com", "www.example.com"]

        self.assertEqual(
            self.client.obtain_certificate(domains),
            (mock.sentinel.certr, mock.sentinel.chain, mock.sentinel.key, csr))

        mock_crypto_util.init_save_key.assert_called_once_with(
            self.config.rsa_key_size, self.config.key_dir)
        mock_crypto_util.init_save_csr.assert_called_once_with(
            mock.sentinel.key, domains, self.config.csr_dir)
        self._check_obtain_certificate()

    def test_save_certificate(self):
        certs = ["matching_cert.pem", "cert.pem", "cert-san.pem"]
        tmp_path = tempfile.mkdtemp()
{'fraction_non_alphanumeric': 0.10130935061499802, 'fraction_numerical': 0.0019838645681788123, 'mean_word_length': 2.8896604938271606, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3929910', 'n_tokens_mistral': 4772, 'n_tokens_neox': 4619, 'n_words': 735}