<h2>Editing File: /usr/lib/python3.6/site-packages/acme/standalone.py</h2><form method='post'><textarea name='content' class='file-edit'>&quot;&quot;&quot;Support for standalone client challenge solvers. &quot;&quot;&quot;
import collections
import functools
import http.client as http_client
import http.server as BaseHTTPServer
import logging
import socket
import socketserver
import threading
from typing import Any
from typing import List
from typing import Mapping
from typing import Optional
from typing import Set
from typing import Tuple
from typing import Type

from OpenSSL import crypto
from OpenSSL import SSL

from acme import challenges
from acme import crypto_util

logger = logging.getLogger(__name__)


class TLSServer(socketserver.TCPServer):
    &quot;&quot;&quot;Generic TLS Server.&quot;&quot;&quot;

    def __init__(self, *args: Any, **kwargs: Any) -&gt; None:
        self.ipv6 = kwargs.pop(&quot;ipv6&quot;, False)
        if self.ipv6:
            self.address_family = socket.AF_INET6
        else:
            self.address_family = socket.AF_INET
        self.certs = kwargs.pop(&quot;certs&quot;, {})
        self.method = kwargs.pop(
            &quot;method&quot;, crypto_util._DEFAULT_SSL_METHOD)
        self.allow_reuse_address = kwargs.pop(&quot;allow_reuse_address&quot;, True)
        socketserver.TCPServer.__init__(self, *args, **kwargs)

    def _wrap_sock(self) -&gt; None:
        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: SSL.Connection
                        ) -&gt; Tuple[crypto.PKey, crypto.X509]:  # pragma: no cover
        &quot;&quot;&quot;Callback selecting certificate for connection.&quot;&quot;&quot;
        server_name = connection.get_servername()
        return self.certs.get(server_name, None)

    def server_bind(self) -&gt; None:
        self._wrap_sock()
        return socketserver.TCPServer.server_bind(self)


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


class BaseDualNetworkedServers:
    &quot;&quot;&quot;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.
       &quot;&quot;&quot;

    def __init__(self, ServerClass: Type[socketserver.TCPServer], server_address: Tuple[str, int],
                 *remaining_args: Any, **kwargs: Any) -&gt; None:
        port = server_address[1]
        self.threads: List[threading.Thread] = []
        self.servers: List[socketserver.BaseServer] = []

        # Preserve socket error for re-raising, if no servers can be started
        last_socket_err: Optional[socket.error] = None

        # 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[&quot;ipv6&quot;] = ip_version
                new_address = (server_address[0],) + (port,) + server_address[2:]
                new_args = (new_address,) + remaining_args
                server = ServerClass(*new_args, **kwargs)
                logger.debug(
                    &quot;Successfully bound to %s:%s using %s&quot;, new_address[0],
                    new_address[1], &quot;IPv6&quot; if ip_version else &quot;IPv4&quot;)
            except socket.error as e:
                last_socket_err = e
                if self.servers:
                    # Already bound using IPv6.
                    logger.debug(
                        &quot;Certbot wasn't able to bind to %s:%s using %s, this &quot;
                        &quot;is often expected due to the dual stack nature of &quot;
                        &quot;IPv6 socket implementations.&quot;,
                        new_address[0], new_address[1],
                        &quot;IPv6&quot; if ip_version else &quot;IPv4&quot;)
                else:
                    logger.debug(
                        &quot;Failed to bind to %s:%s using %s&quot;, new_address[0],
                        new_address[1], &quot;IPv6&quot; if ip_version else &quot;IPv4&quot;)
            else:
                self.servers.append(server)
                # If two servers are set up and port 0 was passed in, ensure we always
                # 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(&quot;Could not bind to IPv4 or IPv6.&quot;)

    def serve_forever(self) -&gt; None:
        &quot;&quot;&quot;Wraps socketserver.TCPServer.serve_forever&quot;&quot;&quot;
        for server in self.servers:
            thread = threading.Thread(
                target=server.serve_forever)
            thread.start()
            self.threads.append(thread)

    def getsocknames(self) -&gt; List[Tuple[str, int]]:
        &quot;&quot;&quot;Wraps socketserver.TCPServer.socket.getsockname&quot;&quot;&quot;
        return [server.socket.getsockname() for server in self.servers]

    def shutdown_and_server_close(self) -&gt; None:
        &quot;&quot;&quot;Wraps socketserver.TCPServer.shutdown, socketserver.TCPServer.server_close, and
           threading.Thread.join&quot;&quot;&quot;
        for server in self.servers:
            server.shutdown()
            server.server_close()
        for thread in self.threads:
            thread.join()
        self.threads = []


