kangshifu-board/integrations/wecom-identity/workshop_registration.py

"""No-argument in-process registration, composed with the reviewed aibot bridge.

No identity JSON is accepted from model tool arguments. The Node worker reuses
the existing Skill key/profile code; its stdin is private transport, not proof.
This does not isolate arbitrary same-UID terminal code from local state.
"""
from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path
import re
import shutil
import stat
import subprocess

TOOL_NAME = 'ksf_register_current_employee'
WORKER = Path('/opt/data/skills/ksf-workshop-runtime/scripts/ksf-identity-bridge.mjs')
TOOL_SCHEMA = {
    'name': TOOL_NAME,
    'description': '登记当前企微消息发送者，返回员工档案键。无需参数；从本回合可信上下文读取，失败停止，不接受身份转述。只写本地档案，不建会话或写业务库。',
    'parameters': {'type': 'object', 'properties': {}, 'additionalProperties': False},
}


class RegistrationError(Exception):
    pass


def _check_private_file(path):
    # Permissions are defensive checks, not a same-UID provenance guarantee.
    for parent in (path, *path.parents):
        if parent.is_symlink():
            raise RegistrationError('REGISTRATION_CONFIG_UNSAFE')
    s = path.lstat()
    if not stat.S_ISREG(s.st_mode):
        raise RegistrationError('REGISTRATION_CONFIG_UNSAFE')
    if os.name == 'posix' and (s.st_uid != os.geteuid() or stat.S_IMODE(s.st_mode) != 0o600):
        raise RegistrationError('REGISTRATION_CONFIG_UNSAFE')


def _read_maintenance_config():
    home = Path(os.environ.get('HERMES_HOME') or Path.home())
    path = home / 'plugins' / 'wecom-identity' / 'registration.json'
    _check_private_file(path)
    if path.stat().st_size > 4096:
        raise RegistrationError('REGISTRATION_CONFIG_INVALID')
    data = json.loads(path.read_text(encoding='utf-8'))
    if not isinstance(data, dict) or set(data) != {'version', 'agentProfileKey', 'botIdSha256'}:
        raise RegistrationError('REGISTRATION_CONFIG_INVALID')
    if data['version'] != 'ksf.current-employee-registration.v1':
        raise RegistrationError('REGISTRATION_CONFIG_INVALID')
    if not isinstance(data['agentProfileKey'], str) or not re.fullmatch(r'agent-account:[a-f0-9]{64}', data['agentProfileKey']):
        raise RegistrationError('REGISTRATION_CONFIG_INVALID')
    if not isinstance(data['botIdSha256'], str) or not re.fullmatch(r'[a-f0-9]{64}', data['botIdSha256']):
        raise RegistrationError('REGISTRATION_CONFIG_INVALID')
    return data


def _current_identity():
    from plugins.platforms.wecom.identity_bridge import current_identity
    return current_identity()  # No model args, reader override, or message-id override.


def _receipt(result, agent_key):
    # Reconstruct the public receipt; do not echo worker stderr or raw account fields.
    if not isinstance(result, dict) or result.get('ok') is not True or result.get('agentProfileKey') != agent_key:
        raise RegistrationError('REGISTRATION_RECEIPT_INVALID')
    p = result.get('profile')
    if not isinstance(p, dict) or not re.fullmatch(r'participant-account:[a-f0-9]{64}', str(p.get('key', ''))):
        raise RegistrationError('REGISTRATION_RECEIPT_INVALID')
    if p.get('identityBasis') != 'channel_context' or not isinstance(p.get('name'), str) or not p['name'].strip():
        raise RegistrationError('REGISTRATION_RECEIPT_INVALID')
    if not isinstance(result.get('changed'), bool) or not isinstance(result.get('needsNameBinding'), bool):
        raise RegistrationError('REGISTRATION_RECEIPT_INVALID')
    iid = result.get('invocationId', '')
    if not isinstance(iid, str) or not re.fullmatch(r'[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}', iid):
        raise RegistrationError('REGISTRATION_RECEIPT_INVALID')
    return {
        'ok': True, 'operation': 'current-employee-registered', 'invocationId': iid,
        'changed': result['changed'], 'agentProfileKey': agent_key,
        'profile': {'key': p['key'], 'name': p['name'], 'identityBasis': 'channel_context'},
        'needsNameBinding': result['needsNameBinding'], 'currentMessageChecked': True,
        'accountAuthenticated': False, 'displayNameAuthenticated': False,
        'networkRequests': 0, 'databaseWrites': 0,
    }


def _register_current_employee_result(args=None, **_kw):
    if args is not None and (not isinstance(args, dict) or args):
        return {'ok': False, 'error': 'NO_ARGUMENTS_ALLOWED'}
    try:
        checked = _current_identity()
        if not isinstance(checked, dict) or checked.get('ok') is not True:
            return {'ok': False, 'error': 'IDENTITY_CONTEXT_UNAVAILABLE'}
        identity = checked.get('identity')
        if not isinstance(identity, dict) or not isinstance(identity.get('botId'), str):
            raise RegistrationError('INVALID_CHANNEL_CONTEXT')
        config = _read_maintenance_config()
        if hashlib.sha256(identity['botId'].encode('utf-8')).hexdigest() != config['botIdSha256']:
            raise RegistrationError('REGISTRATION_INSTANCE_MISMATCH')
        node = shutil.which('node')
        if not node or not WORKER.is_file() or WORKER.is_symlink():
            raise RegistrationError('REGISTRATION_WORKER_UNAVAILABLE')
        request = {'version': config['version'], 'agentProfileKey': config['agentProfileKey'], 'identity': identity}
        proc = subprocess.run([node, str(WORKER)], input=json.dumps(request, ensure_ascii=False),
                              capture_output=True, text=True, encoding='utf-8', timeout=20, check=False)
        if proc.returncode != 0:
            raise RegistrationError('REGISTRATION_WORKER_FAILED')
        if len(proc.stdout) > 16384:
            raise RegistrationError('REGISTRATION_RECEIPT_INVALID')
        return _receipt(json.loads(proc.stdout), config['agentProfileKey'])
    except RegistrationError as exc:
        return {'ok': False, 'error': str(exc)}
    except Exception:
        # Exceptions can contain config/worker output; no raw exception logging.
        return {'ok': False, 'error': 'IDENTITY_REGISTRATION_FAILED'}


def register_current_employee(args=None, **_kw):
    """Hermes tool boundary accepts text; internal callers keep structured data."""
    return json.dumps(_register_current_employee_result(args, **_kw), ensure_ascii=False)


def register(ctx):
    ctx.register_tool(TOOL_NAME, 'wecom', TOOL_SCHEMA, register_current_employee,
                      description=TOOL_SCHEMA['description'], emoji='🪪')