Skip to content
Snippets Groups Projects

Добавлен плагин аутентификации

Merged Andrei Biriukov requested to merge feature/auth into master
2 files
+ 51
0
Compare changes
  • Side-by-side
  • Inline
Files
2
perxis/auth.py 0 → 100644
+ 43
0
 
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
Loading