import logging
from typing import Type

import josepy as jose

from acme import challenges
from acme.challenges import Challenge

logger = logging.getLogger(__name__)

            

Reported by Pylint.

              
import josepy as jose

from acme import challenges
from acme.challenges import Challenge

logger = logging.getLogger(__name__)



            

Reported by Pylint.

              import josepy as jose

from acme import challenges
from acme.challenges import Challenge

logger = logging.getLogger(__name__)


class AnnotatedChallenge(jose.ImmutableMap):

            

Reported by Pylint.

              logger = logging.getLogger(__name__)


class AnnotatedChallenge(jose.ImmutableMap):
    """Client annotated challenge.

    Wraps around server provided challenge and annotates with data
    useful for the client.


            

Reported by Pylint.

                      return getattr(self.challb, name)


class KeyAuthorizationAnnotatedChallenge(AnnotatedChallenge):
    """Client annotated `KeyAuthorizationChallenge` challenge."""
    __slots__ = ('challb', 'domain', 'account_key')

    def response_and_validation(self, *args, **kwargs):
        """Generate response and validation."""

            

Reported by Pylint.

                          self.account_key, *args, **kwargs)


class DNS(AnnotatedChallenge):
    """Client annotated "dns" ACME challenge."""
    __slots__ = ('challb', 'domain')
    acme_type = challenges.DNS

            

Reported by Pylint.

                              errmsg = "PluginStorage file {0} is corrupted.".format(
                    self._storagepath)
                logger.error(errmsg)
                raise errors.PluginStorageError(errmsg)
        self._data = data

    def save(self):
        """Saves PluginStorage content to disk


            

Reported by Pylint.

                      data: Dict[str, Any] = {}
        filedata = ""
        try:
            with open(self._storagepath, 'r') as fh:
                filedata = fh.read()
        except IOError as e:
            errmsg = "Could not read PluginStorage data file: {0} : {1}".format(
                self._storagepath, str(e))
            if os.path.isfile(self._storagepath):

            

Reported by Pylint.

                      try:
            with open(self._storagepath, 'r') as fh:
                filedata = fh.read()
        except IOError as e:
            errmsg = "Could not read PluginStorage data file: {0} : {1}".format(
                self._storagepath, str(e))
            if os.path.isfile(self._storagepath):
                # Only error out if file exists, but cannot be read
                logger.error(errmsg)

            

Reported by Pylint.

              
        try:
            serialized = json.dumps(self._data)
        except TypeError as e:
            errmsg = "Could not serialize PluginStorage data: {0}".format(
                str(e))
            logger.error(errmsg)
            raise errors.PluginStorageError(errmsg)
        try:

            

Reported by Pylint.

                          with os.fdopen(filesystem.open(
                    self._storagepath,
                    os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
                    0o600), 'w') as fh:
                fh.write(serialized)
        except IOError as e:
            errmsg = "Could not write PluginStorage data to file {0} : {1}".format(
                self._storagepath, str(e))
            logger.error(errmsg)

            

Reported by Pylint.

                                  os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
                    0o600), 'w') as fh:
                fh.write(serialized)
        except IOError as e:
            errmsg = "Could not write PluginStorage data to file {0} : {1}".format(
                self._storagepath, str(e))
            logger.error(errmsg)
            raise errors.PluginStorageError(errmsg)


            

Reported by Pylint.

              
import requests

from acme import crypto_util
from acme import errors as acme_errors

logger = logging.getLogger(__name__)



            

Reported by Pylint.

              import requests

from acme import crypto_util
from acme import errors as acme_errors

logger = logging.getLogger(__name__)


class Validator:

            

Reported by Pylint.

              class Validator:
    """Collection of functions to test a live webserver's configuration"""

    def certificate(self, cert, name, alt_host=None, port=443):
        """Verifies the certificate presented at name is cert"""
        if alt_host is None:
            host = socket.gethostbyname(name).encode()
        elif isinstance(alt_host, bytes):
            host = alt_host

            

Reported by Pylint.

              
        return presented_cert.digest("sha256") == cert.digest("sha256")

    def redirect(self, name, port=80, headers=None):
        """Test whether webserver redirects to secure connection."""
        url = "http://{0}:{1}".format(name, port)
        if headers:
            response = requests.get(url, headers=headers, allow_redirects=False)
        else:

            

