"""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='🪪')