身份验证和请求头

身份验证

Shannon Gateway 支持两种认证方式用于所有受保护端点。

API 密钥认证

在 X-API-Key 请求头中包含您的 API 密钥：

curl -H "X-API-Key: sk_test_123456" \
  http://localhost:8080/api/v1/tasks

Bearer Token 认证

也可使用 Authorization: Bearer 请求头传递 API 密钥或 JWT 令牌：

curl -H "Authorization: Bearer sk_test_123456" \
  http://localhost:8080/api/v1/tasks

两种方式等效——选择适合您工具链的方式即可。

认证错误

401 未授权 - 缺少或无效的凭证：

// 缺少认证头
{ "error": "Authentication required" }

// 无效的 API 密钥
{ "error": "Invalid API key" }

// JWT 令牌已过期
{ "error": "Invalid or expired token" }

常见原因：

缺少 X-API-Key 或 Authorization 请求头
无效的 API 密钥格式
禁用的或过期的 API 密钥
JWT 令牌已过期

请求头

必需请求头

X-API-Key

用途：身份验证必需：是（除非 GATEWAY_SKIP_AUTH=1）格式：字符串

X-API-Key: sk_test_123456

Content-Type（POST 请求）

用途：指定请求体格式必需：POST 请求需要格式：application/json

Content-Type: application/json

可选请求头

Idempotency-Key

用途：防止重复任务提交必需：否（推荐用于关键操作）格式：UUID 或唯一字符串 缓存持续时间：24 小时

Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000

示例：

# 第一次请求 - 创建任务
curl -X POST http://localhost:8080/api/v1/tasks \
  -H "X-API-Key: sk_test_123456" \
  -H "Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000" \
  -H "Content-Type: application/json" \
  -d '{"query": "处理支付 #12345"}'

# 重复请求 - 返回缓存响应
curl -X POST http://localhost:8080/api/v1/tasks \
  -H "X-API-Key: sk_test_123456" \
  -H "Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000" \
  -H "Content-Type: application/json" \
  -d '{"query": "处理支付 #12345"}'

traceparent

用途：W3C 分布式追踪必需：否（推荐用于可观测性）格式：{version}-{trace-id}-{parent-id}-{flags}

traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01

格式规范：

version：00（当前版本）
trace-id：32 个十六进制字符（128 位）
parent-id：16 个十六进制字符（64 位）
flags：2 个十六进制字符（采样：01，未采样：00）

示例：

import uuid

def generate_traceparent():
    trace_id = uuid.uuid4().hex + uuid.uuid4().hex  # 32 个字符
    parent_id = uuid.uuid4().hex[:16]                # 16 个字符
    return f"00-{trace_id}-{parent_id}-01"

traceparent = generate_traceparent()
# "00-4bf92f3577b34da6a3ce929d0e0e47360011223344556677-00f067aa0ba902b7-01"

tracestate

用途：供应商特定的追踪上下文必需：否格式：逗号分隔的键值对

tracestate: shannon=task_abc123,vendor=value

Cache-Control

用途：控制缓存行为必需：否格式：标准 HTTP 缓存指令

Cache-Control: no-cache
Cache-Control: max-age=300

Last-Event-ID (SSE only)

Purpose: Resume SSE stream from a specific event Required: No Format: Event ID string — either a Redis stream ID (e.g., 1700000000000-0) or a numeric sequence (e.g., 42)

Last-Event-ID: 1700000000000-0

Used for SSE reconnection:

const eventSource = new EventSource(url, {
  headers: {
    'X-API-Key': 'sk_test_123456',
    'Last-Event-ID': '1700000000000-0'  // Resume from this stream ID (or use a numeric seq)
  }
});

响应头

标准响应头

X-Workflow-ID

用途：Temporal 工作流标识符 出现位置：POST /api/v1/tasks、GET /api/v1/tasks/ 格式：字符串（与 task_id 相同）

X-Workflow-ID: task_01HQZX3Y9K8M2P4N5S7T9W2V

用例：在 Temporal UI 中跟踪工作流执行

WORKFLOW_ID=$(curl -s -X POST ... | grep -i "X-Workflow-ID" | cut -d: -f2)
echo "监控地址: http://localhost:8088/workflows/$WORKFLOW_ID"

X-Session-ID

用途：多轮对话的会话标识符 出现位置：POST /api/v1/tasks 格式：UUID 字符串

X-Session-ID: user-123-chat-session

Content-Type

用途：响应体格式 出现位置：所有 JSON 响应格式：application/json

Content-Type: application/json

对于 SSE：

Content-Type: text/event-stream

速率限制请求头

X-RateLimit-Limit

用途：每个窗口允许的最大请求数 出现位置：所有认证请求格式：整数

X-RateLimit-Limit: 100

