kangshifu-board/docs/unified-database-migration-20261109/release-r2/package/components/research-runtime/scripts/library_channel.py

"""Private SSH stdio channel to the authoritative library. No HTTP write endpoint."""
import sys
sys.dont_write_bytecode = True
import argparse
import base64
import io
import json
import os
from pathlib import Path, PurePosixPath
import re
import subprocess
import zipfile
import voc_db as v

VERSION = 'ksf.library-channel.v1'
READS = ('contract','schema','query','search','aggregate','facets','get','source','verify')
MAX_ARCHIVE = 64 * 1024 * 1024
MAX_EXPANDED = 256 * 1024 * 1024
MAX_WIRE = 90 * 1024 * 1024


class ChannelError(v.LibraryError):
    """Only bounded metadata may leave the channel; never echo SSH stderr/config."""
    def __init__(self, code, **diagnostics):
        super().__init__(code)
        self.diagnostics = diagnostics


def read_input(value):
    """--input is a UTF-8 JSON filename or stdin, not an inline JSON argument."""
    if value == '-':
        text = sys.stdin.read()
    else:
        v.require(not value.lstrip().startswith(('{', '[')), 'READ_INPUT_PATH_EXPECTED')
        try:
            text = Path(value).read_text(encoding='utf-8-sig')
        except FileNotFoundError:
            raise v.LibraryError('READ_INPUT_FILE_NOT_FOUND')
        except UnicodeError:
            raise v.LibraryError('READ_INPUT_INVALID_UTF8')
        except OSError:
            raise v.LibraryError('READ_INPUT_FILE_UNREADABLE')
    try:
        result = json.loads(text.lstrip('\ufeff'))
    except ValueError:
        raise v.LibraryError('READ_INPUT_INVALID_JSON')
    v.require(isinstance(result, dict), 'READ_INPUT_OBJECT_REQUIRED')
    return result

def private_file(path):
    p = Path(path).absolute()
    v.require(p.is_file() and not any(x.is_symlink() for x in (p,*p.parents)), 'CHANNEL_FILE_NOT_FOUND_OR_REDIRECTED')
    if os.name != 'nt':
        v.require(p.stat().st_uid == os.getuid() and p.stat().st_mode & 0o077 == 0, 'CHANNEL_FILE_NOT_PRIVATE')
    return p

def config(path):
    data = json.loads(private_file(path).read_text(encoding='utf-8-sig'))
    v.require(isinstance(data,dict) and set(data)=={'schema','host','port','user','identityFile','knownHostsFile'} and data['schema']==VERSION, 'INVALID_CHANNEL_CONFIG')
    v.require(isinstance(data['host'],str) and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9.:-]{0,252}',data['host']), 'INVALID_CHANNEL_HOST')
    v.require(isinstance(data['user'],str) and re.fullmatch(r'[a-z_][a-z0-9_-]{0,31}',data['user']), 'INVALID_CHANNEL_USER')
    v.require(type(data['port']) is int and 0<data['port']<65536, 'INVALID_CHANNEL_PORT')
    for k in ('identityFile','knownHostsFile'):
        v.require(isinstance(data[k],str) and Path(data[k]).is_absolute(), 'ABSOLUTE_CHANNEL_PATH_REQUIRED')
        private_file(data[k])
    return data

def send(path, request, runner=subprocess.run):
    data = config(path)
    command = ['ssh','-T','-o','BatchMode=yes','-o','IdentitiesOnly=yes','-o','StrictHostKeyChecking=yes',
        '-o','ClearAllForwardings=yes','-o','ConnectTimeout=10','-o','ServerAliveInterval=15','-o','ServerAliveCountMax=2',
        '-o','UserKnownHostsFile='+data['knownHostsFile'],'-i',data['identityFile'],'-p',str(data['port']),
        data['user']+'@'+data['host'],'ksf-library-v1']
    wire=v.canonical({'schema':VERSION,**request}).encode('utf-8')
    v.require(len(wire)<=MAX_WIRE,'CHANNEL_REQUEST_TOO_LARGE')
    try:
        run=runner(command,input=wire,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=300,check=False)
    except subprocess.TimeoutExpired:
        raise ChannelError('CHANNEL_RESULT_UNCERTAIN' if request.get('params',{}).get('apply') else 'CHANNEL_TIMEOUT', stage='ssh-execution')
    except OSError:
        raise ChannelError('CHANNEL_RESULT_UNCERTAIN' if request.get('params',{}).get('apply') else 'CHANNEL_UNAVAILABLE', stage='ssh-execution')
    diagnostics = dict(stage='ssh-response', exitCode=run.returncode,
        stdoutBytes=len(run.stdout), stderrBytes=len(run.stderr),
        stdoutSha256=v.sha(run.stdout), stderrSha256=v.sha(run.stderr))
    applying = request.get('params',{}).get('apply') is True
    if len(run.stdout)>32*1024*1024:
        raise ChannelError('CHANNEL_RESULT_UNCERTAIN' if applying else 'CHANNEL_RESPONSE_TOO_LARGE', **diagnostics)
    try:
        result=json.loads(run.stdout.decode('utf-8-sig'))
    except (ValueError,UnicodeError):
        raise ChannelError('CHANNEL_RESULT_UNCERTAIN' if applying else 'CHANNEL_READ_FAILED', **diagnostics)
    if not isinstance(result,dict) or type(result.get('ok')) is not bool:
        raise ChannelError('CHANNEL_RESULT_UNCERTAIN' if applying else 'INVALID_CHANNEL_RESPONSE', **diagnostics)
    if run.returncode!=0 and result['ok']:
        raise ChannelError('CHANNEL_RESULT_UNCERTAIN' if applying else 'CHANNEL_EXIT_MISMATCH', **diagnostics)
    return result

