Update auth.py

auth.py

@@ -1,71 +1,136 @@
+# auth.py
+"""
+Centralised OAuth 2.0 helper for GitHub, Google Drive and Slack.
+
+Usage
+-----
+from auth import oauth_manager
+
+# 1)  Redirect user to consent page
+auth_url, state = oauth_manager.get_authorization_url(
+    provider="github",
+    redirect_uri="https://your‑app.com/callback"
+)
+
+# 2)  In your callback handler, exchange the code for a token
+token = oauth_manager.fetch_token(
+    provider="github",
+    redirect_uri="https://your‑app.com/callback",
+    authorization_response=request.url  # full URL with ?code=...
+)
+"""
+
+from __future__ import annotations
+
 import os
-from typing import Optional, Dict
+from dataclasses import dataclass
+from typing import Dict, Optional, Tuple
+
 from authlib.integrations.requests_client import OAuth2Session
 
+
+# ------------------------------------------------------------------ #
+# 1  Provider configuration
+# ------------------------------------------------------------------ #
+@dataclass(frozen=True)
+class ProviderConfig:
+    client_id: str | None
+    client_secret: str | None
+    authorize_url: str
+    token_url: str
+    scope: str
+
+
+def _env(name: str) -> str | None:
+    """Shorthand for os.getenv with strip()."""
+    val = os.getenv(name)
+    return val.strip() if val else None
+
+
+PROVIDERS: Dict[str, ProviderConfig] = {
+    "github": ProviderConfig(
+        client_id=_env("GITHUB_CLIENT_ID"),
+        client_secret=_env("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": ProviderConfig(
+        client_id=_env("GOOGLE_CLIENT_ID"),
+        client_secret=_env("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": ProviderConfig(
+        client_id=_env("SLACK_CLIENT_ID"),
+        client_secret=_env("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",
+    ),
+}
+
+
+# ------------------------------------------------------------------ #
+# 2  OAuth manager
+# ------------------------------------------------------------------ #
 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:
+    """Tiny wrapper around Authlib’s OAuth2Session per provider."""
+
+    def __init__(self, providers: Dict[str, ProviderConfig]):
+        self.providers = providers
+
+    # ---------- helpers -------------------------------------------------
+    def _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.")
+        if not cfg:
+            raise KeyError(f"Unsupported provider '{provider}'.")
+        if not (cfg.client_id and cfg.client_secret):
+            raise RuntimeError(
+                f"OAuth credentials for '{provider}' are missing. "
+                "Set the *_CLIENT_ID and *_CLIENT_SECRET env‑vars."
+            )
         return OAuth2Session(
-            cfg['client_id'],
-            cfg['client_secret'],
-            scope=cfg['scope'],
-            redirect_uri=redirect_uri
+            client_id=cfg.client_id,
+            client_secret=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):
+    # ---------- public API ----------------------------------------------
+    def get_authorization_url(
+        self, provider: str, redirect_uri: str, state: Optional[str] = None
+    ) -> Tuple[str, str]:
         """
-        Generate the OAuth2 authorization URL and state.
+        Return (auth_url, state) for the given provider.
 
-        Returns:
-            (authorization_url, state)
+        Pass the *state* back into `fetch_token` to mitigate CSRF.
         """
-        session = self._create_session(provider, redirect_uri)
-        url, state = session.create_authorization_url(self.providers[provider]['authorize_url'], state=state)
-        return url, state
+        sess = self._session(provider, redirect_uri)
+        cfg = self.providers[provider]
+        auth_url, final_state = sess.create_authorization_url(
+            cfg.authorize_url, state=state
+        )
+        return auth_url, final_state
 
-    def fetch_token(self, provider: str, redirect_uri: str, authorization_response: str) -> Dict:
+    def fetch_token(
+        self, provider: str, redirect_uri: str, authorization_response: str
+    ) -> Dict:
         """
-        Exchange the authorization response for an access token.
+        Exchange ?code=… for an access token.
 
-        Returns:
-            Token dict containing access_token, refresh_token, expires_in, etc.
+        Returns the token dict from Authlib (includes 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
+        sess = self._session(provider, redirect_uri)
+        cfg = self.providers[provider]
+        return sess.fetch_token(
+            cfg.token_url,
+            authorization_response=authorization_response,
+            client_secret=cfg.client_secret,  # some providers require it explicitly
         )
-        return token
 
-# Instantiate a global manager
-oauth_manager = OAuthManager()
+
+# Singleton instance used throughout the app
+oauth_manager = OAuthManager(PROVIDERS)