X-RateLimit-Remaining

用途：当前窗口中剩余的请求数 出现位置：所有认证请求格式：整数

X-RateLimit-Remaining: 95

X-RateLimit-Reset

用途：速率限制重置的 Unix 时间戳 出现位置：所有认证请求格式：Unix 时间戳（秒）

X-RateLimit-Reset: 1609459200

示例 - 检查速率限制：

curl -v http://localhost:8080/api/v1/tasks \
  -H "X-API-Key: sk_test_123456" 2>&1 | grep -i "X-RateLimit"

# 输出：
# < X-RateLimit-Limit: 100
# < X-RateLimit-Remaining: 95
# < X-RateLimit-Reset: 1609459200

Retry-After

用途：重试前等待的秒数（429 响应） 出现位置：429 Too Many Requests 响应格式：整数（秒）

Retry-After: 60

示例 - 处理速率限制：

import time
import httpx

response = httpx.post(...)

if response.status_code == 429:
    retry_after = int(response.headers.get("Retry-After", 60))
    print(f"速率限制，等待 {retry_after}s...")
    time.sleep(retry_after)
    # 重试请求...

CORS 请求头

Access-Control-Allow-Origin

用途：CORS 允许的来源 出现位置：所有响应（开发模式）格式：域名或 *

Access-Control-Allow-Origin: *

生产环境：配置特定域名：

// gateway/main.go
w.Header().Set("Access-Control-Allow-Origin", "https://app.example.com")

Access-Control-Allow-Methods

用途：允许的 HTTP 方法 出现位置：CORS 预检响应格式：逗号分隔的方法

Access-Control-Allow-Methods: GET, POST, PUT, DELETE, OPTIONS

Access-Control-Allow-Headers

用途：允许的请求头 出现位置：CORS 预检响应格式：逗号分隔的请求头

Access-Control-Allow-Headers: Content-Type, Authorization, X-API-Key, Idempotency-Key, traceparent

请求头示例

最小请求（GET）

curl http://localhost:8080/api/v1/tasks \
  -H "X-API-Key: sk_test_123456"

完整请求（POST）

curl -X POST http://localhost:8080/api/v1/tasks \
  -H "X-API-Key: sk_test_123456" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: 550e8400-e29b-41d4-a716-446655440000" \
  -H "traceparent: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" \
  -d '{"query": "分析数据"}'

Python - 所有请求头

import httpx
import uuid

def generate_traceparent():
    trace_id = uuid.uuid4().hex + uuid.uuid4().hex
    parent_id = uuid.uuid4().hex[:16]
    return f"00-{trace_id}-{parent_id}-01"

response = httpx.post(
    "http://localhost:8080/api/v1/tasks",
    headers={
        "X-API-Key": "sk_test_123456",
        "Content-Type": "application/json",
        "Idempotency-Key": str(uuid.uuid4()),
        "traceparent": generate_traceparent()
    },
    json={"query": "分析数据"}
)

# 提取响应头
workflow_id = response.headers.get("X-Workflow-ID")
session_id = response.headers.get("X-Session-ID")
rate_limit = response.headers.get("X-RateLimit-Remaining")

print(f"工作流: {workflow_id}")
print(f"会话: {session_id}")
print(f"剩余速率限制: {rate_limit}")

JavaScript - Fetch API

const response = await fetch('http://localhost:8080/api/v1/tasks', {
  method: 'POST',
  headers: {
    'X-API-Key': 'sk_test_123456',
    'Content-Type': 'application/json',
    'Idempotency-Key': crypto.randomUUID()
  },
  body: JSON.stringify({
    query: '分析数据'
  })
});

// 读取响应头
const workflowId = response.headers.get('X-Workflow-ID');
const rateLimit = response.headers.get('X-RateLimit-Remaining');

console.log('工作流:', workflowId);
console.log('剩余速率限制:', rateLimit);

安全最佳实践

1. 保护 API 密钥

永远不要将 API 密钥提交到版本控制：

# ✅ 好 - 使用环境变量
export SHANNON_API_KEY="sk_test_123456"
curl -H "X-API-Key: $SHANNON_API_KEY" ...

# ❌ 坏 - 在脚本中硬编码
curl -H "X-API-Key: sk_test_123456" ...

存储在安全配置中：

import os

API_KEY = os.environ.get("SHANNON_API_KEY")
if not API_KEY:
    raise ValueError("SHANNON_API_KEY not set")

2. 在生产环境中使用 HTTPS

# ✅ 生产环境
BASE_URL = "https://shannon.example.com"

# ⚠️ 仅限开发
BASE_URL = "http://localhost:8080"

3. 定期轮换 API 密钥

-- 禁用旧密钥
UPDATE auth.api_keys SET enabled = false WHERE key = 'sk_test_old';

