Production-ready improvements: security hardening, Redis sessions, retry logic, updated pricing

Changes:
- Removed hardcoded admin credentials (now requires env vars)
- Added Redis session storage with in-memory fallback
- Improved CORS configuration with warnings for development mode
- Added retry_with_backoff decorator for translation API calls
- Updated pricing: Starter=, Pro=, Business=
- Stripe price IDs now loaded from environment variables
- Added redis to requirements.txt
- Updated .env.example with all new configuration options
- Created COMPREHENSIVE_REVIEW_AND_PLAN.md with deployment roadmap
- Frontend: Updated pricing page, new UI components
This commit is contained in:
2025-12-31 10:43:31 +01:00
parent 721b18dbbd
commit c4d6cae735
27 changed files with 7824 additions and 2181 deletions

105
main.py
View File

@@ -41,13 +41,37 @@ logging.basicConfig(
logger = logging.getLogger(__name__)
# ============== Admin Authentication ==============
ADMIN_USERNAME = os.getenv("ADMIN_USERNAME", "admin")
ADMIN_PASSWORD_HASH = os.getenv("ADMIN_PASSWORD_HASH", "") # SHA256 hash of password
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "changeme123") # Default password (change in production!)
ADMIN_USERNAME = os.getenv("ADMIN_USERNAME")
ADMIN_PASSWORD_HASH = os.getenv("ADMIN_PASSWORD_HASH") # SHA256 hash of password (preferred)
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD") # Plain password (use hash in production!)
ADMIN_TOKEN_SECRET = os.getenv("ADMIN_TOKEN_SECRET", secrets.token_hex(32))
# Store active admin sessions (token -> expiry timestamp)
admin_sessions: dict = {}
# Validate admin credentials are configured
if not ADMIN_USERNAME:
logger.warning("⚠️ ADMIN_USERNAME not set - admin endpoints will be disabled")
if not ADMIN_PASSWORD_HASH and not ADMIN_PASSWORD:
logger.warning("⚠️ ADMIN_PASSWORD/ADMIN_PASSWORD_HASH not set - admin endpoints will be disabled")
# Redis connection for sessions (fallback to in-memory if not available)
REDIS_URL = os.getenv("REDIS_URL", "")
_redis_client = None
def get_redis_client():
"""Get Redis client for session storage"""
global _redis_client
if _redis_client is None and REDIS_URL:
try:
import redis
_redis_client = redis.from_url(REDIS_URL, decode_responses=True)
_redis_client.ping()
logger.info("✅ Connected to Redis for session storage")
except Exception as e:
logger.warning(f"⚠️ Redis connection failed: {e}. Using in-memory sessions.")
_redis_client = False # Mark as failed
return _redis_client if _redis_client else None
# In-memory fallback for sessions (not recommended for production)
_memory_sessions: dict = {}
def hash_password(password: str) -> str:
"""Hash password with SHA256"""
@@ -55,28 +79,70 @@ def hash_password(password: str) -> str:
def verify_admin_password(password: str) -> bool:
"""Verify admin password"""
if not ADMIN_PASSWORD_HASH and not ADMIN_PASSWORD:
return False # No credentials configured
if ADMIN_PASSWORD_HASH:
return hash_password(password) == ADMIN_PASSWORD_HASH
return password == ADMIN_PASSWORD
def _get_session_key(token: str) -> str:
"""Get Redis key for session token"""
return f"admin_session:{token}"
def create_admin_token() -> str:
"""Create a new admin session token"""
"""Create a new admin session token with Redis or memory fallback"""
token = secrets.token_urlsafe(32)
# Token expires in 24 hours
admin_sessions[token] = time.time() + (24 * 60 * 60)
expiry = int(time.time()) + (24 * 60 * 60) # 24 hours
redis_client = get_redis_client()
if redis_client:
try:
redis_client.setex(_get_session_key(token), 24 * 60 * 60, str(expiry))
except Exception as e:
logger.warning(f"Redis session save failed: {e}")
_memory_sessions[token] = expiry
else:
_memory_sessions[token] = expiry
return token
def verify_admin_token(token: str) -> bool:
"""Verify admin token is valid and not expired"""
if token not in admin_sessions:
redis_client = get_redis_client()
if redis_client:
try:
expiry = redis_client.get(_get_session_key(token))
if expiry and int(expiry) > time.time():
return True
return False
except Exception as e:
logger.warning(f"Redis session check failed: {e}")
# Fallback to memory
if token not in _memory_sessions:
return False
if time.time() > admin_sessions[token]:
del admin_sessions[token]
if time.time() > _memory_sessions[token]:
del _memory_sessions[token]
return False
return True
def delete_admin_token(token: str):
"""Delete an admin session token"""
redis_client = get_redis_client()
if redis_client:
try:
redis_client.delete(_get_session_key(token))
except Exception:
pass
if token in _memory_sessions:
del _memory_sessions[token]
async def require_admin(authorization: Optional[str] = Header(None)) -> bool:
"""Dependency to require admin authentication"""
if not ADMIN_USERNAME or (not ADMIN_PASSWORD_HASH and not ADMIN_PASSWORD):
raise HTTPException(status_code=503, detail="Admin authentication not configured")
if not authorization:
raise HTTPException(status_code=401, detail="Authorization header required")
@@ -164,11 +230,19 @@ app.add_middleware(SecurityHeadersMiddleware, config={"enable_hsts": os.getenv("
app.add_middleware(RateLimitMiddleware, rate_limit_manager=rate_limit_manager)
# CORS - configure for production
allowed_origins = os.getenv("CORS_ORIGINS", "*").split(",")
# WARNING: Do not use "*" in production! Set CORS_ORIGINS to your actual frontend domains
_cors_env = os.getenv("CORS_ORIGINS", "")
if _cors_env == "*" or not _cors_env:
logger.warning("⚠️ CORS_ORIGINS not properly configured. Using permissive settings for development only!")
allowed_origins = ["*"]
else:
allowed_origins = [origin.strip() for origin in _cors_env.split(",") if origin.strip()]
logger.info(f"✅ CORS configured for origins: {allowed_origins}")
app.add_middleware(
CORSMiddleware,
allow_origins=allowed_origins,
allow_credentials=True,
allow_credentials=True if allowed_origins != ["*"] else False, # Can't use credentials with wildcard
allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
allow_headers=["*"],
expose_headers=["X-Request-ID", "X-Original-Filename", "X-File-Size-MB", "X-Target-Language"]
@@ -891,9 +965,8 @@ async def admin_logout(authorization: Optional[str] = Header(None)):
parts = authorization.split(" ")
if len(parts) == 2 and parts[0].lower() == "bearer":
token = parts[1]
if token in admin_sessions:
del admin_sessions[token]
logger.info("Admin logout successful")
delete_admin_token(token)
logger.info("Admin logout successful")
return {"status": "success", "message": "Logged out"}