115 lines
6.4 KiB
Python
115 lines
6.4 KiB
Python
import base64
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / 'generate-file-access-config.py'
|
|
APPS = ('admin', 'h5', 'mini', 'flutter', 'service', 'package')
|
|
|
|
|
|
class SharedFileAccessConfigurationTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.secret_dir = self.root / 'pj_service_cloud/.file-access'
|
|
self.secret_dir.mkdir(parents=True)
|
|
|
|
def run_script(self, *extra, success=True):
|
|
result = subprocess.run([sys.executable, str(SCRIPT), '--workspace-root', str(self.root),
|
|
'--public-base-url', 'https://files.example/api', *extra],
|
|
text=True, capture_output=True, timeout=15)
|
|
self.assertEqual(result.returncode == 0, success, result.stderr)
|
|
return result
|
|
|
|
def test_existing_production_keys_are_retained_for_every_build_mode(self):
|
|
old = {env: {app: base64.b64encode(hashlib.sha256((env + app).encode()).digest()).decode()
|
|
for app in APPS} for env in ('development', 'test', 'production')}
|
|
(self.secret_dir / 'keys.json').write_text(json.dumps(old))
|
|
for project in ('pj_admin_cloud', 'pj_h5_cloud', 'pj_mini_app_cloud'):
|
|
folder = self.root / project
|
|
folder.mkdir()
|
|
(folder / '.env.local').write_text('VITE_EXISTING_SETTING=keep\n')
|
|
(folder / '.env.development.local').write_text('VITE_API_URL=/existing-api\nVITE_FILE_ACCESS_KEY="old"\n')
|
|
(folder / '.env.production.local').write_text('# Local file access settings; do not commit.\nVITE_FILE_ACCESS_KEY="old"\n')
|
|
for env in old:
|
|
(self.secret_dir / (env + '.yml')).write_text(json.dumps({'file-access': {'mode': 'observe'}}))
|
|
flutter = self.root / 'pj_supermarket_cloud/.file-access'
|
|
flutter.mkdir(parents=True, exist_ok=True)
|
|
(flutter / (env + '.json')).write_text(json.dumps({'FILE_ACCESS_KEY': old[env]['flutter']}))
|
|
result = self.run_script('--environment', 'development', '--mode', 'enforce')
|
|
shared = json.loads((self.secret_dir / 'keys.json').read_text())
|
|
self.assertEqual(shared, old['production'])
|
|
policy = json.loads((self.secret_dir / 'config.yml').read_text())['file-access']
|
|
self.assertEqual(policy['mode'], 'enforce')
|
|
for app in APPS:
|
|
self.assertEqual(policy['clients'][app]['keys']['v1'], shared[app])
|
|
self.assertNotIn(shared[app], result.stdout)
|
|
for app, project in (('admin', 'pj_admin_cloud'), ('h5', 'pj_h5_cloud'), ('mini', 'pj_mini_app_cloud')):
|
|
folder = self.root / project
|
|
self.assertIn(shared[app], (folder / '.env.local').read_text())
|
|
self.assertIn('VITE_EXISTING_SETTING=keep', (folder / '.env.local').read_text())
|
|
self.assertEqual((folder / '.env.development.local').read_text(), 'VITE_API_URL=/existing-api\n')
|
|
self.assertFalse((folder / '.env.production.local').exists())
|
|
flutter = self.root / 'pj_supermarket_cloud/.file-access'
|
|
self.assertEqual(json.loads((flutter / 'config.json').read_text())['FILE_ACCESS_KEY'], shared['flutter'])
|
|
for env in old:
|
|
self.assertFalse((self.secret_dir / (env + '.yml')).exists())
|
|
self.assertFalse((flutter / (env + '.json')).exists())
|
|
self.assertTrue((self.secret_dir / 'migration-backup/pj_service_cloud/.file-access/keys.json').exists())
|
|
before = {path: path.read_bytes() for path in self.root.rglob('*') if path.is_file()}
|
|
self.run_script('--environment', 'test')
|
|
after = {path: path.read_bytes() for path in self.root.rglob('*') if path.is_file()}
|
|
self.assertEqual(before, after)
|
|
|
|
def test_new_workspace_has_one_registry_and_observation_remains_the_default(self):
|
|
self.run_script()
|
|
keys = json.loads((self.secret_dir / 'keys.json').read_text())
|
|
self.assertEqual(set(keys), set(APPS))
|
|
self.assertEqual(len(set(keys.values())), len(APPS))
|
|
for key in keys.values():
|
|
self.assertEqual(len(base64.b64decode(key)), 32)
|
|
policy = json.loads((self.secret_dir / 'config.yml').read_text())['file-access']
|
|
self.assertEqual(policy['mode'], 'observe')
|
|
self.assertTrue(policy['block-browser-navigation'])
|
|
self.assertEqual(policy['public-image-referer-prefixes'], [])
|
|
self.assertEqual(policy['upload']['ttl-seconds'], 300)
|
|
self.assertEqual(policy['upload']['max-file-bytes'], 200 * 1024 * 1024)
|
|
self.assertTrue({'jpg', 'pdf', 'apk', 'docx', 'xlsx', 'zip', 'mp4', 'txt'}.issubset(policy['upload']['allowed-extensions']))
|
|
self.assertEqual(policy['upload']['internal-extensions'], ['log', 'gz'])
|
|
|
|
def test_existing_custom_upload_policy_is_preserved_without_changing_keys(self):
|
|
self.run_script()
|
|
registry = (self.secret_dir / 'keys.json').read_bytes()
|
|
path = self.secret_dir / 'config.yml'
|
|
config = json.loads(path.read_text())
|
|
config['file-access']['public-image-referer-prefixes'] = ['https://files.example/mobile/']
|
|
config['file-access']['upload'] = {'ttl-seconds': 120, 'max-file-bytes': 1024,
|
|
'allowed-extensions': ['png', 'pdf']}
|
|
path.write_text(json.dumps(config))
|
|
self.run_script()
|
|
policy = json.loads(path.read_text())['file-access']['upload']
|
|
self.assertEqual(json.loads(path.read_text())['file-access']['public-image-referer-prefixes'], ['https://files.example/mobile/'])
|
|
self.assertEqual(policy['ttl-seconds'], 120)
|
|
self.assertEqual(policy['max-file-bytes'], 1024)
|
|
self.assertEqual(policy['allowed-extensions'], ['png', 'pdf'])
|
|
self.assertEqual(policy['internal-extensions'], ['log', 'gz'])
|
|
self.assertEqual((self.secret_dir / 'keys.json').read_bytes(), registry)
|
|
|
|
def test_invalid_existing_keys_are_never_silently_replaced(self):
|
|
original = {app: 'not-a-valid-base64-key' for app in APPS}
|
|
registry = self.secret_dir / 'keys.json'
|
|
registry.write_text(json.dumps(original))
|
|
self.run_script(success=False)
|
|
self.assertEqual(json.loads(registry.read_text()), original)
|
|
self.assertFalse((self.secret_dir / 'config.yml').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|