Implement PIN based authentication for the debugger and remove source…

… view.

werkzeug/_compat.py

@@ -13,6 +13,7 @@
 
 
 PY2 = sys.version_info[0] == 2
+WIN = sys.platform.startswith('win')
 
 _identity = lambda x: x
 

werkzeug/debug/__init__.py

@@ -8,13 +8,22 @@
     :copyright: (c) 2014 by the Werkzeug Team, see AUTHORS for more details.
     :license: BSD, see LICENSE for more details.
 """
+import os
+import sys
+import uuid
 import json
+import time
+import getpass
+import hashlib
 import mimetypes
 from os.path import join, dirname, basename, isfile
 from werkzeug.wrappers import BaseRequest as Request, BaseResponse as Response
+from werkzeug.http import parse_cookie
 from werkzeug.debug.tbtools import get_current_traceback, render_console_html
 from werkzeug.debug.console import Console
 from werkzeug.security import gen_salt
+from werkzeug._internal import _log
+from werkzeug._compat import text_type
 
 
 # DEPRECATED
@@ -23,6 +32,9 @@
 from werkzeug.debug.repr import debug_repr  # noqa
 
 
+PIN_TIME = 60 * 60 * 8
+
+
 class _ConsoleFrame(object):
 
     """Helper class so that we can reuse the frame console code for the
@@ -34,6 +46,70 @@ def __init__(self, namespace):
         self.id = 0
 
 
+def get_pin_and_cookie_name(app):
+    """Given an application object this returns a semi-stable 9 digit pin
+    code and a random key.  The hope is that this is stable between
+    restarts to not make debugging particularly frustrating.  If the pin
+    was forcefully disabled this returns `None`.
+
+    Second item in the resulting tuple is the cookie name for remembering.
+    """
+    pin = os.environ.get('WERKZEUG_DEBUG_PIN')
+    rv = None
+    num = None
+
+    # Pin was explicitly disabled
+    if pin == 'off':
+        return None, None
+
+    # Pin was provided explicitly
+    if pin is not None and pin.replace('-', '').isdigit():
+        # If there are separators in the pin, return it directly
+        if '-' in pin:
+            rv = pin
+        else:
+            num = pin
+
+    modname = getattr(app, '__module__',
+                      getattr(app.__class__, '__module__'))
+    bits = [
+        getpass.getuser(),
+        str(uuid.getnode()),
+        modname,
+        getattr(app, '__name__', getattr(app.__class__, '__name__')),
+    ]
+
+    mod = sys.modules.get(modname)
+    bits.append(getattr(mod, '__file__', None))
+
+    h = hashlib.md5()
+    for bit in bits:
+        if not bit:
+            continue
+        if isinstance(bit, text_type):
+            bit = bit.encode('utf-8')
+        h.update(bit)
+
+    if num is None:
+        num = ('%09d' % int(h.hexdigest(), 16))[:9]
+
+    h.update('cookiesalt')
+    cookie_name = '__wzd' + h.hexdigest()[:12]
+
+    # Format the pincode in groups of digits for easier remembering if
+    # we don't have a result yet.
+    if rv is None:
+        for group_size in 5, 4, 3:
+            if len(num) % group_size == 0:
+                rv = '-'.join(num[x:x + group_size].rjust(group_size, '0')
+                              for x in range(0, len(num), group_size))
+                break
+        else:
+            rv = num
+
+    return rv, cookie_name
+
+
 class DebuggedApplication(object):
     """Enables debugging support for a given application::
 
@@ -60,16 +136,19 @@ class DebuggedApplication(object):
     :param show_hidden_frames: by default hidden traceback frames are skipped.
                                You can show them by setting this parameter
                                to `True`.
+    :param pin_security: can be used to disable the pin based security system.
+    :param pin_logging: enables the logging of the pin system.
     """
 
     def __init__(self, app, evalex=False, request_key='werkzeug.request',
                  console_path='/console', console_init_func=None,
-                 show_hidden_frames=False, lodgeit_url=None):
+                 show_hidden_frames=False, lodgeit_url=None,
+                 pin_security=True, pin_logging=True):
         if lodgeit_url is not None:
             from warnings import warn
             warn(DeprecationWarning('Werkzeug now pastes into gists.'))
         if not console_init_func:
-            console_init_func = dict
+            console_init_func = None
         self.app = app
         self.evalex = evalex
         self.frames = {}
