kangshifu-board/server/voc-library/library_receiver.py

"""Forced-command SSH receiver: fixed registry, source validation, no arbitrary commands."""
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 tempfile
import zipfile
import voc_db as v
import library_channel as channel
import agent_batch
import update_library

def hex_sha(value):
    v.require(isinstance(value,str) and re.fullmatch('[a-f0-9]{64}',value),'INVALID_FINGERPRINT');return value

def serve_config(path):
    data=json.loads(channel.private_file(path).read_text(encoding='utf-8-sig'))
    v.require(isinstance(data,dict) and set(data)=={'schema','registry','intake','canPublish'} and data['schema']==channel.VERSION and type(data['canPublish']) is bool,'INVALID_RECEIVER_CONFIG')
    for name in ('registry','intake'):
        v.require(isinstance(data[name],str) and Path(data[name]).is_absolute(),'ABSOLUTE_RECEIVER_PATH_REQUIRED')
        p=Path(data[name]).absolute()
        v.require(p.is_absolute() and p.is_dir() and not any(x.is_symlink() for x in (p,*p.parents)),'INVALID_RECEIVER_DIRECTORY')
        if os.name!='nt':v.require(p.stat().st_uid==os.getuid() and p.stat().st_mode&0o077==0,'RECEIVER_DIRECTORY_NOT_PRIVATE')
    registry,intake=Path(data['registry']).resolve(),Path(data['intake']).resolve()
    v.require(registry!=intake and not registry.is_relative_to(intake) and not intake.is_relative_to(registry),'SEPARATE_INTAKE_REQUIRED')
    return data

def read(config,request):
    v.require(set(request)=={'schema','action','command','params'} and request['command'] in channel.READS,'READ_COMMAND_REQUIRED')
    p=request['params'];v.require(isinstance(p,dict) and not set(p)-{'generation','table','profile','entity','id','input','explain'},'INVALID_READ_PARAMS')
    argv=[sys.executable,'-B','-X','utf8',str(Path(__file__).with_name('voc_db.py')),request['command'],'--registry',config['registry']]
    for k in ('generation','table','entity','id'):
        if k in p:
            v.require(isinstance(p[k],str) and not p[k].startswith('-') and len(p[k])<=1000,'INVALID_READ_VALUE');argv+=['--'+k,p[k]]
    for k in ('profile','explain'):
        if k in p:
            v.require(type(p[k]) is bool,'INVALID_READ_VALUE')
            if p[k]:argv+=['--'+k]
    if 'input' in p:argv+=['--input','-']
    env=dict(os.environ)
    for key in ('KSF_LIBRARY_CHANNEL','KSF_LIBRARY_REGISTRY','KSF_VOC_DB','PYTHONPATH','PYTHONSTARTUP'):env.pop(key,None)
    run=subprocess.run(argv,input=v.canonical(p.get('input')).encode('utf-8'),stdout=subprocess.PIPE,stderr=subprocess.PIPE,env=env,timeout=120,check=False)
    result=json.loads((run.stdout or run.stderr).decode('utf-8'));v.require(isinstance(result,dict) and type(result.get('ok')) is bool,'INVALID_READ_RESULT')
    if result['ok'] and request['command']=='contract':
        result['publicationBinding']={'mode':'ssh-channel','newGenerationsSupported':config['canPublish'],
            'remotePublisherEnabled':config['canPublish'],'currentGeneration':result['generation'],'taskAuthorizationStillRequired':True,
            'note':'Agent经受限SSH通道查询同一registry；新批次在后端重建校验后发布，不共享SQLite副本'}
    result['dataAccessMode']='ssh-channel';return result

def unpack(raw,root):
    with zipfile.ZipFile(io.BytesIO(raw)) as z:
        entries=z.infolist();names=set()
        v.require(0<len(entries)<=2001 and sum(e.file_size for e in entries)<=channel.MAX_EXPANDED,'INVALID_ARCHIVE_SIZE')
        for info in entries:
            name=info.filename;parts=PurePosixPath(name).parts
            v.require(name not in names and not info.is_dir() and not name.startswith('/') and '\\' not in name and ':' not in name and '..' not in parts and
                (name=='batch.json' or name.startswith('raw/')) and (info.external_attr>>16)&0o170000 in (0,0o100000),'UNSAFE_ARCHIVE_MEMBER')
            names.add(name)
        v.require('batch.json' in names,'BATCH_REQUIRED')
        for info in entries:
            dest=root/info.filename;dest.parent.mkdir(parents=True,exist_ok=True,mode=0o700)
            with dest.open('xb') as f:f.write(z.read(info))
            os.chmod(dest,0o600)
    path,manifest,batch,rows,assets=__import__('structured_batch').load(root/'batch.json')
    v.require(names=={'batch.json',*assets},'UNREFERENCED_ARCHIVE_FILES')
    return path

def current(config):
    pointer=json.loads((Path(config['registry'])/'current.json').read_text(encoding='utf-8'))
    return pointer

