# Normally os module should not be imported in certbot codebase except in certbot.compat
        # for the sake of compatibility over Windows and Linux.
        # We make an exception here, since _lock_success is private and called only on Linux.
        from os import stat, fstat  # pylint: disable=os-module-forbidden
        try:
            stat1 = stat(self._path)
        except OSError as err:
            if err.errno == errno.ENOENT:
                return False

            

Reported by Pylint.

                      """Acquire the lock."""
        while self._fd is None:
            # Open the file
            fd = filesystem.open(self._path, os.O_CREAT | os.O_WRONLY, 0o600)
            try:
                self._try_lock(fd)
                if self._lock_success(fd):
                    self._fd = fd
            finally:

            

Reported by Pylint.

                              if self._fd is None:
                    os.close(fd)

    def _try_lock(self, fd: int) -> None:
        """
        Try to acquire the lock file without blocking.
        :param int fd: file descriptor of the opened file to lock
        """
        try:

            

Reported by Pylint.

                              raise errors.LockError('Another instance of Certbot is already running.')
            raise

    def _lock_success(self, fd: int) -> bool:
        """
        Did we successfully grab the lock?
        Because this class deletes the locked file when the lock is
        released, it is possible another process removed and recreated
        the file between us opening the file and acquiring the lock.

            

Reported by Pylint.

                      # Normally os module should not be imported in certbot codebase except in certbot.compat
        # for the sake of compatibility over Windows and Linux.
        # We make an exception here, since _lock_success is private and called only on Linux.
        from os import stat, fstat  # pylint: disable=os-module-forbidden
        try:
            stat1 = stat(self._path)
        except OSError as err:
            if err.errno == errno.ENOENT:
                return False

            

Reported by Pylint.

                      """Acquire the lock"""
        open_mode = os.O_RDWR | os.O_CREAT | os.O_TRUNC

        fd = None
        try:
            # Under Windows, filesystem.open will raise directly an EACCES error
            # if the lock file is already locked.
            fd = filesystem.open(self._path, open_mode, 0o600)
            # This "type: ignore" is currently needed because msvcrt methods

            

Reported by Pylint.

                      try:
            # Under Windows, filesystem.open will raise directly an EACCES error
            # if the lock file is already locked.
            fd = filesystem.open(self._path, open_mode, 0o600)
            # This "type: ignore" is currently needed because msvcrt methods
            # are only defined on Windows. See
            # https://github.com/python/typeshed/blob/16ae4c61201cd8b96b8b22cdfb2ab9e89ba5bcf2/stdlib/msvcrt.pyi.
            msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)  # type: ignore
        except (IOError, OSError) as err:

            

Reported by Pylint.

              
            try:
                os.remove(self._path)
            except OSError as e:
                # If the lock file cannot be removed, it is not a big deal.
                # Likely another instance is acquiring the lock we just released.
                logger.debug(str(e))
        finally:
            self._fd = None

            

Reported by Pylint.

              
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
import josepy as jose
import OpenSSL

from acme import challenges
from acme import client
from acme import crypto_util

            

Reported by Pylint.

              import josepy as jose
import OpenSSL

from acme import challenges
from acme import client
from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone

            

Reported by Pylint.

              import OpenSSL

from acme import challenges
from acme import client
from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone


            

Reported by Pylint.

              
from acme import challenges
from acme import client
from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone

# Constants:

            

Reported by Pylint.

              from acme import challenges
from acme import client
from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone

# Constants:


            

Reported by Pylint.

              from acme import client
from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone

# Constants:

# This is the staging point for ACME-V2 within Let's Encrypt.

            

Reported by Pylint.

              from acme import crypto_util
from acme import errors
from acme import messages
from acme import standalone

# Constants:

# This is the staging point for ACME-V2 within Let's Encrypt.
DIRECTORY_URL = 'https://acme-staging-v02.api.letsencrypt.org/directory'

            

Reported by Pylint.

                      try:
            subprocess.check_output(['docker', '-v'], stderr=subprocess.STDOUT)
        except (subprocess.CalledProcessError, OSError):
            raise ValueError('Error: docker is required in PATH to launch the integration tests on'
                             'boulder, but is not installed or not available for current user.')

        try:
            subprocess.check_output(['docker-compose', '-v'], stderr=subprocess.STDOUT)
        except (subprocess.CalledProcessError, OSError):

            

Reported by Pylint.

                      try:
            subprocess.check_output(['docker-compose', '-v'], stderr=subprocess.STDOUT)
        except (subprocess.CalledProcessError, OSError):
            raise ValueError(
                'Error: docker-compose is required in PATH to launch the integration tests, '
                'but is not installed or not available for current user.'
            )

    # Parameter numprocesses is added to option by pytest-xdist

            

Reported by Pylint.

              See https://docs.pytest.org/en/latest/reference.html#hook-reference
"""
import contextlib
import subprocess
import sys

