157 lines
7.9 KiB
Python
157 lines
7.9 KiB
Python
#!/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()
|