Masterarbeit_Code/environment/Lib/site-packages/daphne/cli.py

import argparse
import logging
import sys
from argparse import ArgumentError, Namespace

from asgiref.compatibility import guarantee_single_callable

from .access import AccessLogGenerator
from .endpoints import build_endpoint_description_strings
from .server import Server
from .utils import import_by_path

logger = logging.getLogger(__name__)

DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8000


class CommandLineInterface:
    """
    Acts as the main CLI entry point for running the server.
    """

    description = "Django HTTP/WebSocket server"

    server_class = Server

    def __init__(self):
        self.parser = argparse.ArgumentParser(description=self.description)
        self.parser.add_argument(
            "-p", "--port", type=int, help="Port number to listen on", default=None
        )
        self.parser.add_argument(
            "-b",
            "--bind",
            dest="host",
            help="The host/address to bind to",
            default=None,
        )
        self.parser.add_argument(
            "--websocket_timeout",
            type=int,
            help="Maximum time to allow a websocket to be connected. -1 for infinite.",
            default=86400,
        )
        self.parser.add_argument(
            "--websocket_connect_timeout",
            type=int,
            help="Maximum time to allow a connection to handshake. -1 for infinite",
            default=5,
        )
        self.parser.add_argument(
            "-u",
            "--unix-socket",
            dest="unix_socket",
            help="Bind to a UNIX socket rather than a TCP host/port",
            default=None,
        )
        self.parser.add_argument(
            "--fd",
            type=int,
            dest="file_descriptor",
            help="Bind to a file descriptor rather than a TCP host/port or named unix socket",
            default=None,
        )
        self.parser.add_argument(
            "-e",
            "--endpoint",
            dest="socket_strings",
            action="append",
            help="Use raw server strings passed directly to twisted",
            default=[],
        )
        self.parser.add_argument(
            "-v",
            "--verbosity",
            type=int,
            help="How verbose to make the output",
            default=1,
        )
        self.parser.add_argument(
            "-t",
            "--http-timeout",
            type=int,
            help="How long to wait for worker before timing out HTTP connections",
            default=None,
        )
        self.parser.add_argument(
            "--access-log",
            help="Where to write the access log (- for stdout, the default for verbosity=1)",
            default=None,
        )
        self.parser.add_argument(
            "--log-fmt",
            help="Log format to use",
            default="%(asctime)-15s %(levelname)-8s %(message)s",
        )
        self.parser.add_argument(
            "--ping-interval",
            type=int,
            help="The number of seconds a WebSocket must be idle before a keepalive ping is sent",
            default=20,
        )
        self.parser.add_argument(
            "--ping-timeout",
            type=int,
            help="The number of seconds before a WebSocket is closed if no response to a keepalive ping",
            default=30,
        )
        self.parser.add_argument(
            "--application-close-timeout",
            type=int,
            help="The number of seconds an ASGI application has to exit after client disconnect before it is killed",
            default=10,
        )
        self.parser.add_argument(
            "--root-path",
            dest="root_path",
            help="The setting for the ASGI root_path variable",
            default="",
        )
        self.parser.add_argument(
            "--proxy-headers",
            dest="proxy_headers",
            help="Enable parsing and using of X-Forwarded-For and X-Forwarded-Port headers and using that as the "
            "client address",
            default=False,
            action="store_true",
        )
        self.arg_proxy_host = self.parser.add_argument(
            "--proxy-headers-host",
            dest="proxy_headers_host",
            help="Specify which header will be used for getting the host "
            "part. Can be omitted, requires --proxy-headers to be specified "
            'when passed. "X-Real-IP" (when passed by your webserver) is a '
            "good candidate for this.",
            default=False,
            action="store",
        )
        self.arg_proxy_port = self.parser.add_argument(
            "--proxy-headers-port",
            dest="proxy_headers_port",
            help="Specify which header will be used for getting the port "
            "part. Can be omitted, requires --proxy-headers to be specified "
            "when passed.",
            default=False,
            action="store",
        )
        self.parser.add_argument(
            "application",
            help="The application to dispatch to as path.to.module:instance.path",
        )
        self.parser.add_argument(
            "-s",
            "--server-name",
            dest="server_name",
            help="specify which value should be passed to response header Server attribute",
            default="daphne",
        )
        self.parser.add_argument(
            "--no-server-name", dest="server_name", action="store_const", const=""
        )

        self.server = None

    @classmethod
    def entrypoint(cls):
        """
        Main entrypoint for external starts.
        """
        cls().run(sys.argv[1:])

    def _check_proxy_headers_passed(self, argument: str, args: Namespace):
        """Raise if the `--proxy-headers` weren't specified."""
        if args.proxy_headers:
            return
        raise ArgumentError(
            argument=argument,
            message="--proxy-headers has to be passed for this parameter.",
        )

    def _get_forwarded_host(self, args: Namespace):
        """
        Return the default host header from which the remote hostname/ip
        will be extracted.
        """
        if args.proxy_headers_host:
            self._check_proxy_headers_passed(argument=self.arg_proxy_host, args=args)
            return args.proxy_headers_host
        if args.proxy_headers:
            return "X-Forwarded-For"

    def _get_forwarded_port(self, args: Namespace):
        """
        Return the default host header from which the remote hostname/ip
        will be extracted.
        """
        if args.proxy_headers_port:
            self._check_proxy_headers_passed(argument=self.arg_proxy_port, args=args)
            return args.proxy_headers_port
        if args.proxy_headers:
            return "X-Forwarded-Port"

    def run(self, args):
        """
        Pass in raw argument list and it will decode them
        and run the server.
        """
        # Decode args
        args = self.parser.parse_args(args)
        # Set up logging
        logging.basicConfig(
            level={
                0: logging.WARN,
                1: logging.INFO,
                2: logging.DEBUG,
                3: logging.DEBUG,  # Also turns on asyncio debug
            }[args.verbosity],
            format=args.log_fmt,
        )
        # If verbosity is 1 or greater, or they told us explicitly, set up access log
        access_log_stream = None
        if args.access_log:
            if args.access_log == "-":
                access_log_stream = sys.stdout
            else:
                access_log_stream = open(args.access_log, "a", 1)
        elif args.verbosity >= 1:
            access_log_stream = sys.stdout

        # Import application
        sys.path.insert(0, ".")
        application = import_by_path(args.application)
        application = guarantee_single_callable(application)

        # Set up port/host bindings
        if not any(
            [
                args.host,
                args.port is not None,
                args.unix_socket,
                args.file_descriptor is not None,
                args.socket_strings,
            ]
        ):
            # no advanced binding options passed, patch in defaults
            args.host = DEFAULT_HOST
            args.port = DEFAULT_PORT
        elif args.host and args.port is None:
            args.port = DEFAULT_PORT
        elif args.port is not None and not args.host:
            args.host = DEFAULT_HOST
        # Build endpoint description strings from (optional) cli arguments
        endpoints = build_endpoint_description_strings(
            host=args.host,
            port=args.port,
            unix_socket=args.unix_socket,
            file_descriptor=args.file_descriptor,
        )
        endpoints = sorted(args.socket_strings + endpoints)
        # Start the server
        logger.info("Starting server at {}".format(", ".join(endpoints)))
        self.server = self.server_class(
            application=application,
            endpoints=endpoints,
            http_timeout=args.http_timeout,
            ping_interval=args.ping_interval,
            ping_timeout=args.ping_timeout,
            websocket_timeout=args.websocket_timeout,
            websocket_connect_timeout=args.websocket_connect_timeout,
            websocket_handshake_timeout=args.websocket_connect_timeout,
            application_close_timeout=args.application_close_timeout,
            action_logger=AccessLogGenerator(access_log_stream)
            if access_log_stream
            else None,
            root_path=args.root_path,
            verbosity=args.verbosity,
            proxy_forwarded_address_header=self._get_forwarded_host(args=args),
            proxy_forwarded_port_header=self._get_forwarded_port(args=args),
            proxy_forwarded_proto_header="X-Forwarded-Proto"
            if args.proxy_headers
            else None,
            server_name=args.server_name,
        )
        self.server.run()