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

"""Check the agent's own research and encrypted business entries; never save business records."""
import argparse
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import uuid

sys.dont_write_bytecode = True
from bootstrap import HERE, Stopped, atomic_create, encode, need, plain, private_json, sha, verify
import voc


def activate_research(data_dir):
    """Persist one non-secret variable, preserving all other .env bytes and a private backup."""
    file = plain(data_dir / '.env')
    before = file.read_bytes() if file.exists() else None
    expected = str(data_dir / 'ksf-library-channel/channel.json')
    lines = (before or b'').decode('utf-8-sig').splitlines()
    matches = [re.fullmatch(r'\s*(?:export\s+)?KSF_LIBRARY_CHANNEL\s*=\s*(.*?)\s*', x) for x in lines]
    values = [m.group(1).strip('\"\'') for m in matches if m]
    need(len(values) <= 1 and (not values or values[0] == expected), 'EXISTING_RESEARCH_BINDING_DIFFERS')
    if values:
        return False
    newline = b'\r\n' if b'\r\n' in (before or b'') else b'\n'
    raw = before or b''
    if raw and not raw.endswith(b'\n'):
        raw += newline
    raw += ('KSF_LIBRARY_CHANNEL=' + expected).encode() + newline
    if before is not None:
        backup = data_dir / 'ksf-enterprise-workshop' / ('env-before-' + uuid.uuid4().hex)
        atomic_create(backup, before)
    temp = file.with_name('.env.ksf-' + uuid.uuid4().hex)
    try:
        atomic_create(temp, raw)
        need((file.read_bytes() if file.exists() else None) == before, 'ENV_CONCURRENT_CHANGE')
        os.replace(temp, file)
    finally:
        if temp.exists():
            temp.unlink()
    return True


def run_json(argv, env, runner=subprocess.run):
    result = runner(argv, env=env, capture_output=True, timeout=180)
    try:
        value = json.loads(result.stdout if result.returncode == 0 else result.stderr)
    except (ValueError, UnicodeError):
        raise Stopped('ENTRY_RESPONSE_INVALID') from None
    if result.returncode != 0 or value.get('ok') is not True:
        code = value.get('error', {}).get('code', '')
        raise Stopped(code if re.fullmatch('[A-Z][A-Z0-9_]{1,90}', str(code)) else 'ENTRY_FAILED')
    return value


def check(data_dir, output, activate=False, runner=subprocess.run):
    result = verify(HERE, data_dir)
    need(sys.platform == 'linux', 'TARGET_LINUX_RUNTIME_REQUIRED')
    voc_result = voc.verify(HERE, data_dir, runner)
    env = dict(os.environ)
    env['KSF_LIBRARY_CHANNEL'] = str(data_dir / 'ksf-library-channel/channel.json')
    env['PYTHONDONTWRITEBYTECODE'] = '1'
    channel = private_json(Path(env['KSF_LIBRARY_CHANNEL']))
    need(channel.get('schema') == 'ksf.library-channel.v1', 'INVALID_CHANNEL_CONFIG')
    # The installed library enforces key/known_hosts ownership and pinned host-key checking.
    library = run_json([sys.executable, '-B', str(data_dir / 'skills/ksf-voc-library/scripts/query_voc.py'),
                        'contract'], env, runner)
    binding = library.get('publicationBinding', {})
    generation = library.get('generation')
    need(binding.get('mode') == 'ssh-channel' and binding.get('backend') == 'parse-postgresql'
         and re.fullmatch('[a-f0-9]{64}', str(generation)) is not None
         and binding.get('currentGeneration') == generation and library.get('readOnly') is True
         and library.get('databaseWrites') == 0, 'RESEARCH_CONTRACT_MISMATCH')
    need(channel.get('host') and channel.get('user'), 'INVALID_CHANNEL_CONFIG')
    request_file = output / ('contract-input-' + uuid.uuid4().hex + '.json')
    atomic_create(request_file, b'{}\n')
    workshop = run_json(['node', str(data_dir / 'skills/ksf-workshop-runtime/scripts/ksf-secure.mjs'),
                         'workshop-contract', '--input', str(request_file), '--request-id',
                         'enterprise-contract-' + uuid.uuid4().hex], env, runner)
    inner = workshop.get('entryResult', {})
    need(workshop.get('credentialLoadedForChild') is True
         and workshop.get('credentialWrittenToDisk') is False
         and inner.get('readOnly') is True and inner.get('result', {}).get('workshop', {}).get('version')
         == 'ksf.workshop.v1', 'WORKSHOP_CONTRACT_MISMATCH')
    changed = activate_research(data_dir) if activate else False
    return {**result, 'researchAuthorized': True, 'businessAuthorized': True,
            'voc': voc_result, 'vocCredentialAvailable': voc_result['runtime']['credentialAvailable'],
            'research': {'mode': binding['mode'], 'backend': binding['backend'], 'generation': generation,
                         'readOnly': True, 'databaseWrites': 0},
            'workshop': {'version': 'ksf.workshop.v1', 'readOnly': True},
            'defaultResearchEnvironmentUpdated': changed, 'hostEnvironmentReloadVerified': False,
            'realSaveReadbackVerified': False, 'dashboardAttributionVerified': False}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--data-dir', type=Path, required=True)
    parser.add_argument('--out', type=Path, required=True)
    parser.add_argument('--activate-research', action='store_true')
    args = parser.parse_args()
    need(args.data_dir.is_absolute() and args.out.is_absolute(), 'ABSOLUTE_PATH_REQUIRED')
    data_dir, output = plain(args.data_dir), plain(args.out)
    need(output not in data_dir.parents and output != data_dir
         and data_dir / 'skills' not in output.parents, 'UNSAFE_RESULT_DIRECTORY')
    output.mkdir(parents=True, exist_ok=True, mode=0o700)
    result = check(data_dir, output, args.activate_research)
    atomic_create(output / ('readiness-' + 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) as error:
        print(json.dumps({'ok': False, 'error': {'code': str(error) if isinstance(error, Stopped)
                                               else 'READINESS_STOPPED'}, 'businessWrites': 0}), file=sys.stderr)
        sys.exit(1)