接口认证与权限

运维 API 通常部署在内网，但仍需认证与鉴权：防止误调用、区分只读与变更权限。本章用 FastAPI 依赖注入（Depends） 实现 API Key 与 Bearer Token 两种常见方式。

密钥一律从环境变量读取，示例中用占位符。

---

API Key：适合脚本与 CI 调用

调用方在 Header 携带固定密钥；服务端校验后放行。服务缺少密钥配置时必须拒绝启动，不能留下任何开发默认值。

#!/usr/bin/env python3
import os
import secrets
from enum import Enum
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import APIKeyHeader

app = FastAPI(title="Ops API with Auth")


def required_env(name: str) -> str:
    """启动时校验关键配置，避免遗漏配置后意外对外开放。"""
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"缺少必填环境变量: {name}")
    return value


# 格式：OPS_API_KEY_ROLES="<readonly-key>:readonly,<operator-key>:operator"
# 实际值由 Secret 管理系统或 CI/CD 注入，代码库不保存任何密钥。
class Role(str, Enum):
    readonly = "readonly"
    operator = "operator"


def load_key_roles() -> dict[str, Role]:
    roles: dict[str, Role] = {}
    for item in required_env("OPS_API_KEY_ROLES").split(","):
        try:
            key, role = item.strip().rsplit(":", 1)
            if not key:
                raise ValueError("API Key 不能为空")
            roles[key] = Role(role)
        except ValueError as exc:
            raise RuntimeError("OPS_API_KEY_ROLES 格式应为 key:role,key:role") from exc
    if not roles:
        raise RuntimeError("OPS_API_KEY_ROLES 至少要配置一个密钥")
    return roles


KEY_ROLES = load_key_roles()
VALID_API_KEYS = frozenset(KEY_ROLES)

api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def verify_api_key(key: str | None = Security(api_key_header)) -> str:
    if not key:
        raise HTTPException(status_code=401, detail="无效或缺失 API Key")
    for valid in VALID_API_KEYS:
        if secrets.compare_digest(key, valid):  # 恒定时间比较，防时序攻击
            return key
    raise HTTPException(status_code=401, detail="无效或缺失 API Key")

@app.get("/health")
def health():
    return {"status": "ok"}  # 探活接口通常不鉴权

@app.get("/ops/hosts", dependencies=[Depends(verify_api_key)])
def list_hosts():
    return {"hosts": ["web-01", "web-02"]}

# 密钥从终端环境读取，避免进入 shell 历史或脚本源码。
export OPS_API_KEY="<由密钥系统注入的只读密钥>"
curl -s -H "X-API-Key: $OPS_API_KEY" http://127.0.0.1:8080/ops/hosts

---

Bearer Token：适合 OAuth2 风格与网关透传

from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

bearer_scheme = HTTPBearer(auto_error=False)
OPS_BEARER_TOKEN = os.environ.get("OPS_BEARER_TOKEN", "")

async def verify_bearer(
    creds: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
) -> str:
    if not creds or creds.scheme.lower() != "bearer":
        raise HTTPException(status_code=401, detail="需要 Bearer Token")
    if not OPS_BEARER_TOKEN or not secrets.compare_digest(creds.credentials, OPS_BEARER_TOKEN):
        raise HTTPException(status_code=401, detail="Token 无效")
    return creds.credentials

@app.post("/ops/restart", dependencies=[Depends(verify_bearer)])
def restart_service(host: str, service: str):
    return {"host": host, "service": service, "status": "submitted"}

curl -s -X POST -H "Authorization: Bearer your-token-here" \
  "http://127.0.0.1:8080/ops/restart?host=10.0.0.1&service=nginx"

---

简单 RBAC：只读 vs 变更

async def require_operator(key: str = Depends(verify_api_key)) -> str:
    if KEY_ROLES.get(key, Role.readonly) != Role.operator:
        raise HTTPException(status_code=403, detail="需要 operator 权限")
    return key

@app.delete("/ops/cache", dependencies=[Depends(require_operator)])
def purge_cache():
    return {"cleared": True}

角色	典型权限
readonly	查询 Pod、主机状态、巡检结果
operator	重启服务、触发同步、执行变更类 Webhook

---

安全 checklist

项	说明
HTTPS	内网也建议 TLS，防中间人嗅探 Key
密钥轮换	支持多 Key 并存，旧 Key 过期前通知调用方
审计	记录 who / path / body 摘要 / 结果
限流	Nginx 或中间件限制单 IP QPS
最小权限	只读接口不要共用 operator Key
启动校验	密钥或角色映射缺失时进程直接失败

注意

不要把真实 API Key 写入仓库或文档。CI 用 Secret 变量注入；本地用 .env 且加入 .gitignore。