File size: 2,932 Bytes
4ae3bf6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import os
from typing import Optional, Dict
from authlib.integrations.requests_client import OAuth2Session
class OAuthManager:
"""
Manages OAuth2 flows for third-party services: GitHub, Google Drive, Slack.
"""
def __init__(self):
self.providers: Dict[str, Dict] = {
'github': {
'client_id': os.getenv('GITHUB_CLIENT_ID'),
'client_secret': os.getenv('GITHUB_CLIENT_SECRET'),
'authorize_url': 'https://github.com/login/oauth/authorize',
'token_url': 'https://github.com/login/oauth/access_token',
'scope': 'repo read:org'
},
'google': {
'client_id': os.getenv('GOOGLE_CLIENT_ID'),
'client_secret': os.getenv('GOOGLE_CLIENT_SECRET'),
'authorize_url': 'https://accounts.google.com/o/oauth2/auth',
'token_url': 'https://oauth2.googleapis.com/token',
'scope': 'openid email profile https://www.googleapis.com/auth/drive.readonly'
},
'slack': {
'client_id': os.getenv('SLACK_CLIENT_ID'),
'client_secret': os.getenv('SLACK_CLIENT_SECRET'),
'authorize_url': 'https://slack.com/oauth/v2/authorize',
'token_url': 'https://slack.com/api/oauth.v2.access',
'scope': 'channels:read chat:write'
}
}
def _create_session(self, provider: str, redirect_uri: str) -> OAuth2Session:
cfg = self.providers.get(provider)
if not cfg or not cfg['client_id'] or not cfg['client_secret']:
raise RuntimeError(f"OAuth credentials for '{provider}' are not configured.")
return OAuth2Session(
cfg['client_id'],
cfg['client_secret'],
scope=cfg['scope'],
redirect_uri=redirect_uri
)
def get_authorization_url(self, provider: str, redirect_uri: str, state: Optional[str] = None) -> (str, str):
"""
Generate the OAuth2 authorization URL and state.
Returns:
(authorization_url, state)
"""
session = self._create_session(provider, redirect_uri)
url, state = session.create_authorization_url(self.providers[provider]['authorize_url'], state=state)
return url, state
def fetch_token(self, provider: str, redirect_uri: str, authorization_response: str) -> Dict:
"""
Exchange the authorization response for an access token.
Returns:
Token dict containing access_token, refresh_token, expires_in, etc.
"""
session = self._create_session(provider, redirect_uri)
token = session.fetch_token(
self.providers[provider]['token_url'],
authorization_response=authorization_response
)
return token
# Instantiate a global manager
oauth_manager = OAuthManager()
|