Esquire media/master

aiopenapi3/v20/glue.py

@@ -1,6 +1,5 @@
 from typing import List, Union, cast
 import json
-import urllib.parse
 
 import httpx
 import pydantic
@@ -11,6 +10,11 @@
 
 from .parameter import Parameter
 
+try:
+    import httpx_auth
+except:
+    httpx_auth = None
+
 
 class Request(RequestBase):
     @property
@@ -76,9 +80,12 @@ def _prepare_security(self):
             )
 
     def _prepare_secschemes(self, scheme: str, value: Union[str, List[str]]):
-        """
-        https://swagger.io/specification/v2/#security-scheme-object
-        """
+        if httpx_auth is not None:
+            return self._prepare_secschemes_extra(scheme, value)
+        else:
+            return self._prepare_secschemes_default(scheme, value)
+
+    def _prepare_secschemes_default(self, scheme: str, value: Union[str, List[str]]):
         ss = self.root.securityDefinitions[scheme]
 
         if ss.type == "basic":
@@ -95,6 +102,23 @@ def _prepare_secschemes(self, scheme: str, value: Union[str, List[str]]):
                 # apiKey in query header data
                 self.req.headers[ss.name] = value
 
+    def _prepare_secschemes_extra(self, scheme: str, value: Union[str, List[str]]):
+        ss = self.root.securityDefinitions[scheme]
+
+        if ss.type == "basic":
+            value = cast(List[str], value)
+            self.req.auth = httpx_auth.Basic(*value)
+
+        value = cast(str, value)
+        if ss.type == "apiKey":
+            if ss.in_ == "query":
+                # apiKey in query parameter
+                self.req.auth = httpx_auth.QueryApiKey(value, ss.name)
+
+            if ss.in_ == "header":
+                # apiKey in query header data
+                self.req.auth = httpx_auth.HeaderApiKey(value, ss.name)
+
     def _prepare_parameters(self, provided):
         provided = provided or dict()
         possible = {_.name: _ for _ in self.operation.parameters + self.root.paths[self.path].parameters}
@@ -284,4 +308,9 @@ def _process(self, result):
 
 
 class AsyncRequest(Request, AsyncRequestBase):
-    pass
+    def _prepare_secschemes(self, scheme: str, value: Union[str, List[str]]):
+        """
+        httpx_auth does not support async yet
+        https://github.com/Colin-b/httpx_auth/pull/48
+        """
+        return self._prepare_secschemes_default(scheme, value)

aiopenapi3/v30/glue.py

@@ -3,6 +3,22 @@
 import urllib.parse
 
 import httpx
+try:
+    import httpx_auth
+    from httpx_auth.authentication import SupportMultiAuth
+except:
+    httpx_auth = None
+
+if httpx_auth is not None:
+    import inspect
+
+    HTTPX_AUTH_METHODS = {
+        name.lower(): getattr(httpx_auth, name)
+        for name in httpx_auth.__all__
+        if inspect.isclass((class_ := getattr(httpx_auth, name)))
+        if issubclass(class_, httpx.Auth)
+    }
+
 import pydantic
 import pydantic.json
 
@@ -76,17 +92,25 @@ def _prepare_security(self):
             )
 
     def _prepare_secschemes(self, scheme: str, value: Union[str, List[str]]):
-        ss = self.root.components.securitySchemes[scheme]
+        if httpx_auth is not None:
+            return self._prepare_secschemes_extra(scheme, value)
+        else:
+            return self._prepare_secschemes_default(scheme, value)
 
-        if ss.type == "http" and ss.scheme_ == "basic":
-            self.req.auth = httpx.BasicAuth(*value)
+    def _prepare_secschemes_default(self, scheme: str, value: Union[str, List[str]]):
+        ss = self.root.components.securitySchemes[scheme]
 
-        if ss.type == "http" and ss.scheme_ == "digest":
-            self.req.auth = httpx.DigestAuth(*value)
+        if ss.type == "http":
+            if ss.scheme_ == "basic":
+                self.req.auth = httpx.BasicAuth(*value)
+            elif ss.scheme_ == "digest":
+                self.req.auth = httpx.DigestAuth(*value)
+            elif ss.scheme_ == "bearer":
+                self.req.headers["Authorization"] = f"Bearer {value:s}"
+            else:
+                raise ValueError(f"Authentication {ss.type}/{ss.scheme_} is not supported.")
 
         value = cast(str, value)
-        if ss.type == "http" and ss.scheme_ == "bearer":
-            self.req.headers["Authorization"] = "Bearer {}".format(value)
 
         if ss.type == "mutualTLS":
             # TLS Client certificates (mutualTLS)
@@ -104,6 +128,81 @@ def _prepare_secschemes(self, scheme: str, value: Union[str, List[str]]):
             if ss.in_ == "cookie":
                 self.req.cookies = {ss.name: value}
 