Reported by Pylint.

              
        return True

    def any_redirect(self, name, port=80, headers=None):
        """Test whether webserver redirects."""
        url = "http://{0}:{1}".format(name, port)
        if headers:
            response = requests.get(url, headers=headers, allow_redirects=False)
        else:

            

Reported by Pylint.

              
        return response.status_code in range(300, 309)

    def hsts(self, name):
        """Test for HTTP Strict Transport Security header"""
        headers = requests.get("https://" + name).headers
        hsts_header = headers.get("strict-transport-security")

        if not hsts_header:

            

Reported by Pylint.

              
import pyparsing

from certbot import errors
from certbot.compat import os
from certbot_nginx._internal import nginxparser
from certbot_nginx._internal import obj
from certbot_nginx._internal.nginxparser import UnspacedList


            

Reported by Pylint.

              import pyparsing

from certbot import errors
from certbot.compat import os
from certbot_nginx._internal import nginxparser
from certbot_nginx._internal import obj
from certbot_nginx._internal.nginxparser import UnspacedList

logger = logging.getLogger(__name__)

            

Reported by Pylint.

                      self.config_root = self._find_config_root()

        # Parse nginx.conf and included files.
        # TODO: Check sites-available/ as well. For now, the configurator does
        # not enable sites from there.
        self.load()

    def load(self):
        """Loads Nginx files into a parsed tree.

            

Reported by Pylint.

              
        """
        files = glob.glob(filepath) # nginx on unix calls glob(3) for this
                                    # XXX Windows nginx uses FindFirstFile, and
                                    # should have a narrower call here
        trees = []
        for item in files:
            if item in self.parsed and not override:
                continue

            

Reported by Pylint.

                      :returns: A vhost object for the newly created vhost
        :rtype: :class:`~certbot_nginx._internal.obj.VirtualHost`
        """
        # TODO: https://github.com/certbot/certbot/issues/5185
        # put it in the same file as the template, at the same level
        new_vhost = copy.deepcopy(vhost_template)

        enclosing_block = self.parsed[vhost_template.filep]
        for index in vhost_template.path[:-1]:

            

Reported by Pylint.

                      _apply_global_addr_ssl(addr_to_ssl, parsed_server)
        return parsed_server

    def has_ssl_on_directive(self, vhost):
        """Does vhost have ssl on for all ports?

        :param :class:`~certbot_nginx._internal.obj.VirtualHost` vhost: The vhost in question

        :returns: True if 'ssl on' directive is included

            

Reported by Pylint.

              from types import TracebackType
from typing import IO

from acme import messages
from certbot import errors
from certbot import util
from certbot._internal import constants
from certbot.compat import os


            

Reported by Pylint.

                  :rtype: tuple

    """
    # TODO: logs might contain sensitive data such as contents of the
    # private key! #525
    util.set_up_core_dir(config.logs_dir, 0o700, config.strict_permissions)
    log_file_path = os.path.join(config.logs_dir, logfile)
    try:
        handler = logging.handlers.RotatingFileHandler(

            

Reported by Pylint.

                  # rotate on each invocation, rollover only possible when maxBytes
    # is nonzero and backupCount is nonzero, so we set maxBytes as big
    # as possible not to overrun in single CLI invocation (1MB).
    handler.doRollover()  # TODO: creates empty letsencrypt.log.1 file
    handler.setLevel(logging.DEBUG)
    handler_formatter = logging.Formatter(fmt=fmt)
    handler.setFormatter(handler_formatter)
    return handler, log_file_path


            

Reported by Pylint.

                      elif isinstance(handler, MemoryHandler):
            memory_handler = handler
    msg = 'Previously configured logging handlers have been removed!'
    assert memory_handler is not None and stderr_handler is not None, msg

    root_logger.addHandler(file_handler)
    root_logger.removeHandler(memory_handler)
    temp_handler = getattr(memory_handler, 'target', None)
    memory_handler.setTarget(file_handler)  # pylint: disable=no-member

            

Reported by Bandit.

                      memory_handler.flush(force=True)


def post_arg_parse_except_hook(exc_type: type, exc_value: BaseException, trace: TracebackType,
                               debug: bool, quiet: bool, log_path: str):
    """Logs fatal exceptions and reports them to the user.

    If debug is True, the full exception and traceback is shown to the
    user, otherwise, it is suppressed. sys.exit is always called with a

            

