v3.0.2

Signed-off-by: Dinger <quantdinger@gmail.com>

Author: quantdinger-lab

backend_api_python/app/services/billing_service.py

@@ -283,7 +283,7 @@ def purchase_membership(self, user_id: int, plan: str) -> Tuple[bool, str, Dict[
                         updated_at = NOW()
                     WHERE id = ?
                     """,
-                    (vip_expires_at, vip_plan, 1 if vip_is_lifetime else 0, user_id),
+                    (vip_expires_at, vip_plan, bool(vip_is_lifetime), user_id),
                 )
 
                 # Credits grants

backend_api_python/app/services/oauth_service.py

@@ -5,7 +5,7 @@
 import secrets
 import requests
 from urllib.parse import urlencode, urlparse
-from datetime import datetime
+from datetime import datetime, timezone, timedelta
 from typing import Tuple, Optional, Dict, Any
 from app.utils.db import get_db_connection
 from app.utils.logger import get_logger
@@ -26,9 +26,121 @@ def get_oauth_service():
 
 class OAuthService:
     """OAuth service for Google and GitHub authentication"""
-    
+
+    # Class-level cache: only create OAuth state table once per process
+    _state_schema_ensured: bool = False
+
     def __init__(self):
         self._load_config()
+
+    def _oauth_state_ttl_minutes(self) -> int:
+        try:
+            return max(5, min(120, int(float(os.getenv("OAUTH_STATE_TTL_MINUTES", "20") or 20))))
+        except Exception:
+            return 20
+
+    def _ensure_oauth_state_schema(self, cur) -> None:
+        """Create qd_oauth_states table if missing.
+
+        OAuth state MUST be shared across Gunicorn workers / replicas; an
+        in-memory dict breaks multi-worker deployments (authorize hits worker
+        A, callback lands on worker B → Invalid state).
+        """
+        if OAuthService._state_schema_ensured:
+            return
+        try:
+            cur.execute(
+                """
+                CREATE TABLE IF NOT EXISTS qd_oauth_states (
+                    state VARCHAR(128) PRIMARY KEY,
+                    provider VARCHAR(20) NOT NULL,
+                    redirect TEXT,
+                    created_at TIMESTAMP DEFAULT NOW(),
+                    expires_at TIMESTAMP NOT NULL
+                )
+                """
+            )
+            cur.execute(
+                "CREATE INDEX IF NOT EXISTS idx_oauth_states_expires ON qd_oauth_states(expires_at)"
+            )
+            OAuthService._state_schema_ensured = True
+        except Exception as e:
+            logger.debug(f"_ensure_oauth_state_schema: {e}")
+
+    def _oauth_state_save(self, state: str, provider: str, redirect: Optional[str]) -> None:
+        # Store naive UTC for TIMESTAMP WITHOUT TIME ZONE columns.
+        exp = (
+            datetime.now(timezone.utc) + timedelta(minutes=self._oauth_state_ttl_minutes())
+        ).replace(tzinfo=None)
+        red = (redirect or "").strip() if redirect else ""
+        with get_db_connection() as db:
+            cur = db.cursor()
+            self._ensure_oauth_state_schema(cur)
+            try:
+                cur.execute("DELETE FROM qd_oauth_states WHERE expires_at < NOW()")
+            except Exception:
+                pass
+            # IMPORTANT: include RETURNING state explicitly.  Otherwise the
+            # PostgresCursor wrapper would auto-append RETURNING id, and this
+            # table has no "id" column → INSERT fails with UndefinedColumn.
+            cur.execute(
+                """
+                INSERT INTO qd_oauth_states (state, provider, redirect, expires_at)
+                VALUES (?, ?, ?, ?)
+                ON CONFLICT (state) DO UPDATE
+                  SET provider = EXCLUDED.provider,
+                      redirect = EXCLUDED.redirect,
+                      expires_at = EXCLUDED.expires_at
+                RETURNING state
+                """,
+                (state, provider, red, exp),
+            )
+            try:
+                cur.fetchone()
+            except Exception:
+                pass
+            db.commit()
+            cur.close()
+
+    def _oauth_state_peek_redirect(self, state: str) -> str:
+        if not state:
+            return ""
+        try:
+            with get_db_connection() as db:
+                cur = db.cursor()
+                self._ensure_oauth_state_schema(cur)
+                cur.execute(
+                    "SELECT redirect FROM qd_oauth_states WHERE state = ? AND expires_at > NOW()",
+                    (state,),
+                )
+                row = cur.fetchone()
+                cur.close()
+                if not row:
+                    return ""
+                return (row.get("redirect") or "").strip()
+        except Exception as e:
+            logger.warning(f"OAuth state peek failed: {e}")
+            return ""
+
+    def _oauth_state_consume(self, state: str, provider: str) -> bool:
+        """Delete and validate state in one statement; True iff a row was removed."""
+        if not state:
+            return False
+        try:
+            with get_db_connection() as db:
+                cur = db.cursor()
+                self._ensure_oauth_state_schema(cur)
+                cur.execute(
+                    "DELETE FROM qd_oauth_states WHERE state = ? AND provider = ? AND expires_at > NOW()",
+                    (state, provider),
+                )
+                n = int(getattr(cur, "rowcount", 0) or 0)
+                db.commit()
+                cur.close()
+                return n > 0
+        except Exception as e:
+            logger.error(f"OAuth state consume failed: {e}", exc_info=True)
+            return False
 
     def _load_config(self):
         """Load OAuth configuration from environment variables"""
@@ -57,9 +169,6 @@ def _load_config(self):
             if origin:
                 self.allowed_redirect_origins.add(origin)
 
-        # State storage (in-memory for simplicity, could use Redis in production)
-        self._states = {}
-
     @staticmethod
     def _normalize_origin(url: str) -> str:
         """Return scheme://host[:port] for a URL, empty string if invalid."""
@@ -82,10 +191,7 @@ def is_redirect_allowed(self, redirect_url: str) -> bool:
 
     def peek_state_redirect(self, state: str) -> str:
         """Read the redirect URL associated with a pending OAuth state (no deletion)."""
-        if not state:
-            return ''
-        entry = self._states.get(state) or {}
-        return entry.get('redirect') or ''
+        return self._oauth_state_peek_redirect(state)
 
     # =========================================================================
     # Google OAuth
@@ -106,11 +212,15 @@ def get_google_auth_url(self, state: str = None, redirect_url: str = None) -> Tu
             return '', ''
 
         state = state or secrets.token_urlsafe(32)
-        state_data = {'provider': 'google', 'created_at': datetime.now()}
+        red = ""
         if redirect_url and self.is_redirect_allowed(redirect_url):
-            state_data['redirect'] = redirect_url
-        self._states[state] = state_data
-        
+            red = redirect_url.strip()
+        try:
+            self._oauth_state_save(state, "google", red or None)
+        except Exception as e:
+            logger.error(f"Failed to persist Google OAuth state: {e}")
+            return '', ''
+
         params = {
             'client_id': self.google_client_id,
             'redirect_uri': self.google_redirect_uri,
@@ -135,12 +245,9 @@ def handle_google_callback(self, code: str, state: str) -> Tuple[bool, Dict[str,
         Returns:
             (success, user_info_or_error)
         """
-        # Validate state
-        if state not in self._states or self._states[state].get('provider') != 'google':
+        if not self._oauth_state_consume(state, "google"):
             return False, {'error': 'Invalid state parameter'}
-        
-        del self._states[state]
-        
+
         try:
             # Exchange code for tokens
             token_response = requests.post(
@@ -204,11 +311,15 @@ def get_github_auth_url(self, state: str = None, redirect_url: str = None) -> Tu
             return '', ''
 
         state = state or secrets.token_urlsafe(32)
-        state_data = {'provider': 'github', 'created_at': datetime.now()}
+        red = ""
         if redirect_url and self.is_redirect_allowed(redirect_url):
-            state_data['redirect'] = redirect_url
-        self._states[state] = state_data
-        
+            red = redirect_url.strip()
+        try:
+            self._oauth_state_save(state, "github", red or None)
+        except Exception as e:
+            logger.error(f"Failed to persist GitHub OAuth state: {e}")
+            return '', ''
+
         params = {
             'client_id': self.github_client_id,
             'redirect_uri': self.github_redirect_uri,
@@ -230,12 +341,9 @@ def handle_github_callback(self, code: str, state: str) -> Tuple[bool, Dict[str,
         Returns:
             (success, user_info_or_error)
         """
-        # Validate state
-        if state not in self._states or self._states[state].get('provider') != 'github':
+        if not self._oauth_state_consume(state, "github"):
             return False, {'error': 'Invalid state parameter'}
-        
-        del self._states[state]
-        
+
         try:
             # Exchange code for token
             token_response = requests.post(
@@ -591,12 +699,16 @@ def unlink_oauth(self, user_id: int, provider: str) -> Tuple[bool, str]:
     # =========================================================================
 
     def cleanup_expired_states(self, max_age_minutes: int = 10):
-        """Clean up expired OAuth states"""
-        cutoff = datetime.now()
-        from datetime import timedelta
-        cutoff = cutoff - timedelta(minutes=max_age_minutes)
-        
-        expired = [k for k, v in self._states.items() 
-                   if v.get('created_at', datetime.now()) < cutoff]
-        for k in expired:
-            del self._states[k]
+        """Delete expired OAuth state rows from the DB.
+        `max_age_minutes` is kept for backward-compat but is ignored — expiry
+        is driven by qd_oauth_states.expires_at which is set on insert.
+        """
+        try:
+            with get_db_connection() as db:
+                cur = db.cursor()
+                self._ensure_oauth_state_schema(cur)
+                cur.execute("DELETE FROM qd_oauth_states WHERE expires_at < NOW()")
+                db.commit()
+                cur.close()
+        except Exception as e:
+            logger.debug(f"cleanup_expired_states: {e}")