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

"""Install a sealed enterprise workshop without copying credentials or identities."""
import argparse
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import shutil
import sys
import tempfile
import uuid

sys.dont_write_bytecode = True
HERE = Path(__file__).absolute().parent
COMPANY = 'FKieRbIoGt'
APP = 'FKieRbIoGt_main'


class Stopped(Exception):
    pass


def need(value, code):
    if not value:
        raise Stopped(code)


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def encode(value):
    return (json.dumps(value, ensure_ascii=False, indent=2) + '\n').encode('utf-8')


def plain(value):
    p = Path(os.path.abspath(value))
    for part in [*reversed(p.parents), p]:
        need(not part.is_symlink(), 'SYMLINK_PATH')
        need(not getattr(part, 'is_junction', lambda: False)(), 'JUNCTION_PATH')
    return p


def safe_name(name):
    return (isinstance(name, str) and bool(name) and '\\' not in name and ':' not in name
            and not name.startswith('/') and name == '/'.join(PurePosixPath(name).parts)
            and all(x not in ('.', '..') for x in PurePosixPath(name).parts))


def private_json(file):
    file = plain(file)
    need(file.is_file(), 'PRIVATE_FILE_MISSING')
    stat = file.stat()
    if os.name != 'nt':
        need(stat.st_uid == os.geteuid() and stat.st_mode & 0o077 == 0, 'PRIVATE_FILE_PERMISSIONS')
    return json.loads(file.read_bytes())


def sealed(root):
    root = plain(root)
    listed = {}
    for line in (root / 'FILES.sha256').read_text('utf-8').splitlines():
        m = re.fullmatch(r'([a-f0-9]{64})  (.+)', line)
        need(m is not None, 'INVALID_MANIFEST')
        digest, name = m.groups()
        need(safe_name(name) and name not in listed, 'INVALID_MANIFEST_PATH')
        p = plain(root / name)
        need(p.is_file() and sha(p.read_bytes()) == digest, 'PACKAGE_HASH_MISMATCH')
        listed[name] = digest
    actual = set()
    for p in root.rglob('*'):
        if p.relative_to(root).parts[0] == '.git':
            continue
        plain(p)
        need(p.is_file() or p.is_dir(), 'SPECIAL_PACKAGE_FILE')
        if p.is_file():
            actual.add(p.relative_to(root).as_posix())
    need(actual == set(listed) | {'FILES.sha256'}, 'UNDECLARED_PACKAGE_FILE')
    cfg = json.loads((root / 'enterprise.json').read_bytes())
    need(cfg['schema'] == 'ksf.enterprise-workshop.v1' and cfg['companyObjectId'] == COMPANY
         and cfg['appId'] == APP, 'PACKAGE_ENTERPRISE_MISMATCH')
    need(cfg['functionsURL'] == 'https://kangshifu.fmode.cn/api/functions'
         and cfg['serverURL'] == 'https://kangshifu.fmode.cn/data', 'PACKAGE_TARGET_MISMATCH')
    entry = 'payload/skills/ksf-workshop-runtime/SKILL.md'
    need(listed.get(entry) == cfg['skill']['entrySha256'], 'SKILL_ENTRY_MISMATCH')
    return cfg, listed


def candidate(root, listed):
    files = {}
    for name in listed:
        if name.startswith('payload/'):
            relative = name[len('payload/'):]
            need(relative.startswith(('skills/ksf-workshop-runtime/', 'skills/ksf-voc-library/scripts/')),
                 'UNEXPECTED_PAYLOAD')
            files[relative] = (root / name).read_bytes()
    skill = {n: b for n, b in files.items() if n.startswith('skills/ksf-workshop-runtime/')}
    need(len(skill) == 45 and len(files) > 45, 'INCOMPLETE_PAYLOAD')
    need(not any('credential-binding' in n or 'channel.json' in n for n in files), 'PRIVATE_PAYLOAD')
    return files


def check_enterprise(current, expected):
    need(isinstance(current, dict), 'INVALID_ENTERPRISE_CONFIGURATION')
    need(current.get('companyObjectId') == expected['companyObjectId']
         and current.get('appId') == expected['appId'], 'ENTERPRISE_MISMATCH')
    need(all(isinstance(current.get(k), str) and current[k].rstrip('/') == expected[k]
             for k in ('serverURL', 'functionsURL')),
         'ENTERPRISE_ENDPOINT_MISMATCH')
    need(current.get('provisionStatus') == 'success', 'ENTERPRISE_NOT_PROVISIONED')


def tree_snapshot(directory):
    if not directory.exists():
        return None
    need(directory.is_dir(), 'NOT_DIRECTORY')
    result = {}
    for p in directory.rglob('*'):
        plain(p)
        need(p.is_file() or p.is_dir(), 'SPECIAL_TARGET_FILE')
        if p.is_file():
            result[p.relative_to(directory).as_posix()] = [sha(p.read_bytes()), p.stat().st_mode & 0o777]
    return result


