# 端到端加密（E2EE）

## 概述

E2EE 是 AUN 协议的可选安全层，提供端到端内容加密。它独立于传输层 TLS 和认证层 JWT，横跨三种网络拓扑（Gateway/Peer/Relay）。

服务端（Gateway、Message 等）无法解密消息内容，仅能看到加密信封的元数据（发送方、接收方、时间戳等）。

## 加密套件

| 套件 | 状态 |
|------|------|
| `P256_HKDF_SHA256_AES_256_GCM` | 当前 Python SDK 已实现 |
| `X25519_HKDF_SHA256_CHACHA20_POLY1305` | 当前 Python SDK 未实现 |

当前 Python SDK 的 `E2EEManager` 只支持单一套件 `P256_HKDF_SHA256_AES_256_GCM`。若对端仅支持其他套件，会在协商阶段收到拒绝。

## SDK Core E2EE API

### 加密出站消息

```python
envelope, encrypted = await client.e2ee.encrypt_outbound(
    peer_aid="peer-agent.aid.com",
    payload={"text": "Hello!"},
    message_id="uuid-string",
    timestamp=1234567890000
)
```

返回值：
- `envelope` — 加密信封（包含 nonce、ciphertext、tag、aad 等）
- `encrypted` — 布尔值，是否成功加密

### 解密入站消息

```python
decrypted_messages = await client.e2ee.process_incoming_messages(raw_messages)
```

自动识别加密消息（`payload.type == "e2ee.encrypted"`），解密后返回明文消息列表。

### 预加载对端证书

```python
await client.e2ee.preload_peer("peer-agent.aid.com")
```

提前获取对端证书，用于后续加密消息时的会话协商。

### 获取缓冲消息

```python
deferred = client.e2ee.drain_deferred_messages()
```

获取会话协商期间缓冲的消息（已解密）。

### 会话失效

```python
client.e2ee.invalidate_session(peer_aid="peer-agent.aid.com")
# 或
client.e2ee.invalidate_session(session_id="uuid")
```

## 会话生命周期

### 状态机

```mermaid
stateDiagram-v2
    [*] --> idle
    idle --> negotiating
    negotiating --> accepted
    negotiating --> failed
    accepted --> active
    active --> rekeying
    rekeying --> active
    active --> closing
    closing --> closed
    failed --> [*]
    closed --> [*]
```

### 会话协商

首次向某个对端发送加密消息时，SDK 自动发起会话协商：

1. 获取对端证书（通过 `auth.download_cert` 或本地缓存）
2. 生成临时 ECDH 密钥对
3. 发送协商消息（`e2ee.negotiate`）
4. 对端接受后，双方通过 ECDH + HKDF 派生共享密钥
5. 会话进入 active 状态

协商期间的消息会被缓冲，协商完成后自动发送。

**e2ee.negotiate 消息字段**（通过 `message.send` 的 `payload` 承载）：

| 字段 | 必需 | 说明 |
|------|:----:|------|
| `type` | ✅ | 固定为 `"e2ee.negotiate"` |
| `version` | ✅ | 协议版本，当前为 `"1"` |
| `session_id` | ✅ | UUID v4，本次会话唯一标识 |
| `supported_suites` | ✅ | 非空数组，支持的加密套件列表 |
| `ephemeral_public_key` | ✅ | Base64 编码的临时 ECDH 公钥 |
| `identity_key_fingerprint` | ✅ | 发起方身份公钥指纹（`sha256:...`） |
| `nonce` | ✅ | Base64 编码的随机值 |
| `signature` | ✅ | 对握手摘要的 ECDSA 签名（摘要字段顺序：version / session_id / supported_suites / ephemeral_public_key / identity_key_fingerprint / nonce） |

**e2ee.accept 消息字段**：

| 字段 | 必需 | 说明 |
|------|:----:|------|
| `type` | ✅ | 固定为 `"e2ee.accept"` |
| `version` | ✅ | 协议版本，当前为 `"1"` |
| `session_id` | ✅ | 与 negotiate 中相同的 UUID v4 |
| `selected_suite` | ✅ | 从 supported_suites 中选定的套件 |
| `ephemeral_public_key` | ✅ | 接收方临时 ECDH 公钥 |
| `identity_key_fingerprint` | ✅ | 接收方身份公钥指纹 |
| `nonce` | ✅ | Base64 编码的随机值 |
| `signature` | ✅ | 对握手摘要的 ECDSA 签名 |

### 协商拒绝（e2ee.reject）

对端无法接受协商时，发送 `e2ee.reject` 消息，`reason` 字段取值：

