import logging
from typing import Optional

import digitalocean

from certbot import errors
from certbot.plugins import dns_common
from certbot.plugins.dns_common import CredentialsConfiguration


            

Reported by Pylint.

              
import digitalocean

from certbot import errors
from certbot.plugins import dns_common
from certbot.plugins.dns_common import CredentialsConfiguration

logger = logging.getLogger(__name__)


            

Reported by Pylint.

              import digitalocean

from certbot import errors
from certbot.plugins import dns_common
from certbot.plugins.dns_common import CredentialsConfiguration

logger = logging.getLogger(__name__)



            

Reported by Pylint.

              
from certbot import errors
from certbot.plugins import dns_common
from certbot.plugins.dns_common import CredentialsConfiguration

logger = logging.getLogger(__name__)


class Authenticator(dns_common.DNSAuthenticator):

            

Reported by Pylint.

                      self.credentials: Optional[CredentialsConfiguration] = None

    @classmethod
    def add_parser_arguments(cls, add):  # pylint: disable=arguments-differ
        super().add_parser_arguments(add)
        add('credentials', help='DigitalOcean credentials INI file.')

    def more_info(self):  # pylint: disable=missing-function-docstring
        return 'This plugin configures a DNS TXT record to respond to a dns-01 challenge using ' + \

            

Reported by Pylint.

                      super().add_parser_arguments(add)
        add('credentials', help='DigitalOcean credentials INI file.')

    def more_info(self):  # pylint: disable=missing-function-docstring
        return 'This plugin configures a DNS TXT record to respond to a dns-01 challenge using ' + \
               'the DigitalOcean API.'

    def _setup_credentials(self):
        self.credentials = self._configure_credentials(

            

Reported by Pylint.

              
        try:
            domain = self._find_domain(domain_name)
        except digitalocean.Error as e:
            hint = None

            if str(e).startswith("Unable to authenticate"):
                hint = 'Did you provide a valid API token?'


            

Reported by Pylint.

                          record_id = result['domain_record']['id']

            logger.debug('Successfully added TXT record with id: %d', record_id)
        except digitalocean.Error as e:
            logger.debug('Error adding TXT record using the DigitalOcean API: %s', e)
            raise errors.PluginError('Error adding TXT record using the DigitalOcean API: {0}'
                                     .format(e))

    def del_txt_record(self, domain_name: str, record_name: str, record_content: str):

            

Reported by Pylint.

              
        try:
            domain = self._find_domain(domain_name)
        except digitalocean.Error as e:
            logger.debug('Error finding domain using the DigitalOcean API: %s', e)
            return

        try:
            domain_records = domain.get_records()

            

Reported by Pylint.

                                              if record.type == 'TXT'
                                and record.name == self._compute_record_name(domain, record_name)
                                and record.data == record_content]
        except digitalocean.Error as e:
            logger.debug('Error getting DNS records using the DigitalOcean API: %s', e)
            return

        for record in matching_records:
            try:

            

Reported by Pylint.

              from typing import Dict
from typing import List

import boto3
from botocore.exceptions import ClientError
from botocore.exceptions import NoCredentialsError

from certbot import errors
from certbot.plugins import dns_common

            

Reported by Pylint.

              from typing import List

import boto3
from botocore.exceptions import ClientError
from botocore.exceptions import NoCredentialsError

from certbot import errors
from certbot.plugins import dns_common


            

Reported by Pylint.

              
import boto3
from botocore.exceptions import ClientError
from botocore.exceptions import NoCredentialsError

from certbot import errors
from certbot.plugins import dns_common

logger = logging.getLogger(__name__)

            

Reported by Pylint.

              from botocore.exceptions import ClientError
from botocore.exceptions import NoCredentialsError

from certbot import errors
from certbot.plugins import dns_common

logger = logging.getLogger(__name__)

