import hashlib from http.client import HTTPConnection from http.cookies import SimpleCookie import json import os from pathlib import Path import tempfile import threading import unittest from unittest.mock import patch from bank_importer import auth, portal_sso from bank_importer.db import connect, migrate, utc_now import server from test_server_auth import Client class PortalSsoTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.db_path = Path(self.temp.name) / 'app.db' self.db = connect(self.db_path) self.addCleanup(self.db.close) migrate(self.db) portal_sso.migrate(self.db) with self.db: self.db.execute('INSERT INTO companies(name,created_at,updated_at) VALUES (?,?,?)', ('测试公司甲', utc_now(), utc_now())) self.db.execute('INSERT INTO companies(name,created_at,updated_at) VALUES (?,?,?)', ('测试公司乙', utc_now(), utc_now())) self.uid = auth.create_user(self.db, 'staff-a', 'UnusedPass123', 'company', company_id=1) self.token = auth.create_session(self.db, self.uid) self.empid = '11111111-2222-3333-4444-555555555555' self.claims = {'active': True, 'identity': {'empid': self.empid, 'local_uid': self.uid}} self.env = patch.dict(os.environ, {'PORTAL_SSO_ENABLED': '1', 'PORTAL_PUBLIC_URL': 'http://portal.test'}) self.env.start() self.addCleanup(self.env.stop) def bind_session(self): with self.db: self.db.execute('INSERT INTO portal_sessions VALUES (?,?,?,?)', (hashlib.sha256(self.token.encode()).hexdigest(), 'a' * 64, self.empid, self.uid)) def test_native_session_does_not_bypass_portal(self): with patch.object(portal_sso, 'call') as call: self.assertIsNone(portal_sso.resolve(self.db, self.token)) call.assert_not_called() def test_native_role_and_company_are_retained(self): self.bind_session() with patch.object(portal_sso, 'call', return_value=self.claims): user = portal_sso.resolve(self.db, self.token) self.assertEqual('company', user['role']) self.assertEqual(1, user['company_id']) def test_revocation_mismatch_and_local_disable_reject(self): self.bind_session() with patch.object(portal_sso, 'call', return_value={'active': False}): self.assertIsNone(portal_sso.resolve(self.db, self.token)) self.assertIsNone(auth.resolve_session(self.db, self.token)) def test_portal_outage_does_not_fallback(self): self.bind_session() with patch.object(portal_sso, 'call', side_effect=TimeoutError): with self.assertRaises(TimeoutError): portal_sso.resolve(self.db, self.token) def test_http_guards_company_scope_and_hidden_credentials(self): self.bind_session() class Quiet(server.AppHandler): def log_message(self, *args): pass with patch.object(server, 'DB_PATH', self.db_path), patch.object(portal_sso, 'call', return_value=self.claims): httpd = server.ThreadingHTTPServer(('127.0.0.1', 0), Quiet) thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() try: client = Client('127.0.0.1', httpd.server_port) self.assertEqual(401, client.get('/api/company/accounts')[0]) client.cookies[server.SESSION_COOKIE] = self.token self.assertEqual(200, client.get('/api/company/accounts?company_id=2')[0]) self.assertEqual(403, client.get('/api/admin/companies')[0]) status, _, data = client.get('/api/me') self.assertEqual(200, status) self.assertEqual(1, json.loads(data)['company_id']) for path in ['/api/login', '/api/logout', '/api/password/change', '/api/admin/users', '/api/admin/users/1/reset-password']: self.assertEqual(403, client.post_json(path, {})[0]) status, _, data = client.get('/company.html') self.assertEqual(200, status) self.assertIn(b'data-return-portal', data) self.assertIn(b'display:none!important', data) self.assertEqual('http://portal.test', client.get('/login-company.html')[1]['location']) finally: httpd.shutdown() httpd.server_close() thread.join() def test_callback_checks_browser_proof_and_pins_local_identity(self): class Quiet(server.AppHandler): def log_message(self, *args): pass def response(operation, payload): if operation == 'exchange': self.assertEqual('a' * 64, payload['verifier']) return {'session_handle': 'b' * 64, 'identity': self.claims['identity']} return self.claims with patch.object(server, 'DB_PATH', self.db_path), patch.object(portal_sso, 'call', side_effect=response): httpd = server.ThreadingHTTPServer(('127.0.0.1', 0), Quiet) thread = threading.Thread(target=httpd.serve_forever, daemon=True) thread.start() try: client = Client('127.0.0.1', httpd.server_port) status, headers, _ = client.get('/portal-sso?theme=night') self.assertEqual(303, status) self.assertIn('/api/sso/apps/finance/authorize?challenge=', headers['location']) client.cookies.clear() status, _, body = client.get('/portal-sso?ticket=' + 'c' * 64) self.assertEqual(401, status) self.assertIn(b'http://portal.test', body) conn = HTTPConnection('127.0.0.1', httpd.server_port) conn.request('GET', '/portal-sso?ticket=' + 'c' * 64, headers={'Cookie': 'cw_portal_proof=' + 'a' * 64 + '.night'}) res = conn.getresponse() self.assertEqual(303, res.status) self.assertEqual('/company.html?theme=night', res.getheader('Location')) for key, value in res.getheaders(): if key.lower() == 'set-cookie': cookies = SimpleCookie(value) if server.SESSION_COOKIE in cookies: client.cookies[server.SESSION_COOKIE] = cookies[server.SESSION_COOKIE].value res.read() conn.close() self.assertEqual(1, json.loads(client.get('/api/me')[2])['company_id']) self.assertEqual(self.empid, self.db.execute('SELECT empid FROM portal_identity WHERE uid=?', (self.uid,)).fetchone()['empid']) finally: httpd.shutdown() httpd.server_close() thread.join()