disabled
  #   li~jeJ
v.jbѿB˿ +    """Support for standalone client challenge solvers. """
import collections
import functools
import logging
import socket
import threading

from six.moves import BaseHTTPServer  # type: ignore
from six.moves import http_client
from six.moves import socketserver  # type: ignore

from acme import challenges
from acme import crypto_util
from acme.magic_typing import List

logger = logging.getLogger(__name__)


class TLSServer(socketserver.TCPServer):
    """Generic TLS Server."""

    def __init__(self, *args, **kwargs):
        self.ipv6 = kwargs.pop("ipv6", False)
        if self.ipv6:
            self.address_family = socket.AF_INET6
        else:
            self.address_family = socket.AF_INET
        self.certs = kwargs.pop("certs", {})
        self.method = kwargs.pop(
            "method", crypto_util._DEFAULT_SSL_METHOD)
        self.allow_reuse_address = kwargs.pop("allow_reuse_address", True)
        socketserver.TCPServer.__init__(self, *args, **kwargs)

    def _wrap_sock(self):
        self.socket = crypto_util.SSLSocket(
            self.socket, cert_selection=self._cert_selection,
            alpn_selection=getattr(self, '_alpn_selection', None),
            method=self.method)

    def _cert_selection(self, connection):  # pragma: no cover
        """Callback selecting certificate for connection."""
        server_name = connection.get_servername()
        return self.certs.get(server_name, None)

    def server_bind(self):
        self._wrap_sock()
        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


class BaseDualNetworkedServers(object):
    """Base class for a pair of IPv6 and IPv4 servers that tries to do everything
       it's asked for both servers, but where failures in one server don't
       affect the other.

       If two servers are instantiated, they will serve on the same port.
       """

    def __init__(self, ServerClass, server_address, *remaining_args, **kwargs):
        port = server_address[1]
        self.threads = [] # type: List[threading.Thread]
        self.servers = [] # type: List[ACMEServerMixin]

        # Must try True first.
        # Ubuntu, for example, will fail to bind to IPv4 if we've already bound
        # to IPv6. But that's ok, since it will accept IPv4 connections on the IPv6
        # socket. On the other hand, FreeBSD will successfully bind to IPv4 on the
        # same port, which means that server will accept the IPv4 connections.
        # If Python is compiled without IPv6, we'll error out but (probably) successfully
        # create the IPv4 server.
        for ip_version in [True, False]:
            try:
                kwargs["ipv6"] = ip_version
                new_address = (server_address[0],) + (port,) + server_address[2:]
                new_args = (new_address,) + remaining_args
                server = ServerClass(*new_args, **kwargs)
                logger.debug(
                    "Successfully bound to %s:%s using %s", new_address[0],
                    new_address[1], "IPv6" if ip_version else "IPv4")
            except socket.error:
                if self.servers:
                    # Already bound using IPv6.
                    logger.debug(
                        "Certbot wasn't able to bind to %s:%s using %s, this "
                        "is often expected due to the dual stack nature of "
                        "IPv6 socket implementations.",
                        new_address[0], new_address[1],
                        "IPv6" if ip_version else "IPv4")
                else:
                    logger.debug(
                        "Failed to bind to %s:%s using %s", new_address[0],
                        new_address[1], "IPv6" if ip_version else "IPv4")
            else:
                self.servers.append(server)
                # If two servers are set up and port 0 was passed i