OpenClaw“龙虾“暗潮:企业如何在员工不知情下完成AI资产清剿与风险闭环

# openclaw_network_scanner_advanced.pyimport socketimport structimport threadingimport ipaddressimport requestsfrom queue import Queue# OpenClaw特征OPENCLAW_PORT = 18789OPENCLAW_HEADERS = {"X-OpenClaw-Version", "Server": "OpenClaw/"}OPENCLAW_PATH = "/api/health"class SynScanner:    def __init__(self, thread_count=100):        self.thread_count = thread_count        self.queue = Queue()        self.results = []        self.lock = threading.Lock()        # 创建原始套接字（需要root/管理员权限）        self.sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)        self.sock.settimeout(0.5)    def checksum(self, msg):        """计算TCP校验和"""        s = 0        for i in range(0, len(msg), 2):            w = (msg[i] << 8) + (msg[i+1])            s = s + w        s = (s >> 16) + (s & 0xffff)        s = s + (s >> 16)        return ~s & 0xffff    def send_syn_packet(self, ip, port):        """发送SYN包"""        # IP头        ip_header = struct.pack('!BBHHHBBH4s4s',            0x45, 0x00, 0x0028, 0x1234, 0x0000, 0x40, socket.IPPROTO_TCP, 0x0000,            socket.inet_aton('0.0.0.0'), socket.inet_aton(ip))        # TCP头        tcp_header = struct.pack('!HHLLBBHHH',            0x1234, port, 0x00000000, 0x00000000,            0x50, 0x02, 0x2000, 0x0000, 0x0000)        # 伪头        pseudo_header = struct.pack('!4s4sBBH',            socket.inet_aton('0.0.0.0'), socket.inet_aton(ip),            0x00, socket.IPPROTO_TCP, len(tcp_header))        # 计算校验和        checksum = self.checksum(pseudo_header + tcp_header)        tcp_header = struct.pack('!HHLLBBHHH',            0x1234, port, 0x00000000, 0x00000000,            0x50, 0x02, 0x2000, checksum, 0x0000)        # 发送数据包        self.sock.sendto(ip_header + tcp_header, (ip, port))    def scan_worker(self):        """扫描工作线程"""        while True:            ip = self.queue.get()            try:                self.send_syn_packet(str(ip), OPENCLAW_PORT)                # 等待响应                try:                    data, addr = self.sock.recvfrom(1024)                    if addr[0] == str(ip) and data[33] == 0x12:  # SYN-ACK响应                        # 验证是否为OpenClaw服务                        try:                            response = requests.get(                                f"http://{ip}:{OPENCLAW_PORT}{OPENCLAW_PATH}",                                timeout=2.0,                                headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}                            )                            if any(header in response.headers for header in OPENCLAW_HEADERS) or "OpenClaw" in response.text:                                with self.lock:                                    self.results.append({                                        "ip": str(ip),                                        "port": OPENCLAW_PORT,                                        "version": response.headers.get("X-OpenClaw-Version", "unknown"),                                        "status": "running"                                    })                        except:                            pass                except socket.timeout:                    pass            finally:                self.queue.task_done()    def scan_subnet(self, subnet):        """扫描指定子网"""        network = ipaddress.IPv4Network(subnet, strict=False)        for ip in network.hosts():            self.queue.put(ip)        # 启动线程        for _ in range(self.thread_count):            t = threading.Thread(target=self.scan_worker, daemon=True)            t.start()        self.queue.join()        return self.resultsif __name__ == "__main__":    scanner = SynScanner(thread_count=200)    subnets = ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"]    all_devices = []    for subnet in subnets:        print(f"正在扫描子网: {subnet}")        devices = scanner.scan_subnet(subnet)        all_devices.extend(devices)        print(f"发现 {len(devices)} 台运行OpenClaw的设备")    # 输出结果    import csv    with open("openclaw_assets.csv", "w", newline="") as f:        writer = csv.DictWriter(f, fieldnames=["ip", "port", "version", "status"])        writer.writeheader()        writer.writerows(all_devices)    print(f"扫描完成，共发现 {len(all_devices)} 台OpenClaw设备")

