fix:图片访问

fix:权限修改

scripts/generate-file-access-config.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+"""Synchronize one file-access key set across every application and build mode."""
+import argparse
+import base64
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+from urllib.parse import urlsplit
+
+APPS = ('admin', 'h5', 'mini', 'flutter', 'service', 'package')
+ENV_FIELD = re.compile(r'^\s*(?:export\s+)?VITE_FILE_ACCESS_(?:APP_ID|KEY_ID|KEY|ALLOWED_ORIGINS)\s*=')
+ENV_COMMENT = '# Shared file access settings (all modes; do not commit).'
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--workspace-root', type=Path, default=Path(__file__).resolve().parents[2])
+    parser.add_argument('--environment', choices=['development', 'test', 'production', 'all'], help=argparse.SUPPRESS)
+    parser.add_argument('--public-base-url', required=True, help='Public gateway base for this deployment')
+    parser.add_argument('--allowed-origin', action='append', help='Owned historical file origin; may be repeated')
+    parser.add_argument('--mode', choices=['observe', 'enforce'], help='Preserve the current mode when omitted')
+    args = parser.parse_args()
+    uri = urlsplit(args.public_base_url)
+    if uri.scheme not in ('http', 'https') or not uri.hostname or uri.username or uri.query or uri.fragment:
+        parser.error('--public-base-url must be an HTTP(S) base without credentials, query or fragment')
+    os.umask(0o077)
+    root = args.workspace_root.resolve()
+    secret_dir = root / 'pj_service_cloud/.file-access'
+    secret_dir.mkdir(parents=True, exist_ok=True)
+    registry = secret_dir / 'keys.json'
+    stored = json.loads(registry.read_text()) if registry.exists() else {}
+    if all(app in stored for app in APPS):
+        keys = {app: stored[app] for app in APPS}
+    elif isinstance(stored.get('production'), dict):
+        # Preserve the deployed keys during migration; never rotate them implicitly.
+        keys = {app: stored['production'].get(app) for app in APPS}
+    elif stored:
+        parser.error('Unrecognized key registry; existing keys have not been changed')
+    else:
+        keys = {app: base64.b64encode(secrets.token_bytes(32)).decode('ascii') for app in APPS}
+    for app, value in keys.items():
+        try:
+            if not isinstance(value, str) or len(base64.b64decode(value, validate=True)) < 32:
+                raise ValueError('Invalid key')
+        except ValueError:
+            parser.error('Invalid existing key for ' + app + '; refusing to generate a replacement')
+    config_path = secret_dir / 'config.yml'
+    try:
+        config = json.loads(config_path.read_text()) if config_path.exists() else {'file-access': {}}
+    except json.JSONDecodeError:
+        parser.error('Preserve the generated JSON-compatible YAML format in config.yml; no files changed')
+    policy = config.setdefault('file-access', {})
+    defaults = {'mode': 'observe', 'block-browser-navigation': True, 'ttl-seconds': 1800,
+                'package-ttl-seconds': 86400, 'future-skew-seconds': 300,
+                'legacy-dart': {'enabled': True, 'extensions': ['jpg', 'jpeg', 'png', 'gif', 'webp', 'bmp', 'svg', 'ico']}}
+    for name, value in defaults.items():
+        policy.setdefault(name, value)
+    if args.mode:
+        policy['mode'] = args.mode
+    policy['public-base-url'] = args.public_base_url.rstrip('/')
+    if args.allowed_origin is not None:
+        policy['allowed-origins'] = list(dict.fromkeys(args.allowed_origin))
+    policy.setdefault('allowed-origins', [])
+    clients = policy.setdefault('clients', {})
+    for app in APPS:
+        client = clients.setdefault(app, {'active-kid': 'v1'})
+        client.setdefault('active-kid', 'v1')
+        client.setdefault('keys', {})['v1'] = keys[app]
+
+    def backup(path):
+        target = secret_dir / 'migration-backup' / path.relative_to(root)
+        if path.exists() and not target.exists():
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_bytes(path.read_bytes())
+            target.chmod(0o600)
+
+    def write(path, content):
+        if path.exists() and path.read_text() == content:
+            return
+        if path.exists():
+            backup(path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        temporary = path.with_name(path.name + '.file-access.tmp')
+        temporary.write_text(content)
+        temporary.chmod(0o600)
+        temporary.replace(path)
+        print('Updated:', path.relative_to(root))
+
+    write(registry, json.dumps(keys, indent=2) + '\n')
+    write(config_path, json.dumps(config, indent=2) + '\n')
+    for app, project in {'admin': 'pj_admin_cloud', 'h5': 'pj_h5_cloud', 'mini': 'pj_mini_app_cloud'}.items():
+        folder = root / project
+        folder.mkdir(parents=True, exist_ok=True)
+        variables = {'VITE_FILE_ACCESS_APP_ID': app, 'VITE_FILE_ACCESS_KEY_ID': 'v1',
+                     'VITE_FILE_ACCESS_KEY': keys[app],
+                     'VITE_FILE_ACCESS_ALLOWED_ORIGINS': ','.join(policy['allowed-origins'])}
+        local = folder / '.env.local'
+        previous = local.read_text() if local.exists() else ''
+        remaining = [line for line in previous.splitlines() if not ENV_FIELD.match(line) and line != ENV_COMMENT]
+        content = '\n'.join(remaining).rstrip()
+        content = (content + '\n\n' if content else '') + ENV_COMMENT + '\n'
+        content += ''.join(f'{name}={json.dumps(value)}\n' for name, value in variables.items())
+        write(local, content)
+        # Mode-specific Vite files override .env.local. Remove only our four fields.
+        for old in folder.glob('.env.*.local'):
+            text = old.read_text()
+            if not any(ENV_FIELD.match(line) for line in text.splitlines()):
+                continue
+            backup(old)
+            remaining = [line for line in text.splitlines(keepends=True) if not ENV_FIELD.match(line)]
+            if any(line.strip() and not line.lstrip().startswith('#') for line in remaining):
+                old.write_text(''.join(remaining))
+            else:
+                old.unlink()
+            print('Removed mode-specific file-access fields:', old.relative_to(root))
+    flutter_dir = root / 'pj_supermarket_cloud/.file-access'
+    write(flutter_dir / 'config.json', json.dumps({
+        'FILE_ACCESS_APP_ID': 'flutter', 'FILE_ACCESS_KEY_ID': 'v1',
+        'FILE_ACCESS_KEY': keys['flutter'], 'FILE_ACCESS_ALLOWED_ORIGINS': ','.join(policy['allowed-origins']),
+    }, indent=2) + '\n')
+    for name in ('development', 'test', 'production'):
+        for old in (secret_dir / (name + '.yml'), flutter_dir / (name + '.json')):
+            if not old.exists():
+                continue
+            text = old.read_text()
+            if old.suffix == '.yml':
+                try:
+                    owned = set(json.loads(text)) == {'file-access'}
+                except json.JSONDecodeError:
+                    owned = re.findall(r'^([a-zA-Z][\w-]*):', text, re.M) == ['file-access']
+            else:
+                owned = set(json.loads(text)).issubset({'FILE_ACCESS_APP_ID', 'FILE_ACCESS_KEY_ID', 'FILE_ACCESS_KEY', 'FILE_ACCESS_ALLOWED_ORIGINS'})
+            if owned:
+                backup(old)
+                old.unlink()
+    if args.environment:
+        print('--environment is retained for compatibility; every environment now uses the same keys.')
+    print('One shared key set is active. Existing keys were retained; migration backups are ignored local files.')
+
+
+if __name__ == '__main__':
+    main()

scripts/tests/test_generate_file_access_config.py

@@ -0,0 +1,91 @@
+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'])
+
+    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()