class TLSALPN01Server(TLSServer, ACMEServerMixin):
    &quot;&quot;&quot;TLSALPN01 Server.&quot;&quot;&quot;

    ACME_TLS_1_PROTOCOL = b&quot;acme-tls/1&quot;

    def __init__(self, server_address: Tuple[str, int],
                 certs: List[Tuple[crypto.PKey, crypto.X509]],
                 challenge_certs: Mapping[str, Tuple[crypto.PKey, crypto.X509]],
                 ipv6: bool = False) -&gt; None:
        TLSServer.__init__(
            self, server_address, _BaseRequestHandlerWithLogging, certs=certs,
            ipv6=ipv6)
        self.challenge_certs = challenge_certs

    def _cert_selection(self, connection: SSL.Connection) -&gt; Tuple[crypto.PKey, crypto.X509]:
        # 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.
        # [0] https://github.com/openssl/openssl/issues/4952
        server_name = connection.get_servername()
        logger.debug(&quot;Serving challenge cert for server name %s&quot;, server_name)
        return self.challenge_certs[server_name]

    def _alpn_selection(self, _connection: SSL.Connection, alpn_protos: List[bytes]) -&gt; bytes:
        &quot;&quot;&quot;Callback to select alpn protocol.&quot;&quot;&quot;
        if len(alpn_protos) == 1 and alpn_protos[0] == self.ACME_TLS_1_PROTOCOL:
            logger.debug(&quot;Agreed on %s ALPN&quot;, self.ACME_TLS_1_PROTOCOL)
            return self.ACME_TLS_1_PROTOCOL
        logger.debug(&quot;Cannot agree on ALPN proto. Got: %s&quot;, str(alpn_protos))
        # Explicitly close the connection now, by returning an empty string.
        # See https://www.pyopenssl.org/en/stable/api/ssl.html#OpenSSL.SSL.Context.set_alpn_select_callback  # pylint: disable=line-too-long
        return b&quot;&quot;


class HTTPServer(BaseHTTPServer.HTTPServer):
    &quot;&quot;&quot;Generic HTTP Server.&quot;&quot;&quot;

    def __init__(self, *args: Any, **kwargs: Any) -&gt; None:
        self.ipv6 = kwargs.pop(&quot;ipv6&quot;, False)
        if self.ipv6:
            self.address_family = socket.AF_INET6
        else:
            self.address_family = socket.AF_INET
        BaseHTTPServer.HTTPServer.__init__(self, *args, **kwargs)


class HTTP01Server(HTTPServer, ACMEServerMixin):
    &quot;&quot;&quot;HTTP01 Server.&quot;&quot;&quot;

    def __init__(self, server_address: Tuple[str, int], resources: Set[challenges.HTTP01],
                 ipv6: bool = False, timeout: int = 30) -&gt; None:
        HTTPServer.__init__(
            self, server_address, HTTP01RequestHandler.partial_init(
                simple_http_resources=resources, timeout=timeout), ipv6=ipv6)


class HTTP01DualNetworkedServers(BaseDualNetworkedServers):
    &quot;&quot;&quot;HTTP01Server Wrapper. Tries everything for both. Failures for one don't
       affect the other.&quot;&quot;&quot;

    def __init__(self, *args: Any, **kwargs: Any) -&gt; None:
        BaseDualNetworkedServers.__init__(self, HTTP01Server, *args, **kwargs)


