Compare commits

..

1 Commits

Author SHA1 Message Date
ي
afcb3d5478 Add token encryption service and Wix token rotation support 2026-05-18 15:57:47 +05:30
3 changed files with 210 additions and 337 deletions

View File

@@ -1,182 +0,0 @@
from __future__ import annotations
import json
from datetime import datetime, timezone
from typing import Optional
from urllib.parse import urlencode
from fastapi import APIRouter, Depends, HTTPException, Query
from fastapi.responses import RedirectResponse
from loguru import logger
from sqlalchemy import text
from sqlalchemy.orm import Session
from services.database import get_db
router = APIRouter(prefix="/v1/social-proxy", tags=["social-proxy"])
def _utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
def _ensure_tables(db: Session) -> None:
# Keep this router backward-compatible on tenant DBs without migrations.
db.execute(text("""
CREATE TABLE IF NOT EXISTS oauth_nonce_sessions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
state TEXT NOT NULL UNIQUE,
nonce TEXT NOT NULL,
user_id TEXT NOT NULL,
platform TEXT NOT NULL,
channel_id INTEGER,
consumed_at TEXT,
expires_at TEXT,
created_at TEXT NOT NULL
)
"""))
db.execute(text("""
CREATE TABLE IF NOT EXISTS social_channels (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id TEXT NOT NULL,
platform TEXT NOT NULL,
platform_account_id TEXT NOT NULL,
token_bundle TEXT NOT NULL,
token_version INTEGER NOT NULL DEFAULT 1,
publication_linkage TEXT,
is_connected INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
UNIQUE(platform, platform_account_id)
)
"""))
def _build_redirect(base_url: str, code: str, message: str, channel_id: Optional[int] = None) -> RedirectResponse:
params = {"code": code, "message": message}
if channel_id is not None:
params["channel_id"] = str(channel_id)
return RedirectResponse(url=f"{base_url}?{urlencode(params)}", status_code=303)
@router.get("/oauth/callback")
def oauth_callback(
state: str = Query(...),
platform: str = Query(...),
account_id: str = Query(...),
token_bundle: str = Query(..., description="Serialized token payload"),
ui_redirect: str = Query("/dashboard/connections"),
db: Session = Depends(get_db),
):
"""Consume OAuth callback, bind to user/platform, and upsert social channel connection."""
_ensure_tables(db)
record = db.execute(
text("""
SELECT id, nonce, user_id, platform, channel_id, consumed_at, expires_at
FROM oauth_nonce_sessions WHERE state = :state
"""),
{"state": state},
).mappings().first()
if not record:
return _build_redirect(ui_redirect, "invalid_state", "Missing OAuth session")
if record["consumed_at"] is not None:
return _build_redirect(ui_redirect, "state_reused", "OAuth state already consumed")
if record["platform"] != platform:
return _build_redirect(ui_redirect, "platform_mismatch", "Platform mismatch")
if record["expires_at"] and record["expires_at"] < _utc_now_iso():
return _build_redirect(ui_redirect, "state_expired", "OAuth session expired")
user_id = record["user_id"]
# Validate token payload is JSON.
try:
parsed_bundle = json.loads(token_bundle)
except json.JSONDecodeError as exc:
raise HTTPException(status_code=400, detail="Invalid token_bundle JSON") from exc
now = _utc_now_iso()
existing = db.execute(
text("""
SELECT id, publication_linkage, token_version
FROM social_channels
WHERE platform = :platform AND platform_account_id = :account_id
"""),
{"platform": platform, "account_id": account_id},
).mappings().first()
if existing:
# Reconnect path: preserve publication linkage and bump token version.
db.execute(
text("""
UPDATE social_channels
SET user_id = :user_id,
token_bundle = :token_bundle,
token_version = :token_version,
is_connected = 1,
updated_at = :updated_at
WHERE id = :id
"""),
{
"id": existing["id"],
"user_id": user_id,
"token_bundle": json.dumps(parsed_bundle),
"token_version": int(existing["token_version"] or 0) + 1,
"updated_at": now,
},
)
channel_id = existing["id"]
result_code = "reconnected"
result_message = "Channel reconnected"
else:
db.execute(
text("""
INSERT INTO social_channels (
user_id, platform, platform_account_id, token_bundle,
token_version, publication_linkage, is_connected, created_at, updated_at
) VALUES (
:user_id, :platform, :account_id, :token_bundle,
1, :publication_linkage, 1, :created_at, :updated_at
)
"""),
{
"user_id": user_id,
"platform": platform,
"account_id": account_id,
"token_bundle": json.dumps(parsed_bundle),
"publication_linkage": None,
"created_at": now,
"updated_at": now,
},
)
channel_id = db.execute(text("SELECT last_insert_rowid()")).scalar_one()
result_code = "connected"
result_message = "Channel connected"
# Bind callback session to concrete channel/user/platform and mark consumed.
db.execute(
text("""
UPDATE oauth_nonce_sessions
SET consumed_at = :consumed_at,
channel_id = :channel_id,
user_id = :user_id,
platform = :platform
WHERE id = :id
"""),
{
"id": record["id"],
"consumed_at": now,
"channel_id": channel_id,
"user_id": user_id,
"platform": platform,
},
)
db.commit()
logger.info(f"OAuth callback complete user={user_id} platform={platform} channel_id={channel_id}")
return _build_redirect(ui_redirect, result_code, result_message, channel_id)

