120 lines
6.7 KiB
Python
120 lines
6.7 KiB
Python
"""Reject unsafe release coordinates and keep secrets out of public release files."""
|
|
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
import re
|
|
|
|
from dotenv import dotenv_values
|
|
|
|
from deploy.gray_config import GrayConfig, write_release
|
|
|
|
|
|
class GrayConfigTests(unittest.TestCase):
|
|
def settings(self):
|
|
return {
|
|
'PUBLIC_ORIGIN': 'https://chat.example.com', 'HOST_PORT': '29013',
|
|
'GRAY_DEPLOY_DIR': '/data/services/gray-logistics-chat',
|
|
'GRAY_PROXY_CONF_DIR': '/etc/nginx/conf.d',
|
|
'GRAY_TLS_CERT_FILE': '/etc/ssl/example/fullchain.pem',
|
|
'GRAY_TLS_KEY_FILE': '/etc/ssl/example/privkey.pem',
|
|
'SOURCE_COMMIT': 'a' * 40, 'BUILD_NUMBER': '7',
|
|
'CHAT_IMAGE': 'baiyue-logistics-chat-gray:' + 'a' * 12 + '-b7',
|
|
'CHAT_IMAGE_ID': 'sha256:' + 'b' * 64,
|
|
'CERTBOT_IMAGE_ID': 'sha256:' + 'c' * 64,
|
|
}
|
|
|
|
def test_unsafe_origins_paths_and_mismatched_images_are_rejected(self):
|
|
cases = (
|
|
('PUBLIC_ORIGIN', 'http://chat.example.com'),
|
|
('PUBLIC_ORIGIN', 'https://user:password@chat.example.com'),
|
|
('PUBLIC_ORIGIN', 'https://chat.example.com/admin'),
|
|
('GRAY_DEPLOY_DIR', '/data/services/../gray-logistics-chat'),
|
|
('GRAY_DEPLOY_DIR', '/data/shared-services'),
|
|
('GRAY_PROXY_CONF_DIR', '/etc/nginx; echo unsafe'),
|
|
('HOST_PORT', '443'), ('CHAT_IMAGE', 'another-service:latest'),
|
|
('SOURCE_COMMIT', 'main'), ('CHAT_IMAGE_ID', 'unverified-image'),
|
|
)
|
|
for key, value in cases:
|
|
with self.subTest(setting=key, value=value):
|
|
settings = self.settings()
|
|
settings[key] = value
|
|
with self.assertRaises(ValueError):
|
|
GrayConfig.from_environment(settings)
|
|
|
|
def test_data_paths_remain_stable_across_releases_and_credentials_stay_separate(self):
|
|
config = GrayConfig.from_environment(self.settings())
|
|
updated = replace(config, release=config.root + '/releases/next-build')
|
|
for release in (config, updated):
|
|
self.assertIn('CHAT_DATA_DIR=' + config.root + '/data\n', release.release_environment())
|
|
self.assertIn('CHAT_LOG_DIR=' + config.root + '/logs\n', release.release_environment())
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'release'
|
|
key = "fixture-${literal}-don't-expand"
|
|
write_release(path, config, key)
|
|
self.assertEqual(dotenv_values(path / 'runtime.env', interpolate=False)['DEEPSEEK_API_KEY'], key)
|
|
for name in ('release.env', 'site.conf', 'compose.yml', 'activate.sh'):
|
|
self.assertNotIn(key, (path / name).read_text(encoding='utf-8'))
|
|
self.assertIn('auth_basic_user_file', (path / 'site.conf').read_text(encoding='utf-8'))
|
|
|
|
def test_public_allowlist_keeps_staff_search_exports_and_order_endpoints_protected(self):
|
|
config = GrayConfig.from_environment(self.settings()).nginx_config()
|
|
self.assertTrue(config.startswith('# Managed for baiyue-logistics-chat-gray;'))
|
|
match = re.search(r'location ~ (\S+) \{([^}]+)\}', config)
|
|
self.assertIsNotNone(match)
|
|
pattern, body = match.groups()
|
|
self.assertIn('auth_basic off;', body)
|
|
self.assertIn('auth_basic "Baiyue gray acceptance";', config)
|
|
for path in ('/healthz','/stream_chat','/api/visitor/session','/api/get_history','/api/get_messages',
|
|
'/api/customer/session','/api/customer/login','/api/customer/logout',
|
|
'/api/del_history','/api/clear_all','/api/visitor/human/request','/api/visitor/human/poll',
|
|
'/api/visitor/human/send','/api/visitor/human/close','/api/speech/transcribe'):
|
|
with self.subTest(public_path=path):self.assertRegex(path,pattern)
|
|
for path in ('/admin/dashboard','/api/export_chat','/api/search_chat','/api/my_orders','/api/tickets',
|
|
'/api/human/queue','/api/human/poll','/api/human/send','/api/staff/uploads/private.txt',
|
|
'/api/visitor/human/accept','/api/customer/users','/api/customer/session/extra',
|
|
'/api/get_history/unknown','/api/unknown'):
|
|
with self.subTest(staff_path=path):self.assertNotRegex(path,pattern)
|
|
example=(Path(__file__).resolve().parents[1]/'deploy/nginx.conf.example').read_text(encoding='utf-8')
|
|
self.assertEqual(re.search(r'location ~ (\S+) \{',example)[1],pattern)
|
|
self.assertIn('location = / {\n auth_basic off;',config)
|
|
|
|
def test_invalid_model_credential_does_not_create_a_release_directory(self):
|
|
config = GrayConfig.from_environment(self.settings())
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'release'
|
|
with self.assertRaises(ValueError):
|
|
write_release(path, config, 'fixture-key\nANOTHER_SETTING=bad')
|
|
self.assertFalse(path.exists())
|
|
|
|
def test_baidu_credentials_only_enter_private_runtime_file(self):
|
|
config = GrayConfig.from_environment(self.settings())
|
|
speech = {'ASR_PROVIDER': 'baidu', 'BAIDU_ASR_API_KEY': 'fixture-${asr-key}',
|
|
'BAIDU_ASR_SECRET_KEY': "fixture-asr-secret's", 'FFMPEG_BIN': r'C:\local\ffmpeg.exe'}
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'release'
|
|
write_release(path, config, 'fixture-model-key', speech)
|
|
runtime = dotenv_values(path / 'runtime.env', interpolate=False)
|
|
self.assertEqual(runtime['ASR_PROVIDER'], 'baidu')
|
|
for key in ('BAIDU_ASR_API_KEY', 'BAIDU_ASR_SECRET_KEY'):
|
|
self.assertEqual(runtime[key], speech[key])
|
|
for file in path.iterdir():
|
|
if file.name != 'runtime.env':
|
|
self.assertNotIn(speech[key], file.read_text(encoding='utf-8'))
|
|
self.assertNotIn('FFMPEG_BIN', runtime)
|
|
|
|
def test_incomplete_or_multiline_speech_credentials_are_rejected_before_writing(self):
|
|
config = GrayConfig.from_environment(self.settings())
|
|
for speech in ({'ASR_PROVIDER': 'baidu'}, {'ASR_PROVIDER': 'unknown'},
|
|
{'ASR_PROVIDER': 'baidu', 'BAIDU_ASR_API_KEY': 'key', 'BAIDU_ASR_SECRET_KEY': 'secret\nOTHER=bad'}):
|
|
with self.subTest(provider=speech['ASR_PROVIDER']), tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'release'
|
|
with self.assertRaises(ValueError):
|
|
write_release(path, config, 'fixture-key', speech)
|
|
self.assertFalse(path.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|