# openclaw_terminal_detector_windows_advanced.ps1# 执行后自动删除自身，不留任何痕迹MyInvocation.MyCommand.Path# 1. 检测进程_.Path -match "openclaw|claw" -or installPaths = @(    "env:APPDATA\OpenClaw",    "env:USERPROFILE\.claw")path in path) {        path    }}# 3. 检测注册表foundRegKeys = @()foreach (regKeys) {    if (Test-Path foundRegKeys += wslInstalled = wslInstances = @()try {    distro in distro)) {            distro --exec which openclaw 2>&1            if (wslInstalled = wslInstances += dockerInstalled = dockerContainers = @()try {    dockerResult -match "openclaw|claw") {        true        dockerResult | Where-Object { result = @{    "hostname" = env:USERNAME    "ip_address" = (Get-NetIPAddress -AddressFamily IPv4 | Where-Object { processes | Select-Object Id, Name, Path, CommandLine    "installPaths" = foundRegKeys    "wslInstalled" = wslInstances    "dockerInstalled" = dockerContainers    "detected" = (foundPaths.Count -gt 0 -or wslInstalled -or logServer = "http://your-log-server/api/openclaw-detection"try {    Invoke-RestMethod -Uri result | ConvertTo-Json -Depth 3) -ContentType "application/json" -TimeoutSec 5} catch {}# 8. 自动删除脚本，不留任何痕迹Remove-Item -Path (ps aux | grep -E "openclaw|node.*openclaw|claw" | grep -v grep)# 2. 检测安装目录INSTALL_PATHS=(    "HOME/.claw")FOUND_PATHS=()for path in "path" ]; then        FOUND_PATHS+=("(npm list -g openclaw 2>/dev/null | grep openclaw)YARN_INSTALLED=(pnpm list -g openclaw 2>/dev/null | grep openclaw)# 4. 检测自启动AUTOSTART=(docker ps -a --format "{{.Names}} {{.Image}}" 2>/dev/null | grep -E "openclaw|claw")# 6. 生成结果RESULT=(hostname)",    "username": "(ip addr | grep 'inet ' | grep -v '127.0.0.1' | awk '{print (echo "(IFS=,; echo "([ -n "([ -n "([ -n "(echo "([ -n "(echo "([ -n "{#FOUND_PATHS[@]} -gt 0 ] || [ -n "YARN_INSTALLED" ] || [ -n "AUTOSTART" ] || [ -n "(date +"%Y-%m-%d %H:%M:%S")"}EOF)# 7. 发送结果到日志服务器curl -X POST -H "Content-Type: application/json" -d "0"

# openclaw_suricata_rules_advanced.yaml# 检测默认端口通信alert tcp any any -> any 18789 (msg:"OpenClaw默认端口通信"; flow:to_server,established; sid:1000001; rev:2;)# 检测OpenClaw API调用alert http any any -> any any (msg:"OpenClaw API调用"; content:"X-OpenClaw-Version"; http_header; sid:1000002; rev:2;)# 检测官方插件市场访问alert http any any -> any any (msg:"OpenClaw插件市场访问"; content:"clawhub.org"; http_host; sid:1000003; rev:2;)# 检测第三方插件市场访问alert http any any -> any any (msg:"OpenClaw第三方插件市场访问"; content:"claw"; http_host; content:"plugin"; http_uri; sid:1000004; rev:1;)# 检测大模型工具调用alert http any any -> any any (msg:"OpenClaw大模型工具调用"; content:"tool_call"; http_client_body; content:"openai"; http_host; sid:1000005; rev:1;)alert http any any -> any any (msg:"OpenClaw大模型工具调用"; content:"function_call"; http_client_body; content:"anthropic"; http_host; sid:1000006; rev:1;)

# openclaw_config_auditor_advanced.pyimport jsonimport osimport paramikoimport csvfrom concurrent.futures import ThreadPoolExecutordef audit_openclaw_config(hostname, username, password=None, key_file=None):    """通过SSH远程审计OpenClaw配置（仅读取，不修改任何文件）"""    ssh = paramiko.SSHClient()    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())    try:        if key_file:            ssh.connect(hostname, username=username, key_filename=key_file, timeout=5)        else:            ssh.connect(hostname, username=username, password=password, timeout=5)        # 1. 读取主配置文件        stdin, stdout, stderr = ssh.exec_command("cat ~/.openclaw/config.json 2>/dev/null")        config_content = stdout.read().decode()        config = json.loads(config_content) if config_content else {}        # 2. 读取已安装插件列表        stdin, stdout, stderr = ssh.exec_command("ls -la ~/.openclaw/plugins/ 2>/dev/null")        plugins = stdout.read().decode().splitlines()        # 3. 读取API密钥配置        stdin, stdout, stderr = ssh.exec_command("ls -la ~/.openclaw/secrets.json 2>/dev/null")        has_secrets = stdout.read().decode() != ""        # 4. 读取执行日志        stdin, stdout, stderr = ssh.exec_command("tail -n 100 ~/.openclaw/logs/execution.log 2>/dev/null")        recent_logs = stdout.read().decode()        # 5. 风险评估        risks = []        risk_score = 0        if config.get("server", {}).get("host") == "0.0.0.0":            risks.append("极高风险：服务绑定0.0.0.0，允许所有IP访问")            risk_score += 40        if not config.get("server", {}).get("auth", {}).get("enabled", False):            risks.append("极高风险：未启用身份认证")            risk_score += 30        if config.get("server", {}).get("allow_remote_execution", True):            risks.append("高风险：允许远程执行命令")            risk_score += 20        if has_secrets:            risks.append("中风险：配置了云端大模型API密钥，可能产生数据泄露")            risk_score += 10        if len(plugins) > 3:            risks.append("中风险：安装了多个非官方插件")            risk_score += 10        # 确定风险等级        if risk_score >= 70:            risk_level = "极高"        elif risk_score >= 40:            risk_level = "高"        elif risk_score >= 20:            risk_level = "中"        else:            risk_level = "低"        return {            "hostname": hostname,            "config": config,            "plugins_count": len(plugins),            "has_secrets": has_secrets,            "risks": risks,            "risk_score": risk_score,            "risk_level": risk_level,            "recent_logs": recent_logs[:1000]  # 只返回最近1000个字符的日志        }    except Exception as e:        return {"hostname": hostname, "error": str(e), "risk_level": "未知"}    finally:        ssh.close()if __name__ == "__main__":    # 从资产清单中读取设备列表    devices = []    with open("openclaw_assets.csv", "r") as f:        reader = csv.DictReader(f)        for row in reader:            devices.append(row["ip"])    # 批量审计（10个线程）    results = []    with ThreadPoolExecutor(max_workers=10) as executor:        futures = [executor.submit(audit_openclaw_config, device, username="admin", key_file="~/.ssh/id_rsa") for device in devices]        for future in futures:            results.append(future.result())    # 输出审计报告    with open("openclaw_risk_assessment.csv", "w", newline="") as f:        writer = csv.DictWriter(f, fieldnames=["hostname", "risk_level", "risk_score", "risks", "plugins_count", "has_secrets"])        writer.writeheader()        for result in results:            writer.writerow({                "hostname": result["hostname"],                "risk_level": result.get("risk_level", "未知"),                "risk_score": result.get("risk_score", 0),                "risks": "; ".join(result.get("risks", [])),                "plugins_count": result.get("plugins_count", 0),                "has_secrets": result.get("has_secrets", False)            })    # 统计风险分布    risk_distribution = {"极高": 0, "高": 0, "中": 0, "低": 0, "未知": 0}    for result in results:        risk_distribution[result.get("risk_level", "未知")] += 1    print("风险分布统计：")    for level, count in risk_distribution.items():        print(f"{level}风险：{count}台")

