129 lines
6.6 KiB
Python
129 lines
6.6 KiB
Python
"""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()
|