"""Staff authorization uses real logins against disposable storage.""" import hashlib import unittest from unittest.mock import patch import test_staff_login as credentials from logistics_chat import db class StaffSessionTests(unittest.TestCase): add_staff = credentials.StaffLoginTests.add_staff login = credentials.StaffLoginTests.login def setUp(self): credentials.StaffLoginTests.setUp(self) self.app.config['RATE_LIMIT_PER_MIN'] = 1000 def signed_in(self): self.add_staff('fixture-password') response = self.login('fixture-password') self.assertEqual(response.status_code, 200) return self.client.get('/api/admin/session').json['csrf_token'] def test_anonymous_staff_pages_and_all_staff_endpoints_are_blocked(self): self.assertEqual(self.client.get('/admin/dashboard').status_code, 302) public = {'static', 'pages.index', 'pages.health', 'admin.admin', 'admin.admin_login', 'chat.stream_chat', 'chat.feedback', 'history.new_chat', 'history.get_history', 'history.get_messages', 'history.del_history', 'history.clear_all', 'files.upload_file', 'files.uploaded_file', 'business.add_ticket', 'business.rate_chat', 'visitor.visitor_session', 'visitor.visitor_human', 'customer.session', 'customer.login', 'customer.logout', 'speech.speech_config', 'speech.speech_transcribe'} for rule in self.app.url_map.iter_rules(): if rule.endpoint in public or rule.rule == '/admin/dashboard': continue method = 'POST' if 'POST' in rule.methods else 'GET' path = rule.rule.replace('', 'fixture.txt') with self.subTest(path=path): self.assertEqual(self.client.open(path, method=method, json={}).status_code, 401) self.assertEqual(self.client.get('/api/future-private-route').status_code, 401) self.assertEqual(self.client.get('/').status_code, 200) self.assertEqual(self.client.get('/healthz').status_code, 200) def test_customer_ticket_and_rating_keep_their_existing_guest_contract(self): self.client.post('/api/visitor/session', json={'key':'a'*64}) self.assertEqual(self.client.post('/api/ticket', json={'content':'fixture ticket'}).status_code, 200) self.assertEqual(self.client.post('/api/rate', json={'sid':999, 'stars':5}).status_code, 404) def test_cookie_and_database_store_24_hour_opaque_sessions(self): self.add_staff('fixture-password') response = self.client.post('/admin/login', json={'username':'staff-fixture', 'password':'fixture-password'}, base_url='https://localhost') cookie = response.headers['Set-Cookie'] for flag in ('HttpOnly', 'Secure', 'SameSite=Lax', 'Max-Age=86400'): self.assertIn(flag, cookie) token = self.client.get_cookie('baiyue_staff').value with self.app.app_context(): row = db.connect().execute('SELECT token_hash FROM staff_sessions').fetchone() self.assertEqual(row[0], hashlib.sha256(token.encode()).hexdigest()) self.assertNotIn(token, self.client.get('/admin/dashboard').text) self.assertEqual(self.client.get('/admin/dashboard').status_code, 200) def test_staff_writes_require_csrf_and_same_origin(self): csrf = self.signed_in() self.assertEqual(self.client.post('/api/human/note', json={}).status_code, 403) self.assertEqual(self.client.post('/admin/logout', headers={'X-CSRF-Token':'bad'}).status_code, 403) self.assertEqual(self.client.post('/admin/logout', headers={'X-CSRF-Token':csrf, 'Origin':'https://evil.test'}).status_code, 403) self.assertEqual(self.client.get('/api/admin/session').status_code, 200) def test_logout_revokes_replayed_cookie_and_preserves_data(self): csrf = self.signed_in() token = self.client.get_cookie('baiyue_staff').value self.assertEqual(self.client.post('/admin/logout', headers={'X-CSRF-Token':csrf}).status_code, 200) self.assertIsNone(self.client.get_cookie('baiyue_staff')) self.client.set_cookie('baiyue_staff', token) self.assertEqual(self.client.get('/api/admin/stats').status_code, 401) with self.app.app_context(): self.assertEqual(db.connect().execute('SELECT COUNT(*) FROM agents').fetchone()[0], 1) def test_login_rotates_current_token(self): self.signed_in() old = self.client.get_cookie('baiyue_staff').value self.login('fixture-password') self.assertNotEqual(old, self.client.get_cookie('baiyue_staff').value) self.client.set_cookie('baiyue_staff', old) self.assertEqual(self.client.get('/api/admin/session').status_code, 401) def test_sessions_expire_at_24_hours_without_renewing_on_requests_or_restart(self): from logistics_chat import create_app with patch('logistics_chat.staff_service.time.time', return_value=1000): self.signed_in() token = self.client.get_cookie('baiyue_staff').value restarted = self.client for elapsed in (8*3600, 23*3600, 24*3600-1): with self.subTest(elapsed=elapsed), patch('logistics_chat.staff_service.time.time', return_value=1000+elapsed): response = self.client.get('/api/admin/session') self.assertEqual(response.status_code, 200) self.assertNotIn('Set-Cookie', response.headers) self.assertEqual(self.client.get_cookie('baiyue_staff').value, token) restarted = create_app(dict(self.app.config)).test_client() restarted.set_cookie('baiyue_staff', token) self.assertEqual(restarted.get('/api/admin/session').status_code, 200) with patch('logistics_chat.staff_service.time.time', return_value=1000+24*3600): for client in (self.client, restarted): self.assertEqual(client.get('/api/admin/session').status_code, 401) self.assertEqual(client.get('/admin/dashboard').status_code, 302) def test_sessions_survive_app_restart_until_explicit_logout(self): from logistics_chat import create_app self.signed_in() token = self.client.get_cookie('baiyue_staff').value restarted = create_app(dict(self.app.config)).test_client() restarted.set_cookie('baiyue_staff', token) response = restarted.get('/api/admin/session') self.assertEqual(response.status_code, 200) self.assertEqual(restarted.post('/admin/logout', headers={'X-CSRF-Token':response.json['csrf_token']}).status_code, 200) self.assertEqual(self.client.get('/api/admin/session').status_code, 401) def test_another_browser_login_does_not_remove_existing_session(self): csrf = self.signed_in() another = self.app.test_client() self.assertEqual(another.post('/admin/login', json={'username':'staff-fixture','password':'fixture-password'}).status_code, 200) self.assertEqual(self.client.get('/api/admin/session').status_code, 200) self.assertEqual(self.client.post('/admin/logout', headers={'X-CSRF-Token':csrf}).status_code, 200) self.assertEqual(another.get('/api/admin/session').status_code, 200) def test_upgrade_preserves_deadlines_and_revokes_unbounded_and_expired_sessions(self): self.signed_in() with self.app.app_context(): connection = db.connect() connection.execute('UPDATE staff_sessions SET expires_at=2000') connection.execute('INSERT INTO staff_sessions SELECT ?,username,auth_stamp,csrf_token,500 FROM staff_sessions', (hashlib.sha256(('b'*64).encode()).hexdigest(),)) connection.execute('INSERT INTO staff_sessions SELECT ?,username,auth_stamp,csrf_token,0 FROM staff_sessions LIMIT 1', (hashlib.sha256(('c'*64).encode()).hexdigest(),)) connection.commit() with patch('logistics_chat.staff_service.time.time', return_value=1000): db.init_db() with patch('logistics_chat.staff_service.time.time', return_value=1500): self.assertEqual(self.client.get('/api/admin/session').status_code, 200) for token in ('b'*64, 'c'*64): other = self.app.test_client() other.set_cookie('baiyue_staff', token) self.assertEqual(other.get('/api/admin/session').status_code, 401) with patch('logistics_chat.staff_service.time.time', return_value=2000): self.assertEqual(self.client.get('/api/admin/session').status_code, 401) def test_password_change_and_account_deletion_invalidate_sessions(self): self.signed_in() with self.app.app_context(): connection = db.connect() connection.execute("UPDATE agents SET password='new-fixture-password'") connection.commit() self.assertEqual(self.client.get('/api/admin/session').status_code, 401) self.login('new-fixture-password') with self.app.app_context(): connection = db.connect() connection.execute('DELETE FROM agents') connection.commit() self.assertEqual(self.client.get('/api/admin/session').status_code, 401) def test_legacy_agent_id_session_schema_is_upgraded_without_changing_credentials(self): from werkzeug.security import generate_password_hash password_hash = generate_password_hash('fixture-password', method='scrypt:1024:8:1') with self.app.app_context(): connection = db.connect() connection.execute('DROP TABLE staff_sessions') connection.execute('DROP TABLE agents') connection.execute('CREATE TABLE agents(id INTEGER PRIMARY KEY, username TEXT UNIQUE, ' "password TEXT, name TEXT, role TEXT NOT NULL DEFAULT 'agent')") connection.execute('INSERT INTO agents VALUES(?,?,?,?,?)', (7,'staff-fixture',password_hash,'测试客服','admin')) connection.execute('CREATE TABLE staff_sessions(id TEXT PRIMARY KEY, agent_id INTEGER NOT NULL, ' 'created_at REAL NOT NULL, expires_at REAL NOT NULL, password_version TEXT NOT NULL)') connection.execute('INSERT INTO staff_sessions VALUES(?,?,?,?,?)', ('legacy-cookie',7,1000,9999999999,'legacy-version')) connection.execute("INSERT INTO tickets(content,phone) VALUES('fixture-ticket','fixture-phone')") connection.commit() db.init_db() db.init_db() self.assertEqual(connection.execute('SELECT id,username,password,name,role FROM agents').fetchone(), (7,'staff-fixture',password_hash,'测试客服','admin')) self.assertEqual(connection.execute('SELECT content FROM tickets').fetchone()[0], 'fixture-ticket') self.assertEqual(self.login('fixture-password').status_code, 200) self.assertEqual(self.client.get('/api/admin/session').status_code, 200) self.assertEqual(self.client.get('/admin/dashboard').status_code, 200) with self.app.app_context(): connection = db.connect() connection.execute("UPDATE agents SET password='new-fixture-password'") connection.commit() self.assertEqual(self.client.get('/api/admin/session').status_code, 401) self.login('new-fixture-password') with self.app.app_context(): connection = db.connect() connection.execute('DELETE FROM agents') connection.commit() self.assertEqual(self.client.get('/api/admin/session').status_code, 401) def test_cross_site_login_and_form_login_are_rejected(self): self.add_staff('fixture-password') self.assertEqual(self.client.post('/admin/login', json={'username':'staff-fixture','password':'fixture-password'}, headers={'Origin':'https://evil.test'}).status_code, 403) self.assertEqual(self.client.post('/admin/login', data={'username':'staff-fixture','password':'fixture-password'}).status_code, 415) def test_restoring_password_or_recreating_account_cannot_restore_old_session(self): for change in ('password', 'account', 'replace'): with self.subTest(change=change): if change == 'password': self.signed_in() else: self.login('fixture-password') with self.app.app_context(): connection = db.connect() if change == 'password': connection.execute("UPDATE agents SET password='temporary-password'") connection.execute("UPDATE agents SET password='fixture-password'") elif change == 'account': connection.execute('DELETE FROM agents') connection.execute('INSERT INTO agents VALUES(?,?,?)', ('staff-fixture','fixture-password','测试客服')) else: for password in ('temporary-password', 'fixture-password'): connection.execute('INSERT OR REPLACE INTO agents VALUES(?,?,?)', ('staff-fixture',password,'测试客服')) connection.commit() self.assertEqual(self.client.get('/api/admin/session').status_code, 401) def test_failed_logins_are_throttled_separately_from_polling(self): self.add_staff('fixture-password') for _ in range(5): self.assertEqual(self.login('wrong').status_code, 401) self.assertEqual(self.login('wrong').status_code, 429) self.assertEqual(self.client.get('/healthz').status_code, 200) def test_random_cookie_and_customer_session_do_not_authorize_staff(self): self.client.set_cookie('baiyue_staff', 'a'*64) self.client.set_cookie('baiyue_customer', 'b'*64) self.client.post('/api/visitor/session', json={'key':'c'*64}) self.assertEqual(self.client.get('/api/get_history').status_code, 200) self.assertEqual(self.client.get('/api/tickets').status_code, 401) def test_proxy_peer_failures_do_not_lock_other_accounts_or_trust_spoofed_headers(self): self.add_staff('fixture-password') with self.app.app_context(): connection = db.connect() connection.execute('INSERT INTO agents VALUES(?,?,?)', ('other-fixture','other-password','Other Staff')) connection.commit() for _ in range(5): self.assertEqual(self.login('wrong').status_code, 401) fresh = self.app.test_client() self.assertEqual(fresh.post('/admin/login', json={'username':'staff-fixture','password':'wrong'}, headers={'X-Forwarded-For':'203.0.113.42'}).status_code, 429) self.assertEqual(fresh.post('/admin/login', json={'username':'other-fixture','password':'other-password'}).status_code, 200)