feat(auth): 后台登录、角色权限与会话访问策略

管理员与客服分角色登录,权限由后端判定;前端补齐登录、人员与组织管理页面。
This commit is contained in:
WangShen007 committed 2026-09-24 21:35:00 +08:00
1 parent 5e0615f36d
commit e24d3e2a65
123 files changed
+25293

No files matched your search

+724
View File
@@ -0,0 +1,724 @@
"""
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 typing import List
from urllib.parse import quote, urlparse
from fastapi import APIRouter, Depends, HTTPException, status, Response
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
import json
from app.database import get_db
from app.models.organization import Organization
from app.models.user import User
from app.models.schemas.organization import (
OrganizationCreate,
OrganizationUpdate,
OrganizationResponse,
OrganizationCreateResponse
)
from app.core.auth import get_current_user, require_permissions
from app.services.auth_sessions import issue_tokens
from app.core.logger import get_logger
from app.models.role import Role
from app.models.permission import Permission, DEFAULT_AGENT_ROLE_PERMISSIONS
from app.repositories.organization import OrganizationRepository
from app.models.agent import Agent
from app.models.session_to_agent import SessionToAgent
from datetime import datetime, timedelta, timezone
from uuid import UUID
from app.core.cors import update_cors_middleware
from app.core.application import app # Import the FastAPI app instance from the new location
# Disposable-address rejection lives in the enterprise module: the hosted signup
# flow is what attracts throwaway signups, and the community edition has no
# reason to carry an 8k-domain blocklist. Absent, every address is accepted.
try:
from app.enterprise.services.email_validation import ensure_not_disposable
HAS_EMAIL_VALIDATION = True
except ImportError:
HAS_EMAIL_VALIDATION = False
DOMAIN_TAKEN_MESSAGE = "That domain is already used by another organization."
LOCALHOST_DOMAIN_MESSAGE = "Localhost can't be used as your domain. Use the domain of the site where you'll add the widget."
LOCALHOST_HOSTS = frozenset({"localhost", "127.0.0.1"})
logger = get_logger(__name__)
router = APIRouter(
tags=["organizations"]
)
def is_localhost_domain(domain: str) -> bool:
"""Whether a domain (bare host, host:port or URL) points at localhost"""
domain = domain.strip()
try:
host = urlparse(domain if "://" in domain else f"//{domain}").hostname
except ValueError:
return False
return (host or "") in LOCALHOST_HOSTS
@router.post("", response_model=OrganizationCreateResponse, status_code=status.HTTP_201_CREATED)
async def create_organization(
org_data: OrganizationCreate,
response: Response,
db: Session = Depends(get_db),
):
"""Create a new organization with an admin user and default roles"""
try:
if HAS_EMAIL_VALIDATION:
ensure_not_disposable(org_data.admin_email)
# Check if any organization exists
existing_orgs = db.query(Organization).first()
# If organizations exist, return 403 Forbidden
if existing_orgs:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Organization already exists"
)
# Create organization
org_repo = OrganizationRepository(db)
organization = org_repo.create_organization(
name=org_data.name,
domain=org_data.domain,
timezone=org_data.timezone,
business_hours=org_data.business_hours
)
# Note: no default agent is created here. New orgs start with zero
# agents so the guided onboarding flow (Create → Teach → Test → Launch)
# is what creates the first agent.
# Get or create default permissions
permissions = {}
for name, description in Permission.default_permissions():
# Try to get existing permission
perm = db.query(Permission).filter(Permission.name == name).first()
if not perm:
perm = Permission(name=name, description=description)
db.add(perm)
db.flush()
permissions[name] = perm
# Create default roles. Exactly one must be is_default: nothing enforces
# that, and get_default_role() just takes .first() with no ordering, so
# two defaults means a newly invited user lands on whichever row the
# database happens to yield. It has to be Agent — the org creator is
# given Admin explicitly below.
admin_role = Role(
name="Admin",
description="Full access to all features",
organization_id=organization.id,
is_default=False
)
admin_role.permissions = list(permissions.values()) # All permissions
db.add(admin_role)
agent_role = Role(
name="Agent",
description="Access to assigned chats and the unclaimed AI queue",
organization_id=organization.id,
is_default=True
)
agent_role.permissions = [
permissions[name] for name in DEFAULT_AGENT_ROLE_PERMISSIONS
]
db.add(agent_role)
db.flush()
# Create admin user with admin role
admin = User(
email=org_data.admin_email,
full_name=org_data.admin_name,
hashed_password=User.get_password_hash(org_data.admin_password),
organization_id=organization.id,
role_id=admin_role.id,
is_active=True
)
db.add(admin)
db.flush()
# Open the admin's first session and set cookies
access_token, refresh_token = issue_tokens(db, admin.id, organization.id)
# Set cookies and return response
response.set_cookie(
key="access_token",
value=access_token,
httponly=True,
secure=True,
samesite="none", # Changed to "none" for cross-domain support (shopifiy)
max_age=1800 # 30 minutes
)
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True,
secure=True,
samesite="none", # Changed to "none" for cross-domain support (shopifiy)
max_age=604800 # 7 days
)
# Set session data with role information
response.set_cookie(
key="user_info",
value=quote(json.dumps({
"id": str(admin.id),
"email": admin.email,
"full_name": admin.full_name,
"organization_id": str(organization.id),
"role": admin_role.to_dict()
}, default=str)),
samesite="none", # Changed to "none" for cross-domain support (shopifiy)
secure=True, # Required when samesite="none"
max_age=604800 # 7 days
)
db.commit()
# Update CORS origins after creating organization
update_cors_middleware(app)
return {
"id": organization.id,
"name": organization.name,
"domain": organization.domain,
"timezone": organization.timezone,
"business_hours": organization.business_hours,
"settings": organization.settings,
"is_active": organization.is_active,
"access_token": access_token,
"refresh_token": refresh_token,
"token_type": "bearer",
"user": {
"id": admin.id,
"email": admin.email,
"full_name": admin.full_name,
"organization_id": organization.id,
"role": admin_role.to_dict()
}
}
except HTTPException:
db.rollback()
raise
except Exception as e:
db.rollback()
logger.error(f"Organization creation failed: {str(e)}")
raise HTTPException(
status_code=500,
detail=f"Failed to create organization: {str(e)}"
)
@router.get("/setup-status", response_model=dict)
async def is_organization_setup(
db: Session = Depends(get_db)
):
"""Check if at least one active organization is set up."""
try:
active_org_exists = db.query(Organization.id)\
.filter(Organization.is_active == True)\
.first() is not None
return {"is_setup": active_org_exists}
except Exception as e:
logger.error(f"Failed to check organization setup status: {str(e)}", exc_info=True)
# In case of error, conservatively assume setup might be needed or report error
# Returning False might lock users out if DB is temporarily unavailable
# A 500 error might be better to indicate a server issue
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to check organization setup status"
)
@router.get("/check-domain/{domain}")
async def check_domain_availability(
domain: str,
db: Session = Depends(get_db)
):
"""Check if an organization domain is available"""
try:
existing_org = db.query(Organization).filter(Organization.domain == domain).first()
return {
"available": not existing_org,
"message": "Domain is available" if not existing_org else "Domain already exists"
}
except Exception as e:
logger.error(f"Failed to check domain availability: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to check domain availability"
)
def get_own_organization_or_404(db: Session, org_id: UUID, current_user: User) -> Organization:
"""Load an organization the caller actually belongs to.
These routes take the org id from the path, so without this a user of one
tenant could read — and with manage_organization, rewrite — another
tenant's record. 404 rather than 403: the existence of another
organization isn't the caller's business either.
"""
if current_user.organization_id != org_id:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Organization not found"
)
org = db.query(Organization).filter(Organization.id == org_id).first()
if not org:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Organization not found"
)
return org
@router.get("/{org_id}", response_model=OrganizationResponse)
async def get_organization(
org_id: UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
):
"""Get organization by ID"""
try:
return get_own_organization_or_404(db, org_id, current_user)
except HTTPException as he:
raise he
except Exception as e:
logger.error(f"Failed to get organization {
org_id}: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to retrieve organization. Please try again later."
)
@router.patch("/{org_id}", response_model=OrganizationResponse)
async def update_organization(
org_id: UUID,
org_data: OrganizationUpdate,
current_user: User = Depends(require_permissions("manage_organization")),
db: Session = Depends(get_db)
):
"""Update organization details including business hours"""
try:
org = get_own_organization_or_404(db, org_id, current_user)
# Update only provided fields
update_data = org_data.model_dump(exclude_unset=True)
# Validate business hours if provided
if 'business_hours' in update_data:
business_hours = update_data['business_hours']
required_days = ['monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday']
required_fields = ['start', 'end', 'enabled']
# Validate all days are present
if not all(day in business_hours for day in required_days):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Business hours must include all days of the week"
)
# Validate each day has required fields
for day in required_days:
if not all(field in business_hours[day] for field in required_fields):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Business hours for {day} must include start, end, and enabled status"
)
# Validate time format (HH:MM)
start = business_hours[day]['start']
end = business_hours[day]['end']
try:
hours, minutes = start.split(':')
if not (0 <= int(hours) <= 23 and 0 <= int(minutes) <= 59):
raise ValueError
hours, minutes = end.split(':')
if not (0 <= int(hours) <= 23 and 0 <= int(minutes) <= 59):
raise ValueError
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid time format for {day}. Use HH:MM format (24-hour)"
)
new_domain = update_data.get('domain')
if new_domain and new_domain.lower() != org.domain.lower() and is_localhost_domain(new_domain):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=LOCALHOST_DOMAIN_MESSAGE
)
if new_domain and OrganizationRepository(db).is_domain_taken(new_domain, exclude_org_id=org.id):
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=DOMAIN_TAKEN_MESSAGE
)
for field, value in update_data.items():
setattr(org, field, value)
try:
db.commit()
except IntegrityError:
# Another org claimed the domain between the check and the write
if 'domain' not in update_data:
raise
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=DOMAIN_TAKEN_MESSAGE
)
db.refresh(org)
# Update CORS origins after updating organization
update_cors_middleware(app)
return org
except HTTPException:
raise
except Exception as e:
db.rollback()
logger.error(f"Failed to update organization {org_id}: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to update organization. Please try again later."
)
# @router.delete("/{org_id}", status_code=status.HTTP_204_NO_CONTENT)
# async def delete_organization(
# org_id: UUID,
# current_user: User = Depends(require_permissions("manage_organization")),
# db: Session = Depends(get_db)
# ):
# """
# Hard delete organization and all related data.
# This operation permanently removes:
# - Users and their notifications, knowledge queue items, session assignments, ratings
# - Customers and their ratings, chat histories
# - Agents and their customizations, widgets, knowledge links, tool links, ratings, chat histories
# - Workflows and workflow nodes
# - Enterprise subscriptions and PayPal orders (if available)
# - AI configurations
# - Knowledge sources and knowledge-to-agent links
# - MCP tools and their agent links
# - User groups
# - Roles and their permission associations
# - Integration data (Shopify shops, Jira tokens)
# - All remaining chat histories
# - The organization itself
# WARNING: This operation cannot be undone.
# """
# try:
# # Check if organization exists
# org = db.query(Organization).filter(Organization.id == org_id).first()
# if not org:
# raise HTTPException(
# status_code=status.HTTP_404_NOT_FOUND,
# detail="Organization not found"
# )
# # Verify user belongs to this organization
# if current_user.organization_id != org_id:
# raise HTTPException(
# status_code=status.HTTP_403_FORBIDDEN,
# detail="You can only delete your own organization"
# )
# logger.info(f"Starting hard delete for organization {org_id}")
# # Import models that might not be available in all environments
# try:
# from app.enterprise.models.subscription import Subscription
# from app.enterprise.models.order import PayPalOrder
# enterprise_available = True
# except ImportError:
# enterprise_available = False
# logger.info("Enterprise models not available")
# # Import all required models
# from app.models.user import User, UserGroup
# from app.models.role import Role
# from app.models.permission import Permission
# from app.models.agent import Agent, AgentCustomization
# from app.models.widget import Widget
# from app.models.ai_config import AIConfig
# from app.models.knowledge import Knowledge
# from app.models.knowledge_to_agent import KnowledgeToAgent
# from app.models.knowledge_queue import KnowledgeQueue
# from app.models.chat_history import ChatHistory
# from app.models.customer import Customer
# from app.models.session_to_agent import SessionToAgent
# from app.models.rating import Rating
# from app.models.workflow import Workflow
# from app.models.workflow_node import WorkflowNode
# from app.models.mcp_tool import MCPTool, MCPToolToAgent
# from app.models.notification import Notification
# from app.models.shopify.shopify_shop import ShopifyShop
# from app.models.jira import JiraToken
# # 1. Delete users and their related data first
# logger.info("Deleting users and related data...")
# users = db.query(User).filter(User.organization_id == org_id).all()
# for user in users:
# # Delete user notifications
# db.query(Notification).filter(Notification.user_id == user.id).delete()
# # Delete user knowledge queue items
# db.query(KnowledgeQueue).filter(KnowledgeQueue.user_id == user.id).delete()
# # Delete user session assignments
# db.query(SessionToAgent).filter(SessionToAgent.user_id == user.id).delete()
# # Delete user ratings
# db.query(Rating).filter(Rating.user_id == user.id).delete()
# # Delete users themselves (this will cascade to user_groups via the association table)
# db.query(User).filter(User.organization_id == org_id).delete()
# # 2. Delete customer-related data
# logger.info("Deleting customer data...")
# customers = db.query(Customer).filter(Customer.organization_id == org_id).all()
# for customer in customers:
# # Delete customer ratings
# db.query(Rating).filter(Rating.customer_id == customer.id).delete()
# # Delete customer chat histories
# db.query(ChatHistory).filter(ChatHistory.customer_id == customer.id).delete()
# # Delete customers
# db.query(Customer).filter(Customer.organization_id == org_id).delete()
# # 3. Delete agents and their related data
# logger.info("Deleting agents and related data...")
# agents = db.query(Agent).filter(Agent.organization_id == org_id).all()
# for agent in agents:
# # Delete agent customizations
# db.query(AgentCustomization).filter(AgentCustomization.agent_id == agent.id).delete()
# # Delete widgets for this agent
# db.query(Widget).filter(Widget.agent_id == agent.id).delete()
# # Delete knowledge-to-agent links
# db.query(KnowledgeToAgent).filter(KnowledgeToAgent.agent_id == agent.id).delete()
# # Delete MCP tool-to-agent links
# db.query(MCPToolToAgent).filter(MCPToolToAgent.agent_id == agent.id).delete()
# # Delete agent session assignments
# db.query(SessionToAgent).filter(SessionToAgent.agent_id == agent.id).delete()
# # Delete agent ratings
# db.query(Rating).filter(Rating.agent_id == agent.id).delete()
# # Delete agent chat histories
# db.query(ChatHistory).filter(ChatHistory.agent_id == agent.id).delete()
# # Delete agents themselves
# db.query(Agent).filter(Agent.organization_id == org_id).delete()
# # 4. Delete workflows and workflow nodes
# logger.info("Deleting workflows...")
# workflows = db.query(Workflow).filter(Workflow.organization_id == org_id).all()
# for workflow in workflows:
# # Delete workflow nodes
# db.query(WorkflowNode).filter(WorkflowNode.workflow_id == workflow.id).delete()
# # Delete workflows
# db.query(Workflow).filter(Workflow.organization_id == org_id).delete()
# # 5. Delete remaining widgets (not associated with agents)
# logger.info("Deleting remaining widgets...")
# db.query(Widget).filter(Widget.organization_id == org_id).delete()
# # 6. Delete enterprise subscriptions and orders if available
# if enterprise_available:
# logger.info("Deleting enterprise subscriptions and orders...")
# # Delete PayPal orders first (they reference subscriptions)
# db.query(PayPalOrder).filter(PayPalOrder.organization_id == org_id).delete()
# # Delete subscriptions
# db.query(Subscription).filter(Subscription.organization_id == org_id).delete()
# # 7. Delete AI configs
# logger.info("Deleting AI configs...")
# db.query(AIConfig).filter(AIConfig.organization_id == org_id).delete()
# # 8. Delete knowledge sources and related data
# logger.info("Deleting knowledge sources...")
# knowledge_sources = db.query(Knowledge).filter(Knowledge.organization_id == org_id).all()
# for knowledge in knowledge_sources:
# # Delete any remaining knowledge-to-agent links
# db.query(KnowledgeToAgent).filter(KnowledgeToAgent.knowledge_id == knowledge.id).delete()
# # Delete knowledge sources
# db.query(Knowledge).filter(Knowledge.organization_id == org_id).delete()
# # 9. Delete remaining knowledge queue items
# logger.info("Deleting knowledge queue items...")
# db.query(KnowledgeQueue).filter(KnowledgeQueue.organization_id == org_id).delete()
# # 10. Delete MCP tools
# logger.info("Deleting MCP tools...")
# mcp_tools = db.query(MCPTool).filter(MCPTool.organization_id == org_id).all()
# for tool in mcp_tools:
# # Delete any remaining MCP tool-to-agent links
# db.query(MCPToolToAgent).filter(MCPToolToAgent.mcp_tool_id == tool.id).delete()
# # Delete MCP tools
# db.query(MCPTool).filter(MCPTool.organization_id == org_id).delete()
# # 11. Delete user groups
# logger.info("Deleting user groups...")
# db.query(UserGroup).filter(UserGroup.organization_id == org_id).delete()
# # 12. Delete roles and their permission associations
# logger.info("Deleting roles and their permissions...")
# # First, directly delete entries from the role_permissions association table
# from app.models.permission import role_permissions
# # Get all role IDs for this organization
# role_ids = [role_id for role_id, in db.query(Role.id).filter(Role.organization_id == org_id).all()]
# if role_ids:
# # Delete from the association table using raw SQL
# # This is necessary because SQLAlchemy ORM doesn't directly support bulk deletion from association tables
# delete_stmt = role_permissions.delete().where(role_permissions.c.role_id.in_(role_ids))
# db.execute(delete_stmt)
# db.flush()
# # Now delete the roles
# db.query(Role).filter(Role.organization_id == org_id).delete()
# # 13. Delete integration-specific data
# logger.info("Deleting integration data...")
# # Delete Shopify shops
# db.query(ShopifyShop).filter(ShopifyShop.organization_id == org_id).delete()
# # Delete Jira tokens
# db.query(JiraToken).filter(JiraToken.organization_id == org_id).delete()
# # 14. Delete any remaining chat histories
# logger.info("Deleting remaining chat histories...")
# db.query(ChatHistory).filter(ChatHistory.organization_id == org_id).delete()
# # 15. Finally, delete the organization itself
# logger.info("Deleting organization...")
# db.delete(org)
# # Commit all changes
# db.commit()
# logger.info(f"Successfully hard deleted organization {org_id}")
# # Update CORS origins after deleting organization
# update_cors_middleware(app)
# return None
# except HTTPException as he:
# db.rollback()
# raise he
# except Exception as e:
# db.rollback()
# logger.error(f"Failed to delete organization {org_id}: {str(e)}", exc_info=True)
# raise HTTPException(
# status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
# detail="Failed to delete organization. Please try again later."
# )
@router.get("/{org_id}/stats")
async def get_organization_stats(
org_id: UUID,
current_user: User = Depends(require_permissions("view_organization")),
db: Session = Depends(get_db)
):
"""Get organization statistics"""
org = get_own_organization_or_404(db, org_id, current_user)
# --- Members ---
total_users = db.query(User).filter(User.organization_id == org_id).count()
active_users = db.query(User).filter(
User.organization_id == org_id,
User.is_active == True
).count()
# Count admins consistently with the team-overview endpoint (users.py):
# a user is an admin if their role grants manage_organization/super_admin,
# or the role is literally named "Admin". A plain Role.name == "Admin"
# match misses custom/renamed roles that still hold admin permissions.
org_users = db.query(User).filter(User.organization_id == org_id).all()
admins = sum(
1
for u in org_users
if u.role and (
{"manage_organization", "super_admin"} & {p.name for p in (u.role.permissions or [])}
or u.role.name == "Admin"
)
)
members_agents = total_users - admins
# --- Active now (online) ---
active_now = db.query(User).filter(
User.organization_id == org_id,
User.is_online == True
).count()
# --- AI agents (live vs draft) ---
agents_total = db.query(Agent).filter(Agent.organization_id == org_id).count()
agents_live = db.query(Agent).filter(
Agent.organization_id == org_id,
Agent.is_active == True
).count()
agents_draft = agents_total - agents_live
# --- Conversations (last 30d vs previous 30d) ---
now = datetime.now(timezone.utc)
d30 = now - timedelta(days=30)
d60 = now - timedelta(days=60)
conversations_30d = db.query(SessionToAgent).filter(
SessionToAgent.organization_id == org_id,
SessionToAgent.assigned_at >= d30
).count()
conversations_prev_30d = db.query(SessionToAgent).filter(
SessionToAgent.organization_id == org_id,
SessionToAgent.assigned_at >= d60,
SessionToAgent.assigned_at < d30
).count()
if conversations_prev_30d:
conversations_change_pct = round(
(conversations_30d - conversations_prev_30d) / conversations_prev_30d * 100
)
else:
conversations_change_pct = 100 if conversations_30d else 0
return {
"total_users": total_users,
"active_users": active_users,
"settings": org.settings,
# Design KPI strip
"members_total": total_users,
"members_admins": admins,
"members_agents": members_agents,
"active_now": active_now,
"agents_total": agents_total,
"agents_live": agents_live,
"agents_draft": agents_draft,
"conversations_30d": conversations_30d,
"conversations_prev_30d": conversations_prev_30d,
"conversations_change_pct": conversations_change_pct,
}
+228
View File
@@ -0,0 +1,228 @@
"""
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, Depends, HTTPException, status, Query
from sqlalchemy.orm import Session
from typing import Optional
from uuid import UUID
from app.core.logger import get_logger
from app.core.redis import get_redis
from app.database import get_db
from app.core.auth import (
PEOPLE_READ_PERMISSIONS,
PEOPLE_WRITE_PERMISSIONS,
get_current_user,
has_any_permission,
)
from app.models.user import User
from app.repositories.people import PeopleRepository
from app.repositories.crm import CrmConnectionRepository, CrmCustomerSyncRepository
from app.models.crm import CrmConnectionStatus
from app.services.crm_sync import sync_customer_to_crm, CrmManualSyncError
from app.models.schemas.people import (
PeopleListResponse, PeopleStats, PersonDetail, PersonListItem, PersonUpdateRequest,
)
from app.models.schemas.crm import CrmCustomerSyncOut, PersonCrmStatus
from app.core.params import PageNumber
# Enterprise gating — People is part of Lead Management, a Pro-plan feature where
# the enterprise module is installed; OSS/community deployments are unrestricted.
try:
from app.enterprise.repositories.plan import PlanRepository
from app.enterprise.services.feature_access import require_accessible_subscription
HAS_ENTERPRISE = True
except ImportError:
HAS_ENTERPRISE = False
router = APIRouter()
logger = get_logger(__name__)
# Ignore a repeat manual "Sync now" for the same person within this window.
MANUAL_SYNC_DEBOUNCE_SECONDS = 10
def _require_people_access(
current_user: User,
db: Session,
permissions: tuple = PEOPLE_READ_PERMISSIONS
) -> None:
# Reading the directory is its own grant (view_people): an agent handling a
# conversation should be able to look up who they are talking to without
# also gaining the right to read every other agent's chats. The org-wide
# inbox permissions imply it, so admins are unaffected. Writes pass
# PEOPLE_WRITE_PERMISSIONS instead. has_any_permission honours the
# super_admin bypass this check used to miss (a super_admin could send
# templates but got 403 here).
if not has_any_permission(current_user, permissions):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions")
# Pro-plan gate (Lead Management) where the enterprise module is present.
if HAS_ENTERPRISE:
subscription = require_accessible_subscription(db, current_user.organization_id)
if not PlanRepository(db).check_feature_availability(str(subscription.plan_id), 'lead_capture'):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="People / Lead Management is not available in your current plan. Please upgrade to Pro.",
)
@router.get("", response_model=PeopleListResponse)
async def list_people(
stage: Optional[str] = Query(None),
search: Optional[str] = Query(None),
page: PageNumber = 1,
page_size: int = Query(20, ge=1, le=100),
view: str = Query("identified", pattern="^(identified|anonymous)$"),
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""Paginated, filterable list of the org's people.
The default view is identified people (real email, phone, or a qualifying
capture); anonymous browser sessions live behind view=anonymous — they are
a lead-capture funnel signal, not directory content."""
_require_people_access(current_user, db)
items, total = PeopleRepository(db).list_people(
current_user.organization_id, stage=stage, search=search, page=page,
page_size=page_size, view=view,
)
return PeopleListResponse(
items=[PersonListItem(**i) for i in items],
total=total, page=page, page_size=page_size,
)
@router.get("/stats", response_model=PeopleStats)
async def people_stats(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""KPI counters for the People page header."""
_require_people_access(current_user, db)
return PeopleStats(**PeopleRepository(db).get_stats(current_user.organization_id))
@router.get("/{customer_id}", response_model=PersonDetail)
async def get_person(
customer_id: UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""Full profile: captured attributes, lifecycle timeline, conversations."""
_require_people_access(current_user, db)
detail = PeopleRepository(db).get_detail(current_user.organization_id, customer_id)
if not detail:
raise HTTPException(status_code=404, detail="Person not found")
return PersonDetail(**detail)
@router.post("/{customer_id}/mark-customer", response_model=PersonDetail)
async def mark_as_customer(
customer_id: UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""Manually promote a person to the Customer stage (phase 1: no automated signal)."""
_require_people_access(current_user, db, PEOPLE_WRITE_PERMISSIONS)
repo = PeopleRepository(db)
customer = repo.get_customer(current_user.organization_id, customer_id)
if not customer:
raise HTTPException(status_code=404, detail="Person not found")
# An anonymous browser session cannot BE a customer — there is no one to
# recognize. Identify them first (the drawer's edit adds a phone/name).
if not repo.is_identified(customer, repo._has_qualified_capture(customer.id)):
raise HTTPException(status_code=400,
detail="Add an email or phone first — this person is anonymous")
repo.mark_customer(current_user.organization_id, customer_id)
detail = repo.get_detail(current_user.organization_id, customer_id)
return PersonDetail(**detail)
@router.patch("/{customer_id}", response_model=PersonDetail)
async def update_person(
customer_id: UUID,
request: PersonUpdateRequest,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""Explicit human edit of a person's name/phone — the identification tool.
This is the one path allowed to CORRECT a phone (the automatic capture
paths are set-if-absent): a mistyped outbound number must be fixable."""
_require_people_access(current_user, db, PEOPLE_WRITE_PERMISSIONS)
repo = PeopleRepository(db)
customer, error = repo.update_person(
current_user.organization_id, customer_id,
full_name=request.full_name, phone=request.phone,
)
if customer is None:
raise HTTPException(status_code=404, detail="Person not found")
if error:
raise HTTPException(status_code=400, detail=error)
detail = repo.get_detail(current_user.organization_id, customer.id)
return PersonDetail(**detail)
def _person_crm_status(db: Session, organization_id, customer_id) -> PersonCrmStatus:
connected = [
c.provider for c in CrmConnectionRepository(db).list_by_org(organization_id)
if c.status == CrmConnectionStatus.ACTIVE.value
]
synced = CrmCustomerSyncRepository(db).list_for_customer(customer_id)
return PersonCrmStatus(
connected_providers=connected,
synced=[CrmCustomerSyncOut.model_validate(s) for s in synced],
)
@router.get("/{customer_id}/crm", response_model=PersonCrmStatus)
async def get_person_crm(
customer_id: UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""CRM state for the drawer: which CRMs the org has connected + where this
person is already synced."""
_require_people_access(current_user, db)
repo = PeopleRepository(db)
customer = repo.get_customer(current_user.organization_id, customer_id)
if not customer:
raise HTTPException(status_code=404, detail="Person not found")
return _person_crm_status(db, current_user.organization_id, customer_id)
@router.post("/{customer_id}/crm-sync", response_model=PersonCrmStatus)
async def sync_person_crm(
customer_id: UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
"""Manual "Sync now": push this person to every connected CRM, synchronously."""
_require_people_access(current_user, db, PEOPLE_WRITE_PERMISSIONS)
repo = PeopleRepository(db)
customer = repo.get_customer(current_user.organization_id, customer_id)
if not customer:
raise HTTPException(status_code=404, detail="Person not found")
# Debounce: a double-click shouldn't fire a second fan-out of CRM writes
# while the first is still in flight. Best-effort (Redis only).
redis_client = get_redis()
if redis_client is not None and not redis_client.set(
f"crm_manual_sync:{customer_id}", "1", nx=True, ex=MANUAL_SYNC_DEBOUNCE_SECONDS):
return _person_crm_status(db, current_user.organization_id, customer_id)
try:
await sync_customer_to_crm(db, customer)
except CrmManualSyncError as e:
raise HTTPException(status_code=400, detail=str(e))
return _person_crm_status(db, current_user.organization_id, customer_id)
+258
View File
@@ -0,0 +1,258 @@
"""
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, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from app.database import get_db
from app.models.user import User
from app.core.auth import (
check_permissions,
require_any_permission,
require_grantable,
require_permissions,
)
from app.repositories.role import RoleRepository
from app.models.permission import Permission
from app.core.params import DbIntId
from app.models.schemas.role import (
RoleCreate,
RoleUpdate,
RoleResponse,
PermissionResponse
)
from typing import List
from uuid import UUID
router = APIRouter()
# Reading the role catalogue is part of two surfaces: the Roles editor
# (manage_roles) and the invite form, which needs the role dropdown
# (manage_users). Gating reads on manage_roles alone breaks user invite.
ROLE_READ_PERMISSIONS = ("manage_roles", "manage_users")
@router.post("", response_model=RoleResponse)
async def create_role(
role_data: RoleCreate,
current_user: User = Depends(require_permissions("manage_roles")),
db: Session = Depends(get_db)
):
"""Create a new role"""
role_repo = RoleRepository(db)
# Check if default role already exists for org
if role_data.is_default:
existing_default = role_repo.get_default_role(current_user.organization_id)
if existing_default:
raise HTTPException(
status_code=400,
detail="Organization already has a default role"
)
# Validate permissions exist
if role_data.permissions:
permission_ids = [p.id for p in role_data.permissions]
existing_permissions = db.query(Permission)\
.filter(Permission.id.in_(permission_ids))\
.all()
if len(existing_permissions) != len(permission_ids):
raise HTTPException(
status_code=400,
detail="Invalid permission"
)
# A new role is a grant too — otherwise the rule is one "create role,
# then assign myself to it" away from being bypassed.
require_grantable(current_user, {p.name for p in existing_permissions})
try:
role = role_repo.create_role(
name=role_data.name,
description=role_data.description,
organization_id=current_user.organization_id,
is_default=role_data.is_default,
permission_ids=[permission.id for permission in role_data.permissions]
)
return role
except IntegrityError:
raise HTTPException(
status_code=400,
detail="Role with this name already exists"
)
@router.get("", response_model=List[RoleResponse])
async def list_roles(
current_user: User = Depends(require_any_permission(*ROLE_READ_PERMISSIONS)),
db: Session = Depends(get_db)
):
"""List all roles in the organization"""
role_repo = RoleRepository(db)
return role_repo.get_roles_by_organization(current_user.organization_id)
@router.get("/{role_id}", response_model=RoleResponse)
async def get_role(
role_id: DbIntId,
current_user: User = Depends(require_any_permission(*ROLE_READ_PERMISSIONS)),
db: Session = Depends(get_db)
):
"""Get role details including permissions"""
role_repo = RoleRepository(db)
role = role_repo.get_role(role_id)
if not role or role.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Role not found")
return role
@router.put("/{role_id}", response_model=RoleResponse)
async def update_role(
role_id: DbIntId,
role_data: RoleUpdate,
current_user: User = Depends(require_permissions("manage_roles")),
db: Session = Depends(get_db)
):
"""Update role details"""
role_repo = RoleRepository(db)
role = role_repo.get_role(role_id)
if not role or role.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Role not found")
# Prevent editing default role
if role.is_default:
raise HTTPException(
status_code=400,
detail="Cannot modify default role"
)
# Check default role constraints
if role_data.is_default:
existing_default = role_repo.get_default_role(current_user.organization_id)
if existing_default and existing_default.id != role_id:
raise HTTPException(
status_code=400,
detail="Organization already has a default role"
)
# Validate permissions if provided
if role_data.permissions:
permission_ids = [p.id for p in role_data.permissions]
existing_permissions = db.query(Permission)\
.filter(Permission.id.in_(permission_ids))\
.all()
if len(existing_permissions) != len(permission_ids):
raise HTTPException(
status_code=400,
detail="Invalid permission"
)
# Only the diff: re-saving a role that already holds a permission the
# caller lacks is legitimate; adding one is the escalation.
already_held = {p.name for p in role.permissions}
require_grantable(
current_user,
{p.name for p in existing_permissions} - already_held,
)
try:
updated_role = role_repo.update_role(
role_id,
**role_data.model_dump(exclude_unset=True) # Changed from dict() to model_dump()
)
return updated_role
except IntegrityError:
raise HTTPException(
status_code=400,
detail="Role with this name already exists"
)
@router.delete("/{role_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_role(
role_id: DbIntId,
current_user: User = Depends(require_permissions("manage_roles")),
db: Session = Depends(get_db)
):
"""Delete a role"""
role_repo = RoleRepository(db)
role = role_repo.get_role(role_id)
if not role or role.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Role not found")
if role.is_default:
raise HTTPException(
status_code=400,
detail="Cannot delete default role"
)
if role_repo.is_role_in_use(role_id):
raise HTTPException(
status_code=400,
detail="Cannot delete role that is assigned to users"
)
role_repo.delete_role(role_id)
@router.post("/{role_id}/permissions/{permission}")
async def add_role_permission(
role_id: DbIntId,
permission: str,
current_user: User = Depends(require_permissions("manage_roles")),
db: Session = Depends(get_db)
):
"""Add a permission to a role"""
role_repo = RoleRepository(db)
role = role_repo.get_role(role_id)
if not role or role.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Role not found")
# Resolve the name before the grant check so an unknown permission still
# reads as 404 rather than "you may not grant that".
if not db.query(Permission).filter(Permission.name == permission).first():
raise HTTPException(status_code=404, detail="Permission not found")
require_grantable(current_user, [permission])
success = role_repo.add_permission(role_id, permission)
if not success:
raise HTTPException(status_code=404, detail="Permission not found")
return {"message": "Permission added to role"}
@router.delete("/{role_id}/permissions/{permission}")
async def remove_role_permission(
role_id: DbIntId,
permission: str,
current_user: User = Depends(require_permissions("manage_roles")),
db: Session = Depends(get_db)
):
"""Remove a permission from a role"""
role_repo = RoleRepository(db)
role = role_repo.get_role(role_id)
if not role or role.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Role not found")
success = role_repo.remove_permission(role_id, permission)
if not success:
raise HTTPException(status_code=404, detail="Permission not found")
return {"message": "Permission removed from role"}
@router.get("/permissions/all", response_model=List[PermissionResponse])
async def list_permissions(
current_user: User = Depends(require_any_permission(*ROLE_READ_PERMISSIONS)),
db: Session = Depends(get_db)
):
"""List all available permissions"""
return db.query(Permission).all()
+527
View File
@@ -0,0 +1,527 @@
"""
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, Depends, Header, HTTPException, status, Request
from sqlalchemy.orm import Session
from datetime import datetime, timezone
import logging
import time
from typing import Optional, Dict, Any
from app.database import get_db
from app.models.widget import Widget
from app.models.customer import Customer
from app.core.config import settings
from app.core.security import (
verify_conversation_token,
get_existing_valid_token_jti,
revoke_token as security_revoke_token,
)
from app.services import conversation_token
from app.repositories.widget_app import WidgetAppRepository
from app.repositories.customer import CustomerRepository
from pydantic import BaseModel, field_validator
import json
router = APIRouter()
logger = logging.getLogger(__name__)
# custom_data is embedded in both the JWT and the customer's stored meta_data, so cap
# its size to keep tokens small and prevent unbounded growth of the DB column.
CUSTOM_DATA_MAX_KEYS = 20
CUSTOM_DATA_MAX_BYTES = 4096
# ============================================================================
# MODELS
# ============================================================================
class GenerateTokenRequest(BaseModel):
"""Request to generate widget token - requires API key in Authorization header"""
widget_id: str
customer_email: Optional[str] = None
customer_name: Optional[str] = None
custom_data: Optional[Dict[str, Any]] = None
ttl_seconds: Optional[int] = 3600 # Default 1 hour
@field_validator('custom_data')
@classmethod
def validate_custom_data_size(cls, value):
if value is None:
return value
if len(value) > CUSTOM_DATA_MAX_KEYS:
raise ValueError(f"custom_data supports at most {CUSTOM_DATA_MAX_KEYS} keys")
size = len(json.dumps(value))
if size > CUSTOM_DATA_MAX_BYTES:
raise ValueError(f"custom_data must be at most {CUSTOM_DATA_MAX_BYTES} bytes when serialized")
return value
class RefreshTokenRequest(BaseModel):
"""Request to rotate a still-valid conversation token"""
widget_id: str
class RevokeTokenRequest(BaseModel):
"""Request to revoke a token"""
token: str
reason: Optional[str] = None
class RevokeTokenResponse(BaseModel):
"""Response for token revocation"""
success: bool
message: str
revoked_at: Optional[str] = None
class TokenData(BaseModel):
"""Token response data"""
token: str
widget_id: str
expires_in: int
expires_at: str
created_at: str
class GenerateTokenResponse(BaseModel):
"""Response with generated token"""
success: bool
data: TokenData
message: Optional[str] = None
# ============================================================================
# DEPENDENCIES
# ============================================================================
def validate_api_key(request: Request, db: Session = Depends(get_db)):
"""
Validate API key from Authorization header (Bearer token) against the
widget_apps table. A plain def, so FastAPI runs it in a worker thread:
a key issued before the lookup column is bcrypt-checked, which must not
hold up the event loop.
"""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Missing or invalid Authorization header. Use 'Bearer YOUR_API_KEY'",
headers={"WWW-Authenticate": "Bearer"},
)
api_key = auth_header[len("Bearer "):]
if not api_key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="API key cannot be empty",
headers={"WWW-Authenticate": "Bearer"},
)
repo = WidgetAppRepository(db)
widget_app = repo.validate_api_key(api_key)
if not widget_app:
logger.warning(f"Invalid API key attempt")
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API key",
headers={"WWW-Authenticate": "Bearer"},
)
# Store widget_app in request state for downstream use
request.state.widget_app = widget_app
return api_key
# ============================================================================
# ENDPOINTS
# ============================================================================
@router.post("/generate-token", response_model=GenerateTokenResponse, status_code=status.HTTP_201_CREATED)
async def generate_widget_token(
body: GenerateTokenRequest,
request: Request,
db: Session = Depends(get_db),
api_key: str = Depends(validate_api_key)
):
"""
Generate a JWT token for widget authentication.
Only calls with valid API key (in Authorization header) will succeed.
Frontend receives this token and must pass it to the widget.
The token is signed with CONVERSATION_SECRET_KEY and can be verified by the widget.
Args:
body: GenerateTokenRequest with widget_id, customer_email, customer_name, ttl_seconds
request: FastAPI request object (contains widget_app from API key validation)
db: Database session
api_key: Validated API key from Authorization header
Returns:
GenerateTokenResponse with JWT token
Raises:
HTTPException 401: Invalid or missing API key
HTTPException 404: Widget not found
HTTPException 400: Invalid widget_id
Example:
POST /api/v1/generate-token
Authorization: Bearer YOUR_API_KEY
{
"widget_id": "widget-123",
"customer_email": "user@example.com",
"customer_name": "John Doe",
"ttl_seconds": 3600
}
Response:
{
"success": true,
"data": {
"token": "eyJhbGciOiJIUzI1NiIs...",
"widget_id": "widget-123",
"expires_in": 3600,
"expires_at": "2024-11-09T15:00:00",
"created_at": "2024-11-09T14:00:00"
}
}
"""
try:
# Get widget_app from request state (set by validate_api_key)
widget_app = request.state.widget_app
# SECURITY: Validate widget exists AND belongs to same organization as the API key
# This prevents cross-organization token generation
widget = db.query(Widget).filter(
Widget.id == body.widget_id,
Widget.organization_id == widget_app.organization_id
).first()
if not widget:
logger.error(f"Widget not found or access denied: {body.widget_id}")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Widget '{body.widget_id}' not found"
)
# Create or update customer
customer = None
if body.customer_email:
# The embedding app supplied a known identity, so this is an
# authenticated/integration customer (excluded from the People/leads
# views), not an organic lead. Shared with the hosted support chat,
# which identifies the signed-in dashboard user the same way.
customer = CustomerRepository(db).identify(
email=body.customer_email,
organization_id=widget.organization_id,
full_name=body.customer_name,
meta_data=body.custom_data,
)
else:
# No email provided - create anonymous customer
# Generate a unique email for anonymous customers
import time
timestamp = int(time.time() * 1000)
anonymous_email = f"anonymous-{timestamp}@{widget.organization_id}.local"
customer = Customer(
email=anonymous_email,
full_name=body.customer_name or "Anonymous",
meta_data=body.custom_data,
organization_id=widget.organization_id
)
db.add(customer)
db.commit()
db.refresh(customer)
logger.info(f"Created anonymous customer: {customer.id}")
# Validate TTL
if not conversation_token.is_ttl_supported(body.ttl_seconds):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=(
f"ttl_seconds must be between {conversation_token.MIN_TTL_SECONDS} and "
f"{conversation_token.MAX_TTL_SECONDS} (1 minute to 24 hours) for security "
"and performance reasons"
)
)
# Check for existing valid token for this customer/widget combination
# This prevents token multiplication when user refreshes page
customer_email = body.customer_email or customer.email
existing_jti = get_existing_valid_token_jti(customer_email, body.widget_id, str(customer.id))
if existing_jti:
# Only the JTI is kept in Redis, so the token itself is re-signed - reusing
# the ID keeps this customer to one revocable session per widget.
logger.info(f"Found existing valid token for customer {customer.id} in widget {body.widget_id}, reusing JTI")
token, expires_at, ttl = conversation_token.mint(
{
"sub": str(customer.id), # Sub is now always the customer_id
"widget_id": body.widget_id,
"customer_email": customer_email,
"customer_name": body.customer_name or customer.full_name,
"customer_id": str(customer.id), # Explicit customer_id claim
"custom_data": body.custom_data,
},
ttl_seconds=body.ttl_seconds,
jti=existing_jti,
)
now = datetime.now(timezone.utc)
if not existing_jti:
logger.info(f"Generated new widget token for widget_id={body.widget_id}, customer_email={customer_email}, expires_in={ttl}s")
else:
logger.info(f"Reused JTI and refreshed Redis TTL for widget_id={body.widget_id}, customer_email={customer_email}, jti={existing_jti[:8]}...")
return GenerateTokenResponse(
success=True,
data=TokenData(
token=token,
widget_id=body.widget_id,
expires_in=ttl,
expires_at=expires_at.isoformat(),
created_at=now.isoformat()
),
message=f"Token generated successfully. Expires in {ttl} seconds."
)
except HTTPException:
raise
except Exception as e:
logger.exception(f"Error generating token: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to generate token"
)
@router.post("/refresh-token", response_model=GenerateTokenResponse)
async def refresh_widget_token(
body: RefreshTokenRequest,
authorization: Optional[str] = Header(None)
):
"""
Rotate a still-valid conversation token, keeping the visitor's identity.
The widget calls this before its token lapses, so a page left open longer than
`ttl_seconds` doesn't silently drop an identified visitor back to anonymous.
No API key: the request is authenticated by the conversation token it carries,
which the caller already holds, so rotating it grants nothing new. An expired or
revoked token is refused - only POST /generate-token (API key) can assert an
identity - and rotation stops once the chain reaches its absolute ceiling.
Args:
body: RefreshTokenRequest with the widget_id the token was issued for
authorization: `Bearer <conversation token>`
Returns:
GenerateTokenResponse with the replacement token
Raises:
HTTPException 401: missing, invalid, expired or revoked conversation token
"""
token = authorization[len("Bearer "):].strip() if authorization and authorization.startswith("Bearer ") else None
state = conversation_token.inspect(token, body.widget_id)
if not state.is_live:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={
"code": (
conversation_token.IDENTITY_EXPIRED_CODE if state.identity_lost
else conversation_token.INVALID_TOKEN_CODE
),
"message": "Conversation token is no longer valid. Issue a new one with /generate-token.",
},
)
rotated = conversation_token.rotate(state.payload)
if not rotated:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={
"code": conversation_token.IDENTITY_EXPIRED_CODE,
"message": "Conversation token has reached its maximum lifetime. Issue a new one with /generate-token.",
},
)
new_token, expires_at, ttl = rotated
logger.info(f"Rotated conversation token for widget_id={body.widget_id}, expires_in={ttl}s")
return GenerateTokenResponse(
success=True,
data=TokenData(
token=new_token,
widget_id=body.widget_id,
expires_in=ttl,
expires_at=expires_at.isoformat(),
created_at=datetime.now(timezone.utc).isoformat()
),
message=f"Token refreshed successfully. Expires in {ttl} seconds."
)
@router.post("/verify-token")
async def verify_widget_token(
token: str,
widget_id: str,
db: Session = Depends(get_db),
api_key: str = Depends(validate_api_key)
):
"""
Verify a widget token (for backend validation).
Args:
token: JWT token to verify
widget_id: Widget ID to match against token
db: Database session
api_key: Validated API key from Authorization header
Returns:
Token data if valid, error if invalid
Example:
POST /api/v1/verify-token?token=eyJ...&widget_id=widget-123
Authorization: Bearer YOUR_API_KEY
"""
try:
# Verify JWT signature AND check Redis revocation status
# This ensures revoked tokens are rejected
payload = verify_conversation_token(token)
if not payload:
logger.warning(f"Token invalid or revoked for widget_id={widget_id}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Token is invalid or has been revoked"
)
# Verify widget_id matches
if payload.get("widget_id") != widget_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Token widget_id does not match provided widget_id"
)
logger.info(f"Token verified successfully for widget_id={widget_id}")
return {
"valid": True,
"token_data": payload
}
except HTTPException:
raise
except Exception as e:
logger.exception(f"Error verifying token: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to verify token"
)
@router.post("/revoke-token", response_model=RevokeTokenResponse, status_code=status.HTTP_200_OK)
async def revoke_widget_token(
body: RevokeTokenRequest,
db: Session = Depends(get_db),
api_key: str = Depends(validate_api_key)
):
"""
Revoke a widget authentication token immediately.
Once revoked, a token becomes invalid even if it hasn't expired naturally.
Use cases:
- User logout: Revoke tokens when user logs out
- Permission changes: Invalidate tokens when user role/permissions change
- Security incident: Revoke compromised tokens
- Manual revocation: Admin/portal revokes a specific token
Args:
body: RevokeTokenRequest with token and optional reason
db: Database session
api_key: Validated API key from Authorization header
Returns:
RevokeTokenResponse indicating success/failure
Example:
POST /api/v1/revoke-token
Authorization: Bearer YOUR_API_KEY
{
"token": "eyJhbGciOiJIUzI1NiIs...",
"reason": "User logged out"
}
Response:
{
"success": true,
"message": "Token revoked successfully",
"revoked_at": "2024-11-09T14:05:00"
}
"""
try:
token = body.token
reason = body.reason or "Manual revocation via API"
if not token:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Token cannot be empty"
)
# Expired tokens stay revocable: only the signature has to hold up here.
payload = conversation_token.claims_ignoring_expiry(token)
if payload is None:
logger.warning("Invalid token format for revocation")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid token format"
)
# Extract token information
jti = payload.get("jti")
if not jti:
logger.warning("Token does not have JTI claim, cannot revoke")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Token does not support revocation (missing JTI claim)"
)
# Revoke the token
security_revoke_token(jti)
revoked_at = datetime.utcnow()
logger.info(
f"Token revoked: jti={jti[:8]}... reason={reason}"
)
return RevokeTokenResponse(
success=True,
message=f"Token revoked successfully. Reason: {reason}",
revoked_at=revoked_at.isoformat()
)
except HTTPException:
raise
except Exception as e:
logger.exception(f"Error revoking token: {str(e)}")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to revoke token"
)
+139
View File
@@ -0,0 +1,139 @@
"""
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, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from app.database import get_db
from app.models.user import User
from app.core.auth import require_permissions
from app.repositories.user_group import UserGroupRepository
from app.models.schemas.user_group import (
UserGroupCreate,
UserGroupUpdate,
UserGroupResponse,
UserGroupWithUsers
)
from uuid import UUID
router = APIRouter()
@router.get("", response_model=List[UserGroupWithUsers])
async def list_groups(
current_user: User = Depends(require_permissions("manage_users")),
db: Session = Depends(get_db)
):
"""List all groups in the organization"""
group_repo = UserGroupRepository(db)
return group_repo.get_groups_by_organization(current_user.organization_id)
@router.post("", response_model=UserGroupResponse)
async def create_group(
group_data: UserGroupCreate,
current_user: User = Depends(require_permissions("manage_users")),
db: Session = Depends(get_db)
):
"""Create a new group"""
group_repo = UserGroupRepository(db)
return group_repo.create_group(
name=group_data.name,
description=group_data.description,
organization_id=current_user.organization_id
)
@router.get("/{group_id}", response_model=UserGroupWithUsers)
async def get_group(
group_id: UUID,
current_user: User = Depends(require_permissions("manage_users")),
db: Session = Depends(get_db)
):
"""Get group details including members"""
group_repo = UserGroupRepository(db)
group = group_repo.get_group(group_id)
if not group or group.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Group not found")
return group
@router.put("/{group_id}", response_model=UserGroupResponse)
async def update_group(
group_id: UUID,
group_data: UserGroupUpdate,
current_user: User = Depends(require_permissions("manage_users")),
db: Session = Depends(get_db)
):
"""Update group details"""
group_repo = UserGroupRepository(db)
group = group_repo.get_group(group_id)
if not group or group.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Group not found")
return group_repo.update_group(group_id, **group_data.dict(exclude_unset=True))
@router.delete("/{group_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_group(
group_id: UUID,
current_user: User = Depends(require_permissions("manage_users")),
db: Session = Depends(get_db)
):
"""Delete a group"""
group_repo = UserGroupRepository(db)
group = group_repo.get_group(group_id)
if not group or group.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Group not found")
group_repo.delete_group(group_id)
@router.post("/{group_id}/users/{user_id}")
async def add_user_to_group(
group_id: UUID,
user_id: UUID,
current_user: User = Depends(require_permissions("manage_users")),
db: Session = Depends(get_db)
):
"""Add a user to a group"""
group_repo = UserGroupRepository(db)
group = group_repo.get_group(group_id)
if not group or group.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Group not found")
success = group_repo.add_user(group_id, user_id)
if not success:
raise HTTPException(status_code=400, detail="Failed to add user to group")
return {"message": "User added to group"}
@router.delete("/{group_id}/users/{user_id}")
async def remove_user_from_group(
group_id: UUID,
user_id: UUID,
current_user: User = Depends(require_permissions("manage_users")),
db: Session = Depends(get_db)
):
"""Remove a user from a group"""
group_repo = UserGroupRepository(db)
group = group_repo.get_group(group_id)
if not group or group.organization_id != current_user.organization_id:
raise HTTPException(status_code=404, detail="Group not found")
success = group_repo.remove_user(group_id, user_id)
if not success:
raise HTTPException(status_code=400, detail="Failed to remove user from group")
return {"message": "User removed from group"}
File diff suppressed because it is too large. Load diff