136 lines
6.8 KiB
Python
136 lines
6.8 KiB
Python
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()
|