-- 创建新密钥
INSERT INTO auth.api_keys (key, user_id, tenant_id, name, enabled)
VALUES ('sk_test_new', 'user-uuid', 'tenant-uuid', 'Rotated Key', true);

4. 实施密钥过期

-- 为密钥添加过期时间
UPDATE auth.api_keys SET expires_at = NOW() + INTERVAL '90 days';

-- 检查过期密钥
SELECT * FROM auth.api_keys WHERE expires_at < NOW();

5. 监控 API 密钥使用

def track_api_usage(api_key: str):
    """记录 API 密钥使用以供监控。"""
    response = httpx.post(...)

    # 记录使用情况
    logger.info("API 请求", extra={
        "api_key": api_key[:10] + "...",  # 仅部分密钥
        "endpoint": "/api/v1/tasks",
        "status": response.status_code,
        "rate_limit_remaining": response.headers.get("X-RateLimit-Remaining")
    })

故障排除

认证失败

问题：收到 401 未授权 解决方案：

检查是否包含 API 密钥：

curl -v http://localhost:8080/api/v1/tasks 2>&1 | grep "X-API-Key"

验证 API 密钥格式：

# 应以 sk_ 开头
echo $API_KEY | grep -E "^sk_"

检查是否禁用了认证：

docker compose exec gateway env | grep GATEWAY_SKIP_AUTH

在数据库中验证密钥：

SELECT key, enabled, expires_at FROM auth.api_keys WHERE key = 'sk_test_123456';

速率限制问题

问题：收到 429 Too Many Requests 解决方案：

检查速率限制请求头：
```
curl -v ... 2>&1 | grep "X-RateLimit"
```

实施指数退避：

if response.status_code == 429:
    retry_after = int(response.headers.get("Retry-After", 60))
    time.sleep(retry_after)

增加速率限制（如需要）：

# 在 .env 中
RATE_LIMIT_RPM=200
RATE_LIMIT_BURST=50

幂等性问题

问题：创建了重复任务 解决方案：

始终包含 Idempotency-Key：

idempotency_key = str(uuid.uuid4())
headers["Idempotency-Key"] = idempotency_key

存储幂等性密钥：

# 存储密钥与请求
db.requests.insert({
    "idempotency_key": idempotency_key,
    "task_id": task_id,
    "created_at": datetime.now()
})

检查 Redis 缓存：

docker compose exec redis redis-cli KEYS "idempotency:*"

REST API 概述

完整 API 参考

提交任务

POST /api/v1/tasks

速率限制

速率限制详情

分布式追踪

追踪设置

Documentation Index

​身份验证

​API 密钥认证

​Bearer Token 认证

​认证错误

​请求头

​必需请求头

​X-API-Key

​Content-Type（POST 请求）

​可选请求头

​Idempotency-Key

​traceparent

​tracestate

​Cache-Control

​Last-Event-ID (SSE only)

​响应头

​标准响应头

​X-Workflow-ID

​X-Session-ID

​Content-Type

​速率限制请求头

​X-RateLimit-Limit

​X-RateLimit-Remaining

​X-RateLimit-Reset

​Retry-After

​CORS 请求头

​Access-Control-Allow-Origin

​Access-Control-Allow-Methods

​Access-Control-Allow-Headers

​请求头示例

​最小请求（GET）

​完整请求（POST）

​Python - 所有请求头

​JavaScript - Fetch API

​安全最佳实践

​1. 保护 API 密钥

​2. 在生产环境中使用 HTTPS

​3. 定期轮换 API 密钥

​4. 实施密钥过期

​5. 监控 API 密钥使用

​故障排除

​认证失败

​速率限制问题

​幂等性问题

​相关文档

REST API 概述

提交任务

速率限制

分布式追踪

身份验证

API 密钥认证

Bearer Token 认证

认证错误

请求头

必需请求头

X-API-Key

Content-Type（POST 请求）

可选请求头

Idempotency-Key

traceparent

tracestate

Cache-Control

Last-Event-ID (SSE only)

响应头

标准响应头

X-Workflow-ID

X-Session-ID

Content-Type

速率限制请求头

X-RateLimit-Limit

X-RateLimit-Remaining

X-RateLimit-Reset

Retry-After

CORS 请求头

Access-Control-Allow-Origin

Access-Control-Allow-Methods

Access-Control-Allow-Headers

请求头示例

最小请求（GET）

完整请求（POST）

Python - 所有请求头

JavaScript - Fetch API

安全最佳实践

1. 保护 API 密钥

2. 在生产环境中使用 HTTPS

3. 定期轮换 API 密钥

4. 实施密钥过期

5. 监控 API 密钥使用

故障排除

认证失败

速率限制问题

幂等性问题

相关文档