Security audit
Security checks across malware telemetry and agentic risk
This appears to be a real P2P agent messaging skill, but it needs Review because it grants broad local and VPS control while handling secrets and public endpoints unsafely.
Install only if you are comfortable giving this skill broad command execution, SSH access to a VPS, persistent background services, and access to OpenClaw/P2P secrets. Use a dedicated VPS and SSH key, rotate any hooks/API/admin credentials after testing, inspect the deploy source before running auto-install, and avoid using it for sensitive files until command injection, plaintext secret storage, endpoint authentication, and file-transfer approval issues are fixed.
def run(cmd, check=True):
"""运行命令"""
print(f"$ {cmd}")
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
if check and result.returncode != 0:
print(f"错误: {result.stderr}")
sys.exit(1)os.chmod(ssh_key, 0o600)
# 测试连接
result = subprocess.run(
f"ssh -i {ssh_key} -o StrictHostKeyChecking=no -o BatchMode=yes ubuntu@{vps_ip} 'echo OK'",
shell=True, capture_output=True, text=True
)echo "部署完成"
"""
result = subprocess.run(
f"ssh -i {ssh_key} ubuntu@{vps_ip} '{script}'",
shell=True, capture_output=True, text=True
)"""获取 API Key"""
print(f"\n[4/6] 获取 API Key")
result = subprocess.run(
f"ssh -i {ssh_key} ubuntu@{vps_ip} 'python3 -c \"import sqlite3; conn = sqlite3.connect(\\\"/opt/agent-p2p/data/portal.db\\\"); cur = conn.cursor(); cur.execute(\\\"SELECT key_id FROM api_keys LIMIT 1\\\"); print(cur.fetchone()[0])\"'",
shell=True, capture_output=True, text=True
)print(f"\n[6/6] 测试连接")
# 测试 Portal
result = subprocess.run(
f"curl -sk https://{domain}/api/portal/info",
shell=True, capture_output=True, text=True
)return False
# 测试 OpenClaw
result = subprocess.run(
f"curl -s -X POST {gateway_url}/hooks/wake -H 'Authorization: Bearer {hooks_token}' -H 'Content-Type: application/json' -d '{{\"text\":\"测试\"}}'",
shell=True, capture_output=True, text=True
)print(f"\n[1/6] 检查域名解析: {domain}")
for i in range(30): # 最多等待 5 分钟
result = subprocess.run(
f"nslookup {domain}",
shell=True, capture_output=True, text=True
)if new_pass:
# 更新 VPS 上的密码
update_cmd = f"ssh -i {args.ssh_key} ubuntu@{args.vps_ip} 'sudo htpasswd -cb /etc/nginx/.htpasswd {admin_user} \"{new_pass}\" && echo 密码已更新'"
result = subprocess.run(update_cmd, shell=True, capture_output=True, text=True)
if result.returncode == 0:
admin_pass = new_pass
print("✅ 密码已更新")if new_pass:
# 更新 VPS 上的密码
update_cmd = f"ssh -i {args.ssh_key} ubuntu@{args.vps_ip} 'sudo htpasswd -cb /etc/nginx/.htpasswd {admin_user} \"{new_pass}\" && echo 密码已更新'"
result = subprocess.run(update_cmd, shell=True, capture_output=True, text=True)
if result.returncode == 0:
admin_pass = new_pass
print("✅ 密码已更新")# 1. 发送到对方 Portal 的 /api/message/receive
try:
resp = requests.post(
f"{to_portal}/api/message/receive",
json={
"api_key": shared_key,file_size: int, file_md5: str, chunks_total: int) -> str:
"""初始化文件传输（简化版：直接返回file_id，无需确认）"""
try:
resp = requests.post(
f"{to_portal}/api/file/initiate",
json={
"api_key": api_key,for attempt in range(max_retries):
try:
resp = requests.post(
f"{to_portal}/api/file/chunk/{file_id}/{chunk_index}",
json={
"api_key": api_key,65/65 vendors flagged this skill as clean.
No suspicious patterns detected.