@@ -79,6 +158,39 @@ def __init__(self, app, evalex=False, request_key='werkzeug.request',
         self.console_init_func = console_init_func
         self.show_hidden_frames = show_hidden_frames
         self.secret = gen_salt(20)
+        self._failed_pin_auth = 0
+
+        self.pin_logging = pin_logging
+        if pin_security:
+            # Print out the pin for the debugger on standard out.
+            if os.environ.get('WERKZEUG_RUN_MAIN') == 'true' and \
+               pin_logging:
+                _log('warning', ' * Debugger is active!')
+                if self.pin is None:
+                    _log('warning', ' * Debugger pin disabled.  '
+                         'DEBUGGER UNSECURED!')
+                else:
+                    _log('info', ' * Debugger pin code: %s' % self.pin)
+        else:
+            self.pin = None
+
+    def _get_pin(self):
+        if not hasattr(self, '_pin'):
+            self._pin, self._pin_cookie = get_pin_and_cookie_name(self.app)
+        return self._pin
+
+    def _set_pin(self, value):
+        self._pin = value
+
+    pin = property(_get_pin, _set_pin)
+    del _get_pin, _set_pin
+
+    @property
+    def pin_cookie_name(self):
+        """The name of the pin cookie."""
+        if not hasattr(self, '_pin_cookie'):
+            self._pin, self._pin_cookie = get_pin_and_cookie_name(self.app)
+        return self._pin_cookie
 
     def debug_application(self, environ, start_response):
         """Run the application and conserve the traceback frames."""
@@ -116,7 +228,9 @@ def debug_application(self, environ, start_response):
                     'response at a point where response headers were already '
                     'sent.\n')
             else:
+                is_trusted = self.is_trusted(environ)
                 yield traceback.render_full(evalex=self.evalex,
+                                            evalex_trusted=is_trusted,
                                             secret=self.secret) \
                     .encode('utf-8', 'replace')
 
@@ -129,19 +243,22 @@ def execute_command(self, request, command, frame):
     def display_console(self, request):
         """Display a standalone shell."""
         if 0 not in self.frames:
-            self.frames[0] = _ConsoleFrame(self.console_init_func())
-        return Response(render_console_html(secret=self.secret),
+            if self.console_init_func is None:
+                ns = {}
+            else:
+                ns = dict(self.console_init_func())
+            ns.setdefault('app', self.app)
+            self.frames[0] = _ConsoleFrame(ns)
+        is_trusted = self.is_trusted(request.environ)
+        return Response(render_console_html(secret=self.secret,
+                                            evalex_trusted=is_trusted),
                         mimetype='text/html')
 
     def paste_traceback(self, request, traceback):
         """Paste the traceback and return a JSON response."""
         rv = traceback.paste()
         return Response(json.dumps(rv), mimetype='application/json')
 
-    def get_source(self, request, frame):
-        """Render the source viewer."""
-        return Response(frame.render_source(), mimetype='text/html')
-
     def get_resource(self, request, filename):
         """Return a static resource from the shared folder."""
         filename = join(dirname(__file__), 'shared', basename(filename))
@@ -155,6 +272,50 @@ def get_resource(self, request, filename):
                 f.close()
         return Response('Not Found', status=404)
 
+    def is_trusted(self, environ):
+        """Checks if the request passed the pin test."""
+        if self.pin is None:
+            return True
+        ts = parse_cookie(environ).get(self.pin_cookie_name, type=int)
+        if ts is None:
+            return False
+        return (time.time() - PIN_TIME) < ts
+
+    def pin_auth(self, request):
+        """Authenticates with the pin."""
+        exhausted = False
+        auth = False
+        if self.is_trusted(request.environ):
+            auth = True
+        elif self._failed_pin_auth > 10:
+            exhausted = True
+        else:
+            entered_pin = request.args.get('pin')
+            if entered_pin == self.pin:
+                self._failed_pin_auth = 0
+                auth = True
+            else:
+                time.sleep(self._failed_pin_auth > 5 and 5.0 or 0.5)
+                self._failed_pin_auth += 1
+                auth = False
+
+        rv = Response(json.dumps({
+            'auth': auth,
+            'exhausted': exhausted,
+        }), mimetype='application/json')
+        if auth:
+            rv.set_cookie(self.pin_cookie_name, str(int(time.time())),
+                          httponly=True)
+        return rv
+
+    def log_pin_request(self):
+        """Log the pin if needed."""
+        if self.pin_logging and self.pin is not None:
+            _log('info', ' * To enable the debugger you need to '
+                 'enter the security pin:')
+            _log('info', ' * Debugger pin code: %s' % self.pin)
+        return Response('')
+
     def __call__(self, environ, start_response):
         """Dispatch the requests."""
         # important: don't ever access a function here that reads the incoming
@@ -173,8 +334,10 @@ def __call__(self, environ, start_response):
             elif cmd == 'paste' and traceback is not None and \
                     secret == self.secret:
                 response = self.paste_traceback(request, traceback)
-            elif cmd == 'source' and frame and self.secret == secret:
-                response = self.get_source(request, frame)
+            elif cmd == 'pinauth' and secret == self.secret:
+                response = self.pin_auth(request)
+            elif cmd == 'printpin' and secret == self.secret:
+                response = self.log_pin_request()
             elif self.evalex and cmd is not None and frame is not None and \
                     self.secret == secret:
                 response = self.execute_command(request, cmd, frame)