+    def _prepare_secschemes_extra(self, scheme: str, value: Union[str, List[str]]):
+        ss = self.root.components.securitySchemes[scheme]
+        auths = []
+
+        if ss.type == "oauth2":
+            # NOTE: refresh_url is not currently supported by httpx_auth
+            # REF: https://github.com/Colin-b/httpx_auth/issues/17
+            if flow := ss.flows.implicit:
+                auths.append(
+                    httpx_auth.OAuth2Implicit(
+                        **value,
+                        authorization_url=flow.authorizationUrl,
+                        scopes=flow.scopes,
+                        # refresh_url=flow.refreshUrl,
+                    )
+                )
+            if flow := ss.flows.password:
+                auths.append(
+                    httpx_auth.OAuth2ResourceOwnerPasswordCredentials(
+                        **value,
+                        token_url=flow.tokenUrl,
+                        scopes=flow.scopes,
+                        # refresh_url=flow.refreshUrl,
+                    )
+                )
+            if flow := ss.flows.clientCredentials:
+                auths.append(
+                    httpx_auth.OAuth2ClientCredentials(
+                        **value,
+                        token_url=flow.tokenUrl,
+                        scopes=flow.scopes,
+                        # refresh_url=flow.refreshUrl,
+                    )
+                )
+            if flow := ss.flows.authorizationCode:
+                auths.append(
+                    httpx_auth.OAuth2AuthorizationCode(
+                        **value,
+                        authorization_url=flow.authorizationUrl,
+                        token_url=flow.tokenUrl,
+                        scopes=flow.scopes,
+                        # refresh_url=flow.refreshUrl,
+                    )
+                )
+
+        if ss.type == "http":
+            if auth := HTTPX_AUTH_METHODS.get(ss.scheme_, None):
+                if isinstance(value, tuple):
+                    auths.append(auth(*value))
+                elif isinstance(value, dict):
+                    auths.append(auth(**value))
+            elif ss.scheme_ == "bearer":
+                auths.append(httpx_auth.HeaderApiKey(f"Bearer {value}", "Authorization"))
+            else:
+                raise ValueError(f"Authentication method {ss.type}/{ss.scheme_} is not supported by httpx-auth")
+
+        value = cast(str, value)
+
+        if ss.type == "mutualTLS":
+            # TLS Client certificates (mutualTLS)
+            self.req.cert = value
+
+        if ss.type == "apiKey":
+            if auth := HTTPX_AUTH_METHODS.get((ss.in_ + ss.type).lower(), None):
+                auths.append(auth(value, ss.name))
+
+            if ss.in_ == "cookie":
+                self.req.cookies = {ss.name: value}
+
+        for auth in auths:
+            if self.req.auth and isinstance(self.req.auth, SupportMultiAuth):
+                self.req.auth += auth
+            else:
+                self.req.auth = auth
+
     def _prepare_parameters(self, provided):
         """
         assigns the parameters provided to the header/path/cookie …

docs/source/advanced.rst

@@ -7,8 +7,9 @@ Advanced usage
 
 Authentication
 ==============
+
 The authentication requirements are part of the definition of an operation, either global or - it it exists - operation scope.
-Authentication can combine/require multiple identifiert as well as providing a choice of a set.
+Authentication can combine/require multiple identifiers as well as providing a choice of a set.
 
 Given the following section of a description document:
 

docs/source/install.rst

@@ -7,3 +7,6 @@ Installation
 .. code:: bash
 
     $ pip install aiopenapi3
+
+
+* aiopenapi3[auth] will install httpx-auth_ which is required to authenticate using oauth2/azuread/. Currently httpx-auth is `limited to Sync <https://github.com/Colin-b/httpx_auth/pull/48>`_ operations.

docs/source/links.rst

@@ -1,4 +1,5 @@
 .. |aiopenapi3| replace:: **aiopenapi3**
 .. _OpenAPI: https://github.com/OAI/OpenAPI-Specification/
 .. _pydantic: https://github.com/pydantic/pydantic
-.. _httpx: https://github.com/encode/httpx/
+.. _httpx: https://github.com/encode/httpx
+.. _httpx-auth: https://github.com/Colin-b/httpx_auth

requirements.txt

@@ -1,9 +1,9 @@
 fastapi~=0.95.0
-httpx~=0.23.3
+httpx~=0.24.0
 hypercorn~=0.14.3
 pydantic~=1.10.7
 pydantic[email]
 pytest~=7.2.2
 PyYAML~=6.0
-uvloop~=0.17.0
+uvloop~=0.17.0; sys_platform != 'win32'
 yarl~=1.8.2

setup.cfg

@@ -69,5 +69,8 @@ tests =
     wtforms
     asgiref
 
+auth =
+    httpx-auth
+
 socks =
     httpx_socks

tests/path_test.py

@@ -155,9 +155,12 @@ def test_paths_security(httpx_mock, with_paths_security):
     request = httpx_mock.get_requests()[-1]
     assert request.headers["Authorization"].split(" ")[1] == base64.b64encode((auth + ":" + auth).encode()).decode()
 
-    api.authenticate(None, digestAuth=(auth, auth))
-    api._.api_v1_auth_login_create(data={}, parameters={})
-    request = httpx_mock.get_requests()[-1]
+    try:
+        import httpx_auth
+    except:
+        api.authenticate(None, digestAuth=(auth, auth))
+        api._.api_v1_auth_login_create(data={}, parameters={})
+        request = httpx_mock.get_requests()[-1]
     # can't test?
 
     api.authenticate(None, bearerAuth=auth)