Security HTTP scheme type support

Author: p1c2u

openapi_core/schema/security_schemes/enums.py

@@ -19,3 +19,9 @@ class ApiKeyLocation(Enum):
     @classmethod
     def has_value(cls, value):
         return (any(value == item.value for item in cls))
+
+
+class HttpAuthScheme(Enum):
+
+    BASIC = 'basic'
+    BEARER = 'bearer'

openapi_core/schema/security_schemes/models.py

@@ -1,6 +1,6 @@
 """OpenAPI core security schemes models module"""
 from openapi_core.schema.security_schemes.enums import (
-    SecuritySchemeType, ApiKeyLocation,
+    SecuritySchemeType, ApiKeyLocation, HttpAuthScheme,
 )
 
 
@@ -16,7 +16,7 @@ def __init__(
         self.description = description
         self.name = name
         self.apikey_in = apikey_in and ApiKeyLocation(apikey_in)
-        self.scheme = scheme
+        self.scheme = scheme and HttpAuthScheme(scheme)
         self.bearer_format = bearer_format
         self.flows = flows
         self.open_id_connect_url = open_id_connect_url

openapi_core/validation/request/validators.py

@@ -1,4 +1,6 @@
 """OpenAPI core validation request validators module"""
+import base64
+import binascii
 from itertools import chain
 from six import iteritems
 import warnings
@@ -103,7 +105,7 @@ def _get_operation(self, request):
     def _get_security(self, request, operation):
         security = operation.security or self.spec.security
         if not security:
-            return
+            return {}
 
         for security_requirement in security:
             data = {
@@ -113,6 +115,8 @@ def _get_security(self, request, operation):
             if all(value for value in data.values()):
                 return data
 
+        return {}
+
     def _get_parameters(self, request, params):
         errors = []
         seen = set()
@@ -195,6 +199,22 @@ def _get_security_value(self, scheme_name, request):
         if scheme.type == SecuritySchemeType.API_KEY:
             source = getattr(request.parameters, scheme.apikey_in.value)
             return source.get(scheme.name)
+        elif scheme.type == SecuritySchemeType.HTTP:
+            auth_header = request.parameters.header.get('Authorization')
+            try:
+                auth_type, encoded_credentials = auth_header.split(' ', 1)
+            except ValueError:
+                raise ValueError('Could not parse authorization header.')
+
+            if auth_type.lower() != scheme.scheme.value:
+                raise ValueError(
+                    'Unknown authorization method %s' % auth_type)
+            try:
+                return base64.b64decode(
+                    encoded_credentials.encode('ascii'), validate=True
+                ).decode('latin1')
+            except binascii.Error:
+                raise ValueError('Invalid base64 encoding.')
 
         warnings.warn("Only api key security scheme type supported")
 

tests/integration/data/v3.0/petstore.yaml

@@ -75,10 +75,6 @@ paths:
       externalDocs:
         url: https://example.com
         description: Find more info here
-      security:
-        - petstore_auth:
-            - write:pets
-            - read:pets
       servers:
         - url: https://development.gigantic-server.com/v1
           description: Development server
@@ -128,6 +124,10 @@ paths:
           schema:
             type: integer
             format: int64
+      security:
+        - petstore_auth:
+            - write:pets
+            - read:pets
       responses:
         '200':
           description: Expected response to a valid request
@@ -371,6 +371,5 @@ components:
       name: api_key
       in: query
     petstore_auth:
-      type: apiKey
-      name: api_key
-      in: header
+      type: http
+      scheme: basic

tests/integration/validation/test_validators.py

@@ -235,9 +235,7 @@ def test_post_pets(self, validator, spec_dict):
                 'user': 123,
             },
         )
-        assert result.security == {
-            'petstore_auth': self.api_key_encoded,
-        }
+        assert result.security == {}
 
         schemas = spec_dict['components']['schemas']
         pet_model = schemas['PetCreate']['x-model']
@@ -251,9 +249,14 @@ def test_post_pets(self, validator, spec_dict):
         assert result.body.address.city == pet_city
 
     def test_get_pet(self, validator):
+        authorization = 'Basic ' + self.api_key_encoded
+        headers = {
+            'Authorization': authorization,
+        }
         request = MockRequest(
             self.host_url, 'get', '/v1/pets/1',
             path_pattern='/v1/pets/{petId}', view_args={'petId': '1'},
+            headers=headers,
         )
 
         result = validator.validate(request)
@@ -265,6 +268,9 @@ def test_get_pet(self, validator):
                 'petId': 1,
             },
         )
+        assert result.security == {
+            'petstore_auth': self.api_key,
+        }
 
 
 class TestPathItemParamsValidator(object):