"""Required official VOC dependency: verify existing installation or install the pinned missing package.""" import argparse import io import json import os from pathlib import Path import shutil import subprocess import sys import tarfile import tempfile import urllib.request import uuid sys.dont_write_bytecode = True from bootstrap import (HERE, Stopped, atomic_create, check_enterprise, encode, need, plain, private_json, safe_name, sealed, sha) def release(root): root = plain(root) cfg, _ = sealed(root) need(cfg['voc']['required'] is True and cfg['voc']['version'] == '0.3.44', 'VOC_REQUIREMENT_INVALID') meta = json.loads((root / cfg['voc']['releaseManifest']).read_bytes()) need(meta['name'] == '@vocmarket/voc-skill' and meta['version'] == '0.3.44' and meta['url'] == 'https://registry.npmjs.org/@vocmarket/voc-skill/-/voc-skill-0.3.44.tgz', 'VOC_RELEASE_INVALID') need(sha((root / cfg['voc']['dependencyLock']).read_bytes()) == meta['dependencyLockSha256'], 'VOC_DEPENDENCY_LOCK_CHANGED') return cfg, meta def fragment(target): return {'mcpServers': {'voc': {'command': 'node', 'args': [str(target / 'mcp/src/server.js')], 'cwd': str(target), 'env': {'VOC_SKILL_NO_UPDATE_CHECK': '1'}}}} def public_hashes(target, meta): result = {} for name, digest in meta['files'].items(): need(safe_name(name), 'INVALID_VOC_PATH') file = plain(target / name) need(file.is_file(), 'VOC_PUBLIC_FILE_MISSING') actual = sha(file.read_bytes()) if actual != digest: need(name == '.mcp.json', 'EXISTING_VOC_FILES_CHANGED') # Preserve the host's existing paths, credentials and additional MCP configuration. value = json.loads(file.read_bytes()).get('mcpServers', {}).get('voc', {}) need(value.get('command') == 'node' and value.get('args') == [str(target / 'mcp/src/server.js')] and value.get('cwd') == str(target), 'EXISTING_VOC_MCP_BINDING_DIFFERS') result[name] = actual return result def plan(root, data_dir): root, data_dir = plain(root), plain(data_dir) cfg, meta = release(root) check_enterprise(private_json(data_dir / '.fmode/config/enterprise/current.json'), cfg) options = [plain(data_dir / 'skills' / name) for name in ('voc-intelligence', 'voc-social-suite')] official = [] for target in options: package = plain(target / 'package.json') if package.is_file(): info = json.loads(package.read_bytes()) if info.get('name') == meta['name']: need(info.get('version') == meta['version'], 'EXISTING_VOC_VERSION_DIFFERS') official.append(target) need(len(official) <= 1, 'DUPLICATE_OFFICIAL_VOC_INSTALLATION') target = official[0] if official else options[0] need(bool(official) or not target.exists(), 'UNKNOWN_EXISTING_VOC_DIRECTORY') hashes = public_hashes(target, meta) if official else None return {'schema': 'ksf.enterprise-voc-plan.v1', 'version': meta['version'], 'target': str(target), 'action': 'verify-existing' if official else 'install-missing', 'publicFiles': len(meta['files']), 'beforePublicHashes': hashes, 'archiveSha256': meta['sha256'], 'packageManifestSha256': sha((root / 'FILES.sha256').read_bytes()), 'businessWrites': 0} def unpack(raw, meta): need(len(raw) == meta['bytes'] and sha(raw) == meta['sha256'], 'VOC_ARCHIVE_HASH_MISMATCH') files = {} with tarfile.open(fileobj=io.BytesIO(raw), mode='r:gz') as z: for member in z.getmembers(): need(member.isfile() and member.name.startswith('package/'), 'UNSAFE_VOC_ARCHIVE') name = member.name[len('package/'):] need(safe_name(name) and name not in files and name in meta['files'], 'UNSAFE_VOC_ARCHIVE_PATH') body = z.extractfile(member).read() need(sha(body) == meta['files'][name], 'VOC_FILE_HASH_MISMATCH') files[name] = body need(set(files) == set(meta['files']), 'VOC_ARCHIVE_INCOMPLETE') return files def runtime(root, target, runner=subprocess.run): node = shutil.which('node') need(node, 'NODE_REQUIRED_FOR_VOC') check = runner([node, str(root / 'check-voc.cjs'), str(target)], capture_output=True, timeout=45) need(check.returncode == 0, 'VOC_LOCAL_RUNTIME_CHECK_FAILED') value = json.loads(check.stdout) need(value.get('ok') is True and value.get('version') == '0.3.44' and value.get('collectionRequests') == 0, 'VOC_CHECK_RECEIPT_INVALID') return value def npm_command(): node, npm = shutil.which('node'), shutil.which('npm') need(node and npm, 'NODE_AND_NPM_REQUIRED_FOR_VOC') version = subprocess.run([node, '-p', 'process.versions.node'], capture_output=True, timeout=15) need(version.returncode == 0 and int(version.stdout.decode().split('.')[0]) >= 18, 'NODE_18_REQUIRED') if os.name == 'nt': cli = Path(node).parent / 'node_modules/npm/bin/npm-cli.js' need(cli.is_file(), 'NPM_CLI_NOT_FOUND') return [node, str(cli)] return [npm] def apply(root, data_dir, expected_plan, output, archive=None, runner=subprocess.run): root, data_dir, output = plain(root), plain(data_dir), plain(output) observed = plan(root, data_dir) cfg, meta = release(root) target = plain(observed['target']) need(observed['packageManifestSha256'] == expected_plan['packageManifestSha256'] and observed['target'] == expected_plan['target'], 'VOC_PLAN_CHANGED') if observed['action'] == 'verify-existing': check = runtime(root, target, runner) result = {'ok': True, 'version': meta['version'], 'action': 'verified-existing', 'target': str(target), 'vocModified': False, 'runtime': check, 'businessWrites': 0} else: need(observed == expected_plan, 'VOC_PLAN_CHANGED') if archive: raw = plain(archive).read_bytes() else: with urllib.request.urlopen(meta['url'], timeout=45) as response: raw = response.read(meta['bytes'] + 1) files = unpack(raw, meta) target.parent.mkdir(parents=True, exist_ok=True, mode=0o700) with tempfile.TemporaryDirectory(prefix='.ksf-voc-stage-', dir=target.parent) as temp: staged = Path(temp) / 'voc' for name, body in files.items(): atomic_create(staged / name, body, 0o644) (staged / '.mcp.json').write_bytes(encode(fragment(target))) atomic_create(staged / 'package-lock.json', (root / cfg['voc']['dependencyLock']).read_bytes(), 0o644) command = npm_command() + ['ci', '--omit=dev', '--ignore-scripts', '--no-audit', '--no-fund', '--registry', 'https://registry.npmjs.org', '--cache', str(output / 'npm-cache')] installed = runner(command, cwd=staged, capture_output=True, timeout=240) need(installed.returncode == 0, 'VOC_DEPENDENCY_INSTALL_FAILED') runtime(root, staged, runner) need(plan(root, data_dir) == expected_plan, 'VOC_CONCURRENT_INSTALL') os.rename(staged, target) result = {**verify(root, data_dir, runner), 'action': 'installed-missing', 'vocModified': True} file = output / 'voc-mcp.fragment.json' body = encode(fragment(target)) if file.exists(): need(file.read_bytes() == body, 'MCP_FRAGMENT_CHANGED') else: atomic_create(file, body) result['mcpFragment'] = str(file) return result def verify(root, data_dir, runner=subprocess.run): observed = plan(root, data_dir) need(observed['action'] == 'verify-existing', 'VOC_NOT_INSTALLED') check = runtime(plain(root), Path(observed['target']), runner) return {'ok': True, 'version': observed['version'], 'target': observed['target'], 'publicFilesVerified': observed['publicFiles'], 'runtime': check, 'hostRegistrationVerified': False, 'liveAccessVerified': False, 'businessWrites': 0} def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('command', choices=['plan', 'apply', 'verify']) p.add_argument('--data-dir', type=Path, required=True) p.add_argument('--out', type=Path, required=True) p.add_argument('--archive', type=Path) a = p.parse_args() need(a.data_dir.is_absolute() and a.out.is_absolute(), 'ABSOLUTE_PATH_REQUIRED') data, output = plain(a.data_dir), plain(a.out) need(output != data and output not in data.parents and data / 'skills' not in output.parents and HERE not in output.parents and output != HERE, 'UNSAFE_RESULT_DIRECTORY') output.mkdir(parents=True, exist_ok=True, mode=0o700) if a.command == 'plan': result = plan(HERE, data) atomic_create(output / 'voc-plan.json', encode(result)) result = {'ok': True, 'planOnly': True, **result} elif a.command == 'apply': result = apply(HERE, data, private_json(output / 'voc-plan.json'), output, a.archive) else: result = verify(HERE, data) atomic_create(output / ('voc-' + a.command + '-' + uuid.uuid4().hex + '.json'), encode(result)) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': try: main() except (Stopped, OSError, ValueError, KeyError, subprocess.TimeoutExpired, tarfile.TarError) as error: print(json.dumps({'ok': False, 'error': {'code': str(error) if isinstance(error, Stopped) else 'VOC_INITIALIZATION_STOPPED'}, 'businessWrites': 0}), file=sys.stderr) sys.exit(1)