| reason | 说明 |
|--------|------|
| `no_common_suite` | 双方无共同加密套件 |
| `unsupported_version` | 对端不支持该 E2EE 版本 |
| `bad_signature` | 握手签名验证失败 |
| `identity_mismatch` | 身份绑定校验失败（证书指纹不匹配） |
| `policy_denied` | 本地策略拒绝建立 E2EE |
| `busy_rekeying` | 对端正在重协商，稍后重试 |

SDK 收到 reject 后抛出 `E2EENegotiationRejectedError`，`reject_reason` 属性为上述值之一。

### 会话关闭（e2ee.close）

任一端可发送 `e2ee.close` 终止会话，`reason` 字段取值：

| reason | 说明 | SDK 异常 |
|--------|------|---------|
| `normal` | 正常关闭 | — |
| `rekey` | 请求用新会话替换当前会话 | — |
| `decrypt_failed` | 连续解密失败 | `E2EEDecryptFailedError` |
| `counter_violation` | 计数器异常或重放嫌疑 | `E2EEBadCounterError` |
| `session_expired` | 会话过期 | `E2EESessionExpiredError` |
| `local_state_lost` | 本地状态丢失，需重新建立 | — |

### 协商前置要求（重要）

`encrypt_outbound()` 首次调用时会自动向对端发送 `e2ee.negotiate` 协商消息，然后轮询**发送方自己的** inbox 等待对端回复 `e2ee.accept`（超时 10 秒）。

这要求**对端必须在后台持续运行消息处理循环**，否则 negotiate 消息无人处理、accept 永远不会发出，导致超时失败。

```python
# 对端必须运行此循环，才能响应 e2ee.negotiate
async def message_loop(client, stop_event, business_msgs=None, start_cursor=0):
    """后台消息处理循环。

    - 自动处理 e2ee.negotiate → 回复 e2ee.accept（会话协商）
    - 自动解密 e2ee.encrypted 消息
    - 解密后的业务消息收集到 business_msgs 列表
    """
    cursor = start_cursor
    while not stop_event.is_set():
        try:
            result = await client.call("message.pull", {
                "after_seq": cursor, "limit": 50,
            })
            raw = result.get("messages", [])
            if raw:
                cursor = max(int(m.get("seq", 0)) for m in raw)
                msgs = await client.e2ee.process_incoming_messages(raw)
                if business_msgs is not None:
                    business_msgs.extend(msgs)
                deferred = client.e2ee.drain_deferred_messages()
                if deferred and business_msgs is not None:
                    business_msgs.extend(deferred)
        except Exception:
            pass
        await asyncio.sleep(0.3)
```

典型调用模式：

```python
# 1. 先启动对端消息循环
stop = asyncio.Event()
inbox = []
task = asyncio.create_task(message_loop(receiver, stop, inbox))

# 2. 再调用 encrypt_outbound（协商会自动完成）
envelope, ok = await sender.e2ee.encrypt_outbound(...)

# 3. 用完后停止循环
stop.set()
task.cancel()
```

### 密文格式

加密消息的 `payload` 结构：

```json
{
    "type": "e2ee.encrypted",
    "version": "1",
    "session_id": "uuid",
    "suite": "P256_HKDF_SHA256_AES_256_GCM",
    "counter": 12,
    "nonce": "base64",
    "ciphertext": "base64",
    "tag": "base64",
    "aad": {
        "from": "my-agent.aid.com",
        "to": "peer-agent.aid.com",
        "message_id": "uuid",
        "timestamp": 1234567890000,
        "session_id": "uuid",
        "counter": 12
    }
}
```

### 重放保护

- 发送方维护单调递增的 `send_counter`
- 接收方拒绝 `counter <= max_seen_counter` 的消息
- 同一 `(session_id, counter, nonce)` 三元组不可重用

**重要副作用**：当 `process_incoming_messages` 遇到 counter 重复的加密消息时，SDK 会**销毁整个 E2EE 会话**（调用 `invalidate_session`），而非静默跳过。这意味着：

- **禁止**对已处理过的加密消息再次调用 `process_incoming_messages`
- 必须用 `after_seq` 游标递增拉取，确保每条消息只处理一次
- 从 `seq=0` 重新拉取全部消息会导致旧加密消息触发 counter 错误，进而销毁当前会话

### 会话持久化

SDK 会持久化 active/rekeying 状态的会话到磁盘（存储在 `{aun_path}/AIDs/{aid}/tokens/meta.json` 的 `e2ee_sessions` 字段）。进程重启后恢复未过期的会话，避免频繁重新协商。

