| 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 |
|
|
| |
| oauth_manager = OAuthManager() |
|
|