Reported by Pylint.

                  # display message the user, otherwise, a lower level like
    # logger.DEBUG should be used
    if debug or not issubclass(exc_type, Exception):
        assert constants.QUIET_LOGGING_LEVEL <= logging.ERROR
        if exc_type is KeyboardInterrupt:
            logger.error('Exiting due to user request.')
            sys.exit(1)
        logger.error('Exiting abnormally:', exc_info=exc_info)
    else:

            

Reported by Bandit.

              """
from typing import Set

from certbot import errors
from certbot.compat import os
from certbot_apache._internal import apache_util
from certbot_apache._internal import assertions
from certbot_apache._internal import interfaces
from certbot_apache._internal import parser

            

Reported by Pylint.

              from typing import Set

from certbot import errors
from certbot.compat import os
from certbot_apache._internal import apache_util
from certbot_apache._internal import assertions
from certbot_apache._internal import interfaces
from certbot_apache._internal import parser
from certbot_apache._internal import parsernode_util as util

            

Reported by Pylint.

                                  )
                )
        except KeyError:
            raise errors.PluginError("Augeas path is required")

    def save(self, msg):
        self.parser.save(msg)

    def find_ancestors(self, name):

            

Reported by Pylint.

                                             filepath=apache_util.get_file_path(path),
                               metadata=metadata)

    def _aug_get_name(self, path):
        """
        Helper function to get name of a configuration block or variable from path.
        """

        # Remove the ending slash if any

            

Reported by Pylint.

                          param_path = "{}/arg[1]".format(self.metadata["augeaspath"])
            self.parser.aug.remove(param_path)
        # Insert new ones
        for pi, param in enumerate(parameters):
            param_path = "{}/arg[{}]".format(self.metadata["augeaspath"], pi+1)
            self.parser.aug.set(param_path, param)

    @property
    def parameters(self):

            

Reported by Pylint.

              
import configobj

from acme import challenges
from certbot import errors
from certbot import interfaces
from certbot.compat import filesystem
from certbot.compat import os
from certbot.display import ops

            

Reported by Pylint.

                          indicate any issue.
        """

        def __validator(filename): # pylint: disable=unused-private-member
            configuration = CredentialsConfiguration(filename, self.dest)

            if required_variables:
                configuration.require(required_variables)


            

Reported by Pylint.

                      :rtype: str
        """

        def __validator(i): # pylint: disable=unused-private-member
            if not i:
                raise errors.PluginError('Please enter your {0}.'.format(label))

        code, response = ops.validated_input(
            __validator,

            

Reported by Pylint.

                      :rtype: str
        """

        def __validator(filename): # pylint: disable=unused-private-member
            if not filename:
                raise errors.PluginError('Please enter a valid path to your {0}.'.format(label))

            filename = os.path.expanduser(filename)


            

Reported by Pylint.

              
        try:
            self.confobj = configobj.ConfigObj(filename)
        except configobj.ConfigObjError as e:
            logger.debug("Error parsing credentials configuration: %s", e, exc_info=True)
            raise errors.PluginError("Error parsing credentials configuration: {0}".format(e))

        self.mapper = mapper


            

Reported by Pylint.

              
class ACMEServerMixin:
    """ACME server common settings mixin."""
    # TODO: c.f. #858
    server_version = "ACME client standalone challenge solver"
    allow_reuse_address = True


class BaseDualNetworkedServers:

            

Reported by Pylint.

                      self.challenge_certs = challenge_certs

    def _cert_selection(self, connection):
        # TODO: We would like to serve challenge cert only if asked for it via
        # ALPN. To do this, we need to retrieve the list of protos from client
        # hello, but this is currently impossible with openssl [0], and ALPN
        # negotiation is done after cert selection.
        # Therefore, currently we always return challenge cert, and terminate
        # handshake in alpn_selection() if ALPN protos are not what we expect.

            

