|
|
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() |
|
|
|
