Be more pluggable

connexion/decorators/security.py

@@ -105,34 +105,11 @@ def deny(*args, **kwargs):
     return deny
 
 
-def get_authorization_info(request, auth_map, required_scopes):
-    authorization = request.headers.get('Authorization')
-    if authorization:
-        try:
-            auth_type, auth_info = authorization.split(None, 1)
-            auth_type = auth_type.lower()
-        except ValueError:
-            raise OAuthProblem(description='Invalid authorization header')
-
-        if auth_type == 'bearer' and 'oauth' in auth_map:
-            return auth_map['oauth'](auth_info, required_scopes)
-
-        if auth_type == 'basic' and 'basic' in auth_map:
-            try:
-                username, password = base64.b64decode(auth_info).decode('latin1').split(':', 1)
-            except Exception:
-                raise OAuthProblem(description='Invalid authorization header')
-            return auth_map['basic'](username, password, required_scopes)
-
-    for loc, name, func in auth_map.get('apikey', []):
-        if loc == 'query':
-            val = request.query.get(name)
-            if val is not None:
-                return func(val, required_scopes)
-        elif loc == 'header':
-            val = request.headers.get(name)
-            if val is not None:
-                return func(val, required_scopes)
+def get_authorization_info(auth_funcs, request, required_scopes):
+    for func in auth_funcs:
+        token_info = func(request, required_scopes)
+        if token_info is not None:
+            return token_info
 
     logger.info("... No auth provided. Aborting with 401.")
     raise OAuthProblem(description='No authorization token provided')
@@ -161,7 +138,19 @@ def validate_scope(required_scopes, token_scopes):
 
 
 def verify_oauth(token_info_func, scope_validate_func):
-    def wrapper(token, required_scopes):
+    def wrapper(request, required_scopes):
+        authorization = request.headers.get('Authorization')
+        if not authorization:
+            return None
+
+        try:
+            auth_type, token = authorization.split(None, 1)
+        except ValueError:
+            raise OAuthProblem(description='Invalid authorization header')
+
+        if auth_type.lower() != 'bearer':
+            return None
+
         token_info = token_info_func(token)
         if token_info is None:
             raise OAuthResponseProblem(
@@ -183,33 +172,60 @@ def wrapper(token, required_scopes):
 
 
 def verify_basic(basic_info_func):
-    def wrapper(username, password, required_scopes):
-        token_info = basic_info_func(username, password, scope=required_scopes)
+    def wrapper(request, required_scopes):
+        authorization = request.headers.get('Authorization')
+        if not authorization:
+            return None
+
+        try:
+            auth_type, user_pass = authorization.split(None, 1)
+        except ValueError:
+            raise OAuthProblem(description='Invalid authorization header')
+
+        if auth_type.lower() != 'basic':
+            return None
+
+        try:
+            username, password = base64.b64decode(user_pass).decode('latin1').split(':', 1)
+        except Exception:
+            raise OAuthProblem(description='Invalid authorization header')
+
+        token_info = basic_info_func(username, password, required_scopes=required_scopes)
         if token_info is None:
             raise OAuthResponseProblem(
-                description='Provided apikey is not valid',
+                description='Provided authorization is not valid',
                 token_response=None
             )
         return token_info
     return wrapper
 
 
-def verify_apikey(apikey_info_func):
-    def wrapper(apikey, required_scopes):
-        token_info = apikey_info_func(apikey, scope=required_scopes)
+def verify_apikey(apikey_info_func, loc, name):
+    def wrapper(request, required_scopes):
+        if loc == 'query':
+            apikey = request.query.get(name)
+        elif loc == 'header':
+            apikey = request.headers.get(name)
+        else:
+            return None
+
+        if apikey is None:
+            return None
+
+        token_info = apikey_info_func(apikey, required_scopes=required_scopes)
         if token_info is None:
             raise OAuthResponseProblem(
-                description='Provided authorization is not valid',
+                description='Provided apikey is not valid',
                 token_response=None
             )
         return token_info
     return wrapper
 
 
-def verify_security(auth_map, required_scopes, function):
+def verify_security(auth_funcs, required_scopes, function):
     @functools.wraps(function)
     def wrapper(request):
-        token_info = get_authorization_info(request, auth_map, required_scopes)
+        token_info = get_authorization_info(auth_funcs, request, required_scopes)
 
         # Fallback to 'uid' for backward compability
         request.context['user'] = token_info.get('sub', token_info.get('uid'))

connexion/operations/secure.py

@@ -76,7 +76,9 @@ def security_decorator(self):
         if not self.security:
             return security_passthrough
 
-        auth_map = {}
+        has_oauth = False
+        has_basic = False
+        auth_funcs = []
         required_scopes = None
         for security_req in self.security:
             if not security_req:
@@ -90,7 +92,7 @@ def security_decorator(self):
             security_scheme = self.security_schemes[scheme_name]
 
             if security_scheme['type'] == 'oauth2':
-                if 'oauth' in auth_map:
+                if has_oauth:
                     logger.warning("... Only one oauth2 scheme per security declaration supported. "
                                    "**DENYING ALL REQUESTS**", extra=vars(self))
                     return security_deny
@@ -102,10 +104,11 @@ def security_decorator(self):
                     logger.warning("... x-tokenInfoFunc missing", extra=vars(self))
                     continue
 
-                auth_map['oauth'] = verify_oauth(token_info_func, scope_validate_func)
+                auth_funcs.append(verify_oauth(token_info_func, scope_validate_func))
+                has_oauth = True
 
             elif security_scheme['type'] == 'basic':
-                if 'basic' in auth_map:
+                if has_basic:
                     logger.warning("... Only one basic scheme per security declaration supported. "
                                    "**DENYING ALL REQUESTS**", extra=vars(self))
                     return security_deny
@@ -115,17 +118,18 @@ def security_decorator(self):
                     logger.warning("... x-basicInfoFunc missing", extra=vars(self))
                     continue
 
-                auth_map['basic'] = verify_basic(basic_info_func)
+                auth_funcs.append(verify_basic(basic_info_func))
+                has_basic = True
 
             elif security_scheme['type'] == 'apiKey':
                 apikey_info_func = get_apikeyinfo_func(security_scheme)
                 if not apikey_info_func:
                     logger.warning("... x-apikeyInfoFunc missing", extra=vars(self))
                     continue
 
-                auth_map.setdefault('apikey', []).append((security_scheme['in'], security_scheme['name'], verify_apikey(apikey_info_func)))
+                auth_funcs.append(verify_apikey(apikey_info_func, security_scheme['in'], security_scheme['name']))
 
-        return functools.partial(verify_security, auth_map, required_scopes)
+        return functools.partial(verify_security, auth_funcs, required_scopes)
 
     def get_mimetype(self):
         return DEFAULT_MIMETYPE