Reported by Pylint.

                      return socketserver.TCPServer.server_bind(self)


class ACMEServerMixin:
    """ACME server common settings mixin."""
    # TODO: c.f. #858
    server_version = "ACME client standalone challenge solver"
    allow_reuse_address = True


            

Reported by Pylint.

                              logger.debug(
                    "Successfully bound to %s:%s using %s", new_address[0],
                    new_address[1], "IPv6" if ip_version else "IPv4")
            except socket.error as e:
                last_socket_err = e
                if self.servers:
                    # Already bound using IPv6.
                    logger.debug(
                        "Certbot wasn't able to bind to %s:%s using %s, this "

            

Reported by Pylint.

                              # bind to the same port for both servers.
                port = server.socket.getsockname()[1]
        if not self.servers:
            if last_socket_err:
                raise last_socket_err
            else: # pragma: no cover
                raise socket.error("Could not bind to IPv4 or IPv6.")

    def serve_forever(self):

            

Reported by Pylint.

              
def call_with_print(command):
    print(command)
    subprocess.check_call(command, shell=True)


def main(args):
    script_dir = os.path.dirname(os.path.abspath(__file__))
    command = [sys.executable, os.path.join(script_dir, 'pip_install_editable.py')]

            

Reported by Bandit.

              #!/usr/bin/env python
# pip installs the requested packages in editable mode and runs unit tests on
# them. Each package is installed and tested in the order they are provided
# before the script moves on to the next package. If CERTBOT_NO_PIN is set not
# set to 1, packages are installed using pinned versions of all of our
# dependencies. See pip_install.py for more information on the versions pinned
# to.
import os
import re

            

Reported by Pylint.

              # to.
import os
import re
import subprocess
import sys

def call_with_print(command):
    print(command)
    subprocess.check_call(command, shell=True)

            

Reported by Bandit.

              import subprocess
import sys

def call_with_print(command):
    print(command)
    subprocess.check_call(command, shell=True)


def main(args):

            

Reported by Pylint.

                  subprocess.check_call(command, shell=True)


def main(args):
    script_dir = os.path.dirname(os.path.abspath(__file__))
    command = [sys.executable, os.path.join(script_dir, 'pip_install_editable.py')]

    for requirement in args:
        current_command = command[:]

            

Reported by Pylint.

                  verified.prepare()
    prepared = verified.available()

    if len(prepared) > 1:
        logger.debug("Multiple candidate plugins: %s", prepared)
        plugin_ep = choose_plugin(list(prepared.values()), question)
        if plugin_ep is None:
            return None
        return plugin_ep.init()

            

Reported by Pylint.

                       config.authenticator, config.installer)


def choose_configurator_plugins(config: configuration.NamespaceConfig,
                                plugins: disco.PluginsRegistry,
                                verb: str) -> Tuple[Optional[interfaces.Installer],
                                                    Optional[interfaces.Authenticator]]:
    """
    Figure out which configurator we're going to use, modifies

            

Reported by Pylint.

                  # Which plugins do we need?
    if verb == "run":
        need_inst = need_auth = True
        from certbot._internal.cli import cli_command
        if req_auth in noninstaller_plugins and not req_inst:
            msg = ('With the {0} plugin, you probably want to use the "certonly" command, eg:{1}'
                   '{1}    {2} certonly --{0}{1}{1}'
                   '(Alternatively, add a --installer flag. See https://eff.org/letsencrypt-plugins'
                   '{1} and "--help plugins" for more information.)'.format(

            

Reported by Pylint.

                  return now


def cli_plugin_requests(config):
    """
    Figure out which plugins the user requested with CLI and config options

    :returns: (requested authenticator string or None, requested installer string or None)
    :rtype: tuple

            

Reported by Pylint.

                                 "your existing configuration.\nThe error was: {1!r}"
                   .format(requested, plugins[requested].problem))
    elif cfg_type == "installer":
        from certbot._internal.cli import cli_command
        msg = ('Certbot doesn\'t know how to automatically configure the web '
          'server on this system. However, it can still get a certificate for '
          'you. Please run "{0} certonly" to do so. You\'ll need to '
          'manually configure your web server to use the resulting '
          'certificate.').format(cli_command)

            

Reported by Pylint.