"""
Copyright 2024-2026 ChatterMate
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 APIRouter, BackgroundTasks, Depends, HTTPException, Request, Response, Header, Query, status
from fastapi.responses import HTMLResponse
from sqlalchemy.orm import Session
from typing import List, Optional
from uuid import UUID
from datetime import datetime
from pathlib import Path
import json
from urllib.parse import urlparse
from app.models.agent import Agent
from app.models.customer import Customer
from app.models.widget import Widget
from app.models.user import User
from app.core.auth import get_current_user
from app.database import get_db
from app.repositories.widget import WidgetRepository
from app.repositories.agent import AgentRepository
from app.core.security import create_conversation_token, verify_conversation_token
from app.services import conversation_token
from app.utils.client_ip import forwarded_client_ip
from app.repositories.customer import CustomerRepository
from app.models.schemas.widget import WidgetCreate, WidgetResponse
from app.core.logger import get_logger
from app.repositories.session_to_agent import SessionToAgentRepository, session_in_scope
from app.services.contact_capture import CONTACT_EMAIL_META_KEY
from app.services.widget_identity import may_stand_for
from app.models.session_to_agent import SessionStatus
from app.core.config import settings
from app.core.s3 import get_s3_signed_url
from app.utils.business_hours import is_within_business_hours
router = APIRouter()
logger = get_logger(__name__)
# The enterprise module can record where each widget is live; without it,
# nothing is recorded.
try:
from app.enterprise.services import widget_install
except ImportError:
widget_install = None
def _identity_expired() -> HTTPException:
"""401 telling the widget its visitor's identity lapsed, not that it never had one.
The embed loader answers this by asking the host page for a fresh token
(`tokenProvider`) and retrying, so an identified visitor is never silently
re-created as an anonymous customer.
"""
return HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={
"code": conversation_token.IDENTITY_EXPIRED_CODE,
"message": "Conversation token expired. Issue a new one with /generate-token.",
},
)
_ASSETS_DIR = Path(settings.WIDGET_ASSETS_DIR)
def _asset_version() -> str:
"""Cache-buster for the widget bundle URLs.
/assets/widget.js is served without a Cache-Control header and its URL never
changes, so browsers fall back to heuristic caching and can keep serving an old
bundle for hours after a release — the widget silently runs last week's code.
Stamping the build's mtime+size onto the URL makes a new build a new URL.
Recomputed per request (a stat is microseconds) so a rebuild takes effect
without restarting the backend.
"""
try:
stat = (_ASSETS_DIR / "widget.js").stat()
return f"{int(stat.st_mtime)}-{stat.st_size}"
except OSError:
# Assets served from somewhere else (CDN, dev server) — nothing to stamp.
return ""
def _widget_html_response(html: str) -> HTMLResponse:
"""The widget HTML embeds a conversation token and the current customization,
and points at a versioned bundle URL. Caching it would hand a visitor someone
else's token from a shared cache, and would keep pinning them to a stale widget
build after a release."""
return HTMLResponse(html, headers={"Cache-Control": "no-store"})
# BACKEND_URL defaults to http://localhost:8000 and neither compose file sets
# it, so a self-hosted install that never overrode it used to serve every visitor
# a widget whose socket dialled the visitor's own machine (#315). Loopback can
# never be right for a browser on someone else's computer, so we fall back to the
# origin the widget HTML was actually fetched over.
_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "0.0.0.0", "::1", "[::1]"})
_warned_about_loopback_backend_url = False
def _is_loopback(url: str) -> bool:
host = urlparse(url).hostname
return host is not None and host.lower() in _LOOPBACK_HOSTS
def _public_origin(request: Optional[Request]) -> Optional[str]:
"""Scheme and host this request reached us on, as the browser sees them.
Reads the forwarding headers our nginx sets rather than ``request.url``, which
only reflects them when the ASGI server is started with proxy headers enabled.
"""
if request is None:
return None
# Several proxies in a chain append to these, and the browser's own value is
# the first entry.
forwarded_host = request.headers.get("x-forwarded-host")
host = (forwarded_host or request.headers.get("host") or "").split(",")[0].strip()
if not host:
return None
forwarded_proto = request.headers.get("x-forwarded-proto")
if forwarded_proto:
scheme = forwarded_proto.split(",")[0].strip()
elif forwarded_host:
# A proxy that rewrites the host but not the protocol (Traefik and some
# Caddy setups, unlike our nginx) leaves request.url.scheme reading "http"
# for the proxy's own hop. Trusting that would hand an https page a ws://
# socket the browser refuses as mixed content, so assume the hop was TLS.
scheme = "https"
else:
scheme = request.url.scheme or "https"
return f"{scheme}://{host}"
def _widget_runtime_config(request: Optional[Request] = None) -> dict:
"""Runtime API/WS URLs to hand the widget iframe as ``window.APP_CONFIG``.
Derived from ``BACKEND_URL`` so a self-hosted widget talks to the configured
backend instead of the vendor cloud. On the hosted deployment BACKEND_URL is
``https://api.chattermate.chat``, which yields exactly the widget's baked-in
defaults, so injecting this is a no-op there.
"""
global _warned_about_loopback_backend_url
api_base = settings.BACKEND_URL.rstrip("/")
if _is_loopback(api_base):
derived = _public_origin(request)
if derived and not _is_loopback(derived):
if not _warned_about_loopback_backend_url:
_warned_about_loopback_backend_url = True
logger.warning(
"BACKEND_URL is %s, which no visitor's browser can reach. "
"Serving widgets against %s instead, taken from the request. "
"Set BACKEND_URL to this install's public API URL.",
api_base, derived,
)
api_base = derived
if api_base.startswith("https://"):
ws_url = "wss://" + api_base[len("https://"):]
elif api_base.startswith("http://"):
ws_url = "ws://" + api_base[len("http://"):]
else:
ws_url = api_base
return {"API_URL": f"{api_base}/api/v1", "WS_URL": ws_url}
@router.post("", response_model=WidgetResponse)
def create_new_widget(
widget: WidgetCreate,
db: Session = Depends(get_db),
current_user=Depends(get_current_user)
):
"""Create a new widget for the organization"""
widget_repo = WidgetRepository(db)
return widget_repo.create_widget(widget, current_user.organization_id)
def _note_install(db: Session, widget: Widget, request: Request, background_tasks: BackgroundTasks) -> None:
"""Note where the widget is live (or blocked), after the response is
sent. Never lets a tracking error break the widget itself."""
if widget_install is None:
return
try:
sighting = widget_install.claim_write(db, widget, request.headers.get("origin"), forwarded_client_ip(request))
if sighting:
background_tasks.add_task(widget_install.record, widget.id, sighting)
except Exception:
logger.exception(f"Install tracking failed for widget {widget.id}")
@router.get("/{widget_id}/data", response_class=HTMLResponse)
async def get_widget_ui(
widget_id: str,
request: Request,
response: Response,
background_tasks: BackgroundTasks,
authorization: Optional[str] = Header(None),
source: Optional[str] = Query(None),
db: Session = Depends(get_db)
):
"""Get widget UI and handle customer authentication"""
widget = db.query(Widget).filter(Widget.id == widget_id).first()
if not widget:
raise HTTPException(status_code=404, detail="Widget not found")
_note_install(db, widget, request, background_tasks)
# Get agent data
agent_repo = AgentRepository(db)
agent = agent_repo.get_by_id(widget.agent_id)
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
# What the header may honestly claim. Only a human-handled agent needs the
# business-hours check: while the AI answers, it answers at 3am too.
presence = {
'mode': 'ai' if agent.ai_replies_enabled else 'human',
'available': True if agent.ai_replies_enabled
else is_within_business_hours(agent.organization),
}
require_token_auth = getattr(agent, 'require_token_auth', False)
customer_id = None
token = None
token_payload = None
# Try to validate existing token if provided
if authorization and authorization.startswith('Bearer '):
token = authorization.split(' ')[1]
state = conversation_token.inspect(token, widget_id)
if state.is_live:
token_payload = state.payload
customer_id = state.payload.get("sub")
elif state.identity_lost:
# A lapsed token that named a customer means a signed-in visitor whose
# identity just ran out. Say so instead of rendering the frame as a
# brand-new anonymous visitor - the loader re-identifies and retries.
raise _identity_expired()
else:
token = None # Invalid token
# SECURITY: If token auth is required, valid token MUST be provided
if require_token_auth:
if not token or customer_id is None:
logger.warning(f"Widget UI request denied: require_token_auth=true but no valid token for widget_id={widget_id}")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Authentication required. Token must be obtained from /api/v1/generate-token endpoint with valid API key."
)
# Has valid token - return widget with existing token
return _widget_html_response(await get_widget_html(
widget_id=widget_id,
agent_name=agent.display_name or agent.name,
agent_customization=agent.customization,
customer_id=customer_id,
initial_token=token,
agent_workflow=bool(agent.use_workflow and agent.active_workflow_id),
allow_attachments=agent.allow_attachments,
ai_replies_enabled=agent.ai_replies_enabled,
presence=presence,
request=request
))
# Token auth NOT required - allow anonymous access
# If we have a valid token, use it; otherwise create a new one
if token and customer_id:
return _widget_html_response(await get_widget_html(
widget_id=widget_id,
agent_name=agent.display_name or agent.name,
agent_customization=agent.customization,
customer_id=customer_id,
initial_token=token,
agent_workflow=bool(agent.use_workflow and agent.active_workflow_id),
allow_attachments=agent.allow_attachments,
ai_replies_enabled=agent.ai_replies_enabled,
presence=presence,
request=request
))
# No valid token - create new one for anonymous access. A live token that
# names no customer yet keeps its conversation key.
token = _mint_widget_token(widget_id, _conversation_key(token_payload), source=source)
return _widget_html_response(await get_widget_html(
widget_id=widget_id,
agent_name=agent.display_name or agent.name,
agent_customization=agent.customization,
customer_id=customer_id,
initial_token=token,
agent_workflow=bool(agent.use_workflow and agent.active_workflow_id),
allow_attachments=agent.allow_attachments,
ai_replies_enabled=agent.ai_replies_enabled,
presence=presence,
request=request
))
async def get_widget_html(widget_id: str, agent_name: str, agent_customization: dict, customer_id: Optional[str] = None, initial_token: Optional[str] = None, agent_workflow: bool = False, allow_attachments: bool = False, ai_replies_enabled: bool = True,
presence: Optional[dict] = None, request: Optional[Request] = None) -> str:
"""Generate widget HTML with embedded data"""
import html
widget_url = settings.VITE_WIDGET_URL
version = _asset_version()
asset_query = f"?v={version}" if version else ""
# Convert AgentCustomization to dict if it's a model instance
customization_dict = {}
if agent_customization:
# Get signed URL for photo if using S3
photo_url = agent_customization.photo_url
if settings.S3_FILE_STORAGE and photo_url:
photo_url = await get_s3_signed_url(photo_url)
customization_dict = {
"chat_background_color": agent_customization.chat_background_color,
"chat_text_color": agent_customization.chat_text_color,
"chat_bubble_color": agent_customization.chat_bubble_color,
"accent_color": agent_customization.accent_color,
"font_family": agent_customization.font_family,
"photo_url": photo_url,
"chat_style": agent_customization.chat_style.value if agent_customization.chat_style else "CHATBOT",
"widget_position": agent_customization.widget_position.value if agent_customization.widget_position else "FLOATING",
"welcome_title": agent_customization.welcome_title,
"welcome_subtitle": agent_customization.welcome_subtitle,
"welcome_message": agent_customization.welcome_message,
"chat_initiation_messages": agent_customization.chat_initiation_messages or [],
"quick_actions": agent_customization.quick_actions or [],
"show_citations": agent_customization.show_citations,
"collect_email": agent_customization.collect_email,
# A human-only agent never produces an AI reply, so disclosing one
# would be untrue — same reason the widget hides it after takeover.
"show_ai_disclaimer": agent_customization.show_ai_disclaimer and ai_replies_enabled,
"allow_new_chat": agent_customization.allow_new_chat,
"customization_metadata": agent_customization.customization_metadata or {}
}
return f"""
Chat Widget
"""
async def get_human_agent_session_info(db: Session, customer_id: str, scope: str) -> dict:
"""Human agent assigned to the customer's open conversation this token may reach"""
human_agent_info = {}
session_repo = SessionToAgentRepository(db)
sessions = session_repo.get_customer_sessions(customer_id, SessionStatus.OPEN, scope=scope)
if sessions and len(sessions) > 0:
session_model,user_full_name, user_profile_pic = sessions[0]
if user_full_name: # If there's a human agent assigned
# HumanAgentResponse signs human_agent_profile_pic on serialization.
# Get human agent info from session
human_agent_info = {
"human_agent_name": user_full_name,
"human_agent_profile_pic": user_profile_pic,
}
return human_agent_info
def _conversation_key(token_payload: Optional[dict]) -> str:
"""Keep the browser's conversation key across re-mints, or start one.
Never an account key: only our backend mints those, for one signed-in
dashboard account, and every token the widget issues is unverified.
"""
key = (token_payload or {}).get(conversation_token.CONVERSATION_KEY_CLAIM)
if key and not conversation_token.is_account_key(key):
return key
return conversation_token.new_conversation_key()
def _mint_widget_token(widget_id: str, conversation_key: str, customer_id: Optional[str] = None,
source: Optional[str] = None, token_payload: Optional[dict] = None) -> str:
"""A conversation token the widget issues itself - unverified, so always scoped.
Prefers a fresh `source` (first-visit Explore flow) over one the visitor's
earlier token carried.
"""
extra_data = {conversation_token.CONVERSATION_KEY_CLAIM: conversation_key}
source = source or (token_payload or {}).get("source")
if widget_id == settings.EXPLORE_WIDGET_ID and source:
extra_data["source"] = source
return create_conversation_token(widget_id=widget_id, customer_id=customer_id, **extra_data)
def _anonymous_customer(customer_repo: CustomerRepository, widget: Widget, conversation_key: str,
claimed_email: Optional[str] = None) -> Customer:
"""A customer of the visitor's own. An address they gave that belongs to
someone they can't stand for is kept where the inbox shows it."""
meta_data = {CONTACT_EMAIL_META_KEY: claimed_email} if claimed_email else None
return customer_repo.create_customer(
CustomerRepository.anonymous_email(), widget.organization_id,
meta_data=meta_data, conversation_key=conversation_key)
def _widget_data_response(widget: Widget, agent: Agent, customer_id: Optional[str], human_agent_info: dict,
token: Optional[str] = None) -> dict:
"""GET /widgets/{id}'s answer; `token` only when the visitor needs a new one."""
# photo_url is signed by AgentCustomizationResponse on serialization.
data = {
"id": widget.id,
"organization_id": widget.organization_id,
"customer_id": customer_id,
"human_agent": human_agent_info,
"agent": {
"id": agent.id,
"name": agent.name,
"display_name": agent.display_name,
"customization": agent.customization,
"workflow": bool(agent.use_workflow and agent.active_workflow_id),
"allow_attachments": agent.allow_attachments
}
}
if token:
data["token"] = token
return data
@router.get("/{widget_id}", response_model=WidgetResponse)
async def get_widget_data(
widget_id: str,
response: Response,
email: Optional[str] = None,
source: Optional[str] = Query(None),
authorization: Optional[str] = Header(None),
db: Session = Depends(get_db)
):
"""Get widget data including agent customization"""
logger.info(f"Getting widget data for widget_id {widget_id}, email {email}, has_token {bool(authorization)}")
# Binds this visitor to a customer: a stored copy would skip that and leave
# the widget holding the page's anonymous token.
response.headers["Cache-Control"] = "no-store"
widget = db.query(Widget).filter(Widget.id == widget_id).first()
if not widget:
raise HTTPException(status_code=404, detail="Widget not found")
agent_repo = AgentRepository(db)
agent = agent_repo.get_by_id(widget.agent_id)
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
require_token_auth = getattr(agent, 'require_token_auth', False)
customer_id = None
token = None
token_payload = None
scope = None
# Try to validate existing token if provided
if authorization and authorization.startswith('Bearer '):
token = authorization.split(' ')[1]
state = conversation_token.inspect(token, widget_id)
if state.is_live:
try:
scope = conversation_token.conversation_scope(state.payload)
token_payload = state.payload
customer_id = state.payload.get("sub")
except ValueError:
token = None
elif state.identity_lost:
# Identity ran out rather than never existing: minting a fresh anonymous
# customer here is what used to turn a signed-in visitor into "Anonymous"
# in the inbox the moment they started a new chat.
raise _identity_expired()
else:
token = None
# SECURITY: If token auth required, must have valid token with customer_id
if require_token_auth:
if not token or customer_id is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized - Token required")
# Check if agent has workflow enabled
agent_has_workflow = bool(agent.use_workflow and agent.active_workflow_id)
# Check if agent has ASK_ANYTHING chat style
is_ask_anything_style = (agent.customization and
agent.customization.chat_style and
agent.customization.chat_style.value == "ASK_ANYTHING")
# Email is optional unless the agent explicitly opts into collecting it before chat.
# When not collecting email we bypass the gate (anonymous customer + token), like
# ASK_ANYTHING / workflow agents already do. NOTE: this runs AFTER the require_token_auth
# 401 check above, so token-protected widgets are unaffected.
email_bypass = not bool(agent.customization and getattr(agent.customization, 'collect_email', False))
customer_repo = CustomerRepository(db)
conversation_key = _conversation_key(token_payload)
# Judged on the customer the socket will act as - the merge survivor - or
# the widget would be told yes here and refused on every connect.
customer = customer_repo.resolve_merged(customer_id) if customer_id not in (None, "None") else None
if customer is not None:
# Once the embedding app signs this person in, a token the widget issued
# (from an email someone typed) may no longer name them - whoever claimed
# the address first would otherwise keep a token for the signed-in user.
# Hand that browser a customer of its own and keep the address it gave.
if not may_stand_for(customer, scope):
logger.info(f"Widget {widget_id}: unverified token names a signed-in customer; starting an anonymous one")
anonymous = _anonymous_customer(customer_repo, widget, conversation_key, claimed_email=customer.email)
new_token = _mint_widget_token(widget_id, conversation_key, customer_id=str(anonymous.id),
source=source, token_payload=token_payload)
return _widget_data_response(widget, agent, str(anonymous.id), {}, new_token)
logger.debug(f"Using existing token for customer_id: {customer_id}")
human_agent_info = await get_human_agent_session_info(db, str(customer.id), scope)
return _widget_data_response(widget, agent, customer_id, human_agent_info)
# A token-protected widget only ever runs on tokens its site issued: one whose
# customer is gone is no reason to issue a widget token of our own.
if require_token_auth:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized - Token required")
# No customer yet - or the token's one no longer exists, which would otherwise
# refuse every connect while this answer kept the token. For workflow agents,
# ASK_ANYTHING style, or email-optional agents, create one with a blank email.
# Only email-collecting regular agents require email up front.
if not (email or agent_has_workflow or is_ask_anything_style or email_bypass):
# Regular agent without email/customer_id - return 401
raise HTTPException(status_code=401, detail="Unauthorized")
# The email is only what the visitor typed. It may still name the existing
# customer - one person in the inbox - because the token minted below is
# scoped to this browser's conversations, not to theirs. Never a person the
# embedding app signed in, though: only its own token may stand for them.
existing = customer_repo.get_customer_by_email(email, widget.organization_id) if email else None
if existing is not None and not existing.is_authenticated:
customer = existing
elif existing is not None:
customer = _anonymous_customer(customer_repo, widget, conversation_key, claimed_email=email)
elif email:
customer = customer_repo.create_customer(email, widget.organization_id, conversation_key=conversation_key)
else:
customer = _anonymous_customer(customer_repo, widget, conversation_key)
human_agent_info = await get_human_agent_session_info(db, str(customer.id), conversation_key)
new_token = _mint_widget_token(widget_id, conversation_key, customer_id=str(customer.id),
source=source, token_payload=token_payload)
return _widget_data_response(widget, agent, str(customer.id), human_agent_info, new_token)
@router.post("/{widget_id}/end-chat")
async def end_chat_acknowledgment(
widget_id: str,
session_id: str,
reason: Optional[str] = None,
description: Optional[str] = None,
authorization: Optional[str] = Header(None),
db: Session = Depends(get_db)
):
"""Acknowledge end_chat event from widget and close session on backend"""
try:
logger.info(f"End chat request received: widget_id={widget_id}, session_id={session_id}, reason={reason}, has_token={bool(authorization)}")
# A conversation token is REQUIRED. This used to validate the token only
# when one was sent, so a request with no Authorization header fell
# straight through and closed any session by id, unauthenticated.
if not authorization or not authorization.startswith('Bearer '):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
try:
token_data = verify_conversation_token(authorization.split(' ')[1])
except Exception:
token_data = None
if not token_data:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
customer_id = token_data.get("sub")
if token_data.get("widget_id") != widget_id:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Widget mismatch")
try:
scope = conversation_token.conversation_scope(token_data)
except ValueError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
# The token proves who the caller is, not that this session is theirs —
# without this any visitor could close another customer's conversation,
# or another browser's conversation of a customer it only claimed to be.
session_repo = SessionToAgentRepository(db)
session = session_repo.get_session(session_id)
if not session or str(session.customer_id) != str(customer_id) or not session_in_scope(session, scope):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found")
success = session_repo.close_session(
session_id=session_id,
reason=reason,
description=description
)
if success:
logger.info(f"End chat acknowledged: widget_id={widget_id}, session_id={session_id}, reason={reason}")
return {
"success": True,
"message": "Chat session closed",
"session_id": session_id,
"closed_at": datetime.utcnow().isoformat()
}
else:
logger.warning(f"Failed to close session: widget_id={widget_id}, session_id={session_id}")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Session not found")
except HTTPException:
raise
except Exception as e:
logger.error(f"Error in end_chat_acknowledgment: {str(e)}")
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to process end chat")
@router.get("", response_model=List[WidgetResponse])
def list_widgets(
db: Session = Depends(get_db),
current_user=Depends(get_current_user)
):
"""List all widgets for the organization"""
widget_repo = WidgetRepository(db)
return widget_repo.get_widgets(current_user.organization_id)
@router.get("/agent/{agent_id}", response_model=List[WidgetResponse])
def get_widgets_by_agent(
agent_id: UUID,
db: Session = Depends(get_db),
current_user=Depends(get_current_user)
):
"""Get all widgets for a specific agent"""
widget_repo = WidgetRepository(db)
widgets = widget_repo.get_widgets_by_agent(agent_id)
# Verify the agent belongs to the user's organization
agent_repo = AgentRepository(db)
agent = agent_repo.get_by_id(agent_id)
if not agent or str(agent.organization_id) != str(current_user.organization_id):
raise HTTPException(status_code=404, detail="Agent not found")
return widgets
@router.delete("/{widget_id}")
def remove_widget(
widget_id: str,
db: Session = Depends(get_db),
current_user=Depends(get_current_user)
):
"""Delete a widget"""
widget_repo = WidgetRepository(db)
widget = widget_repo.get_widget(widget_id)
if not widget or widget.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Widget not found")
widget_repo.delete_widget(widget_id)
return {"message": "Widget deleted"}