# work_env.dspy - 工作环境管理 API(沙箱本地/远程切换 + 目录迁移) # action: get / set / ensure_remote_bwrap user_id = await get_user() if not user_id: return json.dumps({'ok': False, 'error': '未登录'}, ensure_ascii=False) action = (params_kw or {}).get('action', 'get') from pipeline_service.work_env import ( get_work_env, set_work_env, ensure_remote_bwrap, get_org_pubkey, ) from pipeline_service.deploy_account import account_name_for, sanitize_username from app_audit.audit_service import audit_log client_ip = '' try: client_ip = request.get('client_ip', '') or '' except Exception: pass # 查用户 username(账号名 ag_)和 org_id username = user_id org_id = '' try: async with get_sor_context(request._run_ns, 'pipeline') as _sor: _u = await _sor.sqlExe("SELECT username, orgid FROM users WHERE id=${u}$", {'u': user_id}) if _u: username = getattr(_u[0], 'username', user_id) or user_id org_id = getattr(_u[0], 'orgid', '') or '' except Exception: pass if action == 'get': _ot = (params_kw or {}).get('owner_type', '') try: async with get_sor_context(request._run_ns, 'pipeline') as sor: if _ot == 'org': env = await get_work_env(sor, '', org_id) elif _ot == 'user': env = await get_work_env(sor, user_id, '') else: env = await get_work_env(sor, user_id, org_id) return json.dumps({'ok': True, 'env': env, 'account_name': account_name_for(username)}, ensure_ascii=False) except Exception as e: return json.dumps({'ok': False, 'error': str(e)}, ensure_ascii=False) elif action == 'set': owner_type = (params_kw or {}).get('owner_type', 'user') mode = (params_kw or {}).get('mode', 'local') account_name = (params_kw or {}).get('account_name', '') or account_name_for(username) # owner_id 与权限校验 owner_id = (params_kw or {}).get('owner_id', '') if owner_type == 'user': # 用户级:强制只能设自己,不允许设别人 owner_id = user_id elif owner_type == 'org': # 机构级:仅平台超管(owner.superuser)可设置,且只能设自己所在机构 owner_id = org_id if not owner_id: return json.dumps({'ok': False, 'error': '当前用户无机构'}, ensure_ascii=False) async with get_sor_context(request._run_ns, 'pipeline') as _sor: _adm = await _sor.sqlExe( "SELECT 1 FROM userrole WHERE userid=${u}$ AND roleid='owner.superuser' LIMIT 1", {'u': user_id}) if not _adm: return json.dumps({'ok': False, 'error': '只有平台管理员能设置机构级工作环境'}, ensure_ascii=False) else: return json.dumps({'ok': False, 'error': 'owner_type 必须是 user 或 org'}, ensure_ascii=False) remote_config = { 'remote_host': (params_kw or {}).get('remote_host', ''), 'remote_port': (params_kw or {}).get('remote_port', 22), 'remote_user': (params_kw or {}).get('remote_user', ''), 'remote_key_path': (params_kw or {}).get('remote_key_path', ''), 'remote_dir': (params_kw or {}).get('remote_dir', ''), } try: async with get_sor_context(request._run_ns, 'pipeline') as sor: r = await set_work_env(sor, owner_type, owner_id, mode, remote_config, account_name) # 审计:工作环境变更(尤其远程主机变更,高风险) await audit_log(sor, user_id, username, 'work_env_set', target=owner_type + ':' + str(owner_id), detail='mode=' + mode + ' host=' + str(remote_config.get('remote_host', '')), result='ok' if r.get('ok') else 'fail', client_ip=client_ip) return json.dumps(r, ensure_ascii=False) except Exception as e: return json.dumps({'ok': False, 'error': str(e)}, ensure_ascii=False) elif action == 'ensure_remote_bwrap': remote_config = { 'remote_host': (params_kw or {}).get('remote_host', ''), 'remote_port': (params_kw or {}).get('remote_port', 22), 'remote_user': (params_kw or {}).get('remote_user', ''), 'remote_key_path': (params_kw or {}).get('remote_key_path', ''), } try: r = await ensure_remote_bwrap(remote_config) async with get_sor_context(request._run_ns, 'pipeline') as sor: await audit_log(sor, user_id, username, 'remote_bwrap', target=str(remote_config.get('remote_host', '')), detail='部署/校验远程沙箱', result='ok' if r.get('ok') else 'fail', client_ip=client_ip) return json.dumps(r, ensure_ascii=False) except Exception as e: return json.dumps({'ok': False, 'error': str(e)}, ensure_ascii=False) elif action == 'init_remote_env': # 一键初始化远程环境:bwrap + 技能同步 + 依赖安装(可重跑,幂等)。 # 从表单取临时配置(未保存时也可测试);保存后从 DB 取当前配置均可。 from pipeline_service.remote_env_init import init_remote_env as _init_remote_env async with get_sor_context(request._run_ns, 'pipeline') as _sor: _adm = await _sor.sqlExe( "SELECT 1 FROM userrole WHERE userid=${u}$ AND roleid='owner.superuser' LIMIT 1", {'u': user_id}) if not _adm: return json.dumps({'ok': False, 'error': '只有平台管理员能初始化远程环境'}, ensure_ascii=False) remote_config = { 'remote_host': (params_kw or {}).get('remote_host', ''), 'remote_port': int((params_kw or {}).get('remote_port', 22) or 22), 'remote_user': (params_kw or {}).get('remote_user', ''), 'remote_key_path': (params_kw or {}).get('remote_key_path', ''), 'remote_dir': (params_kw or {}).get('remote_dir', ''), } # 未传 key 时用机构级 key(与 set 一致) if not remote_config['remote_key_path'] and org_id: from pipeline_service.work_env import ensure_org_key remote_config['remote_key_path'] = ensure_org_key(org_id).get('key_path', '') try: async with get_sor_context(request._run_ns, 'pipeline') as sor: r = await _init_remote_env(remote_config, org_id=org_id, sor=sor) await audit_log(sor, user_id, username, 'remote_env_init', target=str(remote_config.get('remote_host', '')), detail='技能同步+依赖安装', result='ok' if r.get('ok') else 'fail', client_ip=client_ip) return json.dumps(r, ensure_ascii=False) except Exception as e: return json.dumps({'ok': False, 'error': str(e)}, ensure_ascii=False) elif action == 'init_remote_env_status': # 轮询远程依赖安装进度(无状态:读远程进度文件,服务重启后可继续查)。 from pipeline_service.remote_env_init import get_deps_install_status as _deps_status _ot = (params_kw or {}).get('owner_type', 'org') try: async with get_sor_context(request._run_ns, 'pipeline') as sor: _r = await _deps_status(sor, _ot, user_id if _ot == 'user' else org_id) return json.dumps(_r, ensure_ascii=False) except Exception as e: return json.dumps({'ok': False, 'error': str(e)}, ensure_ascii=False) elif action == 'get_org_pubkey': # 获取机构 public key(供 admin 复制到远程主机 authorized_keys)。仅超管。 try: async with get_sor_context(request._run_ns, 'pipeline') as _sor: _adm = await _sor.sqlExe( "SELECT 1 FROM userrole WHERE userid=${u}$ AND roleid='owner.superuser' LIMIT 1", {'u': user_id}) if not _adm: return json.dumps({'ok': False, 'error': '只有平台管理员能获取机构公钥'}, ensure_ascii=False) _oid = (params_kw or {}).get('org_id', '') or org_id r = get_org_pubkey(_oid) async with get_sor_context(request._run_ns, 'pipeline') as sor: await audit_log(sor, user_id, username, 'org_key_gen', target='org:' + str(_oid), detail='获取/生成机构公钥', result='ok' if r.get('ok') else 'fail', client_ip=client_ip) return json.dumps(r, ensure_ascii=False) except Exception as e: return json.dumps({'ok': False, 'error': str(e)}, ensure_ascii=False) else: return json.dumps({'ok': False, 'error': '未知 action: ' + str(action)}, ensure_ascii=False)