120 lines
4.8 KiB
Python
120 lines
4.8 KiB
Python
"""Guest possession tokens isolate chat data; they never identify an order customer."""
|
|
|
|
import hashlib
|
|
import re
|
|
import secrets
|
|
|
|
from flask import g, has_app_context, jsonify, request
|
|
|
|
from .db import connect
|
|
|
|
|
|
COOKIE_NAME = 'baiyue_visitor'
|
|
KEY_PATTERN = re.compile(r'[0-9a-f]{64}\Z')
|
|
CUSTOMER_PATHS = {
|
|
'/stream_chat', '/api/new_chat', '/api/get_history', '/api/get_messages', '/api/del_history',
|
|
'/api/clear_all', '/api/upload', '/api/ticket', '/api/rate', '/api/feedback',
|
|
'/api/speech/config', '/api/speech/transcribe',
|
|
'/api/customer/session', '/api/customer/login', '/api/customer/logout',
|
|
}
|
|
|
|
|
|
def valid_key(value):
|
|
return isinstance(value, str) and KEY_PATTERN.fullmatch(value) is not None
|
|
|
|
|
|
def visitor_id():
|
|
return getattr(g, 'visitor_id', None) if has_app_context() else None
|
|
|
|
|
|
def customer_tenant_id():
|
|
# Set only after validating a server-side customer session.
|
|
# Visitor cookies, request parameters and headers are deliberately not accepted.
|
|
value = getattr(g, 'customer_tenant_id', None) if has_app_context() else None
|
|
if isinstance(value, bool) or not isinstance(value, (str, int)):
|
|
return None
|
|
tenant = str(value).strip()
|
|
return tenant if tenant and len(tenant) <= 128 else None
|
|
|
|
|
|
def customer_tenant_ids():
|
|
if not has_app_context():
|
|
return ()
|
|
# Also retain compatibility with a trusted single-tenant authentication adapter.
|
|
single = customer_tenant_id()
|
|
if single:
|
|
return (single,)
|
|
values = getattr(g,'customer_tenant_ids',())
|
|
if not isinstance(values,(list,tuple)) or len(values) > 256:
|
|
return ()
|
|
return tuple(sorted(set(value for value in values if isinstance(value,str) and 0 < len(value.strip()) <= 128)))
|
|
|
|
|
|
def attach_visitor():
|
|
key = request.cookies.get(COOKIE_NAME)
|
|
g.visitor_id = hashlib.sha256(key.encode('ascii')).hexdigest() if valid_key(key) else None
|
|
is_customer = request.path in CUSTOMER_PATHS or request.path.startswith('/api/visitor/human/')
|
|
if is_customer and not visitor_id():
|
|
return jsonify({'ok': False, 'e': '访客会话已失效，请刷新页面重试。'}), 401
|
|
if (is_customer or request.path == '/api/visitor/session') and request.method not in ('GET', 'HEAD', 'OPTIONS'):
|
|
origin = request.headers.get('Origin')
|
|
if origin and origin.rstrip('/') != request.host_url.rstrip('/'):
|
|
return jsonify({'ok': False, 'e': '请求来源无效。'}), 403
|
|
|
|
|
|
def start_session(data):
|
|
if not isinstance(data, dict):
|
|
return {'ok': False, 'error': '访客会话格式无效。'}, 400
|
|
key = data.get('key')
|
|
if key is None:
|
|
existing = request.cookies.get(COOKIE_NAME)
|
|
key = existing if valid_key(existing) else secrets.token_hex(32)
|
|
if not valid_key(key):
|
|
return {'ok': False, 'error': '访客标识无效。'}, 400
|
|
response = jsonify({'ok': True, 'key': key})
|
|
response.set_cookie(COOKIE_NAME, key, max_age=365*24*3600, httponly=True,
|
|
secure=request.is_secure, samesite='Lax', path='/')
|
|
response.headers['Cache-Control'] = 'no-store'
|
|
return response
|
|
|
|
|
|
def owns_conversation(conv_id):
|
|
from .customer_service import current_customer_id, current_scope_key
|
|
owner = visitor_id()
|
|
if not owner or not isinstance(conv_id, int) or isinstance(conv_id, bool) or not 0 < conv_id <= 9223372036854775807:
|
|
return False
|
|
connection = connect()
|
|
try:
|
|
return bool(connection.execute('SELECT 1 FROM chat_history WHERE id=? AND visitor_id=? '
|
|
'AND (customer_id IS NULL OR (customer_id=? AND customer_scope=?))',
|
|
(conv_id, owner, current_customer_id(),current_scope_key())).fetchone())
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def owns_human_session(sid):
|
|
from .customer_service import current_customer_id, current_scope_key
|
|
owner = visitor_id()
|
|
if not owner or not isinstance(sid, int) or isinstance(sid, bool) or not 0 < sid <= 9223372036854775807:
|
|
return False
|
|
connection = connect()
|
|
try:
|
|
return bool(connection.execute('SELECT 1 FROM human_sessions s JOIN chat_history h ON h.id=s.conv_id '
|
|
'WHERE s.id=? AND h.visitor_id=? AND (h.customer_id IS NULL OR '
|
|
'(h.customer_id=? AND h.customer_scope=?))',
|
|
(sid, owner, current_customer_id(),current_scope_key())).fetchone())
|
|
finally:
|
|
connection.close()
|
|
|
|
|
|
def owns_upload(name):
|
|
owner = visitor_id()
|
|
if not owner:
|
|
return False
|
|
connection = connect()
|
|
try:
|
|
return bool(connection.execute('SELECT 1 FROM visitor_uploads WHERE filename=? AND visitor_id=?',
|
|
(name, owner)).fetchone())
|
|
finally:
|
|
connection.close()
|