kangshifu-board/deploy/enterprise-workshop/voc.py

"""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)