except ImportError: # pragma: no cover
    from unittest import mock # type: ignore

from certbot_apache._internal import configurator
from certbot_apache._internal import entrypoint


class EntryPointTest(unittest.TestCase):
    """Entrypoint tests"""

            

Reported by Pylint.

                  from unittest import mock # type: ignore

from certbot_apache._internal import configurator
from certbot_apache._internal import entrypoint


class EntryPointTest(unittest.TestCase):
    """Entrypoint tests"""


            

Reported by Pylint.

              
    _multiprocess_can_split_ = True

    def test_get_configurator(self):

        with mock.patch("certbot.util.get_os_info") as mock_info:
            for distro in entrypoint.OVERRIDE_CLASSES:
                return_value = (distro, "whatever")
                if distro == 'fedora_old':

            

Reported by Pylint.

                              self.assertEqual(entrypoint.get_configurator(),
                                 entrypoint.OVERRIDE_CLASSES[distro])

    def test_nonexistent_like(self):
        with mock.patch("certbot.util.get_os_info") as mock_info:
            mock_info.return_value = ("nonexistent", "irrelevant")
            with mock.patch("certbot.util.get_systemd_os_like") as mock_like:
                for like in entrypoint.OVERRIDE_CLASSES:
                    mock_like.return_value = [like]

            