from certbot_integration_tests.utils import acme_server as acme_lib
from certbot_integration_tests.utils import dns_server as dns_lib


            

Reported by Bandit.

                  # Check for runtime compatibility: some tools are required to be available in PATH
    if 'boulder' in config.option.acme_server:
        try:
            subprocess.check_output(['docker', '-v'], stderr=subprocess.STDOUT)
        except (subprocess.CalledProcessError, OSError):
            raise ValueError('Error: docker is required in PATH to launch the integration tests on'
                             'boulder, but is not installed or not available for current user.')

        try:

            

Reported by Bandit.

                  # Check for runtime compatibility: some tools are required to be available in PATH
    if 'boulder' in config.option.acme_server:
        try:
            subprocess.check_output(['docker', '-v'], stderr=subprocess.STDOUT)
        except (subprocess.CalledProcessError, OSError):
            raise ValueError('Error: docker is required in PATH to launch the integration tests on'
                             'boulder, but is not installed or not available for current user.')

        try:

            

Reported by Bandit.

                                           'boulder, but is not installed or not available for current user.')

        try:
            subprocess.check_output(['docker-compose', '-v'], stderr=subprocess.STDOUT)
        except (subprocess.CalledProcessError, OSError):
            raise ValueError(
                'Error: docker-compose is required in PATH to launch the integration tests, '
                'but is not installed or not available for current user.'
            )

            

Reported by Bandit.

                                           'boulder, but is not installed or not available for current user.')

        try:
            subprocess.check_output(['docker-compose', '-v'], stderr=subprocess.STDOUT)
        except (subprocess.CalledProcessError, OSError):
            raise ValueError(
                'Error: docker-compose is required in PATH to launch the integration tests, '
                'but is not installed or not available for current user.'
            )

            

Reported by Bandit.

              
from distutils.version import LooseVersion
import os
import subprocess
import sys

import certbot_integration_tests
# pylint: disable=wildcard-import,unused-wildcard-import
from certbot_integration_tests.utils.constants import *

            

Reported by Bandit.

              from certbot_integration_tests.utils.constants import *


