kangshifu-board/source/install.py

"""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 plan(skills_dir, files):
    actions = []
    for name, content in files.items():
        target = plain_path(skills_dir / name)
        if target.exists():
            verify_component(target, content)
            action = 'verify-only'
        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):
    skills_dir.mkdir(parents=True, exist_ok=True)
    staging = Path(tempfile.mkdtemp(prefix='.ksf-init-', dir=skills_dir))
    created = []
    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')
            else:
                verify_component(target, files[row['skill']])
        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)
        for row in actions:
            verify_component(skills_dir / row['skill'], files[row['skill']])
    except Exception:
        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)
    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)
        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')
    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())