Reported by Pylint.

                                  self.assertEqual(entrypoint.get_configurator(),
                                     entrypoint.OVERRIDE_CLASSES[like])

    def test_nonexistent_generic(self):
        with mock.patch("certbot.util.get_os_info") as mock_info:
            mock_info.return_value = ("nonexistent", "irrelevant")
            with mock.patch("certbot.util.get_systemd_os_like") as mock_like:
                mock_like.return_value = ["unknonwn"]
                self.assertEqual(entrypoint.get_configurator(),

            

Reported by Pylint.

              from __future__ import absolute_import

import errno
import os  # pylint: disable=os-module-forbidden
import stat
import sys
from typing import List

try:

            

Reported by Pylint.

                          # Handle native windows errors into python errors to be consistent with the API
            # of os.open in the situation of a file already existing or locked.
            if err.winerror == winerror.ERROR_FILE_EXISTS:
                raise OSError(errno.EEXIST, err.strerror)
            if err.winerror == winerror.ERROR_SHARING_VIOLATION:
                raise OSError(errno.EACCES, err.strerror)
            raise err
        finally:
            if handle:

            

Reported by Pylint.

                          if err.winerror == winerror.ERROR_FILE_EXISTS:
                raise OSError(errno.EEXIST, err.strerror)
            if err.winerror == winerror.ERROR_SHARING_VIOLATION:
                raise OSError(errno.EACCES, err.strerror)
            raise err
        finally:
            if handle:
                handle.Close()


            

Reported by Pylint.

                      # Handle native windows error into python error to be consistent with the API
        # of os.mkdir in the situation of a directory already existing.
        if err.winerror == winerror.ERROR_ALREADY_EXISTS:
            raise OSError(errno.EEXIST, err.strerror, file_path, err.winerror)
        raise err

    return None



            

Reported by Pylint.

              # it is a decent choice to not have write permissions for group owner and everybody by default.
# We use a class here to avoid needing to define a global variable, and the potential mistakes
# that could happen with this kind of pattern.
class _WindowsUmask:
    """Store the current umask to apply on Windows"""
    def __init__(self):
        self.mask = 0o022



            

Reported by Pylint.

                  logger.info("Running %s command: %s", cmd_name, shell_cmd)

    if POSIX_MODE:
        proc = subprocess.run(shell_cmd, shell=True, stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE, universal_newlines=True,
                              check=False, env=env)
    else:
        line = ['powershell.exe', '-Command', shell_cmd]
        proc = subprocess.run(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

            

Reported by Bandit.

              
import logging
import select
import subprocess
import sys
import warnings
from typing import Optional
from typing import Tuple


            

Reported by Bandit.

                      return

    # https://docs.microsoft.com/en-us/windows/console/setconsolemode
    ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004

    # stdout/stderr will be the same console screen buffer, but this could return None or raise
    try:
        h = GetStdHandle(STD_OUTPUT_HANDLE)
        if h:

            

Reported by Pylint.

              
    # stdout/stderr will be the same console screen buffer, but this could return None or raise
    try:
        h = GetStdHandle(STD_OUTPUT_HANDLE)
        if h:
            h.SetConsoleMode(h.GetConsoleMode() | ENABLE_VIRTUAL_TERMINAL_PROCESSING)
    except pywinerror:
        logger.debug("Failed to set console mode", exc_info=True)


            

Reported by Pylint.

                                            check=False, env=env)
    else:
        line = ['powershell.exe', '-Command', shell_cmd]
        proc = subprocess.run(line, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                              universal_newlines=True, check=False, env=env)

    # universal_newlines causes stdout and stderr to be str objects instead of
    # bytes in Python 3
    out, err = proc.stdout, proc.stderr

            

Reported by Bandit.

              # First round of wrapping: we import statically all public attributes exposed by the os module
# This allows in particular to have pylint, mypy, IDEs be aware that most of os members are
# available in certbot.compat.os.
from os import *  # type: ignore  # pylint: disable=wildcard-import,unused-wildcard-import,redefined-builtin,os-module-forbidden

# Second round of wrapping: we import dynamically all attributes from the os module that have not
# yet been imported by the first round (static import). This covers in particular the case of
# specific python 3.x versions where not all public attributes are in the special __all__ of os,
# and so not in `from os import *`.

            

Reported by Pylint.

              # yet been imported by the first round (static import). This covers in particular the case of
# specific python 3.x versions where not all public attributes are in the special __all__ of os,
# and so not in `from os import *`.
import os as std_os  # pylint: disable=os-module-forbidden
import sys as std_sys

ourselves = std_sys.modules[__name__]
# Adding all of stdlib os to this module confuses Sphinx so we skip this when
# building the documentation.

            

Reported by Pylint.

              ourselves = std_sys.modules[__name__]
# Adding all of stdlib os to this module confuses Sphinx so we skip this when
# building the documentation.
if not std_os.environ.get("CERTBOT_DOCS") == "1":
    for attribute in dir(std_os):
        # Check if the attribute does not already exist in our module. It could
        # be internal attributes of the module (__name__, __doc__), or
        # attributes from standard os already imported with `from os import *`.
        if not hasattr(ourselves, attribute):

            

Reported by Pylint.

              # First round of wrapping: we import statically all public attributes exposed by the os module
# This allows in particular to have pylint, mypy, IDEs be aware that most of os members are
# available in certbot.compat.os.
from os import *  # type: ignore  # pylint: disable=wildcard-import,unused-wildcard-import,redefined-builtin,os-module-forbidden

# Second round of wrapping: we import dynamically all attributes from the os module that have not
# yet been imported by the first round (static import). This covers in particular the case of
# specific python 3.x versions where not all public attributes are in the special __all__ of os,
# and so not in `from os import *`.

            

Reported by Pylint.

              # yet been imported by the first round (static import). This covers in particular the case of
# specific python 3.x versions where not all public attributes are in the special __all__ of os,
# and so not in `from os import *`.
import os as std_os  # pylint: disable=os-module-forbidden
import sys as std_sys

ourselves = std_sys.modules[__name__]
# Adding all of stdlib os to this module confuses Sphinx so we skip this when
# building the documentation.

            

Reported by Pylint.

                      except errors.MissingCommandlineFlag:
            msg = ("You should register before running non-interactively, "
                   "or provide --agree-tos and --email <email_address> flags.")
            raise errors.MissingCommandlineFlag(msg)

        if code != display_util.OK:
            if optional:
                raise errors.Error(
                    "An e-mail address or "

            

Reported by Pylint.

                                       default,
                         message,
                         exc_info=True)
            raise AssertionError('Invalid default "{0}"'.format(default))

    while True:
        code, raw = method(message, default=default, **kwargs)
        if code == display_util.OK:
            try:

            

Reported by Pylint.

                  return valid_domains


def _sort_names(FQDNs):
    """Sort FQDNs by SLD (and if many, by their subdomains)

    :param list FQDNs: list of domain names

    :returns: Sorted list of domain names

            

Reported by Pylint.

                      for i, domain in enumerate(domain_list):
            try:
                domain_list[i] = util.enforce_domain_sanity(domain)
            except errors.ConfigurationError as e:
                invalid_domains[domain] = str(e)

        if invalid_domains:
            retry_message = (
                "One or more of the entered domain names was not valid:"

            

Reported by Pylint.

                  :param list domains: Each domain is a 'str'

    """
    if len(domains) == 1:
        return "https://{0}".format(domains[0])
    elif len(domains) == 2:
        return "https://{dom[0]} and https://{dom[1]}".format(dom=domains)
    elif len(domains) > 2:
        return "{0}{1}{2}".format(

            

Reported by Pylint.

              import subprocess
import sys

from certbot_integration_tests.utils import acme_server

SCRIPT_DIRNAME = os.path.dirname(__file__)


def main(args=None):

            

Reported by Pylint.

              #!/usr/bin/env python
"""
This executable script wraps the apache-conf-test bash script, in order to setup a pebble instance
before its execution. Directory URL is passed through the SERVER environment variable.
"""
import os
import subprocess
import sys


            

Reported by Pylint.

              before its execution. Directory URL is passed through the SERVER environment variable.
"""
import os
import subprocess
import sys

from certbot_integration_tests.utils import acme_server

SCRIPT_DIRNAME = os.path.dirname(__file__)

            

Reported by Bandit.

              SCRIPT_DIRNAME = os.path.dirname(__file__)


def main(args=None):
    if not args:
        args = sys.argv[1:]
    with acme_server.ACMEServer('pebble', [], False) as acme_xdist:
        environ = os.environ.copy()
        environ['SERVER'] = acme_xdist['directory_url']

            

Reported by Pylint.

                      environ['SERVER'] = acme_xdist['directory_url']
        command = [os.path.join(SCRIPT_DIRNAME, 'apache-conf-test')]
        command.extend(args)
        return subprocess.call(command, env=environ)


if __name__ == '__main__':
    sys.exit(main())

            

Reported by Bandit.

              class _ProxyHandler(BaseHTTPServer.BaseHTTPRequestHandler):
    # pylint: disable=missing-function-docstring
    def do_POST(self):
        request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediate-keys/0', verify=False)
        issuer_key = serialization.load_pem_private_key(request.content, None, default_backend())

        request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediates/0', verify=False)
        issuer_cert = x509.load_pem_x509_certificate(request.content, default_backend())


            

Reported by Bandit.

                      request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediate-keys/0', verify=False)
        issuer_key = serialization.load_pem_private_key(request.content, None, default_backend())

        request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediates/0', verify=False)
        issuer_cert = x509.load_pem_x509_certificate(request.content, default_backend())

        content_len = int(self.headers.get('Content-Length'))

        ocsp_request = ocsp.load_der_ocsp_request(self.rfile.read(content_len))

            

Reported by Bandit.

                      ocsp_request = ocsp.load_der_ocsp_request(self.rfile.read(content_len))
        response = requests.get('{0}/cert-status-by-serial/{1}'.format(
            PEBBLE_MANAGEMENT_URL, str(hex(ocsp_request.serial_number)).replace('0x', '')),
            verify=False
        )

        if not response.ok:
            ocsp_response = ocsp.OCSPResponseBuilder.build_unsuccessful(
                ocsp.OCSPResponseStatus.UNAUTHORIZED

            

Reported by Bandit.

                              revocation_time = parser.parse(revoked_at)

            ocsp_response = ocsp.OCSPResponseBuilder().add_response(
                cert=cert, issuer=issuer_cert, algorithm=hashes.SHA1(),
                cert_status=ocsp_status,
                this_update=now, next_update=now + datetime.timedelta(hours=1),
                revocation_time=revocation_time, revocation_reason=revocation_reason
            ).responder_id(
                ocsp.OCSPResponderEncoding.NAME, issuer_cert

            

Reported by Bandit.

              
class _ProxyHandler(BaseHTTPServer.BaseHTTPRequestHandler):
    # pylint: disable=missing-function-docstring
    def do_POST(self):
        request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediate-keys/0', verify=False)
        issuer_key = serialization.load_pem_private_key(request.content, None, default_backend())

        request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediates/0', verify=False)
        issuer_cert = x509.load_pem_x509_certificate(request.content, default_backend())

            

Reported by Pylint.

                              "but SSLSessionTickets is not set.")

if __name__ == '__main__':
    main(*sys.argv[1:])

            

Reported by Pylint.

                              "but SSLSessionTickets is not set.")

if __name__ == '__main__':
    main(*sys.argv[1:])

            

Reported by Pylint.

              #!/usr/bin/env python
# Test script for OpenSSL version checking
from distutils.version import LooseVersion
import sys


def main(openssl_version, apache_version):
    if not openssl_version.strip():
        raise Exception("No OpenSSL version found.")

            

Reported by Pylint.

              import sys


def main(openssl_version, apache_version):
    if not openssl_version.strip():
        raise Exception("No OpenSSL version found.")
    if not apache_version.strip():
        raise Exception("No Apache version found.")
    conf_file_location = "/etc/letsencrypt/options-ssl-apache.conf"

            

Reported by Pylint.

                  if not apache_version.strip():
        raise Exception("No Apache version found.")
    conf_file_location = "/etc/letsencrypt/options-ssl-apache.conf"
    with open(conf_file_location) as f:
        contents = f.read()
    if LooseVersion(apache_version.strip()) < LooseVersion('2.4.11') or \
        LooseVersion(openssl_version.strip()) < LooseVersion('1.0.2l'):
        # should be old version
        # assert SSLSessionTickets not in conf file

            

Reported by Pylint.

              
import zope.interface

from acme.challenges import Challenge
from acme.challenges import ChallengeResponse
from certbot.achallenges import AnnotatedChallenge
from certbot import configuration



            

Reported by Pylint.

              import zope.interface

from acme.challenges import Challenge
from acme.challenges import ChallengeResponse
from certbot.achallenges import AnnotatedChallenge
from certbot import configuration


class AccountStorage(metaclass=ABCMeta):

            

Reported by Pylint.

                      """

    @abstractmethod
    def deploy_cert(self, domain: str, cert_path: str, key_path: str,
                    chain_path: str, fullchain_path: str) -> None:
        """Deploy certificate.

        :param str domain: domain to deploy certificate file
        :param str cert_path: absolute path to the certificate file

            

Reported by Pylint.

              # :func:`IInstaller.deploy_cert` is called.


class GenericUpdater(metaclass=ABCMeta):
    """Interface for update types not currently specified by Certbot.

    This class allows plugins to perform types of updates that Certbot hasn't
    defined (yet).


            

Reported by Pylint.

                      """


class RenewDeployer(metaclass=ABCMeta):
    """Interface for update types run when a lineage is renewed

    This class allows plugins to perform types of updates that need to run at
    lineage renewal that Certbot hasn't defined (yet).


            

Reported by Pylint.

              from typing import List
from typing import Set

from acme import challenges
from certbot import errors
from certbot import interfaces
from certbot._internal import cli
from certbot.achallenges import KeyAuthorizationAnnotatedChallenge as AnnotatedChallenge
from certbot.compat import filesystem

            

Reported by Pylint.

                                  "webroot when using the webroot plugin.")
            return None if index == 0 else known_webroots[index - 1]  # code == display_util.OK

    def _prompt_for_new_webroot(self, domain, allowraise=False):
        code, webroot = ops.validated_directory(
            _validate_webroot,
            "Input the webroot for {0}:".format(domain),
            force_interactive=True)
        if code == display_util.CANCEL:

            

Reported by Pylint.

                          finally:
                filesystem.umask(old_umask)

    def _get_validation_path(self, root_path, achall):
        return os.path.join(root_path, achall.chall.encode("token"))

    def _perform_single(self, achall):
        response, validation = achall.response_and_validation()


            

Reported by Pylint.

                      logger.debug("All challenges cleaned up")


class _WebrootMapAction(argparse.Action):
    """Action class for parsing webroot_map."""

    def __call__(self, parser, namespace, webroot_map, option_string=None):
        for domains, webroot_path in json.loads(webroot_map).items():
            webroot_path = _validate_webroot(webroot_path)

            

Reported by Pylint.

                              (d, webroot_path) for d in cli.add_domains(namespace, domains))


class _WebrootPathAction(argparse.Action):
    """Action class for parsing webroot_path."""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._domain_before_webroot = False

            

Reported by Pylint.