def plan(root, data_dir):
    root, data_dir = plain(root), plain(data_dir)
    need(data_dir.is_absolute() and data_dir != Path(data_dir.anchor), 'INVALID_DATA_DIR')
    need(root != data_dir and root not in data_dir.parents
         and not any(target == root or target in root.parents or root in target.parents
                     for target in (data_dir / 'skills', data_dir / '.fmode',
                                    data_dir / 'ksf-enterprise-workshop')),
         'PACKAGE_TARGET_OVERLAP')
    cfg, listed = sealed(root)
    current_path = data_dir / '.fmode/config/enterprise/current.json'
    current = private_json(current_path)
    check_enterprise(current, cfg)
    files = candidate(root, listed)
    targets = sorted({n.split('/')[1] for n in files})
    if (data_dir / 'skills/ksf-voc-library/SKILL.md').exists():
        raise Stopped('RESEARCH_SKILL_ENTRYPOINT_STILL_ACTIVE')
    snapshots = {n: tree_snapshot(plain(data_dir / 'skills' / n)) for n in targets}
    changes = []
    for name, raw in files.items():
        dest = plain(data_dir / name)
        need(not dest.exists() or dest.is_file(), 'TARGET_NOT_FILE')
        if dest.exists():
            need(sha(dest.read_bytes()) == sha(raw), 'EXISTING_COMPONENT_DIFFERS')
        else:
            # Existing directories may carry private local files, but cannot be a partial/unknown version.
            need(snapshots[name.split('/')[1]] is None, 'INCOMPLETE_EXISTING_COMPONENT')
            changes.append(name)
    binding = encode({'schema': 'ksf.enterprise-workshop-binding.v1', 'companyObjectId': COMPANY,
                      'appId': APP, 'configurationSha256': sha((root / 'enterprise.json').read_bytes()),
                      'packageManifestSha256': sha((root / 'FILES.sha256').read_bytes()),
                      'skill': 'ksf-workshop-runtime', 'version': 'r13'})
    binding_path = plain(data_dir / 'ksf-enterprise-workshop/binding.json')
    binding_before = sha(binding_path.read_bytes()) if binding_path.exists() else None
    if binding_path.exists():
        prior = json.loads((root / 'previous-release.json').read_bytes())
        need(binding_before in (sha(binding), prior['previousBindingSha256']), 'ENTERPRISE_BINDING_CHANGED')
    return {'schema': 'ksf.enterprise-workshop-plan.v1', 'companyObjectId': COMPANY, 'appId': APP,
            'dataDir': str(data_dir), 'packageManifestSha256': sha((root / 'FILES.sha256').read_bytes()),
            'currentPublicFields': {k: current[k] for k in ('companyObjectId', 'appId', 'serverURL', 'functionsURL')},
            'snapshots': snapshots, 'changes': changes, 'bindingExists': binding_path.exists(),
            'bindingSha256': sha(binding), 'bindingBeforeSha256': binding_before,
            'bindingUpdateRequired': binding_before != sha(binding), 'businessWrites': 0}


def atomic_create(file, raw, mode=0o600):
    plain(file)
    file.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    with file.open('xb') as stream:
        os.chmod(file, mode)
        stream.write(raw)
        stream.flush()
        os.fsync(stream.fileno())


