147 lines
8.7 KiB
Python
147 lines
8.7 KiB
Python
"""Independent guests cannot read, mutate or claim each other's chat data."""
|
|
|
|
import io
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from datetime import datetime
|
|
|
|
from logistics_chat import create_app, db
|
|
|
|
|
|
class VisitorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
root = Path(directory.name)
|
|
self.app = create_app({'TESTING':True, 'DATABASE':root/'history.db', 'UPLOAD_DIR':root/'uploads',
|
|
'BACKUP_DIR':root/'backups','LOG_DIR':root/'logs','DEEPSEEK_API_KEY':'','RATE_LIMIT_PER_MIN':1000})
|
|
self.a, self.b = self.app.test_client(), self.app.test_client()
|
|
for client, key in ((self.a,'1'*64),(self.b,'2'*64)):
|
|
self.assertEqual(client.post('/api/visitor/session',json={'key':key}).status_code,200)
|
|
|
|
def conversation(self, client, text):
|
|
cid = client.post('/api/new_chat').json['id']
|
|
with self.app.app_context(): db.save_message(cid,'user',text)
|
|
return cid
|
|
|
|
def human(self, client, cid):
|
|
with patch('logistics_chat.human_service.datetime') as clock:
|
|
clock.now.return_value = datetime(2026,10,30,12)
|
|
return client.post('/api/visitor/human/request',json={'conv_id':cid}).json['session_id']
|
|
|
|
def test_entry_is_public_but_customer_apis_need_a_guest_session_without_basic_login(self):
|
|
fresh = self.app.test_client()
|
|
self.assertEqual(fresh.get('/').status_code,200)
|
|
response = fresh.get('/api/get_history')
|
|
self.assertEqual(response.status_code,401)
|
|
self.assertNotIn('WWW-Authenticate',response.headers)
|
|
self.assertEqual(fresh.post('/api/new_chat').status_code,401)
|
|
self.assertEqual(self.a.get('/api/get_history').json,[])
|
|
|
|
def test_guest_key_is_restorable_but_only_its_hash_is_stored(self):
|
|
cid = self.conversation(self.a,'A_PRIVATE')
|
|
restored = self.app.test_client()
|
|
response = restored.post('/api/visitor/session',json={'key':'1'*64},base_url='https://localhost')
|
|
for flag in ('HttpOnly','Secure','SameSite=Lax'):
|
|
self.assertIn(flag,response.headers['Set-Cookie'])
|
|
self.assertEqual(restored.get('/api/get_history').json[0]['id'],cid)
|
|
with self.app.app_context():
|
|
connection=db.connect()
|
|
owner=connection.execute('SELECT visitor_id FROM chat_history WHERE id=?',(cid,)).fetchone()[0]
|
|
connection.close()
|
|
self.assertEqual(len(owner),64)
|
|
self.assertNotEqual(owner,'1'*64)
|
|
|
|
def test_listing_loading_and_streaming_never_expose_another_guest(self):
|
|
a = self.conversation(self.a,'A_PRIVATE')
|
|
b = self.conversation(self.b,'B_PRIVATE')
|
|
self.assertEqual([row['id'] for row in self.a.get('/api/get_history').json],[a])
|
|
self.assertEqual([row['id'] for row in self.b.get('/api/get_history').json],[b])
|
|
self.assertEqual(self.a.get(f'/api/get_messages?conv_id={b}').json,[])
|
|
self.assertEqual(self.b.get(f'/api/get_messages?conv_id={a}').json,[])
|
|
with patch('logistics_chat.ai_client.post_completion') as model:
|
|
response=self.a.post('/stream_chat',json={'msg':'继续上一段对话','conv_id':b})
|
|
self.assertIn('新建对话',response.get_data(as_text=True))
|
|
self.assertNotIn('B_PRIVATE',response.get_data(as_text=True))
|
|
self.assertEqual(len(self.b.get(f'/api/get_messages?conv_id={b}').json),2)
|
|
model.assert_not_called()
|
|
|
|
def test_delete_and_clear_only_affect_the_current_guest_and_keep_unowned_legacy_data(self):
|
|
a = self.conversation(self.a,'A_PRIVATE')
|
|
b = self.conversation(self.b,'B_PRIVATE')
|
|
sid_a,sid_b=self.human(self.a,a),self.human(self.b,b)
|
|
self.a.post('/api/visitor/human/send',json={'sid':sid_a,'content':'A_SUPPORT'})
|
|
self.b.post('/api/visitor/human/send',json={'sid':sid_b,'content':'B_SUPPORT'})
|
|
with self.app.app_context():
|
|
connection=db.connect()
|
|
legacy=connection.execute("INSERT INTO chat_history(title,ctime) VALUES('LEGACY_PRIVATE','now')").lastrowid
|
|
connection.commit();connection.close()
|
|
self.assertEqual(self.a.delete(f'/api/del_history?id={b}').status_code,404)
|
|
self.assertTrue(self.a.delete('/api/clear_all').json['ok'])
|
|
self.assertEqual(self.a.get('/api/get_history').json,[])
|
|
self.assertEqual(self.b.get('/api/get_history').json[0]['id'],b)
|
|
self.assertEqual(self.b.get(f'/api/visitor/human/poll?sid={sid_b}').json['messages'][0]['content'],'B_SUPPORT')
|
|
with self.app.app_context():
|
|
connection=db.connect()
|
|
self.assertEqual(connection.execute('SELECT id FROM chat_history ORDER BY id').fetchall(),[(b,),(legacy,)])
|
|
connection.close()
|
|
|
|
def test_customer_human_routes_enforce_owner_and_role_and_hide_internal_notes(self):
|
|
a=self.conversation(self.a,'A_PRIVATE')
|
|
sid=self.human(self.a,a)
|
|
self.assertEqual(self.b.post('/api/visitor/human/request',json={'conv_id':a}).status_code,404)
|
|
self.assertEqual(self.b.get(f'/api/visitor/human/poll?sid={sid}').status_code,404)
|
|
for action in ('send','close'):
|
|
self.assertEqual(self.b.post('/api/visitor/human/'+action,json={'sid':sid,'content':'attack'}).status_code,404)
|
|
self.assertEqual(self.a.post('/api/visitor/human/send',json={'sid':sid,'side':'agent','content':'fake agent'}).status_code,403)
|
|
self.a.post('/api/human/note',json={'sid':sid,'note':'STAFF_PRIVATE'})
|
|
response=self.a.get(f'/api/visitor/human/poll?sid={sid}')
|
|
self.assertNotIn('note',response.json)
|
|
self.assertNotIn('STAFF_PRIVATE',response.get_data(as_text=True))
|
|
self.assertEqual(self.b.post('/api/rate',json={'sid':sid,'stars':5}).status_code,404)
|
|
self.assertEqual(self.a.post('/api/rate',json=[1]).status_code,400)
|
|
self.assertEqual(self.a.post('/api/rate',json={'sid':sid,'stars':5}).status_code,200)
|
|
|
|
def test_uploads_are_private_and_legacy_files_are_not_assigned_to_new_visitors(self):
|
|
response=self.a.post('/api/upload',data={'file':(io.BytesIO(b'A_PRIVATE'),'note.txt')})
|
|
url=response.json['url']
|
|
with self.a.get(url) as downloaded:
|
|
self.assertEqual(downloaded.data,b'A_PRIVATE')
|
|
self.assertIn('no-store',downloaded.headers['Cache-Control'])
|
|
self.assertEqual(self.b.get(url).status_code,404)
|
|
self.assertEqual(self.app.test_client().get(url).status_code,404)
|
|
root=Path(self.app.config['UPLOAD_DIR'])
|
|
(root/'legacy.txt').write_text('LEGACY_PRIVATE')
|
|
self.assertEqual(self.a.get('/uploads/legacy.txt').status_code,404)
|
|
with self.a.get('/api/staff'+url) as staff:
|
|
self.assertEqual(staff.data,b'A_PRIVATE')
|
|
|
|
def test_guest_keys_and_client_claims_cannot_unlock_orders(self):
|
|
for message in ('查询15252125','查询订单状态','查询 UNKNOWN20260001'):
|
|
with patch('logistics_chat.order_data.lookup_order') as lookup, patch('logistics_chat.ai_client.post_completion') as model:
|
|
response=self.a.post('/stream_chat',json={'msg':message},headers={'X-Customer-Id':'tenant-a','X-Tenant-Id':'tenant-a'})
|
|
self.assertIn('客户账号登录并关联',response.get_data(as_text=True))
|
|
lookup.assert_not_called();model.assert_not_called()
|
|
|
|
def test_invalid_bootstrap_and_cross_site_requests_cannot_replace_the_guest(self):
|
|
cid=self.conversation(self.a,'A_PRIVATE')
|
|
for key in ('short','x'*64,'1'*65,[],{},123):
|
|
self.assertEqual(self.a.post('/api/visitor/session',json={'key':key}).status_code,400)
|
|
self.assertEqual(self.a.post('/api/visitor/session',json={'key':'2'*64},headers={'Origin':'https://attacker.example'}).status_code,403)
|
|
self.assertEqual(self.a.post('/api/visitor/session',data={'key':'2'*64}).status_code,400)
|
|
self.assertEqual(self.a.post('/api/visitor/session',json={'key':'1'*64,'padding':'x'*2000}).status_code,413)
|
|
self.assertEqual(self.a.get('/api/get_history').json[0]['id'],cid)
|
|
|
|
def test_owned_support_polling_is_not_counted_as_chat_generation_and_bounds_its_cursor(self):
|
|
cid=self.conversation(self.a,'A_PRIVATE')
|
|
sid=self.human(self.a,cid)
|
|
self.app.config['RATE_LIMIT_PER_MIN']=1
|
|
self.app.extensions['rate_hits'].clear()
|
|
self.assertEqual(self.a.get('/api/get_history').status_code,200)
|
|
self.assertEqual(self.a.get('/api/get_history').status_code,429)
|
|
self.assertEqual(self.a.get(f'/api/visitor/human/poll?sid={sid}').status_code,200)
|
|
self.assertEqual(self.b.get(f'/api/visitor/human/poll?sid={sid}').status_code,404)
|
|
self.assertEqual(self.a.get(f'/api/visitor/human/poll?sid={sid}&after=9999999999999999999999999').status_code,400)
|