"""Install two pinned, public Skills into a host-supplied Skills directory. No guessed host directories, credential discovery, account creation or business writes. The official VOC artifact is downloaded from npm and verified before extraction. """ import argparse import hashlib import io import json import os from pathlib import Path, PurePosixPath import re import shutil import subprocess import sys import tarfile import tempfile import urllib.request class InstallError(Exception): pass def require(condition, code): if not condition: raise InstallError(code) def digest(raw): return hashlib.sha256(raw).hexdigest() def encode(value): return (json.dumps(value, ensure_ascii=False, indent=2) + '\n').encode('utf-8') def safe_name(name): parts = PurePosixPath(name).parts return (isinstance(name, str) and bool(name) and '\\' not in name and ':' not in name and not name.startswith('/') and all(p not in ('..', '.') for p in parts) and name == '/'.join(parts)) def plain_path(path): path = Path(os.path.abspath(path)) for part in [*reversed(path.parents), path]: require(not part.is_symlink(), 'SYMLINK_PATH') if part.exists(): require(not hasattr(part, 'is_junction') or not part.is_junction(), 'JUNCTION_PATH') return path def inside(parent, child): return child != parent and child.is_relative_to(parent) def verify_package(package): package = plain_path(package) manifest = package / 'FILES.sha256' require(manifest.is_file(), 'PACKAGE_MANIFEST_MISSING') names = set() for line in manifest.read_text(encoding='utf-8').splitlines(): match = re.fullmatch(r'([a-f0-9]{64}) (.+)', line) require(match is not None, 'INVALID_PACKAGE_MANIFEST') h, name = match.groups() require(safe_name(name) and name not in names, 'INVALID_PACKAGE_PATH') names.add(name) p = plain_path(package / name) require(p.is_file() and digest(p.read_bytes()) == h, 'PACKAGE_HASH_MISMATCH') meta = json.loads((package / 'release.json').read_bytes()) require(meta['schema'] == 'ksf.workshop-onboarding.v1', 'INVALID_RELEASE') return meta def voc_archive_files(raw, expected): require(len(raw) == expected['bytes'] and digest(raw) == expected['sha256'], 'VOC_ARCHIVE_HASH_MISMATCH') result = {} with tarfile.open(fileobj=io.BytesIO(raw), mode='r:gz') as archive: for member in archive.getmembers(): require(member.name.startswith('package/'), 'INVALID_VOC_ARCHIVE_PATH') name = member.name[len('package/'):] require(safe_name(name), 'INVALID_VOC_ARCHIVE_PATH') require(member.isfile(), 'NON_FILE_VOC_ARCHIVE_MEMBER') require(name not in result and name in expected['files'], 'UNEXPECTED_VOC_ARCHIVE_FILE') require(member.size <= 8_000_000, 'VOC_ARCHIVE_MEMBER_TOO_LARGE') raw_file = archive.extractfile(member).read() require(digest(raw_file) == expected['files'][name], 'VOC_FILE_HASH_MISMATCH') result[name] = raw_file require(set(result) == set(expected['files']), 'VOC_ARCHIVE_INCOMPLETE') return result def source_files(package, meta, archive=None): expected = meta['voc'] if archive: raw = plain_path(archive).read_bytes() else: require(expected['url'] == 'https://registry.npmjs.org/@vocmarket/voc-skill/-/voc-skill-0.3.42.tgz', 'UNEXPECTED_VOC_SOURCE') request = urllib.request.Request(expected['url'], headers={'User-Agent': 'ksf-workshop-onboarding/1'}) with urllib.request.urlopen(request, timeout=30) as response: require(response.status == 200, 'VOC_DOWNLOAD_FAILED') raw = response.read(expected['bytes'] + 1) voc = voc_archive_files(raw, expected) workshop = {} for name, h in meta['workshop']['files'].items(): require(safe_name(name), 'INVALID_WORKSHOP_PATH') raw_file = plain_path(package / 'skills/ksf-workshop-runtime' / name).read_bytes() require(digest(raw_file) == h, 'WORKSHOP_FILE_HASH_MISMATCH') workshop[name] = raw_file return {'ksf-workshop-runtime': workshop, 'voc-intelligence': voc} def mcp_fragment(voc_dir): return {'mcpServers': {'voc': {'command': 'node', 'args': [str(voc_dir / 'mcp/src/server.js')], 'cwd': str(voc_dir), 'env': {'VOC_SKILL_NO_UPDATE_CHECK': '1'}}}} def verify_component(target, files): require(target.is_dir(), 'EXISTING_SKILL_NOT_DIRECTORY') for name, raw in files.items(): p = plain_path(target / name) require(p.is_file() and digest(p.read_bytes()) == digest(raw), 'EXISTING_SKILL_DIFFERS') def verify_hashes(target, hashes): for name, expected in hashes.items(): require(safe_name(name), 'INVALID_UPGRADE_PATH') p = plain_path(target / name) require(p.is_file() and digest(p.read_bytes()) == expected, 'UPGRADE_BASELINE_CHANGED') def plan(skills_dir, files, upgrade_from=None): actions = [] for name, content in files.items(): target = plain_path(skills_dir / name) if target.exists(): try: verify_component(target, content) action = 'verify-only' except InstallError: require(name == 'ksf-workshop-runtime' and upgrade_from, 'EXISTING_SKILL_DIFFERS') baseline = None for version, hashes in upgrade_from.items(): try: verify_hashes(target, hashes) baseline = (version, hashes) break except InstallError: continue require(baseline is not None, 'UPGRADE_BASELINE_CHANGED') # Code, locks and unknown local files are never rewritten by this upgrade. for path, raw in content.items(): if path.startswith('scripts/') or path == 'runtime-lock.json': require(digest(raw) == baseline[1].get(path), 'PROTECTED_RUNTIME_CHANGED') require(set(baseline[1]) <= set(content), 'UPGRADE_FILE_SET_CHANGED') additions = sorted(set(content) - set(baseline[1])) for path in additions: require(safe_name(path) and path.startswith('references/') and path.endswith('.md'), 'UPGRADE_NEW_FILE_NOT_REFERENCE') require(not plain_path(target / path).exists(), 'UPGRADE_NEW_FILE_COLLISION') actions.append({'skill': name, 'path': str(target), 'action': 'upgrade', 'files': len(content), 'fromVersion': baseline[0], 'before': baseline[1], 'additions': additions}) continue else: action = 'create' actions.append({'skill': name, 'path': str(target), 'action': action, 'files': len(content)}) return actions def remove_created(path, intended_parent): """Only remove our own newly created directory, after checking its final location.""" path = plain_path(path) intended_parent = plain_path(intended_parent) require(inside(intended_parent, path), 'CLEANUP_PATH_OUTSIDE_TARGET') require(path.name.startswith('.ksf-init-') or path.name in ('ksf-workshop-runtime', 'voc-intelligence'), 'CLEANUP_TARGET_NOT_OWNED') shutil.rmtree(path) def install_files(skills_dir, files, actions, backup_dir=None): skills_dir.mkdir(parents=True, exist_ok=True) staging = Path(tempfile.mkdtemp(prefix='.ksf-init-', dir=skills_dir)) created = [] upgraded = [] try: for row in actions: if row['action'] != 'create': continue staged = staging / row['skill'] for name, raw in files[row['skill']].items(): p = staged / name p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(raw) verify_component(staged, files[row['skill']]) # Recheck both targets before committing either component. for row in actions: target = skills_dir / row['skill'] if row['action'] == 'create': require(not target.exists(), 'CONCURRENT_INSTALL_CHANGE') elif row['action'] == 'upgrade': verify_hashes(target, row['before']) for name in row.get('additions', []): require(not plain_path(target / name).exists(), 'CONCURRENT_INSTALL_CHANGE') else: verify_component(target, files[row['skill']]) # Save every changed public file before replacing any file. Extra files stay untouched. for row in actions: if row['action'] != 'upgrade': continue require(backup_dir is not None, 'UPGRADE_BACKUP_REQUIRED') for name, raw in files[row['skill']].items(): if name not in row['before'] or digest(raw) == row['before'][name]: continue dest = backup_dir / row['skill'] / name dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes((skills_dir / row['skill'] / name).read_bytes()) (backup_dir / 'upgrade-plan.json').write_bytes(encode(row)) for row in actions: if row['action'] == 'create': target = skills_dir / row['skill'] require(not target.exists(), 'CONCURRENT_INSTALL_CHANGE') (staging / row['skill']).rename(target) created.append(target) elif row['action'] == 'upgrade': target = skills_dir / row['skill'] verify_hashes(target, row['before']) names = sorted(files[row['skill']], key=lambda n: n == 'SKILL.md') for name in names: raw = files[row['skill']][name] if name not in row['before']: dest = plain_path(target / name) require(not dest.exists(), 'CONCURRENT_INSTALL_CHANGE') dest.parent.mkdir(parents=True, exist_ok=True) # Exclusive creation preserves an unknown file that arrives after plan. with dest.open('xb') as stream: stream.write(raw) stream.flush() os.fsync(stream.fileno()) upgraded.append((dest, row['skill'], name, digest(raw))) continue if digest(raw) == row['before'][name]: continue dest = plain_path(target / name) require(digest(dest.read_bytes()) == row['before'][name], 'CONCURRENT_INSTALL_CHANGE') temporary = dest.with_name(dest.name + '.ksf-upgrade-new') require(not temporary.exists(), 'UPGRADE_TEMP_EXISTS') try: with temporary.open('xb') as stream: stream.write(raw) stream.flush() os.fsync(stream.fileno()) temporary.chmod(dest.stat().st_mode & 0o777) os.replace(temporary, dest) upgraded.append((dest, row['skill'], name, digest(raw))) finally: if temporary.exists(): temporary.unlink() for row in actions: verify_component(skills_dir / row['skill'], files[row['skill']]) except Exception: for dest, skill, name, after_hash in reversed(upgraded): require(digest(dest.read_bytes()) == after_hash, 'CONCURRENT_CHANGE_PREVENTS_RESTORE') prior = backup_dir / skill / name if prior.is_file(): dest.write_bytes(prior.read_bytes()) else: # Only the newly added public reference has no before-file. require(any(name in r.get('additions', []) and r['skill'] == skill for r in actions), 'MISSING_UPGRADE_BACKUP') dest.unlink() for target in reversed(created): verify_component(target, files[target.name]) remove_created(target, skills_dir) raise finally: if staging.exists(): remove_created(staging, skills_dir) def node_and_npm(): node = shutil.which('node') npm = shutil.which('npm') if not node or not npm: return None version = subprocess.run([node, '-p', 'process.versions.node'], capture_output=True, timeout=15, check=False) require(version.returncode == 0 and int(version.stdout.decode().strip().split('.')[0]) >= 18, 'NODE_18_REQUIRED') if os.name == 'nt': cli = Path(npm).parent / 'node_modules/npm/bin/npm-cli.js' require(cli.is_file(), 'NPM_CLI_NOT_FOUND') return [node, str(cli)] return [npm] def prepare_runtime(package, voc_dir, output_dir): npm = node_and_npm() if not npm: return {'status': 'pending', 'reason': 'NODE_OR_NPM_NOT_AVAILABLE'} # Dependency resolution affects this newly installed VOC directory only. command = npm + ['install', '--omit=dev', '--ignore-scripts', '--no-audit', '--no-fund', '--registry', 'https://registry.npmjs.org', '--cache', str(output_dir / 'npm-cache')] result = subprocess.run(command, cwd=voc_dir, capture_output=True, check=False, timeout=240) # Raw process logs are intentionally not copied into the user-facing receipt. if result.returncode: return {'status': 'pending', 'reason': 'VOC_DEPENDENCY_INSTALL_FAILED', 'exitCode': result.returncode} check = subprocess.run([shutil.which('node'), str(package / 'check-voc.cjs'), str(voc_dir)], cwd=package, capture_output=True, check=False, timeout=40) if check.returncode: return {'status': 'pending', 'reason': 'VOC_LOCAL_CHECK_FAILED', 'exitCode': check.returncode} value = json.loads(check.stdout) require(value.get('ok') is True, 'INVALID_VOC_CHECK_RESULT') return {'status': 'verified', 'check': value} def run(args, package=None): package = Path(package).resolve() if package else Path(__file__).resolve().parent require(Path(args.skills_dir).is_absolute(), 'ABSOLUTE_HOST_SKILLS_DIRECTORY_REQUIRED') skills_dir = plain_path(args.skills_dir) require(skills_dir != Path(skills_dir.anchor), 'FILESYSTEM_ROOT_NOT_ALLOWED') require(not skills_dir.exists() or skills_dir.is_dir(), 'SKILLS_DIRECTORY_REQUIRED') require(not package.is_relative_to(skills_dir) and not skills_dir.is_relative_to(package), 'PACKAGE_TARGET_OVERLAP') require(not args.prepare_runtime or args.apply, 'PREPARE_RUNTIME_REQUIRES_APPLY') if args.apply: require(args.out and Path(args.out).is_absolute(), 'ABSOLUTE_RESULT_DIRECTORY_REQUIRED') output = plain_path(args.out) require(not output.exists(), 'RESULT_DIRECTORY_EXISTS') require(not output.is_relative_to(skills_dir) and not skills_dir.is_relative_to(output), 'RESULT_TARGET_OVERLAP') meta = verify_package(package) files = source_files(package, meta, args.voc_archive) # The npm artifact contains the author's machine-specific MCP paths. # Adapt that one generated configuration after verifying the original artifact. original_mcp_sha = digest(files['voc-intelligence']['.mcp.json']) files['voc-intelligence']['.mcp.json'] = encode(mcp_fragment(skills_dir / 'voc-intelligence')) actions = plan(skills_dir, files, meta.get('upgradeFrom')) if args.apply and any(a['action'] == 'upgrade' for a in actions): require(getattr(args, 'upgrade', False), 'UPGRADE_FLAG_REQUIRED') receipt = {'schema': 'ksf.workshop-onboarding-result.v1', 'ok': True, 'releaseId': meta['releaseId'], 'readOnly': not args.apply, 'actions': actions, 'filesInstalled': False, 'hostSkillsLoaded': 'not-verified', 'vocRuntime': {'status': 'not-checked'}, 'hostToolRegistration': 'not-verified', 'vocLiveAccess': 'not-verified', 'researchLibraryAccess': 'not-verified', 'businessSaveAccess': 'not-verified', 'accountOrEnterpriseCreated': False, 'researchOrBusinessDataWritten': False, 'remoteInstalled': False, 'adaptedVocConfig': {'path': '.mcp.json', 'originalSha256': original_mcp_sha, 'installedSha256': digest(files['voc-intelligence']['.mcp.json'])}, 'note': 'Files, local runtime, host discovery and live access have separate acceptance.'} if not args.apply: return receipt output.mkdir(parents=True, exist_ok=False) try: install_files(skills_dir, files, actions, output / 'before') receipt['filesInstalled'] = True voc_dir = skills_dir / 'voc-intelligence' # A neutral fragment is written to the result directory only; the host owns registration. (output / 'voc-mcp.fragment.json').write_bytes(encode(mcp_fragment(voc_dir))) receipt['mcpFragment'] = str(output / 'voc-mcp.fragment.json') if args.prepare_runtime: try: receipt['vocRuntime'] = prepare_runtime(package, voc_dir, output) except subprocess.TimeoutExpired: receipt['vocRuntime'] = {'status': 'pending', 'reason': 'RUNTIME_PREPARATION_TIMEOUT'} else: receipt['vocRuntime'] = {'status': 'pending', 'reason': 'DEPENDENCIES_NOT_PREPARED'} except Exception as exc: receipt['ok'] = False receipt['errorCode'] = str(exc) if isinstance(exc, InstallError) else 'INSTALL_FAILED' raise finally: (output / 'install-receipt.json').write_bytes(encode(receipt)) return receipt def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--skills-dir', required=True, help='Actual Skills directory provided by the host') parser.add_argument('--out', help='New absolute result directory outside the Skills directory') parser.add_argument('--voc-archive', help='Optional local copy of the exact official npm tarball') parser.add_argument('--apply', action='store_true') parser.add_argument('--prepare-runtime', action='store_true') parser.add_argument('--upgrade', action='store_true', help='Upgrade an integrity-checked supported prior workshop version') args = parser.parse_args() try: receipt = run(args) print(json.dumps(receipt, ensure_ascii=False, indent=2)) return 0 except Exception as exc: code = str(exc) if isinstance(exc, InstallError) else 'INITIALIZATION_FAILED' print(json.dumps({'ok': False, 'errorCode': code, 'note': 'No credentials or raw environment were printed.'})) return 1 if __name__ == '__main__': sys.exit(main())