View File

@@ -9,26 +9,27 @@ from typing import Optional, Dict, Any, List
from datetime import datetime, timedelta
from loguru import logger
from services.database import get_user_db_path
from services.token_crypto_service import TokenCryptoService
class WixOAuthService:
"""Manages Wix OAuth2 authentication flow and token storage."""
def __init__(self, db_path: Optional[str] = None):
self.db_path = db_path
self.token_crypto = TokenCryptoService()
def _get_db_path(self, user_id: str) -> str:
if self.db_path:
return self.db_path
return get_user_db_path(user_id)
def _init_db(self, user_id: str):
"""Initialize database tables for OAuth tokens."""
db_path = self._get_db_path(user_id)
# Ensure directory exists
os.makedirs(os.path.dirname(db_path), exist_ok=True)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
@@ -45,168 +46,133 @@ class WixOAuthService:
member_id TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_active BOOLEAN DEFAULT TRUE
is_active BOOLEAN DEFAULT TRUE,
token_key_version TEXT,
token_key_reference TEXT
)
''')
for column_name, column_def in [
("token_key_version", "TEXT"),
("token_key_reference", "TEXT"),
]:
try:
cursor.execute(f"ALTER TABLE wix_oauth_tokens ADD COLUMN {column_name} {column_def}")
except sqlite3.OperationalError:
pass
conn.commit()
def store_tokens(
self,
user_id: str,
access_token: str,
refresh_token: Optional[str] = None,
expires_in: Optional[int] = None,
token_type: str = 'bearer',
scope: Optional[str] = None,
site_id: Optional[str] = None,
member_id: Optional[str] = None
) -> bool:
"""
Store Wix OAuth tokens in the database.
Args:
user_id: User ID (Clerk string)
access_token: Access token from Wix
refresh_token: Optional refresh token
expires_in: Optional expiration time in seconds
token_type: Token type (default: 'bearer')
scope: Optional OAuth scope
site_id: Optional Wix site ID
member_id: Optional Wix member ID
Returns:
True if tokens were stored successfully
"""
def store_tokens(self, user_id: str, access_token: str, refresh_token: Optional[str] = None,
expires_in: Optional[int] = None, token_type: str = 'bearer', scope: Optional[str] = None,
site_id: Optional[str] = None, member_id: Optional[str] = None) -> bool:
try:
# Ensure DB is initialized for this user
self._init_db(user_id)
db_path = self._get_db_path(user_id)
expires_at = None
if expires_in:
expires_at = datetime.now() + timedelta(seconds=expires_in)
expires_at = datetime.now() + timedelta(seconds=expires_in) if expires_in else None
encrypted_access_token, encrypted_refresh_token = self.token_crypto.encrypt_pair(access_token, refresh_token)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO wix_oauth_tokens
(user_id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (user_id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id))
INSERT INTO wix_oauth_tokens
(user_id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, token_key_version, token_key_reference)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
user_id,
encrypted_access_token,
encrypted_refresh_token,
token_type,
expires_at,
expires_in,
scope,
site_id,
member_id,
self.token_crypto.key_version,
self.token_crypto.key_reference,
))
conn.commit()
logger.info(f"Wix OAuth: Token inserted into database for user {user_id}")
logger.info(f"Wix OAuth: Encrypted token stored for user {user_id}")
return True
except Exception as e:
logger.error(f"Error storing Wix tokens for user {user_id}: {e}")
return False
def get_user_tokens(self, user_id: str) -> List[Dict[str, Any]]:
"""Get all active Wix tokens for a user."""
"""Get all active Wix token rows (encrypted values)."""
try:
# Ensure database tables exist to prevent 'no such table' errors
self._init_db(user_id)
db_path = self._get_db_path(user_id)
if not os.path.exists(db_path):
return []
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, created_at
SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, created_at, token_key_version, token_key_reference
FROM wix_oauth_tokens
WHERE user_id = ? AND is_active = TRUE AND (expires_at IS NULL OR expires_at > datetime('now'))
ORDER BY created_at DESC
''', (user_id,))
tokens = []
for row in cursor.fetchall():
tokens.append({
"id": row[0],
"access_token": row[1],
"refresh_token": row[2],
"token_type": row[3],
"expires_at": row[4],
"expires_in": row[5],
"scope": row[6],
"site_id": row[7],
"member_id": row[8],
"created_at": row[9]
})
return tokens
return [{
"id": row[0], "access_token": row[1], "refresh_token": row[2], "token_type": row[3],
"expires_at": row[4], "expires_in": row[5], "scope": row[6], "site_id": row[7],
"member_id": row[8], "created_at": row[9], "token_key_version": row[10],
"token_key_reference": row[11]
} for row in cursor.fetchall()]
except Exception as e:
logger.error(f"Error getting Wix tokens for user {user_id}: {e}")
return []
def get_user_token_status(self, user_id: str) -> Dict[str, Any]:
"""Get detailed token status for a user including expired tokens."""
try:
# Ensure database tables exist to prevent 'no such table' errors
self._init_db(user_id)
def get_user_tokens_decrypted(self, user_id: str) -> List[Dict[str, Any]]:
"""Decrypt tokens for integration managers and token refresh routines."""
decrypted = []
for token in self.get_user_tokens(user_id):
token_copy = dict(token)
token_copy["access_token"] = self.token_crypto.decrypt_token(token_copy.get("access_token"))
token_copy["refresh_token"] = self.token_crypto.decrypt_token(token_copy.get("refresh_token"))
decrypted.append(token_copy)
return decrypted
def get_user_token_status(self, user_id: str) -> Dict[str, Any]:
try:
self._init_db(user_id)
db_path = self._get_db_path(user_id)
if not os.path.exists(db_path):
return {
"has_tokens": False,
"has_active_tokens": False,
"has_expired_tokens": False,
"active_tokens": [],
"expired_tokens": [],
"total_tokens": 0,
"last_token_date": None
}
return {"has_tokens": False, "has_active_tokens": False, "has_expired_tokens": False,
"active_tokens": [], "expired_tokens": [], "total_tokens": 0, "last_token_date": None}
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
# Get all tokens (active and expired)
cursor.execute('''
SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id, created_at, is_active
SELECT id, access_token, refresh_token, token_type, expires_at, expires_in, scope, site_id, member_id,
created_at, is_active, token_key_version, token_key_reference
FROM wix_oauth_tokens
WHERE user_id = ?
ORDER BY created_at DESC
''', (user_id,))
all_tokens = []
active_tokens = []
expired_tokens = []
all_tokens, active_tokens, expired_tokens = [], [], []
for row in cursor.fetchall():
token_data = {
"id": row[0],
"access_token": row[1],
"refresh_token": row[2],
"token_type": row[3],
"expires_at": row[4],
"expires_in": row[5],
"scope": row[6],
"site_id": row[7],
"member_id": row[8],
"created_at": row[9],
"is_active": bool(row[10])
"id": row[0], "access_token": row[1], "refresh_token": row[2], "token_type": row[3],
"expires_at": row[4], "expires_in": row[5], "scope": row[6], "site_id": row[7],
"member_id": row[8], "created_at": row[9], "is_active": bool(row[10]),
"token_key_version": row[11], "token_key_reference": row[12]
}
all_tokens.append(token_data)
# Determine expiry using robust parsing and is_active flag
is_active_flag = bool(row[10])
not_expired = False
try:
expires_at_val = row[4]
if expires_at_val:
# First try Python parsing
try:
dt = datetime.fromisoformat(expires_at_val) if isinstance(expires_at_val, str) else expires_at_val
not_expired = dt > datetime.now()
except Exception:
# Fallback to SQLite comparison
cursor.execute("SELECT datetime('now') < ?", (expires_at_val,))
not_expired = cursor.fetchone()[0] == 1
else:
# No expiry stored => consider not expired
not_expired = True
except Exception:
not_expired = False
@@ -215,7 +181,7 @@ class WixOAuthService:
active_tokens.append(token_data)
else:
expired_tokens.append(token_data)
return {
"has_tokens": len(all_tokens) > 0,
"has_active_tokens": len(active_tokens) > 0,
@@ -225,81 +191,101 @@ class WixOAuthService:
"total_tokens": len(all_tokens),
"last_token_date": all_tokens[0]["created_at"] if all_tokens else None
}
except Exception as e:
logger.error(f"Error getting Wix token status for user {user_id}: {e}")
return {
"has_tokens": False,
"has_active_tokens": False,
"has_expired_tokens": False,
"active_tokens": [],
"expired_tokens": [],
"total_tokens": 0,
"last_token_date": None,
"error": str(e)
}
def update_tokens(
self,
user_id: str,
access_token: str,
refresh_token: Optional[str] = None,
expires_in: Optional[int] = None
) -> bool:
"""Update tokens for a user (e.g., after refresh)."""
return {"has_tokens": False, "has_active_tokens": False, "has_expired_tokens": False,
"active_tokens": [], "expired_tokens": [], "total_tokens": 0, "last_token_date": None, "error": str(e)}
def update_tokens(self, user_id: str, access_token: str, refresh_token: Optional[str] = None,
expires_in: Optional[int] = None) -> bool:
try:
# Ensure DB initialized for this user
self._init_db(user_id)
db_path = self._get_db_path(user_id)
expires_at = datetime.now() + timedelta(seconds=expires_in) if expires_in else None
encrypted_access_token = self.token_crypto.encrypt_token(access_token)
encrypted_refresh_token = self.token_crypto.encrypt_token(refresh_token) if refresh_token else None
expires_at = None
if expires_in:
expires_at = datetime.now() + timedelta(seconds=expires_in)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
if refresh_token:
cursor.execute('''
UPDATE wix_oauth_tokens
SET access_token = ?, refresh_token = ?, expires_at = ?, expires_in = ?,
is_active = TRUE, updated_at = datetime('now')
WHERE user_id = ? AND refresh_token = ?
''', (access_token, refresh_token, expires_at, expires_in, user_id, refresh_token))
UPDATE wix_oauth_tokens
SET access_token = ?, refresh_token = ?, expires_at = ?, expires_in = ?,
is_active = TRUE, updated_at = datetime('now'), token_key_version = ?, token_key_reference = ?
WHERE user_id = ? AND (refresh_token = ? OR refresh_token = ?)
''', (encrypted_access_token, encrypted_refresh_token, expires_at, expires_in,
self.token_crypto.key_version, self.token_crypto.key_reference,
user_id, encrypted_refresh_token, refresh_token))
else:
cursor.execute('''
UPDATE wix_oauth_tokens
SET access_token = ?, expires_at = ?, expires_in = ?,
is_active = TRUE, updated_at = datetime('now')
UPDATE wix_oauth_tokens
SET access_token = ?, expires_at = ?, expires_in = ?,
is_active = TRUE, updated_at = datetime('now'), token_key_version = ?, token_key_reference = ?
WHERE user_id = ? AND id = (SELECT id FROM wix_oauth_tokens WHERE user_id = ? ORDER BY created_at DESC LIMIT 1)
''', (access_token, expires_at, expires_in, user_id, user_id))
''', (encrypted_access_token, expires_at, expires_in,
self.token_crypto.key_version, self.token_crypto.key_reference, user_id, user_id))
conn.commit()
logger.info(f"Wix OAuth: Tokens updated for user {user_id}")
logger.info(f"Wix OAuth: Encrypted tokens updated for user {user_id}")
return True
except Exception as e:
logger.error(f"Error updating Wix tokens for user {user_id}: {e}")
return False
def rotate_token_encryption(self, user_id: str, batch_size: int = 100) -> Dict[str, int]:
"""Re-encrypt existing token rows in batches for key rotation."""
self._init_db(user_id)
db_path = self._get_db_path(user_id)
rotated, skipped, last_id = 0, 0, 0
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
while True:
cursor.execute('''
SELECT id, access_token, refresh_token
FROM wix_oauth_tokens
WHERE user_id = ? AND id > ?
ORDER BY id ASC
LIMIT ?
''', (user_id, last_id, batch_size))
rows = cursor.fetchall()
if not rows:
break
for row_id, enc_access, enc_refresh in rows:
last_id = row_id
try:
plain_access = self.token_crypto.decrypt_token(enc_access)
plain_refresh = self.token_crypto.decrypt_token(enc_refresh) if enc_refresh else None
except Exception:
skipped += 1
continue
new_access, new_refresh = self.token_crypto.encrypt_pair(plain_access, plain_refresh)
cursor.execute('''
UPDATE wix_oauth_tokens
SET access_token = ?, refresh_token = ?, token_key_version = ?, token_key_reference = ?, updated_at = datetime('now')
WHERE id = ?
''', (new_access, new_refresh, self.token_crypto.key_version, self.token_crypto.key_reference, row_id))
rotated += 1
conn.commit()
logger.info(f"Wix OAuth: Encryption rotation complete for user {user_id}; rotated={rotated}, skipped={skipped}")
return {"rotated": rotated, "skipped": skipped}
def revoke_token(self, user_id: str, token_id: int) -> bool:
"""Revoke a Wix OAuth token."""
try:
db_path = self._get_db_path(user_id)
with sqlite3.connect(db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
UPDATE wix_oauth_tokens
UPDATE wix_oauth_tokens
SET is_active = FALSE, updated_at = datetime('now')
WHERE user_id = ? AND id = ?
''', (user_id, token_id))
conn.commit()
if cursor.rowcount > 0:
logger.info(f"Wix token {token_id} revoked for user {user_id}")
return True
return False
except Exception as e:
logger.error(f"Error revoking Wix token: {e}")
return False

View File

@@ -0,0 +1,69 @@
"""Service for encrypting/decrypting integration tokens with key version metadata."""
import base64
import hashlib
import os
from typing import Optional, Tuple
from cryptography.fernet import Fernet, InvalidToken
from loguru import logger
class TokenCryptoService:
"""Token encryption/decryption service with key version support."""
ENV_KEY = "ALWRITY_TOKEN_ENCRYPTION_KEY"
ENV_KEY_VERSION = "ALWRITY_TOKEN_KEY_VERSION"
def __init__(self):
raw_key = os.getenv(self.ENV_KEY, "")
if raw_key:
self._fernet_key = self._normalize_key(raw_key)
else:
self._fernet_key = self._derive_dev_key()
self._fernet = Fernet(self._fernet_key)
self._key_version = os.getenv(self.ENV_KEY_VERSION, "v1")
self._key_reference = self._fingerprint(self._fernet_key)
@property
def key_version(self) -> str:
return self._key_version
@property
def key_reference(self) -> str:
return self._key_reference
def encrypt_token(self, token: Optional[str]) -> Optional[str]:
if token is None:
return None
return self._fernet.encrypt(token.encode("utf-8")).decode("utf-8")
def decrypt_token(self, encrypted_token: Optional[str]) -> Optional[str]:
if encrypted_token is None:
return None
try:
return self._fernet.decrypt(encrypted_token.encode("utf-8")).decode("utf-8")
except InvalidToken:
logger.error("Token decryption failed due to invalid token/key")
raise
def encrypt_pair(self, access_token: str, refresh_token: Optional[str]) -> Tuple[str, Optional[str]]:
return self.encrypt_token(access_token), self.encrypt_token(refresh_token)
@staticmethod
def _normalize_key(raw_key: str) -> bytes:
raw_key = raw_key.strip()
if len(raw_key) == 44 and raw_key.endswith("="):
return raw_key.encode("utf-8")
digest = hashlib.sha256(raw_key.encode("utf-8")).digest()
return base64.urlsafe_b64encode(digest)
@staticmethod
def _derive_dev_key() -> bytes:
seed = "alwrity-local-token-key"
digest = hashlib.sha256(seed.encode("utf-8")).digest()
return base64.urlsafe_b64encode(digest)
@staticmethod
def _fingerprint(key: bytes) -> str:
return hashlib.sha256(key).hexdigest()[:16]