from __future__ import print_function

import argparse
import os
import subprocess
import sys

DEFAULT_PACKAGES = [
    'certbot', 'acme', 'certbot_apache', 'certbot_dns_cloudflare', 'certbot_dns_cloudxns',

            

Reported by Pylint.

              #!/usr/bin/env python
from __future__ import print_function

import argparse
import os
import subprocess
import sys

DEFAULT_PACKAGES = [

            

Reported by Pylint.

              
import argparse
import os
import subprocess
import sys

DEFAULT_PACKAGES = [
    'certbot', 'acme', 'certbot_apache', 'certbot_dns_cloudflare', 'certbot_dns_cloudxns',
    'certbot_dns_digitalocean', 'certbot_dns_dnsimple', 'certbot_dns_dnsmadeeasy',

            

Reported by Bandit.

              SKIP_PROJECTS_ON_WINDOWS = ['certbot-apache']


def cover(package):
    threshold = COVER_THRESHOLDS.get(package)['windows' if os.name == 'nt' else 'linux']

    pkg_dir = package.replace('_', '-')

    if os.name == 'nt' and pkg_dir in SKIP_PROJECTS_ON_WINDOWS:

            

Reported by Pylint.

                          .format(pkg_dir)))
        return

    subprocess.check_call([sys.executable, '-m', 'pytest',
                           '--cov', pkg_dir, '--cov-append', '--cov-report=', pkg_dir])
    try:
        subprocess.check_call([
            sys.executable, '-m', 'coverage', 'report', '--fail-under',
            str(threshold), '--include', '{0}/*'.format(pkg_dir),

            

Reported by Bandit.

                  subprocess.check_call([sys.executable, '-m', 'pytest',
                           '--cov', pkg_dir, '--cov-append', '--cov-report=', pkg_dir])
    try:
        subprocess.check_call([
            sys.executable, '-m', 'coverage', 'report', '--fail-under',
            str(threshold), '--include', '{0}/*'.format(pkg_dir),
            '--show-missing'])
    except subprocess.CalledProcessError as err:
        print(err)

            

Reported by Bandit.

                      sys.exit(1)


def main():
    description = """
This script is used by tox.ini (and thus by Travis CI and Azure Pipelines) in
order to generate separate stats for each package. It should be removed once
those packages are moved to a separate repo."""
    parser = argparse.ArgumentParser(description=description)

            

Reported by Pylint.

              import subprocess
from unittest import mock

from certbot import errors as le_errors, configuration
from certbot import util as certbot_util
from certbot_apache._internal import entrypoint
from certbot_compatibility_test import errors
from certbot_compatibility_test import util
from certbot_compatibility_test.configurators import common as configurators_common

            

Reported by Pylint.

              from unittest import mock

from certbot import errors as le_errors, configuration
from certbot import util as certbot_util
from certbot_apache._internal import entrypoint
from certbot_compatibility_test import errors
from certbot_compatibility_test import util
from certbot_compatibility_test.configurators import common as configurators_common


            

Reported by Pylint.

              
from certbot import errors as le_errors, configuration
from certbot import util as certbot_util
from certbot_apache._internal import entrypoint
from certbot_compatibility_test import errors
from certbot_compatibility_test import util
from certbot_compatibility_test.configurators import common as configurators_common



            

Reported by Pylint.

                      try:
            subprocess.check_call("apachectl -k restart".split())
        except errors.Error:
            raise errors.Error(
                "Apache failed to load {0} before tests started".format(
                    config))

        return config


            

Reported by Pylint.

              """Provides a common base for Apache proxies"""
import os
import shutil
import subprocess
from unittest import mock

from certbot import errors as le_errors, configuration
from certbot import util as certbot_util
from certbot_apache._internal import entrypoint

            

Reported by Bandit.

                      self._prepare_configurator()

        try:
            subprocess.check_call("apachectl -k restart".split())
        except errors.Error:
            raise errors.Error(
                "Apache failed to load {0} before tests started".format(
                    config))


            

Reported by Bandit.

                  """Returns all and testable domain names in config"""
    all_names = set()
    non_ip_names = set()
    with open(os.path.join(config, "vhosts")) as f:
        for line in f:
            # If parsing a specific vhost
            if line[0].isspace():
                words = line.split()
                if words[0] == "alias":

            

Reported by Pylint.

              import unittest
import warnings

from certbot.compat import os



class ExecuteTest(unittest.TestCase):
    """Tests for certbot.compat.misc.execute_command."""

            

Reported by Pylint.

              
    @classmethod
    def _call(cls, *args, **kwargs):
        from certbot.compat.misc import execute_command
        # execute_command is superseded by execute_command_status
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', category=PendingDeprecationWarning)
            return execute_command(*args, **kwargs)


            

Reported by Pylint.

              
    @classmethod
    def _call(cls, *args, **kwargs):
        from certbot.compat.misc import execute_command_status
        return execute_command_status(*args, **kwargs)

    def _test_common(self, returncode, stdout, stderr):
        given_command = "foo"
        given_name = "foo-hook"

            

Reported by Pylint.

              
    @classmethod
    def _call(cls, *args, **kwargs):
        from certbot.compat.misc import execute_command
        # execute_command is superseded by execute_command_status
        with warnings.catch_warnings():
            warnings.simplefilter('ignore', category=PendingDeprecationWarning)
            return execute_command(*args, **kwargs)


            

Reported by Pylint.

                          warnings.simplefilter('ignore', category=PendingDeprecationWarning)
            return execute_command(*args, **kwargs)

    def test_it(self):
        for returncode in range(0, 2):
            for stdout in ("", "Hello World!",):
                for stderr in ("", "Goodbye Cruel World!"):
                    self._test_common(returncode, stdout, stderr)


            

Reported by Pylint.

              
    @classmethod
    def _call(cls, *args, **kwargs):
        from certbot.compat.misc import execute_command_status
        return execute_command_status(*args, **kwargs)

    def _test_common(self, returncode, stdout, stderr):
        given_command = "foo"
        given_name = "foo-hook"

            

Reported by Pylint.

                          mock_run.return_value.stderr = stderr
            mock_run.return_value.returncode = returncode
            with mock.patch("certbot.compat.misc.logger") as mock_logger:
                self.assertEqual(self._call(given_name, given_command), (returncode, stderr, stdout))

        executed_command = mock_run.call_args[1].get(
            "args", mock_run.call_args[0][0])
        if os.name == 'nt':
            expected_command = ['powershell.exe', '-Command', given_command]

            

Reported by Pylint.

              
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import josepy as jose
import OpenSSL
import pkg_resources

from certbot import configuration
from certbot import util

            

Reported by Pylint.

                  def get_all_names(self) -> Iterable[str]:
        pass

    def deploy_cert(self, domain: str, cert_path: str, key_path: str, chain_path: str,
                    fullchain_path: str) -> None:
        pass

    def enhance(self, domain: str, enhancement: str, options: Optional[List[str]] = None) -> None:
        pass

            

Reported by Pylint.

              
def _guess_loader(filename, loader_pem, loader_der):
    _, ext = os.path.splitext(filename)
    if ext.lower() == '.pem':
        return loader_pem
    elif ext.lower() == '.der':
        return loader_der
    raise ValueError("Loader could not be recognized based on extension")  # pragma: no cover


            

Reported by Pylint.

                  return OpenSSL.crypto.load_privatekey(loader, load_vector(*names))


def make_lineage(config_dir, testfile, ec=False):
    """Creates a lineage defined by testfile.

    This creates the archive, live, and renewal directories if
    necessary and creates a simple lineage.


            

Reported by Pylint.

                      my_lock = lock.LockFile(path)
    try:
        event_out.set()
        assert event_in.wait(timeout=20), 'Timeout while waiting to release the lock.'
    finally:
        my_lock.release()


def lock_and_call(callback, path_to_lock):

            

Reported by Bandit.

                  process.start()

    # Wait confirmation that lock is acquired
    assert receive_event.wait(timeout=10), 'Timeout while waiting to acquire the lock.'
    # Execute the callback
    callback()
    # Trigger unlock from foreign process
    emit_event.set()


            

Reported by Bandit.

              
    # Wait for process termination
    process.join(timeout=10)
    assert process.exitcode == 0


def skip_on_windows(reason):
    """Decorator to skip permanently a test on Windows. A reason is required."""
    def wrapper(function):

            

Reported by Bandit.

              #!/usr/bin/env python
"""Ensures there have been no changes to important certbot-auto files."""

import hashlib
import os


# Relative to the root of the Certbot repo, these files are expected to exist
# and have the SHA-256 hashes contained in this dictionary. These hashes were

            

Reported by Pylint.

              }


def find_repo_root():
    return os.path.dirname(os.path.dirname(os.path.realpath(__file__)))


def sha256_hash(filename):
    hash_object = hashlib.sha256()

            

Reported by Pylint.

                  return os.path.dirname(os.path.dirname(os.path.realpath(__file__)))


def sha256_hash(filename):
    hash_object = hashlib.sha256()
    with open(filename, 'rb') as f:
        hash_object.update(f.read())
    return hash_object.hexdigest()


            

Reported by Pylint.

              
def sha256_hash(filename):
    hash_object = hashlib.sha256()
    with open(filename, 'rb') as f:
        hash_object.update(f.read())
    return hash_object.hexdigest()


def main():

            

Reported by Pylint.

                  return hash_object.hexdigest()


def main():
    repo_root = find_repo_root()
    for filename, expected_hash in EXPECTED_FILES.items():
        filepath = os.path.join(repo_root, filename)
        assert sha256_hash(filepath) == expected_hash, f'unexpected changes to {filepath}'
    print('All certbot-auto files have correct hashes.')

            

Reported by Pylint.

                  repo_root = find_repo_root()
    for filename, expected_hash in EXPECTED_FILES.items():
        filepath = os.path.join(repo_root, filename)
        assert sha256_hash(filepath) == expected_hash, f'unexpected changes to {filepath}'
    print('All certbot-auto files have correct hashes.')


if __name__ == '__main__':
    main()

            

Reported by Bandit.

                      warnings.warn(_DEPRECATION_PLUGIN, DeprecationWarning)
        return True

    if iface == interfaces.Authenticator and interfaces.IAuthenticator.implementedBy(target_class):
        logging.warning(_DEPRECATION_AUTHENTICATOR)
        warnings.warn(_DEPRECATION_AUTHENTICATOR, DeprecationWarning)
        return True

    if iface == interfaces.Installer and interfaces.IInstaller.implementedBy(target_class):

            

Reported by Pylint.

                      warnings.warn(_DEPRECATION_AUTHENTICATOR, DeprecationWarning)
        return True

    if iface == interfaces.Installer and interfaces.IInstaller.implementedBy(target_class):
        logging.warning(_DEPRECATION_INSTALLER)
        warnings.warn(_DEPRECATION_INSTALLER, DeprecationWarning)
        return True

    return False

            

Reported by Pylint.

                      plugins: Dict[str, PluginEntryPoint] = {}
        plugin_paths_string = os.getenv('CERTBOT_PLUGIN_PATH')
        plugin_paths = plugin_paths_string.split(':') if plugin_paths_string else []
        # XXX should ensure this only happens once
        sys.path.extend(plugin_paths)
        for plugin_path in plugin_paths:
            pkg_resources.working_set.add_entry(plugin_path)
        entry_points = itertools.chain(
            pkg_resources.iter_entry_points(

            

Reported by Pylint.

              """Distributions for which prefix will be omitted."""


class PluginEntryPoint:
    """Plugin entry point."""

    # this object is mutable, don't allow it to be hashed!
    __hash__ = None  # type: ignore


            

Reported by Pylint.

                          plugin_ep = cls._load_entry_point(entry_point, plugins, with_prefix=False)
            # entry_point.dist cannot be None here, we would have blown up
            # earlier, however, this assertion is needed for mypy.
            assert entry_point.dist is not None
            if entry_point.dist.key not in PREFIX_FREE_DISTRIBUTIONS:
                prefixed_plugin_ep = cls._load_entry_point(entry_point, plugins, with_prefix=True)
                prefixed_plugin_ep.hidden = True
                message = (
                    "Plugin legacy name {0} may be removed in a future version. "

            

Reported by Bandit.

                      # use list instead of set because PluginEntryPoint is not hashable
        candidates = [plugin_ep for plugin_ep in self._plugins.values()
                      if plugin_ep.initialized and plugin_ep.init() is plugin]
        assert len(candidates) <= 1
        if candidates:
            return candidates[0]
        return None

    def __repr__(self):

            

Reported by Bandit.

              """Module contains classes used by the Nginx Configurator."""
import re

from certbot.plugins import common

ADD_HEADER_DIRECTIVE = 'add_header'

class Addr(common.Addr):
    r"""Represents an Nginx address, i.e. what comes after the 'listen'

            

Reported by Pylint.

                  :param bool ipv6only: Whether the directive includes 'ipv6only=on'

    """
    UNSPECIFIED_IPV4_ADDRESSES = ('', '*', '0.0.0.0')
    CANONICAL_UNSPECIFIED_ADDRESS = UNSPECIFIED_IPV4_ADDRESSES[0]

    def __init__(self, host, port, ssl, default, ipv6, ipv6only):
        super().__init__((host, port))
        self.ssl = ssl

            

Reported by Bandit.

                  UNSPECIFIED_IPV4_ADDRESSES = ('', '*', '0.0.0.0')
    CANONICAL_UNSPECIFIED_ADDRESS = UNSPECIFIED_IPV4_ADDRESSES[0]

    def __init__(self, host, port, ssl, default, ipv6, ipv6only):
        super().__init__((host, port))
        self.ssl = ssl
        self.default = default
        self.ipv6 = ipv6
        self.ipv6only = ipv6only

            

Reported by Pylint.

              
    """

    def __init__(self, filep, addrs, ssl, enabled, names, raw, path):
        """Initialize a VH."""
        self.filep = filep
        self.addrs = addrs
        self.names = names
        self.ssl = ssl

            

Reported by Pylint.

                  def ipv6_enabled(self):
        """Return true if one or more of the listen directives in vhost supports
        IPv6"""
        for a in self.addrs:
            if a.ipv6:
                return True
        return False

    def ipv4_enabled(self):

            

Reported by Pylint.

                      only"""
        if not self.addrs:
            return True
        for a in self.addrs:
            if not a.ipv6:
                return True
        return False

    def display_repr(self):

            

Reported by Pylint.

              from typing import List
from typing import Set

from certbot import errors
from certbot.compat import filesystem
from certbot.compat import os
from certbot.plugins import common
from certbot_apache._internal.obj import VirtualHost  # pylint: disable=unused-import
from certbot_apache._internal.parser import get_aug_path

            

Reported by Pylint.

              from typing import Set

from certbot import errors
from certbot.compat import filesystem
from certbot.compat import os
from certbot.plugins import common
from certbot_apache._internal.obj import VirtualHost  # pylint: disable=unused-import
from certbot_apache._internal.parser import get_aug_path


            

Reported by Pylint.

              
from certbot import errors
from certbot.compat import filesystem
from certbot.compat import os
from certbot.plugins import common
from certbot_apache._internal.obj import VirtualHost  # pylint: disable=unused-import
from certbot_apache._internal.parser import get_aug_path

logger = logging.getLogger(__name__)

            

Reported by Pylint.

              from certbot import errors
from certbot.compat import filesystem
from certbot.compat import os
from certbot.plugins import common
from certbot_apache._internal.obj import VirtualHost  # pylint: disable=unused-import
from certbot_apache._internal.parser import get_aug_path

logger = logging.getLogger(__name__)


            

Reported by Pylint.

                          selected_vhosts += self._relevant_vhosts()

        # Add the challenge configuration
        for vh in selected_vhosts:
            self._set_up_include_directives(vh)

        self.configurator.reverter.register_file_creation(
            True, self.challenge_conf_pre)
        self.configurator.reverter.register_file_creation(

            

Reported by Pylint.

                      name = os.path.join(self.challenge_dir, achall.chall.encode("token"))

        self.configurator.reverter.register_file_creation(True, name)
        with open(name, 'wb') as f:
            f.write(validation.encode())
        filesystem.chmod(name, 0o644)

        return response


            

Reported by Pylint.

                      with open(self.nginx_config_path, 'w') as file:
            file.write(self.nginx_config)

        # pylint: disable=consider-using-with
        process = subprocess.Popen(['nginx', '-c', self.nginx_config_path, '-g', 'daemon off;'])

        assert process.poll() is None
        misc.check_until_timeout('http://localhost:{0}'.format(self.http_01_port))
        return process

            

Reported by Pylint.

              """Module to handle the context of nginx integration tests."""
import os
import subprocess

from certbot_integration_tests.certbot_tests import context as certbot_context
from certbot_integration_tests.nginx_tests import nginx_config as config
from certbot_integration_tests.utils import certbot_call
from certbot_integration_tests.utils import misc


            

Reported by Bandit.

                          file.write(self.nginx_config)

        # pylint: disable=consider-using-with
        process = subprocess.Popen(['nginx', '-c', self.nginx_config_path, '-g', 'daemon off;'])

        assert process.poll() is None
        misc.check_until_timeout('http://localhost:{0}'.format(self.http_01_port))
        return process


            

Reported by Bandit.

                          file.write(self.nginx_config)

        # pylint: disable=consider-using-with
        process = subprocess.Popen(['nginx', '-c', self.nginx_config_path, '-g', 'daemon off;'])

        assert process.poll() is None
        misc.check_until_timeout('http://localhost:{0}'.format(self.http_01_port))
        return process


            

Reported by Bandit.

                      # pylint: disable=consider-using-with
        process = subprocess.Popen(['nginx', '-c', self.nginx_config_path, '-g', 'daemon off;'])

        assert process.poll() is None
        misc.check_until_timeout('http://localhost:{0}'.format(self.http_01_port))
        return process

    def _stop_nginx(self):
        assert self.process.poll() is None

            

Reported by Bandit.

                      return process

    def _stop_nginx(self):
        assert self.process.poll() is None
        self.process.terminate()
        self.process.wait()

            

Reported by Bandit.

              from typing import List
from typing import Tuple

from acme import challenges
from acme import errors as acme_errors
from acme import messages
from certbot import achallenges
from certbot import errors
from certbot._internal import error_handler

            

Reported by Pylint.

              from typing import Tuple

from acme import challenges
from acme import errors as acme_errors
from acme import messages
from certbot import achallenges
from certbot import errors
from certbot._internal import error_handler
from certbot.display import util as display_util

            

Reported by Pylint.

              
from acme import challenges
from acme import errors as acme_errors
from acme import messages
from certbot import achallenges
from certbot import errors
from certbot._internal import error_handler
from certbot.display import util as display_util
from certbot.plugins import common as plugin_common

            

Reported by Pylint.

                              logger.info('Attempting to clean up outstanding challenges...')
                raise error
            # All challenges should have been processed by the authenticator.
            assert len(resps) == len(achalls), 'Some challenges have not been performed.'

            # Inform the ACME CA server that challenges are available for validation.
            for achall, resp in zip(achalls, resps):
                self.acme.answer_challenge(achall.challb, resp)


            

Reported by Bandit.

                          try:
                authzr = self.acme.deactivate_authorization(authzr)
                deactivated.append(authzr)
            except acme_errors.Error as e:
                failed.append(authzr)
                logger.debug('Failed to deactivate authorization %s: %s', authzr.uri, e)

        return (deactivated, failed)


            

Reported by Pylint.

                  chall = challb.chall
    logger.info("%s challenge for %s", chall.typ, domain)

    if isinstance(chall, challenges.KeyAuthorizationChallenge):
        return achallenges.KeyAuthorizationAnnotatedChallenge(
            challb=challb, domain=domain, account_key=account_key)
    elif isinstance(chall, challenges.DNS):
        return achallenges.DNS(challb=challb, domain=domain)
    raise errors.Error(

            

Reported by Pylint.