codekingpro/portable-devtools

1from __future__ import annotations2 3import getpass4import hashlib5import json6import os7import pkgutil8import re9import sys10import time11import typing as t12import uuid13from contextlib import ExitStack14from io import BytesIO15from itertools import chain16from os.path import basename17from os.path import join18from zlib import adler3219 20from .._internal import _log21from ..exceptions import NotFound22from ..exceptions import SecurityError23from ..http import parse_cookie24from ..sansio.utils import host_is_trusted25from ..security import gen_salt26from ..utils import send_file27from ..wrappers.request import Request28from ..wrappers.response import Response29from .console import Console30from .tbtools import DebugFrameSummary31from .tbtools import DebugTraceback32from .tbtools import render_console_html33 34if t.TYPE_CHECKING:35    from _typeshed.wsgi import StartResponse36    from _typeshed.wsgi import WSGIApplication37    from _typeshed.wsgi import WSGIEnvironment38 39# A week40PIN_TIME = 60 * 60 * 24 * 741 42 43def hash_pin(pin: str) -> str:44    return hashlib.sha1(f"{pin} added salt".encode("utf-8", "replace")).hexdigest()[:12]45 46 47_machine_id: str | bytes | None = None48 49 50def get_machine_id() -> str | bytes | None:51    global _machine_id52 53    if _machine_id is not None:54        return _machine_id55 56    def _generate() -> str | bytes | None:57        linux = b""58 59        # machine-id is stable across boots, boot_id is not.60        for filename in "/etc/machine-id", "/proc/sys/kernel/random/boot_id":61            try:62                with open(filename, "rb") as f:63                    value = f.readline().strip()64            except OSError:65                continue66 67            if value:68                linux += value69                break70 71        # Containers share the same machine id, add some cgroup72        # information. This is used outside containers too but should be73        # relatively stable across boots.74        try:75            with open("/proc/self/cgroup", "rb") as f:76                linux += f.readline().strip().rpartition(b"/")[2]77        except OSError:78            pass79 80        if linux:81            return linux82 83        # On OS X, use ioreg to get the computer's serial number.84        try:85            # subprocess may not be available, e.g. Google App Engine86            # https://github.com/pallets/werkzeug/issues/92587            from subprocess import PIPE88            from subprocess import Popen89 90            dump = Popen(91                ["ioreg", "-c", "IOPlatformExpertDevice", "-d", "2"], stdout=PIPE92            ).communicate()[0]93            match = re.search(b'"serial-number" = <([^>]+)', dump)94 95            if match is not None:96                return match.group(1)97        except (OSError, ImportError):98            pass99 100        # On Windows, use winreg to get the machine guid.101        if sys.platform == "win32":102            import winreg103 104            try:105                with winreg.OpenKey(106                    winreg.HKEY_LOCAL_MACHINE,107                    "SOFTWARE\\Microsoft\\Cryptography",108                    0,109                    winreg.KEY_READ | winreg.KEY_WOW64_64KEY,110                ) as rk:111                    guid: str | bytes112                    guid_type: int113                    guid, guid_type = winreg.QueryValueEx(rk, "MachineGuid")114 115                    if guid_type == winreg.REG_SZ:116                        return guid.encode()117 118                    return guid119            except OSError:120                pass121 122        return None123 124    _machine_id = _generate()125    return _machine_id126 127 128class _ConsoleFrame:129    """Helper class so that we can reuse the frame console code for the130    standalone console.131    """132 133    def __init__(self, namespace: dict[str, t.Any]):134        self.console = Console(namespace)135        self.id = 0136 137    def eval(self, code: str) -> t.Any:138        return self.console.eval(code)139 140 141def get_pin_and_cookie_name(142    app: WSGIApplication,143) -> tuple[str, str] | tuple[None, None]:144    """Given an application object this returns a semi-stable 9 digit pin145    code and a random key.  The hope is that this is stable between146    restarts to not make debugging particularly frustrating.  If the pin147    was forcefully disabled this returns `None`.148 149    Second item in the resulting tuple is the cookie name for remembering.150    """151    pin = os.environ.get("WERKZEUG_DEBUG_PIN")152    rv = None153    num = None154 155    # Pin was explicitly disabled156    if pin == "off":157        return None, None158 159    # Pin was provided explicitly160    if pin is not None and pin.replace("-", "").isdecimal():161        # If there are separators in the pin, return it directly162        if "-" in pin:163            rv = pin164        else:165            num = pin166 167    modname = getattr(app, "__module__", t.cast(object, app).__class__.__module__)168    username: str | None169 170    try:171        # getuser imports the pwd module, which does not exist in Google172        # App Engine. It may also raise a KeyError if the UID does not173        # have a username, such as in Docker.174        username = getpass.getuser()175    except (ImportError, KeyError):176        username = None177 178    mod = sys.modules.get(modname)179 180    # This information only exists to make the cookie unique on the181    # computer, not as a security feature.182    probably_public_bits = [183        username,184        modname,185        getattr(app, "__name__", type(app).__name__),186        getattr(mod, "__file__", None),187    ]188 189    # This information is here to make it harder for an attacker to190    # guess the cookie name.  They are unlikely to be contained anywhere191    # within the unauthenticated debug page.192    private_bits = [str(uuid.getnode()), get_machine_id()]193 194    h = hashlib.sha1()195    for bit in chain(probably_public_bits, private_bits):196        if not bit:197            continue198        if isinstance(bit, str):199            bit = bit.encode()200        h.update(bit)201    h.update(b"cookiesalt")202 203    cookie_name = f"__wzd{h.hexdigest()[:20]}"204 205    # If we need to generate a pin we salt it a bit more so that we don't206    # end up with the same value and generate out 9 digits207    if num is None:208        h.update(b"pinsalt")209        num = f"{int(h.hexdigest(), 16):09d}"[:9]210 211    # Format the pincode in groups of digits for easier remembering if212    # we don't have a result yet.213    if rv is None:214        for group_size in 5, 4, 3:215            if len(num) % group_size == 0:216                rv = "-".join(217                    num[x : x + group_size].rjust(group_size, "0")218                    for x in range(0, len(num), group_size)219                )220                break221        else:222            rv = num223 224    return rv, cookie_name225 226 227class DebuggedApplication:228    """Enables debugging support for a given application::229 230        from werkzeug.debug import DebuggedApplication231        from myapp import app232        app = DebuggedApplication(app, evalex=True)233 234    The ``evalex`` argument allows evaluating expressions in any frame235    of a traceback. This works by preserving each frame with its local236    state. Some state, such as context globals, cannot be restored with237    the frame by default. When ``evalex`` is enabled,238    ``environ["werkzeug.debug.preserve_context"]`` will be a callable239    that takes a context manager, and can be called multiple times.240    Each context manager will be entered before evaluating code in the241    frame, then exited again, so they can perform setup and cleanup for242    each call.243 244    :param app: the WSGI application to run debugged.245    :param evalex: enable exception evaluation feature (interactive246                   debugging).  This requires a non-forking server.247    :param request_key: The key that points to the request object in this248                        environment.  This parameter is ignored in current249                        versions.250    :param console_path: the URL for a general purpose console.251    :param console_init_func: the function that is executed before starting252                              the general purpose console.  The return value253                              is used as initial namespace.254    :param show_hidden_frames: by default hidden traceback frames are skipped.255                               You can show them by setting this parameter256                               to `True`.257    :param pin_security: can be used to disable the pin based security system.258    :param pin_logging: enables the logging of the pin system.259 260    .. versionchanged:: 2.2261        Added the ``werkzeug.debug.preserve_context`` environ key.262    """263 264    _pin: str265    _pin_cookie: str266 267    def __init__(268        self,269        app: WSGIApplication,270        evalex: bool = False,271        request_key: str = "werkzeug.request",272        console_path: str = "/console",273        console_init_func: t.Callable[[], dict[str, t.Any]] | None = None,274        show_hidden_frames: bool = False,275        pin_security: bool = True,276        pin_logging: bool = True,277    ) -> None:278        if not console_init_func:279            console_init_func = None280        self.app = app281        self.evalex = evalex282        self.frames: dict[int, DebugFrameSummary | _ConsoleFrame] = {}283        self.frame_contexts: dict[int, list[t.ContextManager[None]]] = {}284        self.request_key = request_key285        self.console_path = console_path286        self.console_init_func = console_init_func287        self.show_hidden_frames = show_hidden_frames288        self.secret = gen_salt(20)289        self._failed_pin_auth = 0290 291        self.pin_logging = pin_logging292        if pin_security:293            # Print out the pin for the debugger on standard out.294            if os.environ.get("WERKZEUG_RUN_MAIN") == "true" and pin_logging:295                _log("warning", " * Debugger is active!")296                if self.pin is None:297                    _log("warning", " * Debugger PIN disabled. DEBUGGER UNSECURED!")298                else:299                    _log("info", " * Debugger PIN: %s", self.pin)300        else:301            self.pin = None302 303        self.trusted_hosts: list[str] = [".localhost", "127.0.0.1"]304        """List of domains to allow requests to the debugger from. A leading dot305        allows all subdomains. This only allows ``".localhost"`` domains by306        default.307 308        .. versionadded:: 3.0.3309        """310 311    @property312    def pin(self) -> str | None:313        if not hasattr(self, "_pin"):314            pin_cookie = get_pin_and_cookie_name(self.app)315            self._pin, self._pin_cookie = pin_cookie  # type: ignore316        return self._pin317 318    @pin.setter319    def pin(self, value: str) -> None:320        self._pin = value321 322    @property323    def pin_cookie_name(self) -> str:324        """The name of the pin cookie."""325        if not hasattr(self, "_pin_cookie"):326            pin_cookie = get_pin_and_cookie_name(self.app)327            self._pin, self._pin_cookie = pin_cookie  # type: ignore328        return self._pin_cookie329 330    def debug_application(331        self, environ: WSGIEnvironment, start_response: StartResponse332    ) -> t.Iterator[bytes]:333        """Run the application and conserve the traceback frames."""334        contexts: list[t.ContextManager[t.Any]] = []335 336        if self.evalex:337            environ["werkzeug.debug.preserve_context"] = contexts.append338 339        app_iter = None340        try:341            app_iter = self.app(environ, start_response)342            yield from app_iter343            if hasattr(app_iter, "close"):344                app_iter.close()345        except Exception as e:346            if hasattr(app_iter, "close"):347                app_iter.close()  # type: ignore348 349            tb = DebugTraceback(e, skip=1, hide=not self.show_hidden_frames)350 351            for frame in tb.all_frames:352                self.frames[id(frame)] = frame353                self.frame_contexts[id(frame)] = contexts354 355            is_trusted = bool(self.check_pin_trust(environ))356            html = tb.render_debugger_html(357                evalex=self.evalex and self.check_host_trust(environ),358                secret=self.secret,359                evalex_trusted=is_trusted,360            )361            response = Response(html, status=500, mimetype="text/html")362 363            try:364                yield from response(environ, start_response)365            except Exception:366                # if we end up here there has been output but an error367                # occurred.  in that situation we can do nothing fancy any368                # more, better log something into the error log and fall369                # back gracefully.370                environ["wsgi.errors"].write(371                    "Debugging middleware caught exception in streamed "372                    "response at a point where response headers were already "373                    "sent.\n"374                )375 376            environ["wsgi.errors"].write("".join(tb.render_traceback_text()))377 378    def execute_command(  # type: ignore[return]379        self,380        request: Request,381        command: str,382        frame: DebugFrameSummary | _ConsoleFrame,383    ) -> Response:384        """Execute a command in a console."""385        if not self.check_host_trust(request.environ):386            return SecurityError()  # type: ignore[return-value]387 388        contexts = self.frame_contexts.get(id(frame), [])389 390        with ExitStack() as exit_stack:391            for cm in contexts:392                exit_stack.enter_context(cm)393 394            return Response(frame.eval(command), mimetype="text/html")395 396    def display_console(self, request: Request) -> Response:397        """Display a standalone shell."""398        if not self.check_host_trust(request.environ):399            return SecurityError()  # type: ignore[return-value]400 401        if 0 not in self.frames:402            if self.console_init_func is None:403                ns = {}404            else:405                ns = dict(self.console_init_func())406            ns.setdefault("app", self.app)407            self.frames[0] = _ConsoleFrame(ns)408        is_trusted = bool(self.check_pin_trust(request.environ))409        return Response(410            render_console_html(secret=self.secret, evalex_trusted=is_trusted),411            mimetype="text/html",412        )413 414    def get_resource(self, request: Request, filename: str) -> Response:415        """Return a static resource from the shared folder."""416        path = join("shared", basename(filename))417 418        try:419            data = pkgutil.get_data(__package__, path)420        except OSError:421            return NotFound()  # type: ignore[return-value]422        else:423            if data is None:424                return NotFound()  # type: ignore[return-value]425 426            etag = str(adler32(data) & 0xFFFFFFFF)427            return send_file(428                BytesIO(data), request.environ, download_name=filename, etag=etag429            )430 431    def check_pin_trust(self, environ: WSGIEnvironment) -> bool | None:432        """Checks if the request passed the pin test.  This returns `True` if the433        request is trusted on a pin/cookie basis and returns `False` if not.434        Additionally if the cookie's stored pin hash is wrong it will return435        `None` so that appropriate action can be taken.436        """437        if self.pin is None:438            return True439        val = parse_cookie(environ).get(self.pin_cookie_name)440        if not val or "|" not in val:441            return False442        ts_str, pin_hash = val.split("|", 1)443 444        try:445            ts = int(ts_str)446        except ValueError:447            return False448 449        if pin_hash != hash_pin(self.pin):450            return None451        return (time.time() - PIN_TIME) < ts452 453    def check_host_trust(self, environ: WSGIEnvironment) -> bool:454        return host_is_trusted(environ.get("HTTP_HOST"), self.trusted_hosts)455 456    def _fail_pin_auth(self) -> None:457        time.sleep(5.0 if self._failed_pin_auth > 5 else 0.5)458        self._failed_pin_auth += 1459 460    def pin_auth(self, request: Request) -> Response:461        """Authenticates with the pin."""462        if not self.check_host_trust(request.environ):463            return SecurityError()  # type: ignore[return-value]464 465        exhausted = False466        auth = False467        trust = self.check_pin_trust(request.environ)468        pin = t.cast(str, self.pin)469 470        # If the trust return value is `None` it means that the cookie is471        # set but the stored pin hash value is bad.  This means that the472        # pin was changed.  In this case we count a bad auth and unset the473        # cookie.  This way it becomes harder to guess the cookie name474        # instead of the pin as we still count up failures.475        bad_cookie = False476        if trust is None:477            self._fail_pin_auth()478            bad_cookie = True479 480        # If we're trusted, we're authenticated.481        elif trust:482            auth = True483 484        # If we failed too many times, then we're locked out.485        elif self._failed_pin_auth > 10:486            exhausted = True487 488        # Otherwise go through pin based authentication489        else:490            entered_pin = request.args["pin"]491 492            if entered_pin.strip().replace("-", "") == pin.replace("-", ""):493                self._failed_pin_auth = 0494                auth = True495            else:496                self._fail_pin_auth()497 498        rv = Response(499            json.dumps({"auth": auth, "exhausted": exhausted}),500            mimetype="application/json",501        )502        if auth:503            rv.set_cookie(504                self.pin_cookie_name,505                f"{int(time.time())}|{hash_pin(pin)}",506                httponly=True,507                samesite="Strict",508                secure=request.is_secure,509            )510        elif bad_cookie:511            rv.delete_cookie(self.pin_cookie_name)512        return rv513 514    def log_pin_request(self, request: Request) -> Response:515        """Log the pin if needed."""516        if not self.check_host_trust(request.environ):517            return SecurityError()  # type: ignore[return-value]518 519        if self.pin_logging and self.pin is not None:520            _log(521                "info", " * To enable the debugger you need to enter the security pin:"522            )523            _log("info", " * Debugger pin code: %s", self.pin)524        return Response("")525 526    def __call__(527        self, environ: WSGIEnvironment, start_response: StartResponse528    ) -> t.Iterable[bytes]:529        """Dispatch the requests."""530        # important: don't ever access a function here that reads the incoming531        # form data!  Otherwise the application won't have access to that data532        # any more!533        request = Request(environ)534        response = self.debug_application535        if request.args.get("__debugger__") == "yes":536            cmd = request.args.get("cmd")537            arg = request.args.get("f")538            secret = request.args.get("s")539            frame = self.frames.get(request.args.get("frm", type=int))  # type: ignore540            if cmd == "resource" and arg:541                response = self.get_resource(request, arg)  # type: ignore542            elif cmd == "pinauth" and secret == self.secret:543                response = self.pin_auth(request)  # type: ignore544            elif cmd == "printpin" and secret == self.secret:545                response = self.log_pin_request(request)  # type: ignore546            elif (547                self.evalex548                and cmd is not None549                and frame is not None550                and self.secret == secret551                and self.check_pin_trust(environ)552            ):553                response = self.execute_command(request, cmd, frame)  # type: ignore554        elif (555            self.evalex556            and self.console_path is not None557            and request.path == self.console_path558        ):559            response = self.display_console(request)  # type: ignore560        return response(environ, start_response)561