"""Nine real rclone experiments. Writes only to a newly created project sandbox.""" import argparse import datetime import hashlib import json import os import platform import re import secrets import shutil import subprocess from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BIN = ROOT / 'tools' / 'bin' / ('rclone.exe' if os.name == 'nt' else 'rclone') SCENES = [ ('inventory', '01 · Инвентаризация', 'Проверяем состав источника до переноса и получаем JSON для отчёта.'), ('copy', '02 · Копирование без уборки', 'Обновляем файлы назначения, сохраняя лишний legacy.txt.'), ('filters', '03 · Отбор документов', 'Переносим только docs/**: медиа, журнал и cache остаются за пределами задачи.'), ('dryrun', '04 · План синхронизации', 'Смотрим план будущих изменений. --dry-run не изменяет целевые файлы.'), ('sync', '05 · Зеркало источника', 'На учебной копии удаляем лишнее и получаем точное зеркало источника.'), ('check', '06 · Поиск повреждения', 'Намеренно меняем один файл, обнаруживаем несовпадение, затем восстанавливаем.'), ('backup', '07 · История и возврат файла', 'Сохраняем заменённые/удалённые файлы в backup-dir и возвращаем старый отчёт отдельно.'), ('crypt', '08 · Зашифрованное хранилище', 'Шифруем локальную папку через crypt и проверяем расшифрованную копию по SHA-256.'), ('limit', '09 · Лимит удаления', 'Запрещаем удаления через --max-delete 0: лишние файлы дают ненулевой код выхода.'), ] def snapshot(folder): if not folder.exists(): return [] return [{'path': p.relative_to(folder).as_posix(), 'bytes': p.stat().st_size, 'sha256': hashlib.sha256(p.read_bytes()).hexdigest()} for p in sorted(folder.rglob('*')) if p.is_file()] def validate_name(name): if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_-]{0,47}', name): raise ValueError('Run name: 1–48 ASCII letters, digits, underscore or hyphen; no paths.') return name def new_run(name): validate_name(name) runtime = ROOT / '.runtime' if runtime.exists() and runtime.resolve() != runtime: raise ValueError('Refusing redirected .runtime directory') runtime.mkdir(exist_ok=True) run = runtime / name run.mkdir(exist_ok=False) return run def assert_inside(run, path): if not path.resolve().is_relative_to(run.resolve()) or path.resolve() == run.resolve(): raise ValueError('Operation outside the isolated run refused') return path def execute(run, config, arguments, expected=(0,)): # No shell interpolation and no user-selected filesystem paths or remotes. env = {k: v for k, v in os.environ.items() if not k.upper().startswith('RCLONE_')} result = subprocess.run([str(BIN), '--config', str(config), '--color', 'NEVER', '--retries', '1', '--low-level-retries', '1', *map(str, arguments)], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=60) def scrub(text): return text.replace(str(run), '$LAB').replace(run.as_posix(), '$LAB') record = {'args': [scrub(str(v)) for v in arguments], 'exitCode': result.returncode, 'expectedExitCodes': list(expected), 'stdout': scrub(result.stdout), 'stderr': scrub(result.stderr)} if result.returncode not in expected: raise RuntimeError(json.dumps(record, ensure_ascii=False)) return record def main(): parser = argparse.ArgumentParser() parser.add_argument('--run', required=True, help='New sandbox name, never an existing path') parser.add_argument('--publish-recording', action='store_true', help='Replace public recording with this verified local run') options = parser.parse_args() if not BIN.is_file(): raise SystemExit('First run: python tools/install.py') version = subprocess.check_output([str(BIN), 'version'], text=True).splitlines()[0] if version != 'rclone v1.76.0': raise SystemExit(f'Expected rclone v1.76.0, got {version}') run = new_run(options.run) config = run / 'empty.conf' config.write_text('', encoding='utf-8') results = [] for key, title, purpose in SCENES: case = assert_inside(run, run / key) source, target = case / 'source', case / 'target' case.mkdir() shutil.copytree(ROOT / 'fixtures' / 'source', source) target.mkdir() if key in ('copy', 'dryrun', 'sync', 'backup', 'limit'): (target / 'docs').mkdir() (target / 'docs' / 'report.csv').write_text('month,revenue\n2026-09,900\n', encoding='utf-8') (target / 'legacy.txt').write_text('Historical destination-only file\n', encoding='utf-8') commands = [] before = snapshot(target) run_cmd = lambda args, expected=(0,): commands.append(execute(run, config, args, expected)) extra = {} if key == 'inventory': run_cmd(['lsjson', source, '--recursive', '--files-only', '--hash']) run_cmd(['size', source, '--json']) assert len(json.loads(commands[0]['stdout'])) == len(snapshot(source)) elif key == 'copy': run_cmd(['copy', source, target, '--checksum', '--log-level', 'INFO']) assert (target / 'legacy.txt').exists() assert len(snapshot(target)) == len(snapshot(source)) + 1 elif key == 'filters': rules = assert_inside(run, case / 'documents.filter') rules.write_text('+ /docs/**\n- **\n', encoding='utf-8') run_cmd(['copy', source, target, '--filter-from', rules, '--log-level', 'INFO']) assert len(snapshot(target)) == 2 and all(v['path'].startswith('docs/') for v in snapshot(target)) elif key == 'dryrun': run_cmd(['sync', source, target, '--checksum', '--dry-run', '--log-level', 'INFO']) assert snapshot(target) == before elif key == 'sync': run_cmd(['sync', source, target, '--checksum', '--log-level', 'INFO']) assert snapshot(source) == snapshot(target) elif key == 'check': run_cmd(['copy', source, target, '--checksum']) (target / 'docs' / 'report.csv').write_text('month,revenue\n2026-10,999\n', encoding='utf-8') before = snapshot(target) combined = case / 'comparison.txt' run_cmd(['check', source, target, '--combined', combined], (1,)) extra['comparison'] = combined.read_text(encoding='utf-8') run_cmd(['hashsum', 'SHA-256', source, '--include', 'docs/report.csv']) run_cmd(['copy', source, target, '--checksum']) run_cmd(['check', source, target]) assert snapshot(source) == snapshot(target) elif key == 'backup': archive, recovered = case / 'archive' / 'run01', case / 'recovered' old_hash = hashlib.sha256((target / 'docs' / 'report.csv').read_bytes()).hexdigest() run_cmd(['sync', source, target, '--checksum', '--backup-dir', archive, '--log-level', 'INFO']) run_cmd(['copyto', archive / 'docs' / 'report.csv', recovered / 'report-old.csv']) extra['archive'] = snapshot(archive) extra['recovered'] = snapshot(recovered) assert extra['recovered'][0]['sha256'] == old_hash assert len(extra['archive']) == 2 elif key == 'crypt': private = case / 'private' private.mkdir(mode=0o700) crypt_config = private / 'crypt.conf' password = secrets.token_urlsafe(32) obscure = subprocess.run([str(BIN), 'obscure', '-'], input=password+'\n', capture_output=True, text=True, check=True).stdout.strip() cipher = case / 'cipher' crypt_config.write_text('[vault]\ntype = crypt\nremote = '+cipher.as_posix()+'\nfilename_encryption = standard\ndirectory_name_encryption = true\npassword = '+obscure+'\n', encoding='utf-8') if os.name != 'nt': crypt_config.chmod(0o600) commands.append(execute(run, crypt_config, ['copy', source, 'vault:'])) commands.append(execute(run, crypt_config, ['copy', 'vault:', target])) commands.append(execute(run, crypt_config, ['cryptcheck', source, 'vault:'])) extra['cipher'] = snapshot(cipher) extra['secretNotice'] = 'Случайный учебный пароль и конфигурация остаются только внутри .runtime. В записи их нет.' assert snapshot(source) == snapshot(target) assert {v['path'] for v in extra['cipher']}.isdisjoint({v['path'] for v in snapshot(source)}) elif key == 'limit': run_cmd(['sync', source, target, '--checksum', '--max-delete', '0', '--log-level', 'INFO'], (7,)) assert (target / 'legacy.txt').exists() extra['warning'] = 'Лимит остановил удаление; новые файлы уже могли скопироваться. Это не транзакция и не откат.' after = snapshot(target) results.append({'id': key, 'title': title, 'purpose': purpose, 'source': snapshot(source), 'before': before, 'after': after, 'commands': commands, 'extra': extra, 'passed': True}) print(f'PASS {key}: {len(commands)} commands') report = {'schemaVersion': 1, 'kind': 'recorded-local-cli-run', 'rcloneVersion': version, 'platform': platform.system(), 'architecture': platform.machine(), 'recordedAt': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'runName': options.run, 'cloudAccountsUsed': False, 'passed': True, 'scenarioCount': len(results), 'scenarios': results} evidence = run / 'evidence.json' evidence.write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n', encoding='utf-8') if options.publish_recording: for destination in (ROOT / 'public' / 'data', ROOT / 'evidence'): destination.mkdir(parents=True, exist_ok=True) (destination / 'recorded-run.json').write_bytes(evidence.read_bytes()) print(f'PASS all 9; evidence: .runtime/{options.run}/evidence.json') if __name__ == '__main__': main()