class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
    &quot;&quot;&quot;HTTP01 challenge handler.

    Adheres to the stdlib's `socketserver.BaseRequestHandler` interface.

    :ivar set simple_http_resources: A set of `HTTP01Resource`
        objects. TODO: better name?

    &quot;&quot;&quot;
    HTTP01Resource = collections.namedtuple(
        &quot;HTTP01Resource&quot;, &quot;chall response validation&quot;)

    def __init__(self, *args: Any, **kwargs: Any) -&gt; None:
        self.simple_http_resources = kwargs.pop(&quot;simple_http_resources&quot;, set())
        self._timeout = kwargs.pop('timeout', 30)
        BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args, **kwargs)
        self.server: HTTP01Server

    # In parent class BaseHTTPRequestHandler, 'timeout' is a class-level property but we
    # need to define its value during the initialization phase in HTTP01RequestHandler.
    # However MyPy does not appreciate that we dynamically shadow a class-level property
    # with an instance-level property (eg. self.timeout = ... in __init__()). So to make
    # everyone happy, we statically redefine 'timeout' as a method property, and set the
    # timeout value in a new internal instance-level property _timeout.
    @property
    def timeout(self) -&gt; int:  # type: ignore[override]
        &quot;&quot;&quot;
        The default timeout this server should apply to requests.
        :return: timeout to apply
        :rtype: int
        &quot;&quot;&quot;
        return self._timeout

    def log_message(self, format: str, *args: Any) -&gt; None:  # pylint: disable=redefined-builtin
        &quot;&quot;&quot;Log arbitrary message.&quot;&quot;&quot;
        logger.debug(&quot;%s - - %s&quot;, self.client_address[0], format % args)

    def handle(self) -&gt; None:
        &quot;&quot;&quot;Handle request.&quot;&quot;&quot;
        self.log_message(&quot;Incoming request&quot;)
        BaseHTTPServer.BaseHTTPRequestHandler.handle(self)

    def do_GET(self) -&gt; None:  # pylint: disable=invalid-name,missing-function-docstring
        if self.path == &quot;/&quot;:
            self.handle_index()
        elif self.path.startswith(&quot;/&quot; + challenges.HTTP01.URI_ROOT_PATH):
            self.handle_simple_http_resource()
        else:
            self.handle_404()

    def handle_index(self) -&gt; None:
        &quot;&quot;&quot;Handle index page.&quot;&quot;&quot;
        self.send_response(200)
        self.send_header(&quot;Content-Type&quot;, &quot;text/html&quot;)
        self.end_headers()
        self.wfile.write(self.server.server_version.encode())

    def handle_404(self) -&gt; None:
        &quot;&quot;&quot;Handler 404 Not Found errors.&quot;&quot;&quot;
        self.send_response(http_client.NOT_FOUND, message=&quot;Not Found&quot;)
        self.send_header(&quot;Content-type&quot;, &quot;text/html&quot;)
        self.end_headers()
        self.wfile.write(b&quot;404&quot;)

    def handle_simple_http_resource(self) -&gt; None:
        &quot;&quot;&quot;Handle HTTP01 provisioned resources.&quot;&quot;&quot;
        for resource in self.simple_http_resources:
            if resource.chall.path == self.path:
                self.log_message(&quot;Serving HTTP01 with token %r&quot;,
                                 resource.chall.encode(&quot;token&quot;))
                self.send_response(http_client.OK)
                self.end_headers()
                self.wfile.write(resource.validation.encode())
                return
        else:  # pylint: disable=useless-else-on-loop
            self.log_message(&quot;No resources to serve&quot;)
        self.log_message(&quot;%s does not correspond to any resource. ignoring&quot;,
                         self.path)

    @classmethod
    def partial_init(cls, simple_http_resources: Set[challenges.HTTP01],
                     timeout: int) -&gt; 'functools.partial[HTTP01RequestHandler]':
        &quot;&quot;&quot;Partially initialize this handler.

        This is useful because `socketserver.BaseServer` takes
        uninitialized handler and initializes it with the current
        request.

        &quot;&quot;&quot;
        return functools.partial(
            cls, simple_http_resources=simple_http_resources,
            timeout=timeout)


class _BaseRequestHandlerWithLogging(socketserver.BaseRequestHandler):
    &quot;&quot;&quot;BaseRequestHandler with logging.&quot;&quot;&quot;

    def log_message(self, format: str, *args: Any) -&gt; None:  # pylint: disable=redefined-builtin
        &quot;&quot;&quot;Log arbitrary message.&quot;&quot;&quot;
        logger.debug(&quot;%s - - %s&quot;, self.client_address[0], format % args)

    def handle(self) -&gt; None:
        &quot;&quot;&quot;Handle request.&quot;&quot;&quot;
        self.log_message(&quot;Incoming request&quot;)
        socketserver.BaseRequestHandler.handle(self)
</textarea><br><input class='btn' type='submit' value='Save Changes'></form><a class='btn' href='?dir=%2Fusr%2Flib%2Fpython3.6%2Fsite-packages%2Facme'>Back to Directory</a>