# openclaw_disable_windows_advanced.ps1# 完全隐形，不弹窗、不提示、不影响员工正常工作# 1. 停止所有OpenClaw相关进程Get-Process -Name "openclaw", "node", "claw" -ErrorAction SilentlyContinue |     Where-Object { _.CommandLine -match "openclaw|claw" } |    Stop-Process -Force -ErrorAction SilentlyContinue# 2. 删除自启动项Remove-Item -Path "HKCU:\Software\Microsoft\Windows\CurrentVersion\Run\*OpenClaw*" -Force -ErrorAction SilentlyContinueRemove-Item -Path "HKLM:\Software\Microsoft\Windows\CurrentVersion\Run\*OpenClaw*" -Force -ErrorAction SilentlyContinueRemove-Item -Path "HKCU:\Software\Microsoft\Windows\CurrentVersion\Run\*Claw*" -Force -ErrorAction SilentlyContinue# 3. 清理WSL中的OpenClawtry {    distro in distro)) {            wsl -d distro --exec pkill -f "node.*openclaw" 2>&1 | Out-Null            wsl -d distro --exec systemctl disable openclaw 2>&1 | Out-Null        }    }} catch {}# 4. 清理Docker中的OpenClaw容器try {    docker ps -a --filter "name=openclaw" --filter "name=claw" -q | ForEach-Object {        docker stop _ 2>&1 | Out-Null    }} catch {}# 5. 防火墙阻断默认端口和已知插件市场New-NetFirewallRule -DisplayName "Block OpenClaw Default Port" -Direction Inbound -LocalPort 18789 -Protocol TCP -Action Block -ErrorAction SilentlyContinueNew-NetFirewallRule -DisplayName "Block OpenClaw Default Port" -Direction Outbound -LocalPort 18789 -Protocol TCP -Action Block -ErrorAction SilentlyContinueNew-NetFirewallRule -DisplayName "Block OpenClaw Plugin Market" -Direction Outbound -RemoteHost "clawhub.org" -Action Block -ErrorAction SilentlyContinue

#!/bin/bash# openclaw_disable_linux_advanced.sh# 完全隐形，不影响员工正常工作# 1. 停止所有OpenClaw相关进程pkill -9 -f "openclaw" 2>/dev/nullpkill -9 -f "node.*openclaw" 2>/dev/nullpkill -9 -f "claw" 2>/dev/null# 2. 禁用systemd服务systemctl stop openclaw 2>/dev/nullsystemctl disable openclaw 2>/dev/nullrm -f /etc/systemd/system/openclaw.service 2>/dev/null# 3. 删除自启动脚本rm -f ~/.config/autostart/openclaw.desktop 2>/dev/nullrm -f /etc/init.d/openclaw 2>/dev/null# 4. 清理Docker容器docker ps -a --filter "name=openclaw" --filter "name=claw" -q | xargs -r docker stop 2>/dev/nulldocker ps -a --filter "name=openclaw" --filter "name=claw" -q | xargs -r docker rm 2>/dev/null# 5. 防火墙阻断默认端口和插件市场iptables -A INPUT -p tcp --dport 18789 -j DROP 2>/dev/nulliptables -A OUTPUT -p tcp --dport 18789 -j DROP 2>/dev/nulliptables -A OUTPUT -d clawhub.org -j DROP 2>/dev/nullufw deny 18789/tcp 2>/dev/nullufw deny out to clawhub.org 2>/dev/null