def status(config,request):
    v.require(set(request)=={'schema','action','params'} and isinstance(request['params'],dict) and set(request['params'])=={'archiveSha256'},'INVALID_STATUS_PARAMS')
    digest=hex_sha(request['params']['archiveSha256']);folder=Path(config['intake'])/digest
    pointer=current(config);result={'ok':True,'readOnly':True,'databaseWrites':0,'archiveSha256':digest,'currentGeneration':pointer['generation'],'publicationFound':False}
    staged=folder/'staging.json'
    if staged.is_file():
        meta=json.loads(staged.read_text(encoding='utf-8'))
        entry=next((x for x in pointer.get('history',[]) if x.get('incomingGeneration')==meta['generation']),None)
        if entry:result.update(publicationFound=True,publishedGeneration=entry['generation'],delta=entry.get('delta',{}))
    return result

def intake(config,request):
    v.require(config['canPublish'],'PUBLICATION_DISABLED')
    v.require(set(request)=={'schema','action','params','archiveBase64'},'INVALID_INTAKE_REQUEST')
    p=request['params'];v.require(isinstance(p,dict) and set(p)=={'archiveSha256','expectedGeneration','label','apply'} and type(p['apply']) is bool,'INVALID_INTAKE_PARAMS')
    digest=hex_sha(p['archiveSha256']);expected=hex_sha(p['expectedGeneration'])
    v.require(isinstance(p['label'],str) and 0<len(p['label'])<=200,'INVALID_LABEL')
    v.require(current(config)['generation']==expected,'LIBRARY_GENERATION_CHANGED')
    v.require(isinstance(request['archiveBase64'],str),'INVALID_ARCHIVE_ENCODING')
    raw=base64.b64decode(request['archiveBase64'],validate=True)
    v.require(len(raw)<=channel.MAX_ARCHIVE and v.sha(raw)==digest,'ARCHIVE_HASH_MISMATCH')
    root=Path(config['intake']);folder=root/digest
    if not p['apply']:
        with tempfile.TemporaryDirectory(prefix='.validate-',dir=root) as td:
            batch=unpack(raw,Path(td));plan=agent_batch.build_batch(batch,Path(td)/'staging.sqlite')
        return {**plan,'archiveSha256':digest,'temporaryFilesUsed':True,'persistentFilesWritten':False,'remotePublished':False,'currentGeneration':expected}
    if not folder.exists():
        with tempfile.TemporaryDirectory(prefix='.receive-',dir=root) as td:
            td=Path(td);unpack(raw,td)
            (td/'transport.zip').write_bytes(raw);os.chmod(td/'transport.zip',0o600)
            # Preserve validated sources before constructing/publishing; retries reuse identical bytes.
            td.rename(folder)
            td.mkdir()  # TemporaryDirectory removes only its empty original path.
    else:
        v.require(not folder.is_symlink() and (folder/'transport.zip').is_file() and agent_batch.digest(folder/'transport.zip')==digest,'INTAKE_ID_COLLISION')
    stage=folder/'staging.sqlite';receipt=folder/'staging.json'
    if stage.exists():
        v.require(receipt.is_file(),'STAGING_RECEIPT_REQUIRED')
        built=json.loads(receipt.read_text(encoding='utf-8'))
        v.require(agent_batch.digest(stage)==built['databaseSha256'],'STAGING_HASH_CHANGED')
    else:
        built=agent_batch.build_batch(folder/'batch.json',stage,True)
        with receipt.open('x',encoding='utf-8') as f:f.write(v.canonical(built)+'\n')
        os.chmod(receipt,0o600)
    update_library.verified(stage)
    published=update_library.publish(stage,config['registry'],True,p['label'],expected_generation=expected)
    return {**published,'archiveSha256':digest,'remotePublished':True,'sourceArchivePreserved':True,'productionImported':True,
        'databaseWrites':'new-generation-only','readOnly':False,'activeLibraryModified':bool(published.get('applied'))}

def dispatch(config,request):
    v.require(isinstance(request,dict) and request.get('schema')==channel.VERSION,'INVALID_PROTOCOL')
    action=request.get('action')
    if action=='read':return read(config,request)
    if action=='status':return status(config,request)
    if action=='intake':return intake(config,request)
    raise v.LibraryError('ACTION_NOT_ALLOWED')

def main():
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('--config',required=True);a=p.parse_args()
    applying=False
    try:
        v.require(os.environ.get('SSH_ORIGINAL_COMMAND') in (None,'','ksf-library-v1'),'SSH_COMMAND_NOT_ALLOWED')
        config=serve_config(a.config);wire=sys.stdin.buffer.read(channel.MAX_WIRE+1);v.require(len(wire)<=channel.MAX_WIRE,'CHANNEL_REQUEST_TOO_LARGE')
        request=json.loads(wire.decode('utf-8'));applying=request.get('action')=='intake' and request.get('params',{}).get('apply') is True
        result=dispatch(config,request);print(v.canonical(result));return 0 if result.get('ok') else 1
    except Exception as e:
        code=str(e) if isinstance(e,v.LibraryError) else 'RECEIVER_OPERATION_FAILED'
        print(v.canonical({'ok':False,'error':{'code':code},'readOnly':not applying,'databaseWrites':None if applying else 0,'publicationOutcome':'not-confirmed'}));return 1

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