def apply(root, data_dir, expected_plan):
    root, data_dir = plain(root), plain(data_dir)
    observed = plan(root, data_dir)
    if not observed['changes'] and observed['bindingExists'] and not observed['bindingUpdateRequired']:
        need(observed['packageManifestSha256'] == expected_plan['packageManifestSha256']
             and observed['dataDir'] == expected_plan['dataDir']
             and observed['currentPublicFields'] == expected_plan['currentPublicFields'], 'PLAN_CHANGED')
        return {**verify(root, data_dir), 'installedFiles': 0, 'applied': False}
    need(observed == expected_plan, 'PLAN_CHANGED')
    _, listed = sealed(root)
    files = candidate(root, listed)
    lock = plain(data_dir / '.ksf-enterprise-bootstrap.lock')
    created = []
    created_snapshots = {}
    replaced_binding = None
    descriptor = os.open(lock, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    os.close(descriptor)
    try:
        need(plan(root, data_dir) == expected_plan, 'PLAN_CHANGED')
        # Existing components are verify-only. All new component trees are staged before any rename.
        with tempfile.TemporaryDirectory(prefix='.ksf-enterprise-stage-', dir=data_dir) as tmp:
            staging = Path(tmp)
            for relative in expected_plan['changes']:
                atomic_create(staging / relative, files[relative], 0o644)
            for name, before in expected_plan['snapshots'].items():
                target = plain(data_dir / 'skills' / name)
                need(tree_snapshot(target) == before, 'CONCURRENT_INSTALL')
                if before is None:
                    target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
                    created_snapshots[target] = tree_snapshot(staging / 'skills' / name)
                    os.rename(staging / 'skills' / name, target)
                    created.append(target)
            bind = plain(data_dir / 'ksf-enterprise-workshop/binding.json')
            if expected_plan['bindingUpdateRequired']:
                raw = encode({'schema': 'ksf.enterprise-workshop-binding.v1', 'companyObjectId': COMPANY,
                              'appId': APP, 'configurationSha256': sha((root / 'enterprise.json').read_bytes()),
                              'packageManifestSha256': expected_plan['packageManifestSha256'],
                              'skill': 'ksf-workshop-runtime', 'version': 'r13'})
                if not bind.exists():
                    atomic_create(bind, raw)
                    created.append(bind)
                else:
                    before = bind.read_bytes()
                    need(sha(before) == expected_plan['bindingBeforeSha256'], 'CONCURRENT_INSTALL')
                    atomic_create(bind.with_name('binding-before-' + uuid.uuid4().hex + '.json'), before)
                    temp_bind = bind.with_name('.binding-' + uuid.uuid4().hex)
                    atomic_create(temp_bind, raw)
                    try:
                        need(bind.read_bytes() == before, 'CONCURRENT_INSTALL')
                        os.replace(temp_bind, bind)
                        replaced_binding = (bind, before)
                    finally:
                        if temp_bind.exists():
                            temp_bind.unlink()
        result = verify(root, data_dir)
        return {**result, 'installedFiles': len(expected_plan['changes']), 'applied': True,
                'existingComponentsModified': False, 'identitiesCopied': False,
                'credentialsCopied': False, 'vocModified': False}
    except Exception:
        if replaced_binding:
            bind, before = replaced_binding
            if sha(bind.read_bytes()) == expected_plan['bindingSha256']:
                rollback = bind.with_name('.binding-rollback-' + uuid.uuid4().hex)
                atomic_create(rollback, before)
                os.replace(rollback, bind)
        # Only remove newly created, still-identical payload files. Preserve concurrent changes.
        for p in reversed(created):
            if p.is_file():
                if sha(p.read_bytes()) == expected_plan['bindingSha256']:
                    p.unlink()
            elif p.is_dir() and tree_snapshot(p) == created_snapshots[p]:
                need(p.parent == data_dir / 'skills', 'ROLLBACK_PATH_OUTSIDE_TARGET')
                shutil.rmtree(p)
        raise
    finally:
        lock.unlink()


def verify(root, data_dir):
    result = plan(root, data_dir)
    need(not result['changes'] and result['bindingExists'] and not result['bindingUpdateRequired'],
         'INSTALLATION_INCOMPLETE')
    return {'ok': True, 'companyObjectId': COMPANY, 'appId': APP, 'version': 'r13',
            'filesVerified': len(candidate(plain(root), sealed(root)[1])), 'businessWrites': 0,
            'agentLoaded': False, 'researchAuthorized': False, 'businessAuthorized': False,
            'participantBound': False, 'automaticLifecycleHookVerified': False}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('command', choices=['plan', 'apply', 'verify'])
    parser.add_argument('--data-dir', required=True, type=Path)
    parser.add_argument('--out', required=True, type=Path)
    args = parser.parse_args()
    need(args.data_dir.is_absolute() and args.out.is_absolute(), 'ABSOLUTE_PATH_REQUIRED')
    data_dir, out = plain(args.data_dir), plain(args.out)
    need(out != data_dir and out not in data_dir.parents and not (data_dir / 'skills') in out.parents,
         'UNSAFE_RESULT_DIRECTORY')
    out.mkdir(parents=True, exist_ok=True, mode=0o700)
    if args.command == 'plan':
        record = plan(HERE, data_dir)
        atomic_create(out / 'plan.json', encode(record))
        result = {'ok': True, 'readOnly': True, 'changes': len(record['changes']), 'businessWrites': 0}
    elif args.command == 'apply':
        result = apply(HERE, data_dir, private_json(out / 'plan.json'))
        atomic_create(out / ('apply-' + uuid.uuid4().hex + '.json'), encode(result))
    else:
        result = verify(HERE, data_dir)
        atomic_create(out / ('verify-' + 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) as error:
        print(json.dumps({'ok': False, 'error': {'code': str(error) if isinstance(error, Stopped)
                                               else 'BOOTSTRAP_STOPPED'}, 'businessWrites': 0}), file=sys.stderr)
        sys.exit(1)