"""Exercise the server factory using disposable state/log paths and no model key.""" import io import json import logging import os from pathlib import Path import shlex import tempfile import unittest from unittest.mock import patch from deploy import wsgi from logistics_chat import create_app from waitress.adjustments import Adjustments from waitress.proxy_headers import proxy_headers_middleware class DeploymentTests(unittest.TestCase): def test_fresh_server_persists_business_data_and_uploads_outside_checkout(self): directory = tempfile.TemporaryDirectory() self.addCleanup(directory.cleanup) root = Path(directory.name) state, logs = root / 'state', root / 'logs' with patch.object(wsgi, 'DATA_DIR', state), patch.object(wsgi, 'LOG_DIR', logs), \ patch.dict(os.environ, {'DEEPSEEK_API_KEY': ''}): app = wsgi.create_app() def close_handlers(): for handler in list(app.logger.handlers): if isinstance(handler, logging.FileHandler): handler.close() app.logger.removeHandler(handler) self.addCleanup(close_handlers) client = app.test_client() client.post('/api/visitor/session', json={'key':'1'*64}) for path in ('/', '/admin/dashboard', '/api/tickets', '/api/my_orders?phone=123'): self.assertEqual(client.get(path).status_code, 200, path) self.assertEqual(client.post('/admin/login', json={'username': 'demo', 'password': 'test-password'}).status_code, 401) self.assertEqual(client.post('/api/ticket', json={'content': 'Acceptance ticket', 'phone': '123'}).status_code, 200) self.assertEqual(client.get('/api/tickets').json[0]['content'], 'Acceptance ticket') uploaded = client.post('/api/upload', data={'file': (io.BytesIO(b'acceptance'), 'note.txt')}) self.assertEqual(uploaded.status_code, 200) self.assertEqual((state / 'uploads' / uploaded.json['url'].rsplit('/', 1)[1]).read_bytes(), b'acceptance') self.assertTrue((state / 'history.db').is_file()) self.assertEqual(len(list((state / 'backups').glob('history_*.db'))), 1) self.assertTrue((logs / 'app.log').is_file()) self.assertEqual(app.config['HOST'], '127.0.0.1') class ProxyIdentityTests(unittest.TestCase): def application(self, runtime): root = Path(__file__).resolve().parents[1] if runtime == 'docker': command = next(line[4:] for line in (root/'Dockerfile').read_text(encoding='utf-8').splitlines() if line.startswith('CMD ')) args = json.loads(command)[3:] else: command = next(line[len('ExecStart='):] for line in (root/'deploy/baiyue-logistics-chat.service').read_text(encoding='utf-8').splitlines() if line.startswith('ExecStart=')) args = shlex.split(command)[3:] options, _ = Adjustments.parse_args(args) options.pop('call'); options.pop('help') settings = Adjustments(**options) folder = tempfile.TemporaryDirectory(); self.addCleanup(folder.cleanup) data = Path(folder.name) app = create_app({'TESTING':True,'DATABASE':data/'history.db','UPLOAD_DIR':data/'uploads', 'BACKUP_DIR':data/'backups','LOG_DIR':data/'logs','RATE_LIMIT_PER_MIN':1000}) app.config['PROXY_TEST_PEER'] = '172.18.0.1' if runtime == 'docker' else '127.0.0.1' app.wsgi_app = proxy_headers_middleware(app.wsgi_app, trusted_proxy=settings.trusted_proxy, trusted_proxy_count=settings.trusted_proxy_count, trusted_proxy_headers=settings.trusted_proxy_headers) return app def clients(self, app): clients = [] for address in ('198.51.100.10', '198.51.100.20'): client = app.test_client() client.environ_base['REMOTE_ADDR'] = app.config['PROXY_TEST_PEER'] client.environ_base['HTTP_X_FORWARDED_FOR'] = address client.post('/api/visitor/session', json={'key':str(len(clients)+1)*64}) clients.append(client) app.extensions['rate_hits'].clear() return clients def test_proxy_clients_do_not_share_request_rate_limits(self): for runtime in ('docker', 'systemd'): with self.subTest(runtime=runtime): app = self.application(runtime) a, b = self.clients(app) app.config['RATE_LIMIT_PER_MIN'] = 2 self.assertEqual(a.get('/api/get_history').status_code, 200) self.assertEqual(a.get('/api/get_history').status_code, 200) self.assertEqual(a.get('/api/get_history').status_code, 429) self.assertEqual(b.get('/api/get_history').status_code, 200) def test_proxy_clients_do_not_share_failed_login_budgets(self): for runtime in ('docker', 'systemd'): with self.subTest(runtime=runtime): app = self.application(runtime) a, b = self.clients(app) for _ in range(5): self.assertEqual(a.post('/api/customer/login',json={'username':'admin','password':'wrong'}).status_code,403) self.assertEqual(a.post('/api/customer/login',json={'username':'admin','password':'123456'}).status_code,429) self.assertTrue(b.post('/api/customer/login',json={'username':'admin','password':'123456'}).json.get('authenticated')) def test_only_the_last_forwarded_address_is_used_and_other_proxy_claims_are_ignored(self): from flask import request for runtime in ('docker', 'systemd'): with self.subTest(runtime=runtime): app = self.application(runtime) @app.get('/proxy-probe') def probe(): return {'ip':request.remote_addr,'host':request.host,'scheme':request.scheme} response = app.test_client().get('/proxy-probe',base_url='https://chat.example.com', environ_overrides={'REMOTE_ADDR':app.config['PROXY_TEST_PEER']},headers={ 'X-Forwarded-For':'203.0.113.99, 198.51.100.10', 'X-Forwarded-Host':'evil.example','X-Forwarded-Proto':'http'}) self.assertEqual(response.json,{'ip':'198.51.100.10','host':'chat.example.com','scheme':'https'}) if runtime == 'systemd': untrusted = app.test_client().get('/proxy-probe',environ_overrides={'REMOTE_ADDR':'203.0.113.1'}, headers={'X-Forwarded-For':'198.51.100.10'}) self.assertEqual(untrusted.json['ip'],'203.0.113.1') if __name__ == '__main__': unittest.main()