Security providers and security models retouch

Author: p1c2u

openapi_core/schema/security_requirements/generators.py

@@ -12,7 +12,4 @@ def __init__(self, dereferencer):
     def generate(self, security_spec):
         security_deref = self.dereferencer.dereference(security_spec)
         for security_requirement_spec in security_deref:
-            name = next(iter(security_requirement_spec))
-            scope_names = security_requirement_spec[name]
-
-            yield SecurityRequirement(name, scope_names=scope_names)
+            yield SecurityRequirement(security_requirement_spec)

openapi_core/schema/security_requirements/models.py

@@ -1,9 +1,6 @@
 """OpenAPI core security requirements models module"""
 
 
-class SecurityRequirement(object):
+class SecurityRequirement(dict):
     """Represents an OpenAPI Security Requirement."""
-
-    def __init__(self, name, scope_names=None):
-        self.name = name
-        self.scope_names = scope_names or []
+    pass

openapi_core/security/__init__.py

@@ -0,0 +1,5 @@
+from openapi_core.exceptions import OpenAPIError
+
+
+class SecurityError(OpenAPIError):
+    pass

openapi_core/security/exceptions.py

@@ -0,0 +1,19 @@
+from openapi_core.schema.security_schemes.enums import SecuritySchemeType
+from openapi_core.security.providers import (
+    ApiKeyProvider, HttpProvider, UnsupportedProvider,
+)
+
+
+class SecurityProviderFactory(object):
+
+    PROVIDERS = {
+        SecuritySchemeType.API_KEY: ApiKeyProvider,
+        SecuritySchemeType.HTTP: HttpProvider,
+    }
+
+    def create(self, scheme):
+        if scheme.type == SecuritySchemeType.API_KEY:
+            return ApiKeyProvider(scheme)
+        elif scheme.type == SecuritySchemeType.HTTP:
+            return HttpProvider(scheme)
+        return UnsupportedProvider(scheme)

openapi_core/security/factories.py

@@ -0,0 +1,47 @@
+import base64
+import binascii
+import warnings
+
+from openapi_core.security.exceptions import SecurityError
+
+
+class BaseProvider(object):
+
+    def __init__(self, scheme):
+        self.scheme = scheme
+
+
+class UnsupportedProvider(BaseProvider):
+
+    def __call__(self, request):
+        warnings.warn("Unsupported scheme type")
+
+
+class ApiKeyProvider(BaseProvider):
+
+    def __call__(self, request):
+        source = getattr(request.parameters, self.scheme.apikey_in.value)
+        if self.scheme.name not in source:
+            raise SecurityError("Missing api key parameter.")
+        return source.get(self.scheme.name)
+
+
+class HttpProvider(BaseProvider):
+
+    def __call__(self, request):
+        if 'Authorization' not in request.parameters.header:
+            raise SecurityError('Missing authorization header.')
+        auth_header = request.parameters.header['Authorization']
+        try:
+            auth_type, encoded_credentials = auth_header.split(' ', 1)
+        except ValueError:
+            raise SecurityError('Could not parse authorization header.')
+
+        if auth_type.lower() != self.scheme.scheme.value:
+            raise SecurityError(
+                'Unknown authorization method %s' % auth_type)
+        try:
+            return base64.b64decode(
+                encoded_credentials.encode('ascii')).decode('latin1')
+        except binascii.Error:
+            raise SecurityError('Invalid base64 encoding.')

openapi_core/security/providers.py

@@ -0,0 +1,15 @@
+"""OpenAPI core validation exceptions module"""
+import attr
+
+from openapi_core.exceptions import OpenAPIError
+
+
+class ValidationError(OpenAPIError):
+    pass
+
+
+@attr.s(hash=True)
+class InvalidSecurity(ValidationError):
+
+    def __str__(self):
+        return "Security not valid for any requirement"

openapi_core/validation/exceptions.py

