Add support for direct granting of OAuth tokens and add tests

This allows a client (when authorized in a whitelist) to send direct credentials via a Basic auth header and therefore bypass the OAuth approval UI for that user.

auth/auth.py

@@ -72,24 +72,37 @@ def _validate_and_apply_oauth_token(token):
   identity_changed.send(app, identity=new_identity)
 
 
-def _process_basic_auth(auth):
+def _parse_basic_auth_header(auth):
   normalized = [part.strip() for part in auth.split(' ') if part]
   if normalized[0].lower() != 'basic' or len(normalized) != 2:
     logger.debug('Invalid basic auth format.')
-    return
+    return None
 
-  credentials = [part.decode('utf-8') for part in b64decode(normalized[1]).split(':', 1)]
+  logger.debug('Found basic auth header: %s', auth)
+  try:
+    credentials = [part.decode('utf-8') for part in b64decode(normalized[1]).split(':', 1)]
+  except TypeError:
+    logger.exception('Exception when parsing basic auth header')
+    return None
 
   if len(credentials) != 2:
     logger.debug('Invalid basic auth credential format.')
+    return None
 
-  elif credentials[0] == '$token':
+  return credentials
+
+
+def _process_basic_auth(auth):
+  credentials = _parse_basic_auth_header(auth)
+  if credentials is None:
+    return
+
+  if credentials[0] == '$token':
     # Use as token auth
     try:
       token = model.token.load_token_data(credentials[1])
       logger.debug('Successfully validated token: %s', credentials[1])
       set_validated_token(token)
-
       identity_changed.send(app, identity=Identity(token.code, 'token'))
       return
 
@@ -117,7 +130,6 @@ def _process_basic_auth(auth):
   else:
     (authenticated, _) = authentication.verify_and_link_user(credentials[0], credentials[1],
                                                              basic_auth=True)
-
     if authenticated:
       logger.debug('Successfully validated user: %s', authenticated.username)
       set_authenticated_user(authenticated)
@@ -130,6 +142,23 @@ def _process_basic_auth(auth):
   logger.debug('Basic auth present but could not be validated.')
 
 
+def has_basic_auth(username):
+  auth = request.headers.get('authorization', '')
+  if not auth:
+    return False
+
+  credentials = _parse_basic_auth_header(auth)
+  if not credentials:
+    return False
+
+  (authenticated, _) = authentication.verify_and_link_user(credentials[0], credentials[1],
+                                                           basic_auth=True)
+  if not authenticated:
+    return False
+
+  return authenticated.username == username
+
+
 def generate_signed_token(grants, user_context):
   ser = SecureCookieSessionInterface().get_signing_serializer(app)
   data_to_sign = {
@@ -209,6 +238,22 @@ def process_auth(func):
   return wrapper
 
 
+def process_auth_or_cookie(func):
+  @wraps(func)
+  def wrapper(*args, **kwargs):
+    auth = request.headers.get('authorization', '')
+
+    if auth:
+      logger.debug('Validating auth header: %s', auth)
+      _process_basic_auth(auth)
+    else:
+      logger.debug('No auth header.')
+      _load_user_from_cookie()
+
+    return func(*args, **kwargs)
+  return wrapper
+
+
 def require_session_login(func):
   @wraps(func)
   def wrapper(*args, **kwargs):