⚡ CFM Redis - 跨框架实时通信

Shell command execution function detected

subprocess.run

Shell command execution function detected

os.system

The skill explicitly demonstrates and encourages the use of shell command execution via `subprocess.run` and `os.system` for integration and troubleshooting. It also defines a 'command' message type, indicating its design to transmit commands between agents. While intended for legitimate purposes, this capability could be abused by a compromised agent to execute arbitrary commands.

subprocess.run(["python3", "~/.shared/cfm/cfm_check.py", ...]), os.system("redis-cli ping"), "type": "command"

The `discover_agents` function uses the `redis_client.keys()` command, which can be a blocking operation and lead to performance degradation or denial of service on a busy Redis server if called frequently or on a large dataset.

keys = self.redis_client.keys("cfm:*:registered")

The skill provides an example of how an agent can use `requests.post` to send messages to an external webhook URL. This demonstrates the capability for external data transfer, which could be repurposed for exfiltration if the `webhook_url` is controlled by an attacker.

requests.post(webhook_url, json=new_msgs)

The skill instructs installing a cron job that runs autonomously every 5 minutes to check for messages and trigger processing. This establishes a persistent agent execution mechanism that operates without user interaction.

### 方式1：Cron任务（简单）
```bash
# 每5分钟检查一次
*/5 * * * * cd ~/.shared/cfm && python3 cfm_check.py myagent
```

The discover_agents() function uses a Redis wildcard key scan (keys('cfm:*:registered')) to enumerate all registered agents on the Redis instance. This enables network reconnaissance of all agents and their statuses.

def discover_agents(self) -> list:
    """发现其他Agent"""
    keys = self.redis_client.keys("cfm:*:registered")
    agents = []
    for key in keys:
        agent_id = key.split(":")[1]
        if agent_id != self.agent_id:
            info = self.redis_client.hgetall(key)
            agents.append({"id": agent_id, **info})

The Redis connection is established without any authentication credentials, ACLs, or TLS. Any process with network access to the Redis port can publish messages to any agent's inbox channel, impersonate any agent identity, or read all stored message history.

self.redis_client = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)

def quick_send(to_agent: str, message: str, from_agent: str = "cli") -> str:
    r = redis.Redis(decode_responses=True)

The Webhook integration pattern sends message content to a configurable webhook_url using requests.post(). If an attacker can influence the webhook_url parameter or message content processed to derive URLs, this could enable SSRF attacks against internal services.

def check_and_trigger(agent_id, webhook_url):
    with CFMMessenger(agent_id) as m:
        msgs = m.get_messages(limit=5)
        new_msgs = [msg for msg in msgs if msg.get("from") != agent_id]
        if new_msgs:
            # 触发webhook
            import requests
            requests.post(webhook_url, json=new_msgs)

Messages of type 'file' use the content field as a file path reference. If agents process these messages by reading the referenced file path, an attacker publishing a 'file' message with a sensitive path (e.g., ~/.ssh/id_rsa, /etc/passwd) could trigger exfiltration of sensitive files.

- `file` - 文件引用（content为文件路径）

The skill establishes a Redis Pub/Sub communication channel between agents where message content from any publisher is consumed and acted upon by subscribing agents. Messages of type 'command' are explicitly supported, enabling attacker-controlled prompts or commands to be injected into agent inboxes and processed by other agents without validation or sanitization.

### 消息类型
- `text` - 普通文本
- `command` - 命令消息
- `response` - 响应消息
- `file` - 文件引用（content为文件路径）