INSTRUCTIONS = (

            

Reported by Pylint.

              from botocore.exceptions import NoCredentialsError

from certbot import errors
from certbot.plugins import dns_common

logger = logging.getLogger(__name__)

INSTRUCTIONS = (
    "To use certbot-dns-route53, configure credentials as described at "

            

Reported by Pylint.

                      pass

    def perform(self, achalls):
        self._attempt_cleanup = True

        try:
            change_ids = [
                self._change_txt_record("UPSERT",
                  achall.validation_domain_name(achall.domain),

            

Reported by Pylint.

                          raise errors.PluginError("\n".join([str(e), INSTRUCTIONS]))
        return [achall.response(achall.account_key) for achall in achalls]

    def _cleanup(self, domain, validation_name, validation):
        try:
            self._change_txt_record("DELETE", validation_name, validation)
        except (NoCredentialsError, ClientError) as e:
            logger.debug('Encountered error during cleanup: %s', e, exc_info=True)


            

Reported by Pylint.

                      self.r53 = boto3.client("route53")
        self._resource_records: DefaultDict[str, List[Dict[str, str]]] = collections.defaultdict(list)

    def more_info(self):  # pylint: disable=missing-function-docstring
        return "Solve a DNS01 challenge using AWS Route53"

    def _setup_credentials(self):
        pass


            

Reported by Pylint.

                  def _perform(self, domain, validation_name, validation):
        pass

    def perform(self, achalls):
        self._attempt_cleanup = True

        try:
            change_ids = [
                self._change_txt_record("UPSERT",

            

Reported by Pylint.

              
            for change_id in change_ids:
                self._wait_for_change(change_id)
        except (NoCredentialsError, ClientError) as e:
            logger.debug('Encountered error during perform: %s', e, exc_info=True)
            raise errors.PluginError("\n".join([str(e), INSTRUCTIONS]))
        return [achall.response(achall.account_key) for achall in achalls]

    def _cleanup(self, domain, validation_name, validation):

            

Reported by Pylint.

                              result = self._handle_http_error(e, domain_name)

                if result:
                    raise result
            except Exception as e:  # pylint: disable=broad-except
                result = self._handle_general_error(e, domain_name)

                if result:
                    raise result  # pylint: disable=raising-bad-type

            

Reported by Pylint.

                              result = self._handle_general_error(e, domain_name)

                if result:
                    raise result  # pylint: disable=raising-bad-type

        raise errors.PluginError('Unable to determine zone identifier for {0} using zone names: {1}'
                                 .format(domain, domain_name_guesses))

    def _handle_http_error(self, e, domain_name):

            

Reported by Pylint.

              
        try:
            self.provider.create_record(rtype='TXT', name=record_name, content=record_content)
        except RequestException as e:
            logger.debug('Encountered error adding TXT record: %s', e, exc_info=True)
            raise errors.PluginError('Error adding TXT record: {0}'.format(e))

    def del_txt_record(self, domain, record_name, record_content):
        """

            

Reported by Pylint.

                      """
        try:
            self._find_domain_id(domain)
        except errors.PluginError as e:
            logger.debug('Encountered error finding domain_id during deletion: %s', e,
                         exc_info=True)
            return

        try:

            

Reported by Pylint.

              
        try:
            self.provider.delete_record(rtype='TXT', name=record_name, content=record_content)
        except RequestException as e:
            logger.debug('Encountered error deleting TXT record: %s', e, exc_info=True)

    def _find_domain_id(self, domain):
        """
        Find the domain_id for a given domain.

            

Reported by Pylint.

                              self.provider.authenticate()

                return  # If `authenticate` doesn't throw an exception, we've found the right name
            except HTTPError as e:
                result = self._handle_http_error(e, domain_name)

                if result:
                    raise result
            except Exception as e:  # pylint: disable=broad-except

            

Reported by Pylint.

              
                if result:
                    raise result
            except Exception as e:  # pylint: disable=broad-except
                result = self._handle_general_error(e, domain_name)

                if result:
                    raise result  # pylint: disable=raising-bad-type


            

Reported by Pylint.

                      raise errors.PluginError('Unable to determine zone identifier for {0} using zone names: {1}'
                                 .format(domain, domain_name_guesses))

    def _handle_http_error(self, e, domain_name):
        return errors.PluginError('Error determining zone identifier for {0}: {1}.'
                                  .format(domain_name, e))

    def _handle_general_error(self, e, domain_name):
        if not str(e).startswith('No domain found'):

            

Reported by Pylint.

                      raise errors.PluginError('Unable to determine zone identifier for {0} using zone names: {1}'
                                 .format(domain, domain_name_guesses))

    def _handle_http_error(self, e, domain_name):
        return errors.PluginError('Error determining zone identifier for {0}: {1}.'
                                  .format(domain_name, e))

    def _handle_general_error(self, e, domain_name):
        if not str(e).startswith('No domain found'):

            

Reported by Pylint.

                      return errors.PluginError('Error determining zone identifier for {0}: {1}.'
                                  .format(domain_name, e))

    def _handle_general_error(self, e, domain_name):
        if not str(e).startswith('No domain found'):
            return errors.PluginError('Unexpected error determining zone identifier for {0}: {1}'
                                      .format(domain_name, e))
        return None


            

Reported by Pylint.

                      self._proxy = http_proxy
        self._workspace = tempfile.mkdtemp()
        self._processes: List[subprocess.Popen] = []
        self._stdout = sys.stdout if stdout else open(os.devnull, 'w') # pylint: disable=consider-using-with
        self._dns_server = dns_server
        self._http_01_port = http_01_port
        if http_01_port != DEFAULT_HTTP_01_PORT:
            if self._acme_type != 'pebble' or self._proxy:
                raise ValueError('setting http_01_port is not currently supported '

            

Reported by Pylint.

                      if not env:
            env = os.environ
        stdout = sys.stderr if force_stderr else self._stdout
        # pylint: disable=consider-using-with
        process = subprocess.Popen(
            command, stdout=stdout, stderr=subprocess.STDOUT, cwd=cwd, env=env
        )
        self._processes.append(process)
        return process

            

Reported by Pylint.

              import os
from os.path import join
import shutil
import subprocess
import sys
import tempfile
import time
from typing import List


            

Reported by Bandit.

              from certbot_integration_tests.utils.constants import *


class ACMEServer:
    """
    ACMEServer configures and handles the lifecycle of an ACME CA server and an HTTP reverse proxy
    instance, to allow parallel execution of integration tests against the unique http-01 port
    expected by the ACME CA server.
    Typically all pytest integration tests will be executed in this context.

            

Reported by Pylint.

                  ACMEServer is also a context manager, and so can be used to ensure ACME server is
    started/stopped upon context enter/exit.
    """
    def __init__(self, acme_server, nodes, http_proxy=True, stdout=False,
                 dns_server=None, http_01_port=DEFAULT_HTTP_01_PORT):
        """
        Create an ACMEServer instance.
        :param str acme_server: the type of acme server used (boulder-v1, boulder-v2 or pebble)
        :param list nodes: list of node names that will be setup by pytest xdist

            

Reported by Pylint.

                              self._prepare_pebble_server()
            if self._acme_type == 'boulder':
                self._prepare_boulder_server()
        except BaseException as e:
            self.stop()
            raise e

    def stop(self):
        """Stop the test stack, and clean its resources"""

            

Reported by Pylint.

                          for process in self._processes:
                try:
                    process.terminate()
                except OSError as e:
                    # Process may be not started yet, so no PID and terminate fails.
                    # Then the process never started, and the situation is acceptable.
                    if e.errno != errno.ESRCH:
                        raise
            for process in self._processes:

            

Reported by Pylint.

                      # useless ImportError, in the case where cryptography dependency is too old to support
        # ocsp, but Boulder is used instead of Pebble, so pebble_ocsp_server is not used. This is
        # the typical situation of integration-certbot-oldest tox testenv.
        from certbot_integration_tests.utils import pebble_ocsp_server
        self._launch_process([sys.executable, pebble_ocsp_server.__file__])

        # Wait for the ACME CA server to be up.
        print('=> Waiting for pebble instance to respond...')
        misc.check_until_timeout(self.acme_xdist['directory_url'])

            

Reported by Pylint.

                      if self._dns_server:
            # Change Boulder config to use the provided DNS server
            for suffix in ["", "-remote-a", "-remote-b"]:
                with open(join(instance_path, 'test/config/va{}.json'.format(suffix)), 'r') as f:
                    config = json.loads(f.read())
                config['va']['dnsResolvers'] = [self._dns_server]
                with open(join(instance_path, 'test/config/va{}.json'.format(suffix)), 'w') as f:
                    f.write(json.dumps(config, indent=2, separators=(',', ': ')))


            

Reported by Pylint.

                              with open(join(instance_path, 'test/config/va{}.json'.format(suffix)), 'r') as f:
                    config = json.loads(f.read())
                config['va']['dnsResolvers'] = [self._dns_server]
                with open(join(instance_path, 'test/config/va{}.json'.format(suffix)), 'w') as f:
                    f.write(json.dumps(config, indent=2, separators=(',', ': ')))

        try:
            # Launch the Boulder server
            self._launch_process(['docker-compose', 'up', '--force-recreate'], cwd=instance_path)

            

Reported by Pylint.

              from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives import serialization
import pytz
import requests

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

            

Reported by Pylint.

                  with open(cert_path, 'rb') as file_handler:
        cert = x509.load_pem_x509_certificate(file_handler.read(), default_backend())
    builder = ocsp.OCSPRequestBuilder()
    builder = builder.add_certificate(cert, issuer, hashes.SHA1())
    request = builder.build()
    request_binary = request.public_bytes(serialization.Encoding.DER)
    try:
        response = requests.post(url, data=request_binary,
                                 headers={'Content-Type': 'application/ocsp-request'},

            

Reported by Bandit.

              from datetime import timedelta
import logging
import re
import subprocess
from subprocess import PIPE
from typing import Optional
from typing import Tuple

from cryptography import x509

            

Reported by Bandit.

              import logging
import re
import subprocess
from subprocess import PIPE
from typing import Optional
from typing import Tuple

from cryptography import x509
from cryptography.exceptions import InvalidSignature

            

Reported by Bandit.

                              return

            # New versions of openssl want -header var=val, old ones want -header var val
            test_host_format = subprocess.run(["openssl", "ocsp", "-header", "var", "val"],
                                     stdout=PIPE, stderr=PIPE, universal_newlines=True,
                                     check=False, env=util.env_no_snap_for_external_calls())
            if "Missing =" in test_host_format.stderr:
                self.host_args = lambda host: ["Host=" + host]
            else:

            

Reported by Bandit.

                              return

            # New versions of openssl want -header var=val, old ones want -header var val
            test_host_format = subprocess.run(["openssl", "ocsp", "-header", "var", "val"],
                                     stdout=PIPE, stderr=PIPE, universal_newlines=True,
                                     check=False, env=util.env_no_snap_for_external_calls())
            if "Missing =" in test_host_format.stderr:
                self.host_args = lambda host: ["Host=" + host]
            else:

            

Reported by Bandit.

                          return self._check_ocsp_openssl_bin(cert_path, chain_path, host, url, timeout)
        return _check_ocsp_cryptography(cert_path, chain_path, url, timeout)

    def _check_ocsp_openssl_bin(self, cert_path: str, chain_path: str,
                                host: str, url: str, timeout: int) -> bool:
        # Minimal implementation of proxy selection logic as seen in, e.g., cURL
        # Some things that won't work, but may well be in use somewhere:
        # - username and password for proxy authentication
        # - proxies accepting TLS connections

            

Reported by Pylint.

                      # - proxies accepting TLS connections
        # - proxy exclusion through NO_PROXY
        env_http_proxy = getenv('http_proxy')
        env_HTTP_PROXY = getenv('HTTP_PROXY')
        proxy_host = None
        if env_http_proxy is not None or env_HTTP_PROXY is not None:
            proxy_host = env_http_proxy if env_http_proxy is not None else env_HTTP_PROXY
        if proxy_host is None:
            url_opts = ["-url", url]

            

Reported by Pylint.

                  # Check OCSP signature
    try:
        _check_ocsp_response(response_ocsp, request, issuer, cert_path)
    except UnsupportedAlgorithm as e:
        logger.warning(str(e))
    except errors.Error as e:
        logger.warning(str(e))
    except InvalidSignature:
        logger.warning('Invalid signature on OCSP response for %s', cert_path)

            

Reported by Pylint.

                      _check_ocsp_response(response_ocsp, request, issuer, cert_path)
    except UnsupportedAlgorithm as e:
        logger.warning(str(e))
    except errors.Error as e:
        logger.warning(str(e))
    except InvalidSignature:
        logger.warning('Invalid signature on OCSP response for %s', cert_path)
    except AssertionError as error:
        logger.warning('Invalid OCSP response for %s: %s.', cert_path, str(error))

            

Reported by Pylint.

                          new_nodes.append(nodeclass(primary=primary_res[0],
                                       secondary=secondary_res[0]))  # pragma: no cover
        elif pass_primary:
            for c in secondary_res:
                new_nodes.append(nodeclass(primary=primary_res[0],
                                           secondary=c))
        elif pass_secondary:
            for c in primary_res:
                new_nodes.append(nodeclass(primary=c,

            

Reported by Pylint.

                              new_nodes.append(nodeclass(primary=primary_res[0],
                                           secondary=c))
        elif pass_secondary:
            for c in primary_res:
                new_nodes.append(nodeclass(primary=c,
                                           secondary=secondary_res[0]))
        else:
            assert len(primary_res) == len(secondary_res)
            matches = self._create_matching_list(primary_res, secondary_res)

            

Reported by Pylint.

                              new_nodes.append(nodeclass(primary=c,
                                           secondary=secondary_res[0]))
        else:
            assert len(primary_res) == len(secondary_res)
            matches = self._create_matching_list(primary_res, secondary_res)
            for p, s in matches:
                new_nodes.append(nodeclass(primary=p, secondary=s))

        return new_nodes

            

Reported by Bandit.

                      else:
            assert len(primary_res) == len(secondary_res)
            matches = self._create_matching_list(primary_res, secondary_res)
            for p, s in matches:
                new_nodes.append(nodeclass(primary=p, secondary=s))

        return new_nodes



            

Reported by Pylint.

                      else:
            assert len(primary_res) == len(secondary_res)
            matches = self._create_matching_list(primary_res, secondary_res)
            for p, s in matches:
                new_nodes.append(nodeclass(primary=p, secondary=s))

        return new_nodes



            

Reported by Pylint.

                      secondary = kwargs.pop("secondary")

        if primary or secondary:
            assert primary and secondary
            self.primary = primary
            self.secondary = secondary
        else:
            self.primary = augeasparser.AugeasCommentNode(**kwargs)
            self.secondary = apacheparser.ApacheCommentNode(**kwargs)

            

Reported by Bandit.

                      secondary = kwargs.pop("secondary")

        if primary or secondary:
            assert primary and secondary
            self.primary = primary
            self.secondary = secondary
        else:
            self.primary = augeasparser.AugeasDirectiveNode(**kwargs)
            self.secondary = apacheparser.ApacheDirectiveNode(**kwargs)

            

Reported by Bandit.

                      secondary = kwargs.pop("secondary")

        if primary or secondary:
            assert primary and secondary
            self.primary = primary
            self.secondary = secondary
        else:
            self.primary = augeasparser.AugeasBlockNode(**kwargs)
            self.secondary = apacheparser.ApacheBlockNode(**kwargs)

            

Reported by Bandit.

                      new_comment = DualCommentNode(primary=primary_new, secondary=secondary_new)
        return new_comment

    def _create_matching_list(self, primary_list, secondary_list):
        """ Matches the list of primary_list to a list of secondary_list and
        returns a list of tuples. This is used to create results for find_
        methods.

        This helper function exists, because we cannot ensure that the list of

            

Reported by Pylint.

                      """

        matched = []
        for p in primary_list:
            match = None
            for s in secondary_list:
                try:
                    assertions.assertEqual(p, s)
                    match = s

            

Reported by Pylint.