WangShen007/replypier
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

build: 打通 FastAPI 与 Vue 最小工程、依赖与数据库连接

Browse Source 
后端入口、配置、Redis 与 Alembic 迁移就位；前端 Vite 工程可安装可构建。
This commit is contained in:
WangShen007 committed 2026-09-22 20:15:00 +08:00
1 parent ec5835ca77
commit 977783c126
167 files changed
+26082
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/core/application.py
+43
View File
@@ -0,0 +1,43 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from fastapi import FastAPI
from app.core.error_handlers import install_error_handlers
from app.core.logger import get_logger
logger = get_logger(__name__)
# Create FastAPI app instance
app = FastAPI(
title="ReplyPier API",
description="AI-Powered Customer Support Platform",
version="1.0.0"
)
# 5xx bodies must not carry our own error text; see error_handlers.
install_error_handlers(app)
def initialize_cors_listener():
"""
Initialize the CORS listener for multi-worker environments
This function is called during application startup
"""
try:
from app.core.cors import start_cors_listener
start_cors_listener(app)
logger.info("CORS listener initialized")
except Exception as e:
logger.error(f"Failed to initialize CORS listener: {str(e)}")
backend/app/core/auth.py
+517
View File
@@ -0,0 +1,517 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from fastapi import Depends, HTTPException, status, Cookie, Request
from fastapi.security import HTTPBearer
from sqlalchemy.orm import Session
from typing import Callable, Iterable, Optional, List
from uuid import UUID
from app.database import get_db
from app.models.user import User
from app.models.organization import Organization
from app.services.auth_sessions import ACCESS, live_payload
from app.core.logger import get_logger
from app.models.role import Role
logger = get_logger(__name__)
# Enterprise: Personal Access Tokens (PATs). PATs are an enterprise-only feature, so the
# resolver lives in the enterprise package. In the community edition that package is absent
# and this falls back to a no-op, meaning "rpat_" bearer tokens simply fail normal JWT
# verification like any other invalid token. See app/enterprise/services/pat.py.
#
# The enterprise resolver is imported lazily (on first use) rather than at module load: the
# enterprise package pulls in the API routers, which import this module, so an eager import
# here would create a circular import during application start-up.
PAT_TOKEN_PREFIX = "rpat_"
_pat_resolver = None
_pat_resolver_loaded = False
def _resolve_pat_user(token: str, db: Session) -> Optional[User]:
global _pat_resolver, _pat_resolver_loaded
if not _pat_resolver_loaded:
_pat_resolver_loaded = True
try:
from app.enterprise.services.pat import resolve_pat_user
_pat_resolver = resolve_pat_user
except ImportError: # community edition / enterprise package not installed
_pat_resolver = None
if _pat_resolver is None:
return None
return _pat_resolver(token, db)
def check_permissions(user: User, required_permissions: List[str]) -> bool:
"""Check if user has required permissions through their role"""
if not user.role or not user.role.permissions:
return False
user_permissions = [p.name for p in user.role.permissions]
# If user has super_admin permission, they have access to everything
if "super_admin" in user_permissions:
return True
return all(perm in user_permissions for perm in required_permissions)
def require_grantable(current_user: User, permission_names: Iterable[str]) -> None:
"""You cannot grant a permission you do not hold yourself.
Without this, manage_roles was equivalent to super_admin — and in fact so
was *any* authenticated session, since the /roles endpoints carried no
permission check at all: POST /roles/{my_role_id}/permissions/super_admin
was enough. check_permissions short-circuits on super_admin, so an org
owner granting anything is unaffected.
Lives here rather than in api/roles.py because granting is not confined to
that router — the chat-scope toggles on the user form widen a role too.
"""
ungrantable = sorted(
name for name in permission_names
if not check_permissions(current_user, [name])
)
if ungrantable:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Cannot grant a permission you do not hold: {', '.join(ungrantable)}"
)
def require_permissions(*required_permissions: str):
"""Dependency to check user permissions"""
async def permission_checker(
current_user: User = Depends(get_current_user)
) -> User:
if not check_permissions(current_user, list(required_permissions)):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
return permission_checker
# Seeing conversations. Either scope grants it — which require_permissions,
# whose semantics are AND, cannot express.
#
# manage_all_chats, not "manage_chats" — the latter appears in several hand-
# rolled checks but is not in Permission.default_permissions(), so it has
# never matched anyone.
CHAT_VIEW_PERMISSIONS = (
"view_all_chats",
"view_assigned_chats",
"view_unassigned_chats",
)
# Acting on a conversation: taking it over, reassigning it, sending outbound.
# Deliberately NOT derived from INBOX_PERMISSIONS — folding the view grants in
# here would let a read-only role reassign other people's conversations.
CHAT_MANAGE_PERMISSIONS = ("manage_all_chats", "manage_assigned_chats")
# Working the inbox, in the sense the surfaces around it mean it: an agent who
# can open a conversation must also be able to read the channel accounts and
# message templates needed to answer one, or the feature exists only in theory.
# This used to be just the two org-wide grants, which excluded the very agents
# the comment described.
INBOX_PERMISSIONS = CHAT_VIEW_PERMISSIONS + CHAT_MANAGE_PERMISSIONS
# The narrower org-wide scope, for the places that genuinely mean "can see the
# whole org's chats" rather than "works the inbox".
ORG_CHAT_PERMISSIONS = ("view_all_chats", "manage_all_chats")
# The people directory.
PEOPLE_READ_PERMISSIONS = ("view_people",) + INBOX_PERMISSIONS
# Correcting a person's details, marking them a customer and pushing them to a
# connected CRM are inbox work, not administration — people.py documents the
# PATCH as the agent's identification tool. So view_people grants both; there
# is no separate read-only directory role today.
PEOPLE_WRITE_PERMISSIONS = PEOPLE_READ_PERMISSIONS
def has_any_permission(user: User, permissions: Iterable[str]) -> bool:
"""True when the user holds AT LEAST ONE of `permissions`.
The OR counterpart to check_permissions, and it honours the same
super_admin bypass — hand-rolled `isdisjoint` checks kept forgetting it,
so a super_admin could send templates but got 403 from People.
"""
if not user.role or not user.role.permissions:
return False
held = {p.name for p in user.role.permissions}
return "super_admin" in held or not held.isdisjoint(set(permissions))
def require_any_permission(*permissions: str):
"""Dependency: the user needs any one of `permissions`."""
async def permission_checker(
current_user: User = Depends(get_current_user)
) -> User:
if not has_any_permission(current_user, permissions):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
return permission_checker
def request_token(request: Request) -> Optional[str]:
"""The bearer credential a request carries: the access_token cookie, else
an Authorization: Bearer header."""
cookie_token = request.cookies.get('access_token')
# Ensure we got a valid string, not a mock or None
if cookie_token and isinstance(cookie_token, str):
return cookie_token
auth_header = request.headers.get('Authorization')
if auth_header and auth_header.startswith('Bearer '):
return auth_header.split(' ')[1]
return None
def _unauthorized(detail: str) -> HTTPException:
return HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=detail,
headers={"WWW-Authenticate": "Bearer"},
)
def _jwt_user(access_token: str, db: Session) -> User:
"""The active user a live access JWT belongs to, or a 401.
live_payload rejects refresh tokens and tokens whose session was revoked
(signed out, or ended by a password change).
"""
payload = live_payload(db, access_token, ACCESS)
if payload is None:
raise _unauthorized("Invalid authentication token")
user_id = payload.get("sub")
if user_id is None:
raise _unauthorized("Invalid token payload")
user = db.query(User).filter(User.id == UUID(str(user_id))).first()
if user is None:
raise _unauthorized("User not found")
if not user.is_active:
raise _unauthorized("User is inactive")
return user
async def get_current_user(
request: Request,
access_token: Optional[str] = Cookie(None),
db: Session = Depends(get_db)
) -> User:
"""Get current user from access token cookie"""
try:
if not access_token or not isinstance(access_token, str):
access_token = request_token(request)
if not access_token:
raise _unauthorized("Not authenticated")
# Enterprise: a Personal Access Token resolves directly to its owning user.
if access_token.startswith(PAT_TOKEN_PREFIX):
pat_user = _resolve_pat_user(access_token, db)
if pat_user is not None and pat_user.is_active:
return pat_user
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or revoked access token",
headers={"WWW-Authenticate": "Bearer"},
)
return _jwt_user(access_token, db)
except HTTPException:
raise
except Exception as e:
logger.error(f"Authentication error: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
async def get_current_organization(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
) -> Organization:
"""Get current user's organization"""
if not current_user.organization_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User is not associated with any organization"
)
organization = db.query(Organization).filter(
Organization.id == current_user.organization_id,
Organization.is_active == True
).first()
if not organization:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Organization not found or inactive"
)
return organization
def require_permission(permission: str):
"""Dependency for checking if user has specific permission"""
async def permission_dependency(
current_user: User = Depends(get_current_user)
) -> User:
if not current_user.role or not current_user.role.has_permission(permission):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"User does not have required permission: {permission}"
)
return current_user
return permission_dependency
def require_subscription_management(
current_user: User = Depends(get_current_user)
) -> User:
"""Dependency for checking if user can manage subscriptions"""
if not current_user.role or not current_user.role.can_manage_subscription():
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User does not have permission to manage subscriptions"
)
return current_user
# Unified Authentication Functions
def get_auth_info_from_request(request: Request) -> dict:
"""Determine if this is a Shopify request based on the URL path"""
return {"is_shopify": "/shopify" in str(request.url)}
async def get_unified_auth(request: Request, db: Session = Depends(get_db)) -> dict:
"""Authenticate a JWT, a PAT, or a Shopify session token.
Authentication only — it deliberately holds no permission check. Wrap it in
require_unified_permissions / require_any_unified_permission to declare what
a route needs.
"""
from app.services.shopify_session import require_shopify_or_jwt_auth
try:
# Check if this is a Shopify-related endpoint or has query params indicating Shopify context
url_str = str(request.url)
query_params = dict(request.query_params)
is_shopify_context = (
"/shopify" in url_str or
url_str.endswith("/shopify") or
"shop" in query_params or
"host" in query_params
)
if is_shopify_context:
# For Shopify endpoints or context, use the shopify/JWT auth
auth_result = await require_shopify_or_jwt_auth(request, db)
return {
"auth_type": auth_result.get("auth_type", "shopify"),
"organization_id": auth_result["organization_id"],
"user_id": auth_result.get("user_id"),
"current_user": auth_result.get("current_user")
}
else:
# For regular endpoints, use JWT auth - inline the logic to avoid calling dependency as function
try:
# Get token from cookie or Authorization header
access_token = request_token(request)
if not access_token:
raise _unauthorized("Not authenticated")
# Enterprise: a Personal Access Token resolves directly to its owning user.
if access_token.startswith(PAT_TOKEN_PREFIX):
current_user = _resolve_pat_user(access_token, db)
if current_user is None or not current_user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or revoked access token",
headers={"WWW-Authenticate": "Bearer"},
)
return {
"auth_type": "pat",
"organization_id": current_user.organization_id,
"user_id": current_user.id,
"current_user": current_user
}
current_user = _jwt_user(access_token, db)
return {
"auth_type": "jwt",
"organization_id": current_user.organization_id, # Keep as UUID
"user_id": current_user.id,
"current_user": current_user
}
except HTTPException:
raise
except Exception as e:
logger.error(f"JWT authentication error in unified auth: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
except HTTPException:
# Re-raise HTTP exceptions as-is (don't wrap them)
raise
except Exception as e:
logger.error(f"Authentication error: {str(e)}")
raise HTTPException(status_code=401, detail="Authentication failed")
def _authorize_unified(auth_info: dict, permitted: Callable[[User], bool]) -> dict:
"""Apply a permission check to a get_unified_auth result.
A Shopify shop session authenticates a *store*, not a person, so it carries
no `current_user` and there is no role to check — those requests pass
through exactly as they did before authorization moved out of the helper.
"""
current_user = auth_info.get("current_user")
if current_user is None:
return auth_info
if not permitted(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return auth_info
def require_unified_permissions(*required_permissions: str):
"""get_unified_auth + AND-semantics permissions, for JWT/PAT callers.
get_unified_auth used to hardcode a manage_agents check, which meant every
route sharing the helper inherited it — including reading your own org's
subscription, which locked every non-admin out of the whole app. The helper
now only authenticates; each route declares what it needs.
"""
async def permission_checker(
auth_info: dict = Depends(get_unified_auth)
) -> dict:
return _authorize_unified(
auth_info,
lambda user: check_permissions(user, list(required_permissions)),
)
return permission_checker
def require_any_unified_permission(*permissions: str):
"""The OR counterpart of require_unified_permissions."""
async def permission_checker(
auth_info: dict = Depends(get_unified_auth)
) -> dict:
return _authorize_unified(
auth_info,
lambda user: has_any_permission(user, permissions),
)
return permission_checker
async def get_unified_chat_auth(request: Request, db: Session = Depends(get_db)) -> dict:
"""Unified authentication that handles both regular JWT and Shopify session tokens for chat endpoints"""
from app.services.shopify_session import require_shopify_or_jwt_auth
try:
if "/shopify" in str(request.url):
# For Shopify endpoints, use the shopify auth
auth_result = await require_shopify_or_jwt_auth(request, db)
return {
"auth_type": auth_result.get("auth_type", "shopify"),
"organization_id": auth_result["organization_id"],
"user_id": auth_result.get("user_id"),
"current_user": auth_result.get("current_user")
}
else:
# For regular endpoints, use JWT auth with chat permissions - inline the logic
try:
# Get token from cookie or Authorization header
access_token = request_token(request)
if not access_token:
raise _unauthorized("Not authenticated")
# Enterprise: a Personal Access Token resolves directly to its owning user.
if access_token.startswith(PAT_TOKEN_PREFIX):
current_user = _resolve_pat_user(access_token, db)
if current_user is None or not current_user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or revoked access token",
headers={"WWW-Authenticate": "Bearer"},
)
pat_permissions = {p.name for p in current_user.role.permissions}
pat_is_super_admin = "super_admin" in pat_permissions
pat_can_view_all = pat_is_super_admin or "view_all_chats" in pat_permissions
pat_can_view_assigned = pat_is_super_admin or "view_assigned_chats" in pat_permissions
pat_can_view_unassigned = pat_is_super_admin or "view_unassigned_chats" in pat_permissions
if not (pat_can_view_all or pat_can_view_assigned or pat_can_view_unassigned):
raise HTTPException(status_code=403, detail="Not enough permissions")
return {
"auth_type": "pat",
"organization_id": current_user.organization_id,
"user_id": current_user.id,
"current_user": current_user,
"can_view_all": pat_can_view_all,
"can_view_assigned": pat_can_view_assigned,
"can_view_unassigned": pat_can_view_unassigned
}
current_user = _jwt_user(access_token, db)
# Check chat permissions
user_permissions = {p.name for p in current_user.role.permissions}
is_super_admin = "super_admin" in user_permissions
can_view_all = is_super_admin or "view_all_chats" in user_permissions
can_view_assigned = is_super_admin or "view_assigned_chats" in user_permissions
# The unclaimed AI queue — sessions no human has taken yet
can_view_unassigned = is_super_admin or "view_unassigned_chats" in user_permissions
if not (can_view_all or can_view_assigned or can_view_unassigned):
raise HTTPException(
status_code=403,
detail="Not enough permissions"
)
return {
"auth_type": "jwt",
"organization_id": current_user.organization_id, # Keep as UUID
"user_id": current_user.id,
"current_user": current_user,
"can_view_all": can_view_all,
"can_view_assigned": can_view_assigned,
"can_view_unassigned": can_view_unassigned
}
except HTTPException:
raise
except Exception as e:
logger.error(f"JWT authentication error in unified chat auth: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
except HTTPException:
# Re-raise HTTP exceptions as-is (don't wrap them)
raise
except Exception as e:
logger.error(f"Chat authentication error: {str(e)}")
raise HTTPException(status_code=401, detail="Authentication failed")
backend/app/core/auth_utils.py
+224
View File
@@ -0,0 +1,224 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from typing import Optional, Tuple
from uuid import UUID
import http.cookies
from urllib.parse import parse_qs, urlparse
from app.core.security import verify_conversation_token
from app.core.logger import get_logger
from app.database import get_db, SessionLocal
from app.models.user import User
from sqlalchemy.orm import Session
from app.core.cors import is_first_party_origin, origin_of
from app.models.widget import Widget
from app.repositories.customer import CustomerRepository
from app.services.auth_sessions import ACCESS, REFRESH, issue_tokens, live_payload
logger = get_logger(__name__)
async def refresh_access_token(refresh_token: str, db: Session) -> str | None:
"""Refresh access token using refresh token"""
try:
payload = live_payload(db, refresh_token, REFRESH)
if not payload:
return None
user_id = payload.get("sub")
org_id = payload.get("org")
if not user_id or not org_id:
return None
# Verify user exists and is active
user = db.query(User).filter(User.id == UUID(user_id)).first()
if not user or not user.is_active:
return None
# Generate new access token for the same session (a legacy token
# without one gets a new session)
access_token, _ = issue_tokens(db, user_id, org_id, sid=payload.get("sid"))
db.commit()
return access_token
except Exception as e:
logger.error(f"Token refresh failed: {str(e)}")
return None
def _is_own_origin(origin: str, host: str) -> bool:
"""Whether Origin is the API's own origin (the host the request was sent to)."""
own = origin_of(origin)
return bool(own and host) and own == origin_of(f"{urlparse(origin).scheme}://{host}")
def cookie_socket_allowed(environ: dict) -> bool:
"""Whether a socket handshake may sign in with the dashboard's cookies.
Sockets get no CORS check and the cookies are SameSite=None, so a page on
another site (an org's signup domain is unverified) could otherwise open
the socket as whoever is signed in. socket.io authenticates a namespace
with the headers of the first handshake request, so all of it must have
come from our own pages:
- never a JSONP handshake: a <script> tag sends the cookies without an Origin
- with an Origin, a first-party one or the API's own origin
- without one, only a same-origin request (browsers omit Origin on
same-origin GETs, e.g. polling behind a single-host proxy)
"""
if "j" in parse_qs(environ.get("QUERY_STRING", "")):
return False
origin = environ.get("HTTP_ORIGIN")
if origin:
return is_first_party_origin(origin) or _is_own_origin(origin, environ.get("HTTP_HOST", ""))
return environ.get("HTTP_SEC_FETCH_SITE") == "same-origin"
async def authenticate_socket(sid: str, environ: dict) -> Tuple[Optional[str], Optional[int], Optional[int]]:
"""
Authenticate socket connection using cookies
Returns: (access_token, user_id, org_id)
"""
if not cookie_socket_allowed(environ):
logger.warning(f"Refused cookie socket auth for sid {sid} from origin {environ.get('HTTP_ORIGIN')}")
return None, None, None
try:
# Extract cookies from environ
cookie_str = environ.get('HTTP_COOKIE', environ.get('headers', {}).get('Cookie', ''))
cookies = http.cookies.SimpleCookie()
cookies.load(cookie_str)
# Get access token from cookies
access_token = None
if 'access_token' in cookies:
access_token = cookies['access_token'].value
if not access_token:
logger.info("No access token found in cookies")
# Try to refresh using refresh token
if 'refresh_token' in cookies:
refresh_token = cookies['refresh_token'].value
with SessionLocal() as db:
access_token = await refresh_access_token(refresh_token, db)
if not access_token:
return None, None, None
# Verify the token is a live access token of a user who can still sign in
with SessionLocal() as db:
payload = live_payload(db, access_token, ACCESS)
if not payload:
return None, None, None
user_id = payload.get('sub')
org_id = payload.get('org')
user = db.query(User).filter(User.id == UUID(str(user_id))).first()
if not user or not user.is_active:
return None, None, None
return access_token, user_id, org_id
except Exception as e:
logger.error(f"Authentication error for sid {sid}: {str(e)}")
return None, None, None
async def authenticate_socket_conversation_token(sid: str, auth: dict) -> Tuple[Optional[str], Optional[str], Optional[str], Optional[str]]:
"""
Authenticate widget socket connection using conversation token.
Supports both token-required and optional-token authentication modes.
Returns: (widget_id, org_id, customer_id, conversation_token)
Returns (None, None, None, None) if authentication fails
"""
try:
if not auth or not isinstance(auth, dict):
logger.warning(f"Socket auth failed: No auth data for sid {sid}")
return None, None, None, None
conversation_token = None
# Get token from Socket.IO auth data
conversation_token = auth.get('conversation_token', '')
if not conversation_token:
logger.info("No conversation token found in auth data or cookies")
return None, None, None, None
# Verify token and get info
token_data = verify_conversation_token(conversation_token)
if not token_data:
return None, None, None, None
widget_id = token_data.get('widget_id')
customer_id = token_data.get('sub')
token_type = token_data.get('type')
# Verify token type
if token_type != "conversation":
logger.info(f"Invalid token type: {token_type}")
return None, None, None, None
logger.info(f"Authenticated widget {widget_id} for customer {customer_id}")
# Get widget to verify and get org_id
with SessionLocal() as db:
widget = db.query(Widget).filter(Widget.id == widget_id).first()
if not widget:
return None, None, None, None
org_id = widget.organization_id
# Resolve merged customers: if this device's token still carries an
# anonymous customer that was merged into an identified one (lead
# capture matched an existing email), continue as the merged target so
# returning visitors keep their identity and history.
if customer_id:
survivor = CustomerRepository(db).resolve_merged(customer_id)
if survivor is not None:
customer_id = str(survivor.id)
return widget_id, org_id, customer_id, conversation_token
except Exception as e:
logger.error(f"Widget authentication error for sid {sid}: {str(e)}")
return None, None, None, None
def widget_socket_identity(session: Optional[dict]) -> Tuple[Optional[str], Optional[str], Optional[str]]:
"""Identity of an already-connected widget socket, from its server-side session.
The conversation token authenticates the *connection*: the /widget connect
handler runs authenticate_socket_conversation_token once and stores what it
resolved. Events arriving on a live connection read that back instead of
re-verifying the token, because the token's lifetime is not the connection's
lifetime — re-checking tore down working chats the moment the TTL passed
(mid-conversation, and on end_chat when the visitor closed the widget), and
it did the same whenever Redis lost the token's JTI. Expiry and revocation
still take effect: the next connect is authenticated normally.
Returns (None, None, None) when there is no authenticated session for the sid.
"""
if not session:
return None, None, None
widget_id = session.get('widget_id')
org_id = session.get('org_id')
if not widget_id or not org_id:
return None, None, None
return widget_id, org_id, session.get('customer_id')
backend/app/core/config.py
+471
View File
@@ -0,0 +1,471 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import os
import json
from pydantic_settings import BaseSettings
from typing import List, Literal, NamedTuple, Optional
from dotenv import load_dotenv
from pathlib import Path
from pydantic import field_validator
from app.core.logger import get_logger
# Get the absolute path to the backend directory (parent of app directory)
BACKEND_DIR = Path(__file__).resolve().parent.parent.parent
# ENV_FILE points the backend at a different env file. The e2e suite sets it to
# e2e/.env.e2e so nothing from a developer's backend/.env (dev DB, dev Redis,
# real provider keys) leaks into a test run.
ENV_FILE = Path(os.getenv("ENV_FILE") or BACKEND_DIR / ".env")
load_dotenv(ENV_FILE)
DEFAULT_CORS = ["https://replypier.site", "http://localhost:5173", "http://localhost:8000"]
class Settings(BaseSettings):
PROJECT_NAME: str = "ReplyPier"
VERSION: str = "0.1.0"
API_V1_STR: str = "/api/v1"
# Database
DATABASE_URL: str = os.getenv(
"DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/replypier")
# Redis
REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")
REDIS_ENABLED: bool = os.getenv("REDIS_ENABLED", "false").lower() == "true"
# JWT
SECRET_KEY: str = os.getenv("JWT_SECRET_KEY", "your-secret-key")
ALGORITHM: str = os.getenv("JWT_ALGORITHM", "HS256")
CONVERSATION_SECRET_KEY: str = os.getenv(
"CONVERSATION_SECRET_KEY", "your-conversation-secret-key")
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# CORS Configuration
CORS_ORIGINS: List[str] = os.getenv("CORS_ORIGINS", DEFAULT_CORS)
# Firebase config
FIREBASE_CREDENTIALS: str = os.getenv(
"FIREBASE_CREDENTIALS", "app/config/firebase-config.json")
FRONTEND_URL: str = os.getenv("FRONTEND_URL", "http://localhost:5173")
BACKEND_URL: str = os.getenv("BACKEND_URL", "http://localhost:8000")
# Where the widget bundle (widget.js / widget.css) is served from at /assets.
# The e2e suite points this at a bundle built from the current source, so the
# committed production build in backend/assets is never rebuilt locally.
WIDGET_ASSETS_DIR: str = os.getenv("WIDGET_ASSETS_DIR", str(BACKEND_DIR / "assets"))
VITE_WIDGET_URL: str = os.getenv("VITE_WIDGET_URL", "http://localhost:5173")
# app.core.encryption owns the actual key loading (reads the env var directly and
# refuses to start without it outside development). This mirror exists only so
# check_secret_configuration can audit it, hence no default: unset is itself
# one of the states that audit flags.
ENCRYPTION_KEY: str = os.getenv("ENCRYPTION_KEY", "")
# Downgrades the public-default ENCRYPTION_KEY startup error to a warning, for
# deployments whose existing data is encrypted under that key. See
# verify_secret_configuration.
ALLOW_INSECURE_ENCRYPTION_KEY: bool = os.getenv(
"ALLOW_INSECURE_ENCRYPTION_KEY", "false").lower() == "true"
# SMTP Settings
SMTP_SERVER: str = os.getenv("SMTP_SERVER", "smtp.gmail.com")
SMTP_PORT: int = int(os.getenv("SMTP_PORT", "587"))
SMTP_USERNAME: str = os.getenv("SMTP_USERNAME", "your-email@gmail.com")
SMTP_PASSWORD: str = os.getenv("SMTP_PASSWORD", "your-password")
FROM_EMAIL: str = os.getenv("FROM_EMAIL", "noreply@replypier.site")
FROM_NAME: str = os.getenv("FROM_NAME", "ReplyPier")
# Shopify
SHOPIFY_API_KEY: str = os.getenv("SHOPIFY_API_KEY", "")
SHOPIFY_API_SECRET: str = os.getenv("SHOPIFY_API_SECRET", "")
SHOPIFY_API_VERSION: str = os.getenv("SHOPIFY_API_VERSION", "2025-10")
# Slack. The OAuth redirect URI is derived from BACKEND_URL, not configured.
SLACK_CLIENT_ID: str = os.getenv("SLACK_CLIENT_ID", "")
SLACK_CLIENT_SECRET: str = os.getenv("SLACK_CLIENT_SECRET", "")
SLACK_SIGNING_SECRET: str = os.getenv("SLACK_SIGNING_SECRET", "")
# Jira. Unlike the others the redirect URI IS configured, because it must
# match the callback registered on the Atlassian app exactly. Self-hosters
# supply their own OAuth app creds.
JIRA_CLIENT_ID: str = os.getenv("JIRA_CLIENT_ID", "")
JIRA_CLIENT_SECRET: str = os.getenv("JIRA_CLIENT_SECRET", "")
JIRA_REDIRECT_URI: str = os.getenv("JIRA_REDIRECT_URI", "")
# CRM lead push (HubSpot / Pipedrive). OAuth redirect URIs are derived from
# BACKEND_URL, not configured. Self-hosters supply their own OAuth app creds.
HUBSPOT_CLIENT_ID: str = os.getenv("HUBSPOT_CLIENT_ID", "")
HUBSPOT_CLIENT_SECRET: str = os.getenv("HUBSPOT_CLIENT_SECRET", "")
PIPEDRIVE_CLIENT_ID: str = os.getenv("PIPEDRIVE_CLIENT_ID", "")
PIPEDRIVE_CLIENT_SECRET: str = os.getenv("PIPEDRIVE_CLIENT_SECRET", "")
# Meta (WhatsApp Cloud API, Messenger, Instagram) — one app, shared webhook.
# Self-hosters supply their own app; the cloud supplies its approved app.
META_APP_ID: str = os.getenv("META_APP_ID", "")
META_APP_SECRET: str = os.getenv("META_APP_SECRET", "")
# Our own random token echoed back during webhook GET verification
META_WEBHOOK_VERIFY_TOKEN: str = os.getenv("META_WEBHOOK_VERIFY_TOKEN", "")
META_GRAPH_VERSION: str = os.getenv("META_GRAPH_VERSION", "v21.0")
# WhatsApp Embedded Signup config id (cloud onboarding convenience)
META_CONFIG_ID: str = os.getenv("META_CONFIG_ID", "")
# Facebook Login for Business config id, for connecting a Page (Messenger and
# Instagram DM both ride on the Page's token). A separate configuration from
# META_CONFIG_ID: it requests pages_messaging + pages_show_list and returns a
# user token, not a WhatsApp signup.
META_MESSENGER_CONFIG_ID: str = os.getenv("META_MESSENGER_CONFIG_ID", "")
# Instagram API with Instagram Login: its own app id/secret, separate from
# the Facebook ones above. This flow needs no Facebook Page — the business
# signs in with Instagram and we get an Instagram user token.
INSTAGRAM_APP_ID: str = os.getenv("INSTAGRAM_APP_ID", "")
INSTAGRAM_APP_SECRET: str = os.getenv("INSTAGRAM_APP_SECRET", "")
# Comma-separated emails allowed to use one-click Meta signup. Empty means
# everyone, which is the end state — this exists so the flow can be exercised
# in production while the Meta app is still in App Review, since until it is
# approved the login only works for people with a role on the app anyway.
SIGNUP_ALLOWED_EMAILS: str = os.getenv("SIGNUP_ALLOWED_EMAILS", "")
VERIFY_SSL_CERTIFICATES: bool = os.getenv("VERIFY_SSL_CERTIFICATES", "true").lower() == "true"
# Which addresses the AI base URL, MCP server URLs and ticket databases may
# point at (see app/core/outbound_policy.py). "any" suits a self-hosted
# install reaching its own network; "public_only" refuses private,
# loopback, link-local and reserved targets on a shared deployment.
OUTBOUND_URL_POLICY: Literal["any", "public_only"] = os.getenv("OUTBOUND_URL_POLICY", "any")
# STDIO MCP connectors start a process on this server from a command an org
# admin types. Fine on your own install; on a shared deployment it hands
# every customer a shell, so turn it off there.
MCP_STDIO_ENABLED: bool = os.getenv("MCP_STDIO_ENABLED", "true").lower() == "true"
ENVIRONMENT: str = os.getenv("ENVIRONMENT", "development")
TRIAL_DAYS: int = 7 # 7-day trial period
# S3 Configuration
S3_FILE_STORAGE: bool = os.getenv("S3_FILE_STORAGE", "false").lower() == "true"
S3_BUCKET: str = os.getenv("S3_BUCKET", "replypier-uploads")
S3_REGION: str = os.getenv("S3_REGION", "us-east-1")
# None, not "", when unset: boto3 only falls back to its default credential
# chain (env → shared config → IAM instance/container role) if these are
# None. An empty string is treated as an explicit credential, so requests
# get signed with a blank access key and fail with InvalidAccessKeyId —
# which is what blocks EC2/ECS/EKS deployments from using an attached role.
AWS_ACCESS_KEY_ID: Optional[str] = os.getenv("AWS_ACCESS_KEY_ID") or None
AWS_SECRET_ACCESS_KEY: Optional[str] = os.getenv("AWS_SECRET_ACCESS_KEY") or None
# Presigned URLs are regenerated on every response, so this only needs to
# outlive a single page render. Hard-capped at S3_MAX_PRESIGN_SECONDS.
S3_PRESIGN_EXPIRY_SECONDS: int = int(os.getenv("S3_PRESIGN_EXPIRY_SECONDS", "3600"))
# Enhanced Website Knowledge Base Configuration
KB_MAX_DEPTH: int = int(os.getenv("KB_MAX_DEPTH", "5"))
KB_MAX_LINKS: int = int(os.getenv("KB_MAX_LINKS", "25"))
KB_MIN_CONTENT_LENGTH: int = int(os.getenv("KB_MIN_CONTENT_LENGTH", "100"))
KB_TIMEOUT: int = int(os.getenv("KB_TIMEOUT", "30"))
KB_MAX_RETRIES: int = int(os.getenv("KB_MAX_RETRIES", "3"))
KB_MAX_WORKERS: int = int(os.getenv("KB_MAX_WORKERS", "5"))
KB_BATCH_SIZE: int = int(os.getenv("KB_BATCH_SIZE", "5"))
# A website source that has not finished reading in this long fails rather
# than running on (one hit the worker's memory limit after three hours).
KB_SOURCE_TIME_LIMIT_SECONDS: int = int(os.getenv("KB_SOURCE_TIME_LIMIT_SECONDS", "1800"))
# Pages a PDF may have (app/knowledge/pdf_limits.py): every page is
# embedded, and a scanned one OCR'd, so bytes alone don't bound the work.
KB_PDF_MAX_PAGES: int = int(os.getenv("KB_PDF_MAX_PAGES", "300"))
# Hard ceiling on a single agent run (model + tool calls). A stuck run is
# cancelled instead of hanging the chat handler forever (issue #269).
AGENT_RUN_TIMEOUT: int = int(os.getenv("AGENT_RUN_TIMEOUT", "90"))
# agno's own run reporting to api.agno.com (see app.core.agno_telemetry).
# Off unless an operator opts in.
AGNO_TELEMETRY: bool = os.getenv("AGNO_TELEMETRY", "false").lower() == "true"
AGNO_MONITOR: bool = os.getenv("AGNO_MONITOR", "false").lower() == "true"
# Tool calls allowed in a single turn before the agent must answer from what
# it already has. Reaching this is no longer fatal — the agent answers without
# further tools rather than returning nothing (see app.utils.agno_patches) —
# so the budget can stay tight. AGENT_RUN_TIMEOUT is the real safety net.
AGENT_TOOL_CALL_LIMIT: int = int(os.getenv("AGENT_TOOL_CALL_LIMIT", "5"))
# Chars of previous-turn history re-sent to the model each turn. Tool
# results from earlier turns are always stubbed out (see
# app.agents.history_memory); this bounds what remains by dropping the oldest
# turns. Stored history is never touched.
AGENT_HISTORY_MAX_CHARS: int = int(os.getenv("AGENT_HISTORY_MAX_CHARS", "40000"))
# Knowledge chunking. The search embedder (bge-small, 512-token window) only
# sees roughly this much text per row, so a bigger chunk is not searchable
# past this point — and crawled text is URL-dense (~3 chars/token).
KNOWLEDGE_CHUNK_SIZE: int = int(os.getenv("KNOWLEDGE_CHUNK_SIZE", "1100"))
KNOWLEDGE_CHUNK_OVERLAP: int = int(os.getenv("KNOWLEDGE_CHUNK_OVERLAP", "120"))
# Hits returned per knowledge search, and the ceiling on the rendered result
# (whole pages are dropped past it, a page is never cut mid-chunk).
KNOWLEDGE_SEARCH_RESULTS: int = int(os.getenv("KNOWLEDGE_SEARCH_RESULTS", "6"))
KNOWLEDGE_SEARCH_MAX_CHARS: int = int(os.getenv("KNOWLEDGE_SEARCH_MAX_CHARS", "15000"))
# Knowledge base content summarization settings
KNOWLEDGE_SUMMARY_ENABLED: bool = os.getenv("KNOWLEDGE_SUMMARY_ENABLED", "false").lower() == "true"
KNOWLEDGE_SUMMARY_MODEL_TYPE: str = os.getenv("KNOWLEDGE_SUMMARY_MODEL_TYPE", "GROQ")
KNOWLEDGE_SUMMARY_MODEL_NAME: str = os.getenv("KNOWLEDGE_SUMMARY_MODEL_NAME", "llama-3.1-8b-instant")
KNOWLEDGE_SUMMARY_API_KEY: str = os.getenv("KNOWLEDGE_SUMMARY_API_KEY", "")
KNOWLEDGE_SUMMARY_MAX_TOKENS: int = int(os.getenv("KNOWLEDGE_SUMMARY_MAX_TOKENS", "4000"))
# Required when KNOWLEDGE_SUMMARY_MODEL_TYPE=OPENAI_COMPATIBLE, ignored otherwise.
KNOWLEDGE_SUMMARY_BASE_URL: str = os.getenv("KNOWLEDGE_SUMMARY_BASE_URL", "")
# Global chat guardrails (platform-owned; deliberately NOT tenant-configurable).
# GUARDRAIL_POLICY_ENABLED -> prepend the code-owned policy block to every
# chat agent system prompt.
# GUARDRAIL_INBOUND_ACTION -> pre-LLM injection check on visitor messages:
# "off" (never block), "template_only" (block
# only literal chat-template tokens a human
# never types), "strict" (block every strong
# injection signal). Non-blocking hits are
# always still counted.
# GUARDRAIL_OUTPUT_CHECK_ENABLED -> scan replies for leaked policy text and
# count scope refusals.
# GUARDRAIL_EVENTS_ENABLED -> persist trigger events to guardrail_events.
# GUARDRAIL_STORE_EXCERPT -> store an encrypted 300-char excerpt of the
# offending text on each event for review.
GUARDRAIL_POLICY_ENABLED: bool = os.getenv("GUARDRAIL_POLICY_ENABLED", "true").lower() == "true"
GUARDRAIL_INBOUND_ACTION: str = os.getenv("GUARDRAIL_INBOUND_ACTION", "template_only")
# Long self-contained exercise briefs with no business context. Defaults to
# "block": prompt text demonstrably failed to hold these in production, and
# blocking here also avoids paying for the inference. "count" or "off" to
# relax without a deploy.
GUARDRAIL_OFFTOPIC_ACTION: str = os.getenv("GUARDRAIL_OFFTOPIC_ACTION", "block")
GUARDRAIL_OUTPUT_CHECK_ENABLED: bool = os.getenv("GUARDRAIL_OUTPUT_CHECK_ENABLED", "true").lower() == "true"
GUARDRAIL_EVENTS_ENABLED: bool = os.getenv("GUARDRAIL_EVENTS_ENABLED", "true").lower() == "true"
GUARDRAIL_STORE_EXCERPT: bool = os.getenv("GUARDRAIL_STORE_EXCERPT", "true").lower() == "true"
# Help center (public FAQ site)
# How the public help center URL is advertised (live_url):
# "path" -> {BACKEND_URL}/help/{slug}, served same-origin as the API.
# Works on localhost/self-host with no DNS/TLS/proxy. Default.
# "subdomain" -> https://{slug}.<HELP_CENTER_BASE_DOMAIN> (cloud). MUST be set
# on cloud so subdomain help centers keep their branded URL.
# A verified custom domain always takes precedence over both. Host-based dispatch
# (subdomains + custom domains) stays active regardless; only path dispatch is gated
# to "path" mode.
HELP_CENTER_PUBLIC_MODE: str = os.getenv("HELP_CENTER_PUBLIC_MODE", "path")
# Base domain serving {slug}.<base> help centers.
HELP_CENTER_BASE_DOMAIN: str = os.getenv("HELP_CENTER_BASE_DOMAIN", "help.replypier.site")
# CNAME target customers point their custom help-center domain at.
HELP_CENTER_CNAME_TARGET: str = os.getenv("HELP_CENTER_CNAME_TARGET", "cname.replypier.site")
# IPs the CNAME target resolves to — accepted when a provider flattens the
# CNAME into A/AAAA records (comma-separated).
HELP_CENTER_TARGET_IPS: frozenset = frozenset(
ip.strip() for ip in os.getenv("HELP_CENTER_TARGET_IPS", "").split(",") if ip.strip()
)
# FAQ generation cost caps (per source / per LLM call) and import fetch limits.
FAQ_MAX_PAGES_PER_SOURCE: int = int(os.getenv("FAQ_MAX_PAGES_PER_SOURCE", "300"))
FAQ_MAX_BATCH_CHARS: int = int(os.getenv("FAQ_MAX_BATCH_CHARS", "15000"))
# Ceiling for context-window-derived batch sizing (see utils/model_context.py)
# — a quality guard for very-large-context models, not a token limit.
FAQ_MAX_BATCH_CHARS_CEILING: int = int(os.getenv("FAQ_MAX_BATCH_CHARS_CEILING", "60000"))
# Force a context-window size (tokens) for exotic/self-hosted models; 0 = auto.
FAQ_CONTEXT_TOKENS_OVERRIDE: int = int(os.getenv("FAQ_CONTEXT_TOKENS_OVERRIDE", "0"))
# Meter FAQ generation credits even for orgs on their own API key
# (default: hosted REPLYPIER model only).
FAQ_METER_OWN_KEY: bool = os.getenv("FAQ_METER_OWN_KEY", "false").lower() == "true"
FAQ_IMPORT_MAX_PAGE_CHARS: int = int(os.getenv("FAQ_IMPORT_MAX_PAGE_CHARS", "100000"))
FAQ_IMPORT_FETCH_TIMEOUT: int = int(os.getenv("FAQ_IMPORT_FETCH_TIMEOUT", "30"))
# Article-mode import (crawl linked pages, no LLM): crawl and re-host caps.
# High enough to pull a whole mid-size help center in one pass; a deliberate
# one-time migration in a background worker, so slowness is acceptable.
FAQ_ARTICLE_IMPORT_MAX_PAGES: int = int(os.getenv("FAQ_ARTICLE_IMPORT_MAX_PAGES", "200"))
FAQ_ARTICLE_IMPORT_MAX_IMAGES: int = int(os.getenv("FAQ_ARTICLE_IMPORT_MAX_IMAGES", "10"))
# Category/section listing pages to follow for the full per-category article
# list (help-center homepages truncate each section to a few articles).
FAQ_ARTICLE_IMPORT_MAX_CATEGORIES: int = int(os.getenv("FAQ_ARTICLE_IMPORT_MAX_CATEGORIES", "20"))
# A 'processing' FAQ job whose progress hasn't advanced in this long is
# treated as dead (worker crashed/killed): excluded from active-job polling
# and reaped on the next enqueue. Generous — must exceed the slowest single
# LLM batch / page fetch so a live-but-slow job is never killed.
FAQ_JOB_STALE_SECONDS: int = int(os.getenv("FAQ_JOB_STALE_SECONDS", "600"))
# FAQ generation/import jobs processed concurrently across ALL orgs. Default
# 1 = strictly one business at a time (each job does LLM calls + vector-DB
# reads); raise for more throughput on a bigger host.
MAX_CONCURRENT_FAQ_JOBS: int = int(os.getenv("MAX_CONCURRENT_FAQ_JOBS", "1"))
# AI ticket triage/investigation runs processed concurrently across ALL
# orgs (each run does LLM calls; investigations later add MCP subprocesses).
MAX_CONCURRENT_INVESTIGATIONS: int = int(os.getenv("MAX_CONCURRENT_INVESTIGATIONS", "2"))
# Subdomain labels reserved for infrastructure — must mirror the DNS/nginx
# records that exist on the base domain, hence env-configurable.
HELP_CENTER_RESERVED_SLUGS: frozenset = frozenset(
s.strip() for s in os.getenv(
"HELP_CENTER_RESERVED_SLUGS",
"www,api,app,help,mail,admin,staging,cname,status",
).split(",") if s.strip()
)
# Embedding Model Configuration
EMBEDDING_BATCH_SIZE: int = int(os.getenv("EMBEDDING_BATCH_SIZE", "32"))
EMBEDDING_MAX_WORKERS: int = int(os.getenv("EMBEDDING_MAX_WORKERS", "4"))
# FastEmbed Configuration
FASTEMBED_MODEL: str = os.getenv("FASTEMBED_MODEL", "BAAI/bge-small-en-v1.5")
# Embedding Optimization Configuration
ENABLE_IMMEDIATE_EMBEDDING: bool = os.getenv("ENABLE_IMMEDIATE_EMBEDDING", "true").lower() == "true"
# Embedding Safety Configuration (for Docker environments)
EMBEDDING_SINGLE_THREADED: bool = os.getenv("EMBEDDING_SINGLE_THREADED", "true").lower() == "true"
EMBEDDING_SEQUENTIAL_FALLBACK: bool = os.getenv("EMBEDDING_SEQUENTIAL_FALLBACK", "true").lower() == "true"
# Explore View Configuration
EXPLORE_SOURCE_ORG_ID: str = os.getenv("EXPLORE_SOURCE_ORG_ID", "bab82aab-d095-46f8-bf16-da638671bcf4")
EXPLORE_AGENT_ID: str = os.getenv("EXPLORE_AGENT_ID", "b20188ee-2800-41d0-8bf1-8fc291ab0076")
EXPLORE_USER_ID: str = os.getenv("EXPLORE_USER_ID", "154540a3-6177-4b1b-aab2-f23f0ef74ac7")
EXPLORE_WIDGET_ID: str = os.getenv("EXPLORE_WIDGET_ID", "397046dc-0093-4499-ab45-a0afe3c3ee14")
@field_validator("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", mode="after")
@classmethod
def _blank_credential_is_unset(cls, v: Optional[str]) -> Optional[str]:
"""Coerce a blank credential to None so boto3 uses its default chain.
The field default already does this, but BaseSettings reads the
environment itself and overrides it — a var that is present but empty
(`AWS_ACCESS_KEY_ID=` in .env, or an env_file entry with no value)
lands as "" and would be signed with as an explicit blank credential.
"""
return v or None
model_config = {
"case_sensitive": True,
"env_file": ENV_FILE,
"extra": "allow", # This allows extra fields from .env
}
settings = Settings()
def webhooks_base_url() -> str:
"""Where platforms reach this backend's inbound webhooks."""
return f"{settings.BACKEND_URL.rstrip('/')}{settings.API_V1_STR}/webhooks"
logger = get_logger(__name__)
# Environments where the public defaults are acceptable, because nothing durable or
# reachable is protected by them. Anything else is treated as a real deployment.
_THROWAWAY_ENVIRONMENTS = frozenset({"development", "test", "testing"})
# Shell one-liners producing a real value for each kind of secret, quoted back in
# the startup error so the fix does not require going and finding the docs.
# ENCRYPTION_KEY_HINT is public because app.core.encryption reports the same fix
# when the key is missing at load time, and the two must not drift apart.
ENCRYPTION_KEY_HINT = (
'python -c "import base64;from cryptography.fernet import Fernet;'
'print(base64.b64encode(Fernet.generate_key()).decode())"'
)
_HEX_SECRET_HINT = "openssl rand -hex 32"
class _ManagedSecret(NamedTuple):
"""A secret the app refuses to serve traffic without.
One record per secret rather than parallel dicts keyed by setting name: adding
a secret should be one line, and a half-filled entry should not surface as a
KeyError while building the very error meant to explain the problem.
"""
# What the operator sets, which is not always the setting name.
env_var: str
# Value once shipped in this repo (config default / .env.example). Public, so
# it stays listed after being removed as a default — deployments copied it.
public_default: str
# Shell one-liner producing a real replacement.
generate: str
_MANAGED_SECRETS = {
"SECRET_KEY": _ManagedSecret(
"JWT_SECRET_KEY", "your-secret-key", _HEX_SECRET_HINT),
"CONVERSATION_SECRET_KEY": _ManagedSecret(
"CONVERSATION_SECRET_KEY", "your-conversation-secret-key", _HEX_SECRET_HINT),
"ENCRYPTION_KEY": _ManagedSecret(
"ENCRYPTION_KEY",
"RFQ4SzhyRTVYdGtsLUxsc25SaDB0QlZpbTdQRmlVRlpsZUlCaFRlU2Vxbz0=",
ENCRYPTION_KEY_HINT),
}
# .env.example placeholders - not secret either, and they mean "never configured"
_PLACEHOLDER_VALUES = {
"your_jwt_secret_key_here",
"your_conversation_secret_key_here",
"your_fernet_encryption_key_here",
}
def is_throwaway_environment(config: Settings = settings) -> bool:
"""True for local development and test runs, where the public defaults are
harmless. Defined here so every check of "is this a real deployment" agrees;
app.core.encryption asks the same question before generating a throwaway key.
"""
return config.ENVIRONMENT.lower() in _THROWAWAY_ENVIRONMENTS
def check_secret_configuration(config: Settings = settings) -> list[str]:
"""Names of the auth/encryption secrets that are missing or still at a public
default/placeholder. Empty in development, which runs on the defaults on purpose.
Pure audit: verify_secret_configuration decides what to do about the result.
"""
if is_throwaway_environment(config):
return []
return [
name for name, secret in _MANAGED_SECRETS.items()
if not getattr(config, name, None)
or getattr(config, name) == secret.public_default
or getattr(config, name) in _PLACEHOLDER_VALUES
]
def verify_secret_configuration(config: Settings = settings) -> None:
"""Refuse to start on secrets anyone can look up in this repo.
Every value in _MANAGED_SECRETS and _PLACEHOLDER_VALUES is public, so a
deployment using one has no auth boundary at all — tokens can be forged and
stored credentials decrypted by anyone. A warning was not enough: it scrolls
past in the boot log and the deployment stays exposed indefinitely.
ENCRYPTION_KEY alone gets an escape hatch. An instance predating the required
key has real data encrypted under the public one, and refusing to boot would
lock it out of that data — worse than the exposure it already lives with.
Rotating a JWT secret only ends sessions, so those get no exemption.
"""
insecure = check_secret_configuration(config)
if "ENCRYPTION_KEY" in insecure and config.ALLOW_INSECURE_ENCRYPTION_KEY:
logger.warning(
"ENCRYPTION_KEY is the public default and ALLOW_INSECURE_ENCRYPTION_KEY "
"is set, so startup continues. Conversations and stored credentials in "
"this database can be decrypted by anyone who has a copy of it. Generate "
"a real key (%s), re-encrypt what the old key wrote, then unset this.",
ENCRYPTION_KEY_HINT,
)
insecure = [name for name in insecure if name != "ENCRYPTION_KEY"]
if not insecure:
return
names = ", ".join(insecure)
verb = "is" if len(insecure) == 1 else "are"
fixes = "\n".join(
f" {_MANAGED_SECRETS[name].env_var}=$({_MANAGED_SECRETS[name].generate})"
for name in insecure
)
raise RuntimeError(
f"Refusing to start: {names} {verb} unset or still set to a value published "
"in this repository, so anyone can forge tokens or decrypt stored data. "
f"Set real values and restart:\n{fixes}"
)
backend/app/core/cors.py
+387
View File
@@ -0,0 +1,387 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import json
import os
import time
import asyncio
import traceback
import re
from urllib.parse import urlparse
from typing import Iterable, List, Optional, Set
from functools import lru_cache
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from starlette.datastructures import Headers, MutableHeaders
from starlette.types import Message, Send
from app.core.config import settings
from app.core.logger import get_logger
from app.utils.domain_check import site_host
logger = get_logger(__name__)
# Base CORS origins that are always allowed
BASE_CORS_ORIGINS: list = settings.CORS_ORIGINS
_CREDENTIALS_HEADER = "access-control-allow-credentials"
_DEFAULT_PORTS = {"http": 80, "https": 443}
def origin_of(url: Optional[str]) -> Optional[str]:
"""The origin of a URL the way a browser sends it: lowercase, no path and
no default port, so https://App.example.com:443/ is https://app.example.com."""
parsed = urlparse((url or "").strip())
if parsed.scheme not in _DEFAULT_PORTS or not parsed.hostname:
return None
try:
port = parsed.port
except ValueError:
return None
netloc = parsed.hostname if port in (None, _DEFAULT_PORTS[parsed.scheme]) else f"{parsed.hostname}:{port}"
return f"{parsed.scheme}://{netloc}".lower()
@lru_cache(maxsize=1)
def first_party_origins() -> Set[str]:
"""Origins that may use the dashboard's sign-in cookies: the dashboard
(FRONTEND_URL) and what the operator listed in CORS_ORIGINS.
Never an org's domain or a help centre. An org domain is whatever someone
typed at signup, unverified, so trusting it with credentials would let a
page on any signed-up domain read a signed-in user's account. Read from
settings once: every /agent socket connect asks.
"""
dashboard = origin_of(settings.FRONTEND_URL)
if not dashboard:
logger.error(f"FRONTEND_URL {settings.FRONTEND_URL!r} is not an http(s) URL; the dashboard can't sign in cross-origin")
origins = {dashboard, *(origin_of(url) for url in BASE_CORS_ORIGINS)}
return {o for o in origins if o}
def is_first_party_origin(origin: Optional[str]) -> bool:
return origin_of(origin) in first_party_origins()
def with_configured_origins(origins: Iterable[str]) -> List[str]:
"""The given origins plus the operator's own and the API's, sorted. A
cached list from Redis was built from an earlier config, and must never drop
the dashboard. The API's own origin is where the widget's iframe snippet and
path-mode help centres run, and their sockets send it as Origin."""
own = {origin_of(settings.BACKEND_URL)} - {None}
return sorted(set(origins) | set(BASE_CORS_ORIGINS) | first_party_origins() | own)
class FirstPartyCredentialsCORSMiddleware(CORSMiddleware):
"""CORS for every allowed origin, credentials only for first-party ones.
Org domains and help centres still call the API cross-origin (older
widget loaders fetch /data from customers' sites), but without cookies:
Access-Control-Allow-Credentials is only sent to `credentialed_origins`,
so a browser never lets another site read a cookie-authenticated response.
"""
def __init__(self, app, credentialed_origins: Iterable[str] = (), **kwargs) -> None:
kwargs["allow_credentials"] = True
super().__init__(app, **kwargs)
self.credentialed_origins = frozenset(filter(None, map(origin_of, credentialed_origins)))
def _credentialed(self, origin: str) -> bool:
return origin_of(origin) in self.credentialed_origins
def preflight_response(self, request_headers: Headers):
response = super().preflight_response(request_headers)
if not self._credentialed(request_headers["origin"]) and _CREDENTIALS_HEADER in response.headers:
del response.headers[_CREDENTIALS_HEADER]
return response
async def send(self, message: Message, send: Send, request_headers: Headers) -> None:
if self._credentialed(request_headers["Origin"]):
await super().send(message, send, request_headers)
return
async def without_credentials(outgoing: Message) -> None:
if outgoing["type"] == "http.response.start":
headers = MutableHeaders(scope=outgoing)
if _CREDENTIALS_HEADER in headers:
del headers[_CREDENTIALS_HEADER]
await send(outgoing)
await super().send(message, without_credentials, request_headers)
@lru_cache(maxsize=1)
def get_cors_origins() -> Set[str]:
"""
Get all allowed CORS origins including organization domains.
This function is cached to improve performance.
"""
from app.repositories.organization import OrganizationRepository
from app.database import SessionLocal
# Start with base origins
all_origins = set(BASE_CORS_ORIGINS)
# Add organization domains
try:
db = SessionLocal()
org_repo = OrganizationRepository(db)
orgs = org_repo.get_active_organizations()
for org in orgs:
# The site with and without www., over http and https. Normalised:
# a domain typed as "www.Acme.com" must admit https://acme.com
# (browsers send Origin lowercase), not www.www.Acme.com.
host = site_host(org.domain or "")
if not host:
continue
for scheme in ("https", "http"):
all_origins.add(f"{scheme}://{host}")
all_origins.add(f"{scheme}://www.{host}")
# Help-center hosts embed the org's chat widget, which fetches
# /widgets/{id}/data and opens a socket from the public help-center
# origin — so those origins must be allowed too. The base-domain
# wildcard is also covered by get_cors_origin_regex() for FastAPI, but
# python-socketio can't match a regex, so enumerate the live slugs here.
from app.repositories.help_center import HelpCenterRepository
hc_repo = HelpCenterRepository(db)
base = settings.HELP_CENTER_BASE_DOMAIN
# In local dev the help center is reached on a non-standard port (e.g.
# :8000), which the browser puts in the Origin header. socket.io matches
# origins exactly (no regex), so add that port variant when configured.
dev_port = urlparse(settings.BACKEND_URL).port
for slug in hc_repo.list_enabled_slugs():
for scheme in ("https", "http"):
all_origins.add(f"{scheme}://{slug}.{base}")
if dev_port:
all_origins.add(f"{scheme}://{slug}.{base}:{dev_port}")
for domain in hc_repo.list_verified_domains():
all_origins.add(f"https://{domain}")
all_origins.add(f"http://{domain}")
except Exception as e:
logger.error(f"Error fetching organization domains: {e}")
finally:
if 'db' in locals():
db.close()
return all_origins
@lru_cache(maxsize=1)
def get_cors_origin_regex() -> str:
"""Regex allowing every help-center subdomain (`*.{base_domain}`) for the
FastAPI CORS middleware, so a new slug works without a cache refresh.
Socket.IO can't use this (no regex support) — it relies on the enumerated
origins in get_cors_origins()."""
base = re.escape(settings.HELP_CENTER_BASE_DOMAIN)
return rf"^https?://([a-z0-9-]+\.)?{base}(:\d+)?$"
def refresh_cors_origins() -> None:
"""
Clear the CORS origins cache to force a refresh
"""
get_cors_origins.cache_clear()
def update_local_cors(app: FastAPI, new_origins: List[str]) -> bool:
"""
Update local CORS middleware with new origins
Args:
app: FastAPI application instance
new_origins: List of CORS origins to set
Returns:
bool: True if CORS middleware was found and updated, False otherwise
"""
middleware_found = False
new_origins = with_configured_origins(new_origins)
# Update the registered CORS middleware in place. `user_middleware` holds
# the class plus its kwargs, so match on the class itself; adding another
# layer instead would leave the old ones answering for removed origins.
for middleware in getattr(app, 'user_middleware', []):
if isinstance(middleware.cls, type) and issubclass(middleware.cls, CORSMiddleware):
# The publishing worker also hears its own Redis broadcast, and
# startup re-applies the cached list; skip the rebuild when nothing changed
if set(middleware.kwargs.get('allow_origins') or ()) != set(new_origins):
middleware.kwargs['allow_origins'] = new_origins
# Starlette rebuilds the stack from user_middleware on the next request
app.middleware_stack = None
logger.info(f"Updated FastAPI CORS origins: {new_origins}")
middleware_found = True
break
# No CORS middleware registered yet: add one
if not middleware_found:
logger.warning("Could not find CORS middleware in app.user_middleware, applying global update")
try:
# Force FastAPI to rebuild the middleware stack
app.middleware_stack = None
# Re-add the middleware with updated origins. Keep the help-center
# subdomain regex here too, or a rebuild would silently drop the
# *.{base_domain} wildcard that the embedded widget relies on.
app.add_middleware(
FirstPartyCredentialsCORSMiddleware,
credentialed_origins=first_party_origins(),
allow_origins=new_origins,
allow_origin_regex=get_cors_origin_regex(),
allow_methods=["*"],
allow_headers=["*"],
)
logger.info(f"Added new CORS middleware with origins: {new_origins}")
middleware_found = True
except Exception as e:
logger.error(f"Failed to rebuild middleware stack: {str(e)}")
# Update Socket.IO CORS settings
try:
from app.core.socketio import sio
sio.eio.cors_allowed_origins = new_origins
logger.info(f"Updated Socket.IO CORS origins: {new_origins}")
except Exception as e:
logger.error(f"Failed to update Socket.IO CORS origins: {str(e)}")
return middleware_found
def update_cors_middleware(app: FastAPI) -> None:
"""
Update the CORS middleware with new origins and notify all workers
"""
logger.info("Refreshing CORS origins")
# Get fresh CORS origins
refresh_cors_origins()
new_origins = list(get_cors_origins())
# Update local instance
update_local_cors(app, new_origins)
# Store updated origins in Redis with timestamp for other workers
try:
from app.core.redis import get_redis
redis_client = get_redis()
if redis_client:
# Store CORS origins with timestamp in Redis
redis_client.set(
"cors:origins",
json.dumps({"origins": new_origins, "updated_at": time.time()})
)
# Publish event to notify other workers
redis_client.publish("cors:update", "refresh")
logger.info("Published CORS update to other workers via Redis")
else:
logger.warning("Redis not available, CORS update will only apply to this worker")
except Exception as e:
logger.error(f"Failed to publish CORS update to Redis: {str(e)}")
async def listen_for_cors_updates(app: FastAPI):
"""
Listen for CORS update notifications from Redis using an efficient pattern
"""
from app.core.redis import get_redis
redis_client = get_redis()
if not redis_client:
logger.warning("Redis not available, skipping CORS update listener")
return
pubsub = None
try:
# Use ignore_subscribe_messages to reduce noise
pubsub = redis_client.pubsub(ignore_subscribe_messages=True)
pubsub.subscribe("cors:update")
logger.info("Started CORS update listener")
# Apply any existing CORS config on startup
try:
cached = redis_client.get("cors:origins")
if cached:
cached_data = json.loads(cached)
if isinstance(cached_data, dict) and "origins" in cached_data:
update_local_cors(app, cached_data["origins"])
except Exception:
# Best-effort; continue
pass
# Non-blocking loop using get_message without blocking the event loop
while True:
try:
# timeout=0.0 ensures this call is non-blocking in redis-py
message = pubsub.get_message(timeout=0.0)
if message:
try:
origins_data = redis_client.get("cors:origins")
if origins_data:
origins_data = json.loads(origins_data)
if origins_data and "origins" in origins_data:
logger.info("Received CORS update notification")
update_local_cors(app, origins_data["origins"])
except Exception as inner_e:
logger.error(f"Error processing CORS update: {str(inner_e)}")
# Yield control to the event loop and poll at a light cadence
await asyncio.sleep(0.5)
except Exception as loop_e:
# TimeoutError or transient connection issues: log at warning and continue
logger.warning(f"CORS listener transient error: {str(loop_e)}")
await asyncio.sleep(1.0)
except Exception as e:
traceback.print_exc()
logger.error(f"Error in CORS update listener: {str(e)}")
finally:
# Try to clean up subscription
try:
if pubsub is not None:
pubsub.unsubscribe("cors:update")
pubsub.close()
except Exception:
pass
def start_cors_listener(app: FastAPI):
"""
Start background task to listen for CORS updates
"""
try:
# Check if Redis is available before starting listener
from app.core.redis import get_redis
redis_client = get_redis()
if redis_client:
# Create background task with proper error handling
background_task = asyncio.create_task(listen_for_cors_updates(app))
# Add exception handler to restart if it fails
def handle_exception(task):
if not task.cancelled() and task.exception():
logger.error(f"CORS listener crashed: {task.exception()}")
# Wait a moment before restarting
asyncio.create_task(restart_listener(app))
async def restart_listener(app):
await asyncio.sleep(5) # Wait before restarting
logger.info("Restarting CORS listener")
new_task = asyncio.create_task(listen_for_cors_updates(app))
new_task.add_done_callback(handle_exception)
background_task.add_done_callback(handle_exception)
logger.info("Started CORS update listener task")
else:
logger.warning("Redis not available, CORS synchronization between workers disabled")
except Exception as e:
logger.error(f"Failed to start CORS listener: {str(e)}")
backend/app/core/default_templates.py
+116
View File
@@ -0,0 +1,116 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from typing import Dict, List
from app.models.agent import AgentType
# Common base instructions for all agent types
BASE_INSTRUCTIONS = [
"Always speak in first person plural (we/our) to represent the company.",
"Be professional yet approachable in your responses.",
"Take ownership of customer concerns and show empathy.",
"Use phrases like 'we understand', 'our team', 'we can help'.",
"If you don't know something, say 'Let me check with our team'.",
"Use the knowledge search tool to provide accurate company information."
]
DEFAULT_BUSINESS_HOURS = {
"monday": {
"start": "09:00",
"end": "17:00",
"enabled": True
},
"tuesday": {
"start": "09:00",
"end": "17:00",
"enabled": True
},
"wednesday": {
"start": "09:00",
"end": "17:00",
"enabled": True
},
"thursday": {
"start": "09:00",
"end": "17:00",
"enabled": True
},
"friday": {
"start": "09:00",
"end": "17:00",
"enabled": True
},
"saturday": {
"start": "09:00",
"end": "17:00",
"enabled": False
},
"sunday": {
"start": "09:00",
"end": "17:00",
"enabled": False
}
}
DEFAULT_TEMPLATES: Dict[AgentType, Dict] = {
AgentType.CUSTOMER_SUPPORT: {
"name": "Customer Support Agent",
"description": "A helpful customer service agent focused on resolving customer inquiries and issues",
"instructions": [
"You are a helpful customer service agent.",
*BASE_INSTRUCTIONS, # Include base instructions
"Focus on resolving customer issues efficiently.",
"Ask clarifying questions when needed.",
"Provide clear step-by-step solutions.",
"Show empathy and understanding.",
"Follow up to ensure customer satisfaction.",
"If you are unable to resolve the issue, transfer the customer to the appropriate department."
],
"tools": []
},
AgentType.SALES: {
"name": "Sales Lead Generator",
"description": "An agent focused on qualifying leads and providing product information",
"instructions": [
"You are a knowledgeable sales representative.",
*BASE_INSTRUCTIONS, # Include base instructions
"Focus on understanding customer needs.",
"Provide relevant product information.",
"Highlight key benefits and features.",
"Ask qualifying questions.",
"Handle objections professionally.",
"Guide prospects through the sales process.",
"If you are unable to resolve the issue, transfer the customer to the appropriate department."
],
"tools": []
},
AgentType.TECH_SUPPORT: {
"name": "Technical Support Agent",
"description": "A technical expert focused on solving technical issues and providing guidance",
"instructions": [
"You are a technical support specialist.",
*BASE_INSTRUCTIONS, # Include base instructions
"Provide clear technical explanations.",
"Guide users through troubleshooting steps.",
"Use technical knowledge base when available.",
"Explain solutions in user-friendly terms.",
"Verify problem resolution.",
"Document solutions for future reference.",
"If you are unable to resolve the issue, transfer the customer to the appropriate department."
],
"tools": []
}
}
backend/app/core/edition.py
+54
View File
@@ -0,0 +1,54 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
# NO_ENTERPRISE=1 runs the backend as the community edition even when the
# enterprise submodule is checked out.
#
# CI and self-hosted builds have app/enterprise empty, so every
# `try: import app.enterprise` falls back to community mode. A developer
# checkout has the submodule populated, which switches on plan and
# subscription gating. Setting this makes the import fail the same way, so the
# unit suite (`NO_ENTERPRISE=1 python -m pytest tests/`) and the e2e OSS stack
# behave exactly like a community install.
#
# An env var rather than a flag because the blocker must be installed before
# anything imports app.enterprise — app/__init__.py calls apply() first thing.
import os
import sys
class _BlockEnterprise:
"""Meta-path finder that makes app.enterprise look un-checked-out."""
def find_spec(self, fullname, path=None, target=None):
if fullname == "app.enterprise" or fullname.startswith("app.enterprise."):
# ModuleNotFoundError, not bare ImportError: that is what an
# un-checked-out submodule actually raises, and it is what
# pytest.importorskip and the app's own guards look for.
raise ModuleNotFoundError(
f"No module named {fullname!r} (disabled by NO_ENTERPRISE=1)",
name=fullname,
)
return None
def apply() -> None:
"""Install the enterprise blocker when NO_ENTERPRISE=1 (idempotent)."""
if os.getenv("NO_ENTERPRISE") != "1":
return
if not any(isinstance(finder, _BlockEnterprise) for finder in sys.meta_path):
sys.meta_path.insert(0, _BlockEnterprise())
backend/app/core/encryption.py
+223
View File
@@ -0,0 +1,223 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Encryption of user content at rest.
Conversation content (chat messages, AI-written session summaries, the agno agent
memory blob) is stored as ciphertext so that a database dump, a stolen backup or a
stray `psql` session does not expose it. The application decrypts transparently.
Stored format is ``enc:<key_id>:<fernet-token>``. The key id is what makes key
rotation possible without a schema change: a future key is added to KEYS as ``v2``,
CURRENT_KEY_ID moves to it, and ``v1`` stays in the registry so old rows keep
reading while a backfill re-encrypts them.
This module deliberately depends on nothing but the standard library, cryptography,
SQLAlchemy and app.core.{logger,config} (both light), so the backfill script can load
it by file path on the host without importing the `app` package — whose __init__
pulls in the FastAPI app and the whole ML stack.
"""
import base64
import json
import os
from functools import lru_cache
from typing import Any, Dict, Optional
from cryptography.fernet import Fernet
from sqlalchemy import Text
from sqlalchemy.types import TypeDecorator
from app.core.config import ENCRYPTION_KEY_HINT, is_throwaway_environment
from app.core.logger import get_logger
logger = get_logger(__name__)
# Wire format. Nothing outside this module should restate these.
ENCRYPTED_PREFIX = "enc"
SEPARATOR = ":"
CURRENT_KEY_ID = "v1"
# Key under which an encrypted JSON blob is stored, so an encrypted dict is
# distinguishable from a plaintext one written before this change.
JSON_MARKER = "__enc__"
ENCRYPTION_KEY_ENV = "ENCRYPTION_KEY"
# Indented for the multi-line error messages below; the command itself is defined
# once in app.core.config so this and the startup audit cannot drift apart.
GENERATE_KEY_HINT = f" {ENCRYPTION_KEY_HINT}"
def load_key() -> bytes:
"""Return the raw Fernet key from ``ENCRYPTION_KEY``.
The env var holds the Fernet key base64-encoded a second time, which is the
format already in use for API-key encryption — decoding must stay identical or
every existing encrypted API key becomes unreadable.
Outside development and tests a missing or malformed key is fatal: chat
messages, session summaries and agent memory are all encrypted with it.
Usability of the key is checked here rather than left to ``Fernet()`` in
get_keys(), so that the .env.example placeholder and the common near-miss of
pasting a raw ``Fernet.generate_key()`` (base64-encoded once, not twice) both
surface the message below instead of a bare "must be 32 url-safe base64-encoded
bytes" from deep inside the library.
"""
env_key = os.getenv(ENCRYPTION_KEY_ENV)
if env_key:
try:
key = base64.b64decode(env_key)
Fernet(key)
return key
except Exception as e:
logger.error(f"Invalid encryption key format: {str(e)}")
if not is_throwaway_environment():
raise RuntimeError(
f"{ENCRYPTION_KEY_ENV} is set but is not a usable Fernet key. "
"Encrypted data cannot be read; refusing to start with a "
f"different key. Generate one with:\n{GENERATE_KEY_HINT}"
) from e
if not is_throwaway_environment():
raise RuntimeError(
f"{ENCRYPTION_KEY_ENV} is not set. It encrypts chat messages, session "
"summaries and agent memory at rest, so starting without it would make "
f"existing conversations unreadable. Generate one with:\n{GENERATE_KEY_HINT}"
)
key = Fernet.generate_key()
logger.info(
f"Generated a throwaway encryption key. Set "
f"{ENCRYPTION_KEY_ENV}={base64.b64encode(key).decode()} to keep data readable"
)
return key
@lru_cache(maxsize=1)
def get_keys() -> Dict[str, Fernet]:
"""Every key we can still decrypt with, by key id. Add an id here to rotate.
Loaded on first use rather than at import: a module import must not depend on
environment setup, and the explicit check belongs in verify_encryption_key()
at startup where the failure is legible.
"""
return {CURRENT_KEY_ID: Fernet(load_key())}
def verify_encryption_key() -> None:
"""Fail fast at startup if the key is missing or unusable.
Without this the first customer message would be the thing that discovers a
misconfigured deployment — after the app has already accepted traffic.
"""
get_keys() # raises with an actionable message if the key is missing or invalid
if decrypt_value(encrypt_value("verify")) != "verify":
raise RuntimeError(f"{ENCRYPTION_KEY_ENV} does not round-trip; refusing to start")
def is_encrypted(value: Any) -> bool:
"""True when the value is already in the stored ciphertext format."""
return isinstance(value, str) and value.startswith(ENCRYPTED_PREFIX + SEPARATOR)
def encrypt_value(value: Optional[str]) -> Optional[str]:
"""Encrypt a string for storage. Idempotent: already-encrypted input is returned
untouched, so it is safe to call on a value of unknown provenance."""
if value is None:
return None
if is_encrypted(value):
return value
# Outside the try: a missing key raises a RuntimeError explaining itself, and
# wrapping that in "failed to encrypt" would hide the actual problem.
fernet = get_keys()[CURRENT_KEY_ID]
try:
token = fernet.encrypt(str(value).encode()).decode()
except Exception as e:
logger.error(f"Encryption error: {str(e)}")
raise ValueError("Failed to encrypt value")
return SEPARATOR.join((ENCRYPTED_PREFIX, CURRENT_KEY_ID, token))
def decrypt_value(value: Optional[str]) -> Optional[str]:
"""Decrypt a stored string.
Values without the ``enc:`` prefix are returned unchanged — that is a row the
backfill has not reached yet, and reads must keep working throughout the
migration. A key we no longer hold, or a corrupt token, logs and returns None
rather than raising: one damaged row must not fail the whole request.
"""
if not is_encrypted(value):
return value
parts = value.split(SEPARATOR, 2)
if len(parts) != 3:
logger.error("Malformed ciphertext: missing key id or token")
return None
_, key_id, token = parts
fernet = get_keys().get(key_id)
if fernet is None:
logger.error(f"No encryption key registered for key id '{key_id}'")
return None
try:
return fernet.decrypt(token.encode()).decode()
except Exception as e:
logger.error(f"Decryption error: {str(e)}")
return None
def encrypt_json(value: Optional[Any]) -> Optional[Dict[str, str]]:
"""Encrypt a JSON-serialisable object into a single-key marker dict."""
if value is None:
return None
if isinstance(value, dict) and JSON_MARKER in value:
return value
return {JSON_MARKER: encrypt_value(json.dumps(value))}
def decrypt_json(value: Optional[Any]) -> Optional[Any]:
"""Inverse of encrypt_json. A plaintext object written before this change is
returned as-is; an undecryptable blob degrades to None."""
if not isinstance(value, dict) or JSON_MARKER not in value:
return value
plaintext = decrypt_value(value[JSON_MARKER])
if plaintext is None:
return None
try:
return json.loads(plaintext)
except ValueError as e:
logger.error(f"Decrypted blob is not valid JSON: {str(e)}")
return None
class EncryptedText(TypeDecorator):
"""Text column whose value is encrypted at rest and decrypted on read.
Underlying storage stays unbounded text, so applying this to an existing
String/Text column needs no schema migration.
"""
impl = Text
cache_ok = True
def process_bind_param(self, value, dialect):
return encrypt_value(value)
def process_result_value(self, value, dialect):
return decrypt_value(value)
backend/app/core/error_handlers.py
+73
View File
@@ -0,0 +1,73 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from fastapi import FastAPI, HTTPException, Request
from fastapi.exception_handlers import http_exception_handler
from starlette.exceptions import HTTPException as StarletteHTTPException
from starlette.responses import Response
from app.core.logger import get_logger
logger = get_logger(__name__)
# What a 5xx answers with instead of the server's own error text.
INTERNAL_ERROR_DETAIL = "Internal server error"
class ClientSafeHTTPException(HTTPException):
"""A 5xx whose detail is meant for the caller and holds nothing internal.
Raise this for a message we wrote ourselves, or an upstream provider's own
wording that the operator has to act on — "Recipient not on WhatsApp" is
useless in a log and essential on screen. Every other 5xx is assumed to be
carrying server internals and has its detail replaced, which is what makes
the widespread `detail=str(e)` safe by default.
"""
async def sanitized_http_exception_handler(
request: Request, exc: StarletteHTTPException
) -> Response:
"""Answer 5xx with a fixed message and keep the real one in the log.
Routes across the API raise `HTTPException(500, detail=str(e))`, which hands
the caller whatever the exception happened to carry — SQL fragments, file
paths, connection strings. None of it is actionable for a client, and a 5xx
means the fault is ours, so the body gets a constant and the original goes to
the log together with the route that produced it.
4xx detail is written for the caller ("Agent not found", "Invalid token") and
is passed through untouched, as is any `ClientSafeHTTPException`.
"""
if exc.status_code < 500 or isinstance(exc, ClientSafeHTTPException):
return await http_exception_handler(request, exc)
logger.error(
f"{request.method} {request.url.path} -> {exc.status_code}: {exc.detail}"
)
return await http_exception_handler(
request,
StarletteHTTPException(
status_code=exc.status_code,
detail=INTERNAL_ERROR_DETAIL,
headers=exc.headers,
),
)
def install_error_handlers(app: FastAPI) -> None:
"""Register the sanitizer. Called for every FastAPI instance we serve."""
app.add_exception_handler(StarletteHTTPException, sanitized_http_exception_handler)
backend/app/core/exceptions.py
+19
View File
@@ -0,0 +1,19 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
class JiraAuthError(Exception):
"""Exception raised for Jira authentication errors."""
pass
backend/app/core/external_urls.py
+29
View File
@@ -0,0 +1,29 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
# Base URLs of the third-party APIs the channel and CRM adapters call.
#
# Deployments never set these. The e2e suite points them at its recording fake
# (e2e/services/fake-externals) so channel journeys run end to end without
# reaching Telegram, Meta, Slack, … Read once at import, which is also when the
# adapters bind them (Meta's helpers take the base as a default argument).
import os
def external_base(env_var: str, default: str) -> str:
"""`default`, unless `env_var` overrides it; never with a trailing slash."""
return (os.getenv(env_var) or default).rstrip("/")
backend/app/core/file_validation.py
+87
View File
@@ -0,0 +1,87 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Shared upload validation: content-type, size, and magic-byte checks.
The frontend performs its own checks, but those are not a security boundary — the
CLI, AI agents, and raw API callers bypass them, so validate every upload here.
"""
import os
from typing import Tuple
from uuid import uuid4
from fastapi import HTTPException, UploadFile, status
# Magic-byte signatures used across the app.
PDF_MAGIC = b"%PDF-"
async def read_validated(
file: UploadFile,
*,
max_size: int,
allowed_content_types: Tuple[str, ...],
magic_prefix: bytes,
label: str = "file",
) -> bytes:
"""Validate an ``UploadFile`` and return its content as bytes.
Checks the declared content type, the actual byte size, and the leading magic
bytes (content type is advisory since clients can lie; the magic-byte check is
the real gate). Raises ``HTTPException`` (400/413) on any failure.
"""
name = file.filename or label
if file.content_type not in allowed_content_types:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"{name}: unsupported content type",
)
# Determine size from the spooled temp file without buffering into our own memory.
file.file.seek(0, 2)
size = file.file.tell()
file.file.seek(0)
if size > max_size:
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
detail=f"{name}: exceeds the {max_size // 1024 // 1024}MB limit",
)
content = await file.read()
if not content:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"{name}: empty file",
)
if not content.startswith(magic_prefix):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"{name}: not a valid {label}",
)
return content
def safe_filename(filename: str) -> str:
"""Return a filesystem-safe, collision-resistant name from ``filename``.
Strips directory components (prevents path traversal via ``../`` or absolute
paths) and prefixes a UUID so distinct uploads never overwrite each other.
"""
base = os.path.basename(filename or "").replace("\\", "").strip()
if not base or base in (".", ".."):
base = "upload"
return f"{uuid4().hex}_{base}"
backend/app/core/help_center_host.py
+149
View File
@@ -0,0 +1,149 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
Host-based dispatch for the public help center.
Requests whose Host is {slug}.<HELP_CENTER_BASE_DOMAIN> or a DB-verified
custom domain are routed to the public help-center app; everything else
passes through to the main API untouched. The Host header is never trusted
beyond an equality lookup against verified values.
This module is also the single source of truth for host canonicalization and
slug parsing — the resolver in services/help_center_public imports from here
so dispatch and resolution can never disagree.
"""
import asyncio
import re
import time
from typing import Optional
from app.core.config import settings
from app.core.logger import get_logger
logger = get_logger(__name__)
DOMAIN_CACHE_TTL_SECONDS = 60
# Path-based dispatch (self-host): {BACKEND_URL}/help/{slug}/... on the main
# origin. Group 1 is the help-center slug, group 2 the remainder ("/", "/a/x").
_HELP_PATH_RE = re.compile(r"^/help/([^/]+)(/.*)?$")
def normalize_host(raw_host: Optional[str]) -> str:
return (raw_host or "").split(":")[0].strip().lower().rstrip(".")
def slug_for_host(host: str) -> Optional[str]:
"""The {slug} of {slug}.<HELP_CENTER_BASE_DOMAIN>, if this host is one."""
suffix = f".{settings.HELP_CENTER_BASE_DOMAIN}"
if host.endswith(suffix):
label = host[: -len(suffix)]
if label and "." not in label:
return label
return None
class _VerifiedDomainCache:
"""Snapshot of verified custom domains refreshed in the background, so
host dispatch is always a non-blocking set lookup — the middleware fronts
EVERY request, and a synchronous DB refresh here would let a slow database
stall the whole service. Stale-while-revalidate; fails to an empty set
(custom domains temporarily 404; API traffic unaffected)."""
def __init__(self):
self._domains: frozenset = frozenset()
self._fetched_at: float = 0.0
self._refreshing = False
def contains(self, host: str) -> bool:
if time.monotonic() - self._fetched_at >= DOMAIN_CACHE_TTL_SECONDS and not self._refreshing:
self._refreshing = True
try:
asyncio.get_running_loop().create_task(self._refresh_async())
except RuntimeError:
# No running loop (sync test context): refresh inline.
self._refresh_sync()
self._refreshing = False
return host in self._domains
async def _refresh_async(self):
try:
await asyncio.to_thread(self._refresh_sync)
finally:
self._refreshing = False
def _refresh_sync(self):
try:
from app.database import SessionLocal
from app.repositories.help_center import HelpCenterRepository
with SessionLocal() as db:
self._domains = frozenset(HelpCenterRepository(db).list_verified_domains())
self._fetched_at = time.monotonic()
except Exception as e:
# Back off a full TTL rather than re-querying a broken DB per request.
self._fetched_at = time.monotonic()
logger.error(f"Verified-domain cache refresh failed: {e}")
_domain_cache = _VerifiedDomainCache()
def _host_from_scope(scope) -> str:
for name, value in scope.get("headers") or []:
if name == b"host":
return normalize_host(value.decode("latin-1"))
return ""
def is_help_center_host(host: str) -> bool:
if not host:
return False
if slug_for_host(host) is not None:
return True
return _domain_cache.contains(host)
class HelpCenterHostMiddleware:
"""Pure-ASGI dispatcher installed inside the socketio wrapper: help-center
hosts go to the public app, everything else to the main FastAPI app."""
def __init__(self, app, public_app):
self.app = app
self.public_app = public_app
async def __call__(self, scope, receive, send):
if scope["type"] == "http":
# Path dispatch (self-host default): /help/{slug}/... on the main
# origin. Gated to path mode so cloud never serves the /help/ form.
if settings.HELP_CENTER_PUBLIC_MODE == "path":
match = _HELP_PATH_RE.match(scope.get("path", ""))
if match:
slug = match.group(1)
# Drop a trailing slash from the remainder ourselves: letting
# Starlette's redirect_slashes handle it would 307 to a
# Location built from the rewritten (prefix-stripped) path,
# sending the visitor to the origin root without /help/{slug}.
remainder = (match.group(2) or "").rstrip("/") or "/"
scope["help_center"] = {"slug": slug, "base_path": f"/help/{slug}"}
scope["path"] = remainder
scope["raw_path"] = remainder.encode("latin-1")
await self.public_app(scope, receive, send)
return
if is_help_center_host(_host_from_scope(scope)):
await self.public_app(scope, receive, send)
return
await self.app(scope, receive, send)
backend/app/core/local_file_urls.py
+58
View File
@@ -0,0 +1,58 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
"""Short-lived signed URLs for private files in local (non-S3) storage.
The local counterpart of an S3 presigned URL: an HMAC binds the storage key to
an expiry, so a link works for whoever holds it until it expires, and for no
other file. Bearer links rather than session auth because the widget renders
attachments with plain <img>/<a> tags, which cannot send a token header.
"""
import hashlib
import hmac
import time
from typing import Optional
from urllib.parse import quote, urlencode
from app.core.config import settings
LOCAL_FILE_ROUTE = f"{settings.API_V1_STR}/files/local"
# SECRET_KEY also signs JWTs; the purpose prefix keeps the two from ever
# validating each other.
_PURPOSE = b"replypier:local-file-url:v1\n"
def _signature(key: str, expires: int) -> str:
message = _PURPOSE + f"{key}\n{expires}".encode()
return hmac.new(settings.SECRET_KEY.encode(), message, hashlib.sha256).hexdigest()
def sign_local_key(key: str, expires_in: Optional[int] = None) -> str:
"""Relative URL serving ``key`` (a path under uploads/) until it expires.
Lifetime follows S3_PRESIGN_EXPIRY_SECONDS, so both backends behave alike."""
lifetime = settings.S3_PRESIGN_EXPIRY_SECONDS if expires_in is None else expires_in
expires = int(time.time()) + lifetime
query = urlencode({"exp": expires, "sig": _signature(key, expires)})
return f"{LOCAL_FILE_ROUTE}/{quote(key)}?{query}"
def verify_local_signature(key: str, expires: Optional[int], signature: Optional[str]) -> bool:
"""True when ``signature`` was issued for ``key`` and has not expired."""
if expires is None or not signature or expires < time.time():
return False
return hmac.compare_digest(_signature(key, expires), signature)
backend/app/core/logger.py
+73
View File
@@ -0,0 +1,73 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import logging
import sys
from logging.handlers import RotatingFileHandler
import os
# Dictionary to store loggers by name
_loggers = {}
def get_logger(name=None):
"""
Get or create a logger instance.
Args:
name: The name for the logger (typically __name__ from the calling module)
Returns:
A configured logger instance
"""
# Use the provided name or default to 'app'
logger_name = name or 'app'
# Return existing logger if already created
if logger_name in _loggers:
return _loggers[logger_name]
# Create new logger
logger = logging.getLogger(logger_name)
# Only set up handlers if they haven't been set up already
if not logger.handlers:
logger.setLevel(logging.DEBUG)
# Create formatters and handlers
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
# File handler
log_dir = 'logs'
if not os.path.exists(log_dir):
os.makedirs(log_dir)
file_handler = RotatingFileHandler(
f'{log_dir}/app.log',
maxBytes=10485760, # 10MB
backupCount=5
)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# Store logger in dictionary
_loggers[logger_name] = logger
return logger
backend/app/core/model_catalog.py
+201
View File
@@ -0,0 +1,201 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
"""Central catalog of BYO-key AI model providers and their suggested models.
Single source of truth consumed by:
- the ``GET /ai/providers`` endpoint (which populates the frontend selectors), and
- ``validate_model_selection`` in ``app.api.ai_setup``.
The listed models are *suggestions*, not a hard allowlist: every provider here has
``custom_allowed=True``, so an organization may type any model ID the provider
supports. Validation only checks that the provider is known and the model name is
non-empty — the live API-key test is what actually rejects a bad model ID.
Provider model IDs churn and several have near-term shutdowns (see notes). Verify
against each provider's live ``GET /v1/models`` before relying on an ID long-term.
Every provider key MUST match a value in ``app.models.ai_config.AIModelType`` and a
branch in ``app.utils.agno_utils.create_model``.
"""
from typing import Dict, List, TypedDict
class CatalogModel(TypedDict):
value: str # exact model ID passed to the provider API
label: str # human-readable display name
class CatalogProvider(TypedDict):
label: str
requires_api_key: bool
custom_allowed: bool
# Console URL where the user creates/copies their API key for this provider.
api_key_url: str
models: List[CatalogModel]
# Whether this provider needs a user-supplied base URL (e.g. a self-hosted or
# third-party OpenAI-compatible endpoint). Defaults to False when absent.
requires_base_url: bool
def _m(value: str, label: str) -> CatalogModel:
return {"value": value, "label": label}
# Keyed by AIModelType value. REPLYPIER (managed) is intentionally excluded — it
# is handled as a special case in the setup flow and is not user-selectable here.
MODEL_CATALOG: Dict[str, CatalogProvider] = {
"OPENAI": {
"label": "OpenAI",
"requires_api_key": True,
"custom_allowed": True,
"api_key_url": "https://platform.openai.com/api-keys",
"models": [
_m("gpt-4.1", "GPT-4.1"),
_m("gpt-4o", "GPT-4o"),
_m("gpt-4.1-mini", "GPT-4.1 Mini"),
_m("gpt-4o-mini", "GPT-4o Mini"),
_m("o4-mini", "o4-mini"),
],
},
"ANTHROPIC": {
"label": "Anthropic (Claude)",
"requires_api_key": True,
"custom_allowed": True,
"api_key_url": "https://console.anthropic.com/settings/keys",
# Current-gen Claude IDs are complete stable strings — do NOT append a date.
"models": [
_m("claude-opus-4-8", "Claude Opus 4.8"),
_m("claude-sonnet-5", "Claude Sonnet 5"),
_m("claude-sonnet-4-6", "Claude Sonnet 4.6"),
_m("claude-haiku-4-5", "Claude Haiku 4.5"),
],
},
"GOOGLE": {
"label": "Google Gemini",
"requires_api_key": True,
"custom_allowed": True,
"api_key_url": "https://aistudio.google.com/app/apikey",
# gemini-2.0-flash is shut down; use bare 2.5 IDs (avoid -latest/-preview).
"models": [
_m("gemini-2.5-pro", "Gemini 2.5 Pro"),
_m("gemini-2.5-flash", "Gemini 2.5 Flash"),
_m("gemini-2.5-flash-lite", "Gemini 2.5 Flash-Lite"),
_m("gemini-3.5-flash", "Gemini 3.5 Flash"),
],
},
"MISTRAL": {
"label": "Mistral",
"requires_api_key": True,
"custom_allowed": True,
"api_key_url": "https://console.mistral.ai/api-keys",
# -latest aliases auto-advance to the current dated snapshot.
"models": [
_m("mistral-large-latest", "Mistral Large"),
_m("mistral-medium-latest", "Mistral Medium"),
_m("mistral-small-latest", "Mistral Small"),
],
},
"XAI": {
"label": "xAI (Grok)",
"requires_api_key": True,
"custom_allowed": True,
# The Grok inference API uses a single xai-... key from console.x.ai — NOT the
# X/Twitter OAuth app credentials (consumer key / access token / bearer token).
"api_key_url": "https://console.x.ai",
"models": [
_m("grok-4", "Grok 4"),
_m("grok-4-fast-reasoning", "Grok 4 Fast (Reasoning)"),
_m("grok-4-fast-non-reasoning", "Grok 4 Fast (Non-Reasoning)"),
_m("grok-3", "Grok 3"),
],
},
"DEEPSEEK": {
"label": "DeepSeek",
"requires_api_key": True,
"custom_allowed": True,
"api_key_url": "https://platform.deepseek.com/api_keys",
# deepseek-chat/reasoner scheduled for EOL 2026-07-24; successors are
# deepseek-v4-flash / deepseek-v4-pro — add here once live.
"models": [
_m("deepseek-chat", "DeepSeek Chat (V3)"),
_m("deepseek-reasoner", "DeepSeek Reasoner"),
],
},
"GROQ": {
"label": "Groq",
"requires_api_key": True,
"custom_allowed": True,
"api_key_url": "https://console.groq.com/keys",
# gpt-oss-120b is the durable pick. gpt-oss-20b is intentionally NOT listed: the
# smaller model leaks OpenAI "harmony" formatting (calls the response tool as
# `functions/json`), which Groq rejects, making structured output unreliable.
# llama-3.3-70b-versatile is kept for continuity but is deprecated (EOL
# 2026-08-16). Org-prefixed IDs must be passed exactly.
"models": [
_m("openai/gpt-oss-120b", "GPT-OSS 120B"),
_m("llama-3.3-70b-versatile", "Llama 3.3 70B Versatile"),
],
},
"OPENAI_COMPATIBLE": {
# Names the wire protocol, not the field: the endpoint must speak the
# OpenAI chat-completions API (OpenRouter, vLLM, llama.cpp, Ollama, LiteLLM
# and friends all do). The parenthetical is what self-hosters recognise.
"label": "OpenAI-compatible (self-hosted, OpenRouter, Ollama…)",
"requires_api_key": True,
"custom_allowed": True,
"requires_base_url": True,
# No hosted console — points at the provider's own docs for getting a key.
"api_key_url": "",
# No suggested models: the endpoint and model catalog are whatever the
# user's server exposes, so the model name must always be typed in.
"models": [],
},
}
def is_known_provider(provider: str) -> bool:
"""Return True if the provider is a selectable BYO-key provider in the catalog."""
return bool(provider) and provider.upper() in MODEL_CATALOG
def requires_base_url(provider: str) -> bool:
"""Return True if the provider needs a user-supplied base URL.
The catalog is the single source of truth for this, so a new self-hosted or
gateway provider only has to set ``requires_base_url`` on its entry — callers
must not string-compare against a specific provider name.
"""
if not is_known_provider(provider):
return False
return MODEL_CATALOG[provider.upper()].get("requires_base_url", False)
def list_providers() -> List[dict]:
"""Return the catalog as a serializable list for the /ai/providers endpoint."""
return [
{
"value": provider_value,
"label": entry["label"],
"requires_api_key": entry["requires_api_key"],
"custom_allowed": entry["custom_allowed"],
"api_key_url": entry["api_key_url"],
"models": entry["models"],
"requires_base_url": entry.get("requires_base_url", False),
}
for provider_value, entry in MODEL_CATALOG.items()
]
backend/app/core/oauth_state.py
+63
View File
@@ -0,0 +1,63 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
# One-time OAuth `state` values for the install flows (Jira, CRMs, Slack).
#
# The provider sends the browser back to whichever backend worker it lands on, so
# the state lives in Redis when it is enabled. Without Redis (single-worker dev
# setups) an in-process dict stands in. Either way a state expires after
# STATE_TTL_SECONDS and can be redeemed once.
import time
from typing import Optional
from app.core.redis import get_redis
STATE_TTL_SECONDS = 60 * 10
# (kind, state) -> (value, expires_at)
_fallback: dict = {}
def _redis_key(kind: str, state: str) -> str:
return f"{kind}_oauth_state:{state}"
def store_state(kind: str, state: str, value: str) -> None:
"""Remember `value` (usually the org id) for `state` until it is popped or expires."""
redis_client = get_redis()
if redis_client is not None:
redis_client.set(_redis_key(kind, state), value, ex=STATE_TTL_SECONDS)
return
now = time.time()
for key in [k for k, (_, expires_at) in _fallback.items() if expires_at < now]:
_fallback.pop(key, None)
_fallback[(kind, state)] = (value, now + STATE_TTL_SECONDS)
def pop_state(kind: str, state: str) -> Optional[str]:
"""The value stored for `state`, removed so it can't be replayed; None if unknown or expired."""
if not state:
return None
redis_client = get_redis()
if redis_client is not None:
pipe = redis_client.pipeline()
pipe.get(_redis_key(kind, state))
pipe.delete(_redis_key(kind, state))
value, _ = pipe.execute()
return value or None
entry = _fallback.pop((kind, state), None)
return entry[0] if entry and entry[1] >= time.time() else None
backend/app/core/outbound_http.py
+139
View File
@@ -0,0 +1,139 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
"""httpx clients that only dial public addresses, for OUTBOUND_URL_POLICY.
Checking a URL before handing it to an HTTP client leaves two holes: the
client resolves the name again (a 0-TTL record can answer public to the check
and 127.0.0.1 to the client), and it follows redirects nobody checked. The
guard here sits in httpcore's network backend instead, so every TCP connect —
first request, each redirect hop, every reconnect — resolves the host once,
refuses it unless every address is public, and connects to that vetted
address. TLS SNI, certificate checks and the Host header still use the
hostname: httpcore passes it to start_tls and the request separately.
Under policy "any" the factories return None, so callers keep their clients
exactly as before.
"""
from typing import Any, Optional
import anyio
import httpcore
import httpx
from app.core.logger import get_logger
from app.core.outbound_policy import OutboundTargetBlocked, public_only, vetted_addresses
logger = get_logger(__name__)
class OutboundConnectRefused(httpcore.ConnectError):
"""The guard refused an address. Raised as a connect error so httpx, and
every SDK built on it (whatever its version), reports an ordinary
connection failure rather than letting an unknown exception escape; the
policy's OutboundTargetBlocked is its __cause__."""
def _vet(host: str, port: int):
try:
return vetted_addresses(host, port, host)
except OutboundTargetBlocked as e:
logger.warning(f"Refused outbound connection to internal address via {host}")
raise OutboundConnectRefused(str(e)) from e
except OSError as e: # DNS failure: report it the way httpx would
raise httpcore.ConnectError(str(e)) from e
class GuardedBackend(httpcore.NetworkBackend):
"""Sync backend: connect only to vetted public addresses of the host."""
def __init__(self, inner: httpcore.NetworkBackend):
self._inner = inner
def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None):
last_error: Optional[Exception] = None
for address in _vet(host, port):
try:
return self._inner.connect_tcp(
address, port, timeout=timeout,
local_address=local_address, socket_options=socket_options,
)
except (httpcore.ConnectError, httpcore.ConnectTimeout) as e:
last_error = e # try the host's next address, like create_connection
raise last_error
def connect_unix_socket(self, *args, **kwargs):
raise OutboundConnectRefused("Unix sockets are not allowed for outbound targets.")
def sleep(self, seconds: float) -> None:
self._inner.sleep(seconds)
class AsyncGuardedBackend(httpcore.AsyncNetworkBackend):
"""Async counterpart of GuardedBackend."""
def __init__(self, inner: httpcore.AsyncNetworkBackend):
self._inner = inner
async def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None):
last_error: Optional[Exception] = None
for address in await anyio.to_thread.run_sync(_vet, host, port):
try:
return await self._inner.connect_tcp(
address, port, timeout=timeout,
local_address=local_address, socket_options=socket_options,
)
except (httpcore.ConnectError, httpcore.ConnectTimeout) as e:
last_error = e
raise last_error
async def connect_unix_socket(self, *args, **kwargs):
raise OutboundConnectRefused("Unix sockets are not allowed for outbound targets.")
async def sleep(self, seconds: float) -> None:
await self._inner.sleep(seconds)
def _guard(transport, backend_cls):
"""Swap the transport's connection-pool backend for the guarded one.
httpx has no public hook for this; fail loudly rather than hand out an
unguarded client if a library upgrade moves it."""
pool = getattr(transport, "_pool", None)
if not isinstance(pool, (httpcore.ConnectionPool, httpcore.AsyncConnectionPool)) \
or not hasattr(pool, "_network_backend"):
raise RuntimeError("Cannot install the outbound address guard on this httpx version")
pool._network_backend = backend_cls(pool._network_backend)
return transport
def guarded_http_client(**kwargs: Any) -> Optional[httpx.Client]:
"""A sync httpx.Client that dials only public addresses, or None when the
policy is "any". ``kwargs`` go to httpx.Client (not the transport). An
explicit transport also turns off env proxies, which would otherwise
connect on the target's behalf unchecked."""
if not public_only():
return None
return httpx.Client(transport=_guard(httpx.HTTPTransport(), GuardedBackend), **kwargs)
def guarded_async_http_client(**kwargs: Any) -> Optional[httpx.AsyncClient]:
"""Async counterpart of guarded_http_client."""
if not public_only():
return None
return httpx.AsyncClient(
transport=_guard(httpx.AsyncHTTPTransport(), AsyncGuardedBackend), **kwargs
)
backend/app/core/outbound_policy.py
+119
View File
@@ -0,0 +1,119 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
"""Opt-in refusal of user-configured outbound targets on internal networks.
AI base URLs, MCP server URLs and ticket databases are dialled by this server.
Self-hosters legitimately point them at their own network (Ollama on
localhost, a private MCP server or database), so the default policy "any"
allows everything. A shared, multi-tenant deployment sets
OUTBOUND_URL_POLICY=public_only so no tenant can use these features to reach
the server's own network or the cloud metadata endpoint.
Targets are checked when saved and again before connecting. The connection
itself is then pinned: the address that passed the check is the one dialled
(app.core.outbound_http for HTTP clients, pin_outbound_target for databases),
so a hostname that re-resolves elsewhere (DNS rebinding) or a redirect to an
internal host is refused at connect time too.
"""
import ipaddress
import socket
from typing import List, Optional
from fastapi import HTTPException, status
from app.core.config import settings
from app.knowledge.url_safety import is_public_ip, resolves_to_blocked_host
POLICY_ANY = "any"
POLICY_PUBLIC_ONLY = "public_only"
class OutboundTargetBlocked(ValueError):
"""A configured target is internal and the policy is public_only."""
def public_only() -> bool:
return settings.OUTBOUND_URL_POLICY == POLICY_PUBLIC_ONLY
def _blocked(label: str) -> OutboundTargetBlocked:
return OutboundTargetBlocked(
f"{label} must be a public address. This server does not allow "
"private, loopback, link-local or reserved network targets."
)
def _as_url(target: str) -> str:
"""resolves_to_blocked_host parses a URL; give a bare host (a database or
SSH host) the shape it expects. A scheme-less "localhost:11434/v1" would
otherwise parse with no host at all and slip through."""
if "://" in target:
return target
try:
if ipaddress.ip_address(target).version == 6:
return f"//[{target}]"
except ValueError:
pass
return f"//{target}"
def check_outbound_target(target: Optional[str], label: str) -> None:
"""Raise OutboundTargetBlocked when the policy is public_only and
``target`` (a URL or a bare host) is, or resolves to, a private, loopback,
link-local or reserved address. ``label`` names the field in the error."""
if not public_only() or not target:
return
if resolves_to_blocked_host(_as_url(target.strip())):
raise _blocked(label)
def vetted_addresses(host: str, port: Optional[int], label: str) -> List[str]:
"""Resolve ``host`` once and return its addresses for dialling, refusing
it if ANY of them is not public. Unlike the save-time check this fails
closed: an address that can't be parsed is refused, and a name that
doesn't resolve raises socket.gaierror. Callers connect to the returned
addresses directly, so nothing resolves the name a second time."""
infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
addresses = list(dict.fromkeys(info[4][0] for info in infos))
if not addresses:
raise socket.gaierror(f"{host} did not resolve to any address")
for address in addresses:
try:
public = is_public_ip(ipaddress.ip_address(address))
except ValueError:
public = False
if not public:
raise _blocked(label)
return addresses
def pin_outbound_target(host: Optional[str], label: str) -> Optional[str]:
"""For drivers that dial a host themselves (databases, SSH): under
public_only, vet ``host`` and return the address to connect to instead of
the name. None under policy "any", meaning connect as before."""
if not public_only() or not host:
return None
return vetted_addresses(host.strip(), None, label)[0]
def require_public_target(target: Optional[str], label: str) -> None:
"""check_outbound_target for request handlers: a refusal is a 400."""
try:
check_outbound_target(target, label)
except OutboundTargetBlocked as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
backend/app/core/params.py
+32
View File
@@ -0,0 +1,32 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
# Request parameters with the bounds the database implies. An integer id past
# Postgres INTEGER, or a negative LIMIT, makes the query itself fail — a 500 —
# where the caller should get a 422 (or simply nothing found).
from typing import Annotated
from fastapi import Path, Query
MAX_DB_INT = 2_147_483_647
# An integer primary key in the URL path.
DbIntId = Annotated[int, Path(ge=1, le=MAX_DB_INT)]
# A 1-based page number; far past any real list, short of overflowing the OFFSET.
MAX_PAGE = 100_000
PageNumber = Annotated[int, Query(ge=1, le=MAX_PAGE)]
backend/app/core/redis.py
+61
View File
@@ -0,0 +1,61 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import redis
from app.core.config import settings
from app.core.logger import get_logger
logger = get_logger(__name__)
redis_client = None
def init_redis():
"""Initialize Redis connection"""
global redis_client
if not settings.REDIS_ENABLED:
logger.info("Redis is disabled")
return None
redis_url = settings.REDIS_URL
if redis_url and redis_url.startswith("redis://") and ".cache.amazonaws.com" in redis_url:
redis_url = "rediss://" + redis_url[8:]
logger.info(f"Using TLS for Redis connection: {redis_url}")
try:
redis_client = redis.from_url(
redis_url,
socket_timeout=5.0,
socket_connect_timeout=5.0,
decode_responses=True # Automatically decode responses to Python strings
)
# Test connection immediately
redis_client.ping()
logger.info("Redis connection successful")
return redis_client
except Exception as e:
logger.error(f"Failed to initialize Redis client: {str(e)}")
return None
def get_redis():
"""Get Redis client instance"""
global redis_client
if not redis_client and settings.REDIS_ENABLED:
redis_client = init_redis()
return redis_client
backend/app/core/s3.py
+310
View File
@@ -0,0 +1,310 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
"""
S3 Storage Utilities
"""
import asyncio
import re
from functools import lru_cache, partial
import boto3
from app.core.config import settings
from app.core.logger import get_logger
from fastapi import HTTPException
from typing import Optional
from urllib.parse import urlparse, urlunparse
logger = get_logger(__name__)
# AWS rejects any SigV4 presigned URL whose X-Amz-Expires exceeds one week with
# AuthorizationQueryParametersError. Only the legacy SigV2 signer (which
# botocore still selects for us-east-1 and eu-west-1) has no such limit, so a
# longer value happens to work in those regions and 403s everywhere else.
S3_MAX_PRESIGN_SECONDS = 604800
# head_object retries after put_object, to ride out transient read-after-write
# failures rather than failing the whole upload.
_UPLOAD_VERIFY_ATTEMPTS = 3
_UPLOAD_VERIFY_BACKOFF_SECONDS = 0.5
def strip_s3_signature(url: Optional[str]) -> Optional[str]:
"""Reduce a presigned S3 URL back to the bare object URL.
Signed URLs must never reach the database: they expire, and a stored
signature outlives its validity. Clients round-trip response values into
update payloads (the agent customization form does exactly this), so
stripping on the way in is the only reliable guard.
"""
if not url or 'amazonaws.com' not in url:
return url
return urlunparse(urlparse(url)._replace(query='', fragment=''))
# Matches an S3 endpoint in the HOST position, in every shape AWS serves: the
# global endpoint, regional ones (dot or legacy dash), and the virtual-hosted
# forms that put the bucket in front, including dotted bucket names. Anchored at
# both ends so "evil.com/s3.amazonaws.com/x" and "s3.amazonaws.com.attacker.com"
# do not match, and neither does "foos3.amazonaws.com".
_S3_HOST_RE = re.compile(r"(^|\.)s3([.-][a-z0-9-]+)?\.amazonaws\.com$", re.IGNORECASE)
def is_s3_url(url: Optional[str]) -> bool:
"""True when this URL's HOST is an S3 endpoint.
Checks the parsed hostname, never a substring of the whole URL: any
attacker-supplied URL can carry "s3.amazonaws.com" in its path, and treating
that as one of ours is how a delete or a key extraction ends up pointed
somewhere it shouldn't be. A substring test is also INCOMPLETE — it misses
the regional virtual-hosted form url_for_s3_key actually produces
(bucket.s3.<region>.amazonaws.com), which contains no "s3.amazonaws.com".
"""
if not url:
return False
return bool(_S3_HOST_RE.search(urlparse(url).hostname or ""))
def url_for_s3_key(s3_key: str) -> str:
"""The canonical stored URL for an object key. Single source of truth for
the URL shape — _key_from_url must be able to round-trip whatever this
produces.
Buckets whose name contains a dot get path-style: the wildcard on
*.s3.<region>.amazonaws.com only covers one label, so the virtual-hosted
form (my.bucket.s3.<region>.amazonaws.com) fails TLS verification in any
strict client.
"""
if '.' in settings.S3_BUCKET:
return f"https://s3.{settings.S3_REGION}.amazonaws.com/{settings.S3_BUCKET}/{s3_key}"
return f"https://{settings.S3_BUCKET}.s3.{settings.S3_REGION}.amazonaws.com/{s3_key}"
def _key_from_url(s3_url: str) -> str:
"""Extract the object key from either S3 URL format.
Virtual-hosted puts the bucket in the host (bucket.s3...amazonaws.com/key);
path-style puts it in the first path segment, against either the legacy
global endpoint or a regional one (s3.<region>.amazonaws.com/bucket/key).
"""
parsed_url = urlparse(s3_url)
path_parts = parsed_url.path.strip('/').split('/')
if parsed_url.netloc.startswith(f"{settings.S3_BUCKET}."):
return '/'.join(path_parts)
# Path-style: drop the leading bucket segment.
return '/'.join(path_parts[1:])
def owned_s3_key(url: Optional[str]) -> Optional[str]:
"""The object key when ``url`` names an object in OUR bucket, else None.
_key_from_url assumes the URL is ours and reads any S3 URL as a key in our
bucket; use this instead before acting on a URL a user supplied.
"""
if not is_s3_url(url):
return None
parsed = urlparse(url)
host = (parsed.hostname or "").lower()
parts = parsed.path.strip('/').split('/')
bucket = settings.S3_BUCKET.lower()
if host.startswith(f"{bucket}."):
return '/'.join(parts)
if host.startswith(("s3.", "s3-")) and parts[0] == settings.S3_BUCKET:
return '/'.join(parts[1:])
return None
@lru_cache(maxsize=1)
def get_s3_client():
"""Get boto3 S3 client.
Cached: constructing a client resolves session, endpoint and credential
config and costs ~1.3ms, which dwarfs the ~0.05ms presign it is usually
built for. boto3 clients are thread-safe for method calls. Call
get_s3_client.cache_clear() if settings change at runtime (tests).
The credential settings are None (never "") when unset, which is what makes
boto3 fall through to its default chain and pick up an IAM instance or
container role. See the note on AWS_ACCESS_KEY_ID in core/config.py.
"""
return boto3.client(
's3',
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
region_name=settings.S3_REGION
)
def sign_s3_url(s3_url: str, expiration: Optional[int] = None) -> str:
"""
Generate a signed URL for an S3 object.
Synchronous on purpose: generate_presigned_url computes an HMAC locally and
makes no network call, so this is safe to call from response serialization
(see the field_serializers on the response schemas) without an event loop.
Args:
s3_url: The S3 URL of the object
expiration: Lifetime in seconds; defaults to settings.S3_PRESIGN_EXPIRY_SECONDS
and is clamped to S3_MAX_PRESIGN_SECONDS
Returns:
Signed URL for the object, or the input unchanged if it is not signable
"""
try:
if not settings.S3_FILE_STORAGE or not s3_url:
return s3_url
# Inline data URIs (e.g. generated orb avatars stored in photo_url) and any
# non-S3 absolute URL are not S3 objects — return them unchanged.
if s3_url.startswith('data:') or 'amazonaws.com' not in s3_url:
return s3_url
if expiration is None:
expiration = settings.S3_PRESIGN_EXPIRY_SECONDS
expiration = min(expiration, S3_MAX_PRESIGN_SECONDS)
s3_client = get_s3_client()
signed_url = s3_client.generate_presigned_url(
'get_object',
Params={
'Bucket': settings.S3_BUCKET,
'Key': _key_from_url(s3_url)
},
ExpiresIn=expiration
)
return signed_url
except Exception as e:
logger.exception(f"Error generating signed URL: {str(e)}")
return s3_url
def sign_s3_key(s3_key: str, expiration: Optional[int] = None) -> str:
"""Signed URL for an object key, for callers that hold a key rather than a
stored URL (help-center article images resolve by key)."""
return sign_s3_url(url_for_s3_key(s3_key), expiration)
async def get_s3_signed_url(s3_url: str, expiration: Optional[int] = None) -> str:
"""Awaitable shim over sign_s3_url for the existing `await` call sites.
Deprecated: signing never touched the network, so prefer sign_s3_url.
"""
return sign_s3_url(s3_url, expiration)
async def upload_file_to_s3(
file_content: bytes,
folder: str,
filename: str,
content_type: Optional[str] = None
) -> str:
"""
Upload file to S3 bucket
Args:
file_content: The file content as bytes
folder: The S3 folder path
filename: The filename to save as
content_type: Optional MIME type
Returns:
The S3 URL of the uploaded file
Raises:
HTTPException: if the object cannot be stored or verified
The bucket is expected to exist already — provisioning it belongs in
deployment, not on the request path, and the runtime identity should not
hold s3:CreateBucket.
"""
try:
s3_client = get_s3_client()
# Construct S3 key (path)
s3_key = f"{folder}/{filename}"
# Upload to S3
extra_args = {}
if content_type:
extra_args['ContentType'] = content_type
logger.info(f"Putting object to S3: bucket={settings.S3_BUCKET}, key={s3_key}, size={len(file_content)} bytes")
# boto3 is synchronous — offload so a large body never blocks the loop.
await asyncio.to_thread(
partial(
s3_client.put_object,
Bucket=settings.S3_BUCKET,
Key=s3_key,
Body=file_content,
**extra_args
)
)
# Verify file was written
for attempt in range(_UPLOAD_VERIFY_ATTEMPTS):
try:
await asyncio.to_thread(
partial(s3_client.head_object, Bucket=settings.S3_BUCKET, Key=s3_key)
)
break
except Exception as verify_err:
if attempt < _UPLOAD_VERIFY_ATTEMPTS - 1:
await asyncio.sleep(_UPLOAD_VERIFY_BACKOFF_SECONDS)
else:
logger.error(f"File verification failed after retries: {s3_key} - {str(verify_err)}")
raise HTTPException(
status_code=500,
detail="File upload verification failed"
)
url = url_for_s3_key(s3_key)
logger.info(f"Successfully uploaded file to S3: {s3_key}")
return url
except HTTPException:
raise
except Exception as e:
# Deliberately no local-disk fallback: on a containerised deploy the
# uploads dir is not durable, so falling back silently returns a URL
# that works once and 404s after the next restart, with the caller
# none the wiser. Fail the request instead.
logger.exception(f"S3 upload failed: {str(e)}")
raise HTTPException(status_code=500, detail="File upload failed")
async def download_file_from_s3(s3_url: str) -> bytes:
"""Download an object's bytes (worker-side readback of stored uploads).
boto3 is synchronous — offload so the transfer never blocks the event loop."""
def _download() -> bytes:
s3_client = get_s3_client()
response = s3_client.get_object(Bucket=settings.S3_BUCKET, Key=_key_from_url(s3_url))
return response['Body'].read()
return await asyncio.to_thread(_download)
async def delete_file_from_s3(s3_url: str) -> bool:
"""Delete file from S3 bucket"""
try:
s3_client = get_s3_client()
await asyncio.to_thread(
partial(
s3_client.delete_object,
Bucket=settings.S3_BUCKET,
Key=_key_from_url(s3_url)
)
)
return True
except Exception as e:
logger.error(f"Error deleting from S3: {str(e)}")
return False
backend/app/core/security.py
+489
View File
@@ -0,0 +1,489 @@
"""
Copyright 2024-2026 ReplyPier
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import json
from datetime import datetime, timedelta, timezone
from typing import Optional, Dict
import re
import uuid
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings
from app.core.encryption import CURRENT_KEY_ID, get_keys
from app.core.logger import get_logger
import base64
import hashlib
logger = get_logger(__name__)
# Constants
SECRET_KEY = settings.SECRET_KEY
CONVERSATION_SECRET_KEY = settings.CONVERSATION_SECRET_KEY
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# Password policy for passwords one person sets on behalf of another (today:
# an admin resetting an agent's password). Mirrors the strength meter the UI
# draws — minimum length plus three of the four character classes — so the
# form and the API agree on what "strong enough" means.
MIN_PASSWORD_LENGTH = 8
MIN_PASSWORD_CHARACTER_CLASSES = 3
_PASSWORD_CHARACTER_CLASSES = (r"[A-Z]", r"[a-z]", r"[0-9]", r"[^A-Za-z0-9]")
def validate_password_strength(password: str) -> str:
"""Return the password, or raise ValueError describing what it is missing."""
if len(password or "") < MIN_PASSWORD_LENGTH:
raise ValueError(
f"Password must be at least {MIN_PASSWORD_LENGTH} characters long"
)
classes = sum(1 for pattern in _PASSWORD_CHARACTER_CLASSES if re.search(pattern, password))
if classes < MIN_PASSWORD_CHARACTER_CLASSES:
raise ValueError(
f"Password must include at least {MIN_PASSWORD_CHARACTER_CLASSES} of: "
"uppercase letter, lowercase letter, number, special character"
)
return password
def _fernet():
"""The shared encryption-at-rest key, loaded by app.core.encryption.
API keys predate the ``enc:v1:`` wire format and stay bare Fernet tokens, so
they are encrypted here rather than through encrypt_value/decrypt_value — but
the key itself has exactly one owner.
"""
return get_keys()[CURRENT_KEY_ID]
def create_access_token(data: dict) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire, "type": "access"})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def create_refresh_token(data: dict) -> str:
to_encode = data.copy()
expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
to_encode.update({"exp": expire, "type": "refresh"})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(token: str) -> Optional[dict]:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
def encrypt_api_key(api_key: str) -> str:
"""Encrypt API key before storing in database"""
try:
return _fernet().encrypt(api_key.encode()).decode()
except Exception as e:
logger.error(f"Encryption error: {str(e)}")
raise ValueError("Failed to encrypt API key")
def decrypt_credentials(encrypted: str) -> dict:
"""Decrypt a JSON credential blob stored with encrypt_api_key."""
return json.loads(decrypt_api_key(encrypted))
def decrypt_api_key(encrypted_key: str) -> str:
"""Decrypt API key from database"""
try:
return _fernet().decrypt(encrypted_key.encode()).decode()
except Exception as e:
logger.error(f"Decryption error: {str(e)}")
raise ValueError("Failed to decrypt API key")
def create_conversation_token(widget_id: str, customer_id: Optional[str] = None, customer_email: Optional[str] = None, **extra_data) -> str:
"""
Create a JWT token for widget conversations with revocation support via Redis
Includes a JTI (JWT ID) claim for revocation. Token is stored in Redis with TTL.
Optionally stores customer email for session tracking and bulk revocation.
Args:
widget_id: The widget ID
customer_id: Optional customer ID
customer_email: Optional customer email for session management
**extra_data: Additional data to include in token
"""
jti = str(uuid.uuid4())
data = {
"widget_id": widget_id,
"type": "conversation",
"jti": jti
}
# Only include `sub` when a customer_id is present. RFC 7519 (and python-jose)
# require `sub` to be a string — encoding it as JSON null causes decode to fail
# with JWTError("Subject must be a string") for anonymous widget visitors
# (e.g. Explore flow), which silently drops all downstream claims (like `source`).
if customer_id is not None:
data["sub"] = customer_id
# Add email to token if provided
if customer_email:
data["email"] = customer_email
# Add any extra data to the token
data.update(extra_data)
# Set expiration to 30 days
now = datetime.now(timezone.utc)
expire = now + timedelta(days=30)
data.update({"exp": expire})
token = jwt.encode(data, CONVERSATION_SECRET_KEY, algorithm=ALGORITHM)
# Store token JTI in Redis with TTL and optional email mapping (with widget_id)
_store_token_in_redis(jti, int((expire - now).total_seconds()), customer_email, widget_id)
return token
def verify_conversation_token(token: str) -> Optional[dict]:
"""
Verify conversation token signature and check if revoked
Returns:
Token payload dict if valid and not revoked, None otherwise
"""
try:
# verify_sub=False tolerates existing tokens (already in browsers' localStorage)
# that were issued with `sub: null`. New tokens from create_conversation_token
# omit `sub` entirely when anonymous, so this option only matters for in-flight
# tokens until the old 30-day TTLs expire.
payload = jwt.decode(token, CONVERSATION_SECRET_KEY,
algorithms=[ALGORITHM],
options={"verify_sub": False})
if payload.get("type") != "conversation":
return None
# Check if token JTI exists in Redis (if revoked, it won't exist)
jti = payload.get("jti")
if jti and not _is_token_in_redis(jti):
logger.warning(f"Token revoked or expired: jti={jti[:8]}...")
return None
return payload
except JWTError:
return None
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify a password against its hash"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""Hash a password"""
return pwd_context.hash(password)
# ============================================================================
# TOKEN STORAGE IN REDIS - Minimal approach
# ============================================================================
def _store_token_in_redis(jti: str, ttl_seconds: int, email: Optional[str] = None, widget_id: Optional[str] = None) -> bool:
"""
Store token JTI in Redis with TTL and optionally store email/widget_id mapping
Args:
jti: JWT ID for the token
ttl_seconds: Time to live in seconds
email: Optional email to store for session tracking
widget_id: Optional widget ID for widget-specific session tracking
"""
try:
from app.core.redis import redis_client
if redis_client:
# Store the token JTI
redis_client.setex(f"token:{jti}", ttl_seconds, "1")
# Store email mapping if provided (key: user_sessions:email, value: set of JTIs)
if email:
redis_client.sadd(f"user_sessions:{email}", jti)
# Set same TTL for email->JTI mapping
redis_client.expire(f"user_sessions:{email}", ttl_seconds)
# If widget_id provided, also store widget-specific sessions
if widget_id:
widget_key = f"user_sessions:{email}:widget:{widget_id}"
redis_client.sadd(widget_key, jti)
redis_client.expire(widget_key, ttl_seconds)
# Maintain reverse index: track all widget keys for this email
# for efficient bulk deletion without scanning
redis_client.sadd(f"user_sessions:{email}:widget_keys", widget_key)
redis_client.expire(f"user_sessions:{email}:widget_keys", ttl_seconds)
return True
except Exception as e:
logger.warning(f"Failed to store token in Redis: {str(e)}")
return False
def _is_token_in_redis(jti: str) -> bool:
"""Check if token JTI exists in Redis"""
try:
from app.core.redis import redis_client
if redis_client:
return redis_client.exists(f"token:{jti}") > 0
except Exception as e:
logger.debug(f"Failed to check token in Redis: {str(e)}")
# If Redis unavailable, assume valid (don't block requests)
return True
def revoke_token(jti: str) -> bool:
"""Revoke token by deleting JTI from Redis"""
try:
from app.core.redis import redis_client
if redis_client:
redis_client.delete(f"token:{jti}")
logger.info(f"Token revoked: {jti[:8]}...")
return True
except Exception as e:
logger.error(f"Failed to revoke token: {str(e)}")
return False
def revoke_user_sessions(email: str) -> bool:
"""
Revoke all sessions for a user by email.
Args:
email: The user's email address
Returns:
True if successful, False otherwise
"""
try:
from app.core.redis import redis_client
if redis_client:
# Get all JTIs for this email across all widgets
jti_set = redis_client.smembers(f"user_sessions:{email}")
if jti_set:
# Delete each JTI
for jti in jti_set:
redis_client.delete(f"token:{jti.decode() if isinstance(jti, bytes) else jti}")
# Delete the email mapping
redis_client.delete(f"user_sessions:{email}")
# Also delete all widget-specific mappings using the reverse index set
widget_keys_set = f"user_sessions:{email}:widget_keys"
widget_keys = redis_client.smembers(widget_keys_set)
if widget_keys:
for key in widget_keys:
redis_client.delete(key.decode() if isinstance(key, bytes) else key)
redis_client.delete(widget_keys_set)
# NOTE: When creating or deleting widget-specific session keys elsewhere,
# be sure to add/remove the key from the widget_keys_set for this user.
logger.info(f"Revoked {len(jti_set)} sessions for user: {email}")
return True
else:
logger.info(f"No active sessions found for user: {email}")
return True
except Exception as e:
logger.error(f"Failed to revoke user sessions: {str(e)}")
return False
def revoke_user_sessions_by_widget(email: str, widget_id: str) -> bool:
"""
Revoke all sessions for a user in a specific widget.
Args:
email: The user's email address
widget_id: The widget ID to revoke sessions for
Returns:
True if successful, False otherwise
"""
try:
from app.core.redis import redis_client
if redis_client:
# Get all JTIs for this email in this specific widget
jti_set = redis_client.smembers(f"user_sessions:{email}:widget:{widget_id}")
if jti_set:
# Delete each JTI
for jti in jti_set:
jti_str = jti.decode() if isinstance(jti, bytes) else jti
redis_client.delete(f"token:{jti_str}")
# Also remove from general user_sessions set
redis_client.srem(f"user_sessions:{email}", jti_str)
# Delete the widget-specific mapping
redis_client.delete(f"user_sessions:{email}:widget:{widget_id}")
logger.info(f"Revoked {len(jti_set)} sessions for user: {email} in widget: {widget_id}")
return True
else:
logger.info(f"No active sessions found for user: {email} in widget: {widget_id}")
return True
except Exception as e:
logger.error(f"Failed to revoke user sessions by widget: {str(e)}")
return False
def get_user_active_sessions(email: str, widget_id: Optional[str] = None) -> list:
"""
Get all active session JTIs for a user by email.
Args:
email: The user's email address
widget_id: Optional widget ID to get sessions only for that widget
Returns:
List of active JTI strings for the user (filtered by widget if provided)
"""
try:
from app.core.redis import redis_client
if redis_client:
if widget_id:
# Get sessions only for this specific widget
jti_set = redis_client.smembers(f"user_sessions:{email}:widget:{widget_id}")
else:
# Get all sessions for this user across all widgets
jti_set = redis_client.smembers(f"user_sessions:{email}")
return [jti.decode() if isinstance(jti, bytes) else jti for jti in jti_set]
except Exception as e:
logger.error(f"Failed to get user sessions: {str(e)}")
return []
def get_existing_valid_token_jti(email: str, widget_id: str, customer_id: str) -> Optional[str]:
"""
Check if there's an existing valid token JTI for this customer/widget combination.
Returns the JTI (JWT ID) of an existing valid token, not the token itself.
This prevents token multiplication when the same user refreshes the page multiple times.
Args:
email: Customer email
widget_id: Widget ID
customer_id: Customer ID
Returns:
The JTI (JWT ID) string of an existing valid token if found, None otherwise
"""
try:
from app.core.redis import redis_client
if not redis_client:
return None
# Get all active JTIs for this email in this widget
jti_set = redis_client.smembers(f"user_sessions:{email}:widget:{widget_id}")
if not jti_set:
logger.debug(f"No existing sessions found for {email} in widget {widget_id}")
return None
# If we have active tokens, return the first valid one
# In a typical flow, there should only be one per widget per user
for jti in jti_set:
jti_str = jti.decode() if isinstance(jti, bytes) else jti
# Check if the token still exists in Redis (not revoked)
if redis_client.exists(f"token:{jti_str}") > 0:
logger.info(f"Found existing valid token for {email} in widget {widget_id}, reusing it")
return jti_str
logger.debug(f"Active sessions found for {email} in widget {widget_id}, but none are valid in Redis")
return None
except Exception as e:
logger.error(f"Error checking existing tokens: {str(e)}")
return None
# ============================================================================
# WIDGET API KEY MANAGEMENT
# ============================================================================
def generate_widget_api_key() -> str:
"""
Generate a secure random API key with 'wak_' prefix.
Format: wak_<base64url> (e.g., wak_Xu8f3kLp9mN2...)
Entropy: 256 bits (32 random bytes)
Length: ~47 characters total (4 + 43)
Returns:
str: API key in format 'wak_<base64url>'
"""
import secrets
# Generate 32 random bytes (256 bits of entropy)
random_bytes = secrets.token_bytes(32)
# Encode as URL-safe base64 without padding
key_suffix = base64.urlsafe_b64encode(random_bytes).decode('utf-8').rstrip('=')
return f"wak_{key_suffix}"
def hash_widget_api_key(api_key: str) -> str:
"""
Hash API key using bcrypt (same as password hashing).
Args:
api_key: Plain text API key
Returns:
str: bcrypt hash of the API key
"""
return get_password_hash(api_key)
def verify_widget_api_key(plain_key: str, hashed_key: str) -> bool:
"""
Verify API key against its hash.
Args:
plain_key: Plain text API key from request
hashed_key: Stored bcrypt hash
Returns:
bool: True if key matches hash
"""
return verify_password(plain_key, hashed_key)
def widget_api_key_lookup(api_key: str) -> str:
"""
Indexed lookup value for a widget API key: its SHA-256.
A fast hash is safe here because the key is 256 random bits, nothing to
brute-force; it lets a request find its app in one query instead of
bcrypt-checking every app.
"""
return hashlib.sha256(api_key.encode()).hexdigest()