def certbot_test(certbot_args, directory_url, http_01_port, tls_alpn_01_port,
                 config_dir, workspace, force_renew=True):
    """
    Invoke the certbot executable available in PATH in a test context for the given args.
    The test context consists in running certbot in debug mode, with various flags suitable
    for tests (eg. no ssl check, customizable ACME challenge ports and config directory ...).

            

Reported by Pylint.

                  command, env = _prepare_args_env(certbot_args, directory_url, http_01_port, tls_alpn_01_port,
                                     config_dir, workspace, force_renew)

    proc = subprocess.run(command, stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE, check=False, universal_newlines=True,
                          cwd=workspace, env=env)
    print('--> Certbot log output was:')
    print(proc.stderr)
    proc.check_returncode()

            

Reported by Bandit.

              
def _compute_additional_args(workspace, environ, force_renew):
    additional_args = []
    output = subprocess.check_output(['certbot', '--version'],
                                     universal_newlines=True, stderr=subprocess.STDOUT,
                                     cwd=workspace, env=environ)
    # Typical response is: output = 'certbot 0.31.0.dev0'
    version_str = output.split(' ')[1].strip()
    if LooseVersion(version_str) >= LooseVersion('0.30.0'):

            

Reported by Bandit.

              
def _compute_additional_args(workspace, environ, force_renew):
    additional_args = []
    output = subprocess.check_output(['certbot', '--version'],
                                     universal_newlines=True, stderr=subprocess.STDOUT,
                                     cwd=workspace, env=environ)
    # Typical response is: output = 'certbot 0.31.0.dev0'
    version_str = output.split(' ')[1].strip()
    if LooseVersion(version_str) >= LooseVersion('0.30.0'):

            

Reported by Bandit.

                  return additional_args


def _prepare_args_env(certbot_args, directory_url, http_01_port, tls_alpn_01_port,
                      config_dir, workspace, force_renew):

    new_environ = _prepare_environ(workspace)
    additional_args = _compute_additional_args(workspace, new_environ, force_renew)


            

Reported by Pylint.

                  # Invoke certbot in test mode, without capturing output so users see directly the outcome.
    command, env = _prepare_args_env(args, directory_url, http_01_port, tls_alpn_01_port,
                                     config_dir, workspace, True)
    subprocess.check_call(command, universal_newlines=True, cwd=workspace, env=env)


if __name__ == '__main__':
    main()

            

Reported by Bandit.

              
import pkg_resources

from certbot import errors
from certbot import util
from certbot.compat import os

logger = logging.getLogger(__name__)


            

Reported by Pylint.

              import pkg_resources

from certbot import errors
from certbot import util
from certbot.compat import os

logger = logging.getLogger(__name__)



            

Reported by Pylint.

              
from certbot import errors
from certbot import util
from certbot.compat import os

logger = logging.getLogger(__name__)


def get_mod_deps(mod_name):

            

Reported by Pylint.

                      logger.error(
            "Error running command %s for runtime parameters!%s",
            command, os.linesep)
        raise errors.MisconfigurationError(
            "Error accessing loaded Apache parameters: {0}".format(
                command))
    # Small errors that do not impede
    if proc.returncode != 0:
        logger.warning("Error in checking parameter list: %s", stderr)

            

Reported by Pylint.

              import fnmatch
import logging
import re
import subprocess

import pkg_resources

from certbot import errors
from certbot import util

            

Reported by Bandit.

                  return_vars = {}
    # Get list of words in the variable
    a_opts = util.get_var_from_file(varname, filepath).split()
    for i, v in enumerate(a_opts):
        # Handle Define statements and make sure it has an argument
        if v == "-D" and len(a_opts) >= i+2:
            var_parts = a_opts[i+1].partition("=")
            return_vars[var_parts[0]] = var_parts[2]
        elif len(v) > 2 and v.startswith("-D"):

            

Reported by Pylint.

              
    """
    try:
        proc = subprocess.run(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
            check=False,

            

Reported by Bandit.

              """Test certbot_nginx._internal.display_ops."""
import unittest

from certbot.display import util as display_util
from certbot.tests import util as certbot_util
from certbot_nginx._internal import parser
from certbot_nginx._internal.display_ops import select_vhost_multiple
import test_util as util


            

Reported by Pylint.

              import unittest

from certbot.display import util as display_util
from certbot.tests import util as certbot_util
from certbot_nginx._internal import parser
from certbot_nginx._internal.display_ops import select_vhost_multiple
import test_util as util



            

Reported by Pylint.

              
from certbot.display import util as display_util
from certbot.tests import util as certbot_util
from certbot_nginx._internal import parser
from certbot_nginx._internal.display_ops import select_vhost_multiple
import test_util as util


class SelectVhostMultiTest(util.NginxTest):

            

Reported by Pylint.

              from certbot.display import util as display_util
from certbot.tests import util as certbot_util
from certbot_nginx._internal import parser
from certbot_nginx._internal.display_ops import select_vhost_multiple
import test_util as util


class SelectVhostMultiTest(util.NginxTest):
    """Tests for certbot_nginx._internal.display_ops.select_vhost_multiple."""

            

Reported by Pylint.

                      nparser = parser.NginxParser(self.config_path)
        self.vhosts = nparser.get_vhosts()

    def test_select_no_input(self):
        self.assertFalse(select_vhost_multiple([]))

    @certbot_util.patch_display_util()
    def test_select_correct(self, mock_util):
        mock_util().checklist.return_value = (

            

Reported by Pylint.

                      self.assertFalse(select_vhost_multiple([]))

    @certbot_util.patch_display_util()
    def test_select_correct(self, mock_util):
        mock_util().checklist.return_value = (
            display_util.OK, [self.vhosts[3].display_repr(),
                              self.vhosts[2].display_repr()])
        vhs = select_vhost_multiple([self.vhosts[3],
                                     self.vhosts[2],

            

Reported by Pylint.

                      self.assertFalse(self.vhosts[1] in vhs)

    @certbot_util.patch_display_util()
    def test_select_cancel(self, mock_util):
        mock_util().checklist.return_value = (display_util.CANCEL, "whatever")
        vhs = select_vhost_multiple([self.vhosts[2], self.vhosts[3]])
        self.assertFalse(vhs)



            

Reported by Pylint.

              
        # Unfortunately the BIND9 image forces everything to stderr with -g and we can't
        # modify the verbosity.
        # pylint: disable=consider-using-with
        self._output = sys.stderr if show_output else open(os.devnull, "w")

    def start(self):
        """Start the DNS server"""
        try:

            

Reported by Pylint.

                  def _start_bind(self):
        """Launch the BIND9 server as a Docker container"""
        addr_str = "{}:{}".format(BIND_BIND_ADDRESS[0], BIND_BIND_ADDRESS[1])
        # pylint: disable=consider-using-with
        self.process = subprocess.Popen(
            [
                "docker",
                "run",
                "--rm",

            

Reported by Pylint.

                          try:
                self.process.terminate()
                self.process.wait()
            except BaseException as e:
                print("BIND9 did not stop cleanly: {}".format(e), file=sys.stderr)

        shutil.rmtree(self.bind_root, ignore_errors=True)

        if self._output != sys.stderr:

            

Reported by Pylint.

              import os.path
import shutil
import socket
import subprocess
import sys
import tempfile
import time
from typing import Optional


            

Reported by Bandit.

                          try:
                self.process.terminate()
                self.process.wait()
            except BaseException as e:
                print("BIND9 did not stop cleanly: {}".format(e), file=sys.stderr)

        shutil.rmtree(self.bind_root, ignore_errors=True)

        if self._output != sys.stderr:

            

Reported by Pylint.

                      """Launch the BIND9 server as a Docker container"""
        addr_str = "{}:{}".format(BIND_BIND_ADDRESS[0], BIND_BIND_ADDRESS[1])
        # pylint: disable=consider-using-with
        self.process = subprocess.Popen(
            [
                "docker",
                "run",
                "--rm",
                "-p",

            

Reported by Bandit.

                      """Launch the BIND9 server as a Docker container"""
        addr_str = "{}:{}".format(BIND_BIND_ADDRESS[0], BIND_BIND_ADDRESS[1])
        # pylint: disable=consider-using-with
        self.process = subprocess.Popen(
            [
                "docker",
                "run",
                "--rm",
                "-p",

            

Reported by Bandit.

              from typing import Tuple
from typing import Union

import josepy as jose
from OpenSSL import crypto
from OpenSSL import SSL

from acme import errors


            

Reported by Pylint.

                  )

    cert.get_subject().CN = domains[0]
    # TODO: what to put into cert.get_subject()?
    cert.set_issuer(cert.get_subject())

    if force_san or len(domains) > 1:
        extensions.append(crypto.X509Extension(
            b"subjectAltName",

            

Reported by Pylint.

                  :rtype: bytes

    """
    # XXX: returns empty string when no chain is available, which
    # shuts up RenewableCert, but might not be the best solution...

    def _dump_cert(cert):
        if isinstance(cert, jose.ComparableX509):
            cert = cert.wrapped

            

Reported by Pylint.

              _DEFAULT_SSL_METHOD = SSL.SSLv23_METHOD


class _DefaultCertSelection:
    def __init__(self, certs):
        self.certs = certs

    def __call__(self, connection):
        server_name = connection.get_servername()

            

Reported by Pylint.

                      `certs` parameter would be ignored, and therefore must be empty.

    """
    def __init__(self, sock, certs=None,
            method=_DEFAULT_SSL_METHOD, alpn_selection=None,
            cert_selection=None):
        self.sock = sock
        self.alpn_selection = alpn_selection
        self.method = method

            

Reported by Pylint.

                          for part in sans_parts if part.startswith(prefix)]


def gen_ss_cert(key, domains, not_before=None,
                validity=(7 * 24 * 60 * 60), force_san=True, extensions=None):
    """Generate new self-signed certificate.

    :type domains: `list` of `unicode`
    :param OpenSSL.crypto.PKey key:

            

Reported by Pylint.

                  extension is used, unless `force_san` is ``True``.

    """
    assert domains, "Must provide one or more hostnames for the cert."
    cert = crypto.X509()
    cert.set_serial_number(int(binascii.hexlify(os.urandom(16)), 16))
    cert.set_version(2)

    if extensions is None:

            

Reported by Bandit.

                  for _ in range(attempts):
        time.sleep(1)
        try:
            if requests.get(url, verify=False).status_code == 200:
                return
        except requests.exceptions.ConnectionError:
            pass

    raise ValueError('Error, url did not respond after {0} attempts: {1}'.format(attempts, url))

            

Reported by Bandit.

              
    issuers = []
    for i in range(PEBBLE_ALTERNATE_ROOTS + 1):
        request = requests.get(PEBBLE_MANAGEMENT_URL + '/intermediates/{}'.format(i), verify=False)
        issuers.append(load_pem_x509_certificate(request.content, default_backend()))

    return issuers

            

Reported by Bandit.

                      pass
raise ValueError('Error, url did not respond after 10 attempts: {{0}}'.format(url))
'''.format(http_server_root.replace('\\', '\\\\'), http_port))
        os.chmod(auth_script_path, 0o755)

        cleanup_script_path = os.path.join(tempdir, 'cleanup.py')
        with open(cleanup_script_path, 'w') as file_h:
            file_h.write('''\
#!/usr/bin/env python

            

Reported by Bandit.

              well_known = os.path.join('{0}', '.well-known')
shutil.rmtree(well_known)
'''.format(http_server_root.replace('\\', '\\\\')))
        os.chmod(cleanup_script_path, 0o755)

        yield ('{0} {1}'.format(sys.executable, auth_script_path),
               '{0} {1}'.format(sys.executable, cleanup_script_path))
    finally:
        shutil.rmtree(tempdir)

            

Reported by Bandit.

                  :param context: the testing context.
    :return: the `list of x509.Certificate` representing the list of issuers.
    """
    # TODO: in fact, Boulder has alternate chains in config-next/, just not yet in config/.
    if context.acme_server != "pebble":
        raise NotImplementedError()

    _suppress_x509_verification_warnings()


            

Reported by Pylint.

              
def _suppress_x509_verification_warnings():
    try:
        import urllib3
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    except ImportError:
        # Handle old versions of request with vendorized urllib3
        # pylint: disable=no-member
        from requests.packages.urllib3.exceptions import InsecureRequestWarning

            

Reported by Pylint.

                  except ImportError:
        # Handle old versions of request with vendorized urllib3
        # pylint: disable=no-member
        from requests.packages.urllib3.exceptions import InsecureRequestWarning
        requests.packages.urllib3.disable_warnings(InsecureRequestWarning)


def check_until_timeout(url, attempts=30):
    """

            

Reported by Pylint.

              import traceback
from typing import List

import pytz

from certbot import crypto_util
from certbot import errors
from certbot import ocsp
from certbot import util

            

Reported by Pylint.

                      """Return cert as identical_names_cert if it matches,
           or subset_names_cert if it matches as subset
        """
        # TODO: Handle these differently depending on whether they are
        #       expired or still valid?
        identical_names_cert, subset_names_cert = rv
        candidate_names = set(candidate_lineage.names())
        if candidate_names == set(domains):
            identical_names_cert = candidate_lineage

            

Reported by Pylint.

                          renewal_candidate = storage.RenewableCert(renewal_file, config)
            crypto_util.verify_renewable_cert(renewal_candidate)
            parsed_certs.append(renewal_candidate)
        except Exception as e:  # pylint: disable=broad-except
            logger.warning("Renewal configuration file %s produced an "
                           "unexpected error: %s. Skipping.", renewal_file, e)
            logger.debug("Traceback was:\n%s", traceback.format_exc())
            parse_failures.append(renewal_file)


            

Reported by Pylint.

                  :rtype: `tuple` of `storage.RenewableCert` or `None`

    """
    def update_certs_for_domain_matches(candidate_lineage, rv):
        """Return cert as identical_names_cert if it matches,
           or subset_names_cert if it matches as subset
        """
        # TODO: Handle these differently depending on whether they are
        #       expired or still valid?

            

Reported by Pylint.

                      return return_value

    matched = _search_lineages(cli_config, find_matches, [], acceptable_matches)
    if not matched:
        raise errors.Error("No match found for cert-path {0}!".format(cli_config.cert_path))
    elif len(matched) > 1:
        raise errors.OverlappingMatchFound()
    return matched


            

Reported by Pylint.

                  # Verify the directory is there
    util.make_or_verify_dir(configs_dir, mode=0o755)

    rv = initial_rv
    for renewal_file in storage.renewal_conf_files(cli_config):
        try:
            candidate_lineage = storage.RenewableCert(renewal_file, cli_config)
        except (errors.CertStorageError, IOError):
            logger.debug("Renewal conf file %s is broken. Skipping.", renewal_file)

            

Reported by Pylint.

                          logger.debug("Renewal conf file %s is broken. Skipping.", renewal_file)
            logger.debug("Traceback was:\n%s", traceback.format_exc())
            continue
        rv = func(candidate_lineage, rv, *args)
    return rv

            

Reported by Pylint.