**注意**：持久化会话 + 从 `seq=0` 拉取旧消息 = counter 冲突 → 会话销毁。demo 或测试场景如果需要反复运行，必须在开始前清理旧状态：

```python
# 幂等运行模式：清除旧会话 + 跳过旧消息
client.e2ee.invalidate_session(peer_aid=peer_aid)
r = await client.call("message.pull", {"after_seq": 0, "limit": 1})
cursor = r.get("latest_seq", 0)  # 从最新位置开始，跳过所有旧消息
```

## 与业务层的集成

E2EE 在 SDK Core 中提供原语 API，业务层调用者负责组合。

### 发送加密消息

```python
import uuid, time

mid = str(uuid.uuid4())
ts = int(time.time() * 1000)

# 1. 加密
envelope, ok = await client.e2ee.encrypt_outbound(
    "peer-agent.aid.com", {"text": "secret"}, message_id=mid, timestamp=ts
)

# 2. 发送（传加密信封作为 payload）
if ok:
    await client.call("message.send", {
        "to": "peer-agent.aid.com",
        "payload": envelope,
        "encrypted": True,
        "message_id": mid,
        "timestamp": ts,
    })
```

### 接收加密消息

```python
result = await client.call("message.pull", {"after_seq": last_seq})
messages = await client.e2ee.process_incoming_messages(result["messages"])
# messages 中的加密消息已自动解密
```

### 完整双端通信模式

实际应用中，两端都需要后台运行消息循环来处理协商和接收消息：

```python
import asyncio, uuid, time

async def agent_loop(client, stop, inbox, start_cursor=0):
    """后台消息循环：处理 E2EE 协商 + 收集解密后的业务消息。"""
    cursor = start_cursor
    while not stop.is_set():
        try:
            result = await client.call("message.pull", {
                "after_seq": cursor, "limit": 50,
            })
            raw = result.get("messages", [])
            if raw:
                cursor = max(int(m.get("seq", 0)) for m in raw)
                msgs = await client.e2ee.process_incoming_messages(raw)
                inbox.extend(msgs)
                deferred = client.e2ee.drain_deferred_messages()
                if deferred:
                    inbox.extend(deferred)
        except Exception:
            pass
        await asyncio.sleep(0.3)


async def send_e2ee(client, to_aid, payload):
    """加密并发送消息的辅助函数。"""
    mid = str(uuid.uuid4())
    ts = int(time.time() * 1000)
    envelope, encrypted = await client.e2ee.encrypt_outbound(
        peer_aid=to_aid, payload=payload, message_id=mid, timestamp=ts,
    )
    result = await client.call("message.send", {
        "to": to_aid,
        "payload": envelope if encrypted else payload,
        "encrypted": encrypted,
        "message_id": mid,
        "timestamp": ts,
    })
    return result, encrypted


async def main():
    # 清除旧会话（幂等运行）
    client_a.e2ee.invalidate_session(peer_aid=aid_b)
    client_b.e2ee.invalidate_session(peer_aid=aid_a)

    # 获取当前最新 seq 作为起点，跳过旧消息
    r = await client_a.call("message.pull", {"after_seq": 0, "limit": 1})
    cursor_a = r.get("latest_seq", 0)
    r = await client_b.call("message.pull", {"after_seq": 0, "limit": 1})
    cursor_b = r.get("latest_seq", 0)

    # 两端同时启动后台消息循环
    stop_a, stop_b = asyncio.Event(), asyncio.Event()
    inbox_a, inbox_b = [], []
    task_a = asyncio.create_task(agent_loop(client_a, stop_a, inbox_a, cursor_a))
    task_b = asyncio.create_task(agent_loop(client_b, stop_b, inbox_b, cursor_b))

    # A → B 发送加密消息（首次会自动协商）
    await send_e2ee(client_a, aid_b, {"text": "Hello from A!"})
    # inbox_b 中会自动出现解密后的消息

    # B → A 回复（B 也会自动协商反向会话）
    await send_e2ee(client_b, aid_a, {"text": "Reply from B!"})
    # inbox_a 中会自动出现解密后的消息

    # A → B 第二轮（复用已有会话，无需重新协商）
    await send_e2ee(client_a, aid_b, {"text": "Second message, session reused!"})

    # 完成后停止循环
    stop_a.set(); stop_b.set()
    task_a.cancel(); task_b.cancel()
```

## 安全约定

- 协商失败后**禁止**静默降级为明文传输，必须由上层应用显式决定
- 会话密钥不可导出，仅存在于 SDK 运行时和持久化存储中
- Rekeying 由 SDK 自动触发（基于消息计数或时间阈值）
