Add OAuth2Gateway for gRPC

perxis/auth.py

+import time
+
+import grpc
+
+from oauthlib.oauth2 import BackendApplicationClient, OAuth2Token
+from requests_oauthlib import OAuth2Session
+
+
+class OAuth2Gateway(grpc.AuthMetadataPlugin):
+    _token = None
+
+    def __init__(self, client_id: str, client_secret: str, token_url: str, audience: str,
+                 signature_header_key: str = 'Authorization', token_type: str = 'Bearer') -> None:
+
+        self._client_id = client_id
+        self._client_secret = client_secret
+        self._token_url = token_url
+        self._audience = audience
+        self._signature_header_key = signature_header_key
+        self._token_type = token_type
+
+        self.oauth2 = OAuth2Session(client=BackendApplicationClient(client_id=client_id))
+
+    def __call__(self, context: grpc.AuthMetadataContext,
+                 callback: grpc.AuthMetadataPluginCallback) -> None:
+        callback(((self._signature_header_key, f'{self._token_type} {self.token["access_token"]}'),), None)
+
+    @property
+    def token(self) -> OAuth2Token:
+        if self._token is None:
+            self._token = self.oauth2.fetch_token(
+                token_url=self._token_url,
+                client_secret=self._client_secret,
+                audience=self._audience
+            )
+        else:
+            if self._token.expires_at and self._token.expires_at < time.time():
+                self._token = self.oauth2.refresh_token(
+                    token_url=self._token_url,
+                    client_secret=self._client_secret,
+                    audience=self._audience
+                )
+        return self._token

requirements.txt

+certifi==2020.12.5
+chardet==4.0.0
 grpcio==1.36.1
+idna==2.10
+oauthlib==3.1.0
 protobuf==3.15.4
+PyJWT==2.0.1
+requests==2.25.1
+requests-oauthlib==1.3.0
 six==1.15.0
+urllib3==1.26.4