allow access_tokens to bypass ban_networks check

binderhub/app.py

@@ -195,6 +195,32 @@ def _valid_badge_base_url(self, proposal):
         start the new server for the logged in user.""",
         config=True)
 
+    access_tokens = Dict(
+        key_trait=Unicode(),
+        value_trait=Unicode(),
+        help="""Dict of {'hex-encoded sha256 hash of token': 'label'}
+
+        Requests made with these tokens bypass the ban_networks origin check.
+
+        Requests should be made with the original token,
+        while the keys of this dict shall be the (hex-encoded) sha256 hashes of the tokens,
+        not the tokens themselves.
+        The values of this dict identify the tokens,
+        which are logged when authenticated requests are made.
+
+        Tokens can be passed in URL parameters with `?token={token}`
+        or in the Authorization header with:
+
+            Authorization: Bearer {token}
+
+        Create a token and its hash with e.g.:
+
+            token = secrets.token_hex(32)
+            hashed_token = hashlib.sha256(token.encode("ascii")).hexdigest()
+        """,
+        config=True,
+    )
+
     port = Integer(
         8585,
         help="""
@@ -663,6 +689,7 @@ def initialize(self, *args, **kwargs):
 
         self.tornado_settings.update(
             {
+                "access_tokens": self.access_tokens,
                 "log_function": log_request,
                 "push_secret": self.push_secret,
                 "image_prefix": self.image_prefix,

binderhub/base.py

@@ -1,5 +1,6 @@
 """Base classes for request handlers"""
 
+import hashlib
 import json
 from ipaddress import ip_address
 
@@ -29,6 +30,9 @@ def prepare(self):
 
     def check_request_ip(self):
         """Check network block list, if any"""
+        if self.current_user and self.current_user != "anonymous":
+            # don't check request ip when authenticated
+            return
         ban_networks = self.settings.get("ban_networks")
         if self.skip_check_request_ip or not ban_networks:
             return
@@ -45,8 +49,44 @@ def check_request_ip(self):
             )
             raise web.HTTPError(403, f"Requests from {message} are not allowed")
 
+    def get_hashed_token(self):
+        """Lookup access token in Authorization header or ?token= url parameter
+
+        Returns the hashed token, ready for lookup in access_tokens
+        """
+        auth_header = self.request.headers.get("Authorization")
+        token = None
+        if auth_header:
+            kind, *token = auth_header.split(maxsplit=1)
+            if token:
+                token = token[0]
+        if not token:
+            # check url param
+            token = self.get_argument("token", None)
+        if not token:
+            return
+        return hashlib.sha256(token.encode("ascii", "replace")).hexdigest()
+
+    def get_login_url(self):
+        if not self.settings['auth_enabled']:
+            # when auth is not enabled, login redirect doesn't make sense
+            raise web.HTTPError(403)
+        return super().get_login_url()
+
     def get_current_user(self):
         if not self.settings['auth_enabled']:
+            if self.settings["access_tokens"]:
+                hashed_token = self.get_hashed_token()
+                if hashed_token:
+                    owner = self.settings["access_tokens"].get(hashed_token, None)
+                    if owner:
+                        return owner
+                    else:
+                        # Don't treat bad access tokens as anonymous requests,
+                        # explicitly fail
+                        app_log.error(f"Invalid access token: {hashed_token}")
+                        return None
+            # no token, anonymous request
             return 'anonymous'
         return super().get_current_user()