def pack(batch_path):
    from structured_batch import load
    path,manifest,batch,rows,assets=load(batch_path)
    v.require(len(manifest)+sum(p.stat().st_size for p,h in assets.values())<=MAX_EXPANDED,'CHANNEL_BATCH_TOO_LARGE')
    output=io.BytesIO()
    with zipfile.ZipFile(output,'w',zipfile.ZIP_DEFLATED) as z:
        def add(name,body):
            info=zipfile.ZipInfo(name,(2026,1,1,0,0,0));info.external_attr=0o100600<<16
            z.writestr(info,body,compress_type=zipfile.ZIP_DEFLATED)
        v.require(path.read_bytes()==manifest,'BATCH_CHANGED_DURING_PACK');add('batch.json',manifest)
        for name,(p,h) in sorted(assets.items()):
            raw=p.read_bytes();v.require(v.sha(raw)==h,'SOURCE_CHANGED_DURING_PACK');add(name,raw)
    raw=output.getvalue();v.require(len(raw)<=MAX_ARCHIVE,'CHANNEL_ARCHIVE_TOO_LARGE')
    return raw,v.sha(raw)

def read_main(argv):
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('command',choices=READS)
    p.add_argument('--channel',default=os.environ.get('KSF_LIBRARY_CHANNEL'));p.add_argument('--generation')
    p.add_argument('--table');p.add_argument('--profile',action='store_true');p.add_argument('--explain',action='store_true')
    p.add_argument('--entity');p.add_argument('--id');p.add_argument('--input')
    a=p.parse_args(argv)
    try:
        v.require(a.channel,'CHANNEL_NOT_CONFIGURED')
        params={k:getattr(a,k) for k in ('generation','table','entity','id') if getattr(a,k) is not None}
        if a.profile:params['profile']=True
        if a.explain:params['explain']=True
        if a.input:params['input']=read_input(a.input)
        result=send(a.channel,{'action':'read','command':a.command,'params':params})
        print(json.dumps(result,ensure_ascii=False,indent=2));return 0 if result['ok'] else 1
    except (ValueError,OSError) as e:
        code=str(e) if isinstance(e,v.LibraryError) else 'CHANNEL_READ_FAILED'
        error={'code':code}
        if isinstance(e,ChannelError):error['diagnostics']=e.diagnostics
        print(v.canonical({'ok':False,'error':error,'readOnly':True,'databaseWrites':0}),file=sys.stderr);return 1

def main():
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('command',choices=['publish','correct-review','status'])
    p.add_argument('--channel',default=os.environ.get('KSF_LIBRARY_CHANNEL'));p.add_argument('--batch');p.add_argument('--archive-sha256')
    p.add_argument('--expected-generation');p.add_argument('--label',default='Agent资料入库');p.add_argument('--apply',action='store_true')
    a=p.parse_args();fingerprint=None
    try:
        v.require(a.channel,'CHANNEL_NOT_CONFIGURED')
        if a.command=='status':
            v.require(not a.batch and not a.apply and a.archive_sha256,'STATUS_ARGUMENTS_REQUIRED')
            fingerprint=a.archive_sha256;request={'action':'status','params':{'archiveSha256':fingerprint}}
        else:
            v.require(a.batch and a.expected_generation and not a.archive_sha256,'BATCH_AND_EXPECTED_GENERATION_REQUIRED')
            raw,fingerprint=pack(a.batch)
            request={'action':'correct-review' if a.command=='correct-review' else 'intake',
                'params':{'archiveSha256':fingerprint,'expectedGeneration':a.expected_generation,'label':a.label,'apply':a.apply},'archiveBase64':base64.b64encode(raw).decode('ascii')}
        result=send(a.channel,request);print(json.dumps(result,ensure_ascii=False,indent=2));return 0 if result['ok'] else 1
    except (ValueError,OSError) as e:
        code=str(e) if isinstance(e,v.LibraryError) else 'CHANNEL_REQUEST_FAILED'
        print(v.canonical({'ok':False,'error':{'code':code},'archiveSha256':fingerprint,'publicationOutcome':'unknown' if code=='CHANNEL_RESULT_UNCERTAIN' else 'not-confirmed'}),file=sys.stderr);return 1

if __name__=='__main__':sys.exit(main())