#!/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, 'public-image-referer-prefixes': [], 'legacy-dart': {'enabled': True, 'extensions': ['jpg', 'jpeg', 'png', 'gif', 'webp', 'bmp', 'svg', 'ico']}} for name, value in defaults.items(): policy.setdefault(name, value) upload = policy.setdefault('upload', {}) for name, value in { 'ttl-seconds': 300, 'max-file-bytes': 200 * 1024 * 1024, 'allowed-extensions': ['jpg', 'jpeg', 'png', 'gif', 'webp', 'bmp', 'ico', 'svg', 'pdf', 'apk', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx', 'txt', 'csv', 'zip', 'mp4', 'mov', 'webm', 'avi', 'mp3', 'wav', 'm4a', 'aac', 'ogg', 'amr', 'flac'], 'internal-extensions': ['log', 'gz'], }.items(): upload.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()