@@ -16,9 +16,11 @@
 from openapi_core.schema.request_bodies.exceptions import MissingRequestBody
 from openapi_core.schema.security_schemes.enums import SecuritySchemeType
 from openapi_core.schema.servers.exceptions import InvalidServer
+from openapi_core.security.exceptions import SecurityError
 from openapi_core.unmarshalling.schemas.exceptions import (
     UnmarshalError, ValidateError,
 )
+from openapi_core.validation.exceptions import InvalidSecurity
 from openapi_core.validation.request.datatypes import (
     RequestParameters, RequestValidationResult,
 )
@@ -45,8 +47,7 @@ def validate(self, request):
 
         try:
             security = self._get_security(request, operation)
-        # TODO narrow exceptions
-        except Exception as exc:
+        except InvalidSecurity as exc:
             return RequestValidationResult([exc, ], None, None, None)
 
         params, params_errors = self._get_parameters(
@@ -108,14 +109,16 @@ def _get_security(self, request, operation):
             return {}
 
         for security_requirement in security:
-            data = {
-                security_requirement.name: self._get_security_value(
-                    security_requirement.name, request)
-            }
-            if all(value for value in data.values()):
-                return data
+            try:
+                return {
+                    scheme_name: self._get_security_value(
+                        scheme_name, request)
+                    for scheme_name in security_requirement
+                }
+            except SecurityError:
+                continue
 
-        return {}
+        raise InvalidSecurity()
 
     def _get_parameters(self, request, params):
         errors = []
@@ -196,27 +199,10 @@ def _get_security_value(self, scheme_name, request):
         if not scheme:
             return
 
-        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")
+        from openapi_core.security.factories import SecurityProviderFactory
+        security_provider_factory = SecurityProviderFactory()
+        security_provider = security_provider_factory.create(scheme)
+        return security_provider(request)
 
     def _get_parameter_value(self, param, request):
         location = request.parameters[param.location.value]

openapi_core/validation/request/validators.py

@@ -13,6 +13,7 @@ info:
     url: https://opensource.org/licenses/MIT
 security:
   - api_key: []
+  - {}
 servers:
   - url: http://petstore.swagger.io/{version}
     variables:

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

@@ -71,9 +71,8 @@ def test_spec(self, spec, spec_dict):
             assert type(security_req) == SecurityRequirement
 
             security_req_spec = security_spec[idx]
-            name = next(iter(security_req_spec))
-            assert security_req.name == name
-            assert security_req.scope_names == security_req_spec[name]
+            for scheme_name in security_req:
+                security_req[scheme_name] == security_req_spec[scheme_name]
 
         assert spec.get_server_url() == url
 
@@ -115,15 +114,6 @@ def test_spec(self, spec, spec_dict):
                     assert ext_docs.description == ext_docs_spec.get(
                         'description')
 
-                security_spec = operation_spec.get('security')
-                if security_spec:
-                    for idx, sec_req in enumerate(operation.security):
-                        assert type(sec_req) == SecurityRequirement
-                        sec_req_spec = security_spec[idx]
-                        sec_req_nam = next(iter(sec_req_spec))
-                        assert sec_req.name == sec_req_nam
-                        assert sec_req.scope_names == sec_req_spec[sec_req_nam]
-
                 servers_spec = operation_spec.get('servers', [])
                 for idx, server in enumerate(operation.servers):
                     assert type(server) == Server
@@ -146,9 +136,9 @@ def test_spec(self, spec, spec_dict):
                     assert type(security_req) == SecurityRequirement
 
                     security_req_spec = security_spec[idx]
-                    name = next(iter(security_req_spec))
-                    assert security_req.name == name
-                    assert security_req.scope_names == security_req_spec[name]
+                    for scheme_name in security_req:
+                        security_req[scheme_name] == security_req_spec[
+                            scheme_name]
 
                 responses_spec = operation_spec.get('responses')