构建一个可删除云基础设施的 Slack 审批工作流

Block Kit、签名验证，以及防止一次按钮点击演变为事件的各项设计决策。

---

这张截图展示了一个机器人正在请求删除 EBS 卷的权限。点击
Approve Remediation 将创建卷快照，等待快照完成，然后删除卷，并更新消息说明发生了什么。

实现这一功能主要涉及连接代码。而让它安全运行，避免因陈旧点击、重放请求或某人临时保护资源而造成损坏，才是真正有趣的部分。

本文将通过 FinOps Sentinel 的 Slack 适配器，详细讲解上述两部分。

---

问题的形态

Slack 的交互性实际上是两个独立的通道，看起来像是一次对话：

Your app ──── incoming webhook ────▶ Slack channel
                                          │
                                     user clicks
                                          │
Your app ◀─── HTTP POST ──────────────────┘
             (a completely new request, from Slack's servers)

Enter fullscreen mode Exit fullscreen mode

点击以来自公网的未认证 POST 形式到达你注册的任何 URL。请求中没有任何信息能证明它来自 Slack，或者确实是人类点击的。这就是安全问题的核心，后续一切都由此展开。

---

第一部分：设置 Slack 应用

创建应用并获取 webhook

1. api.slack.com/apps → Create New App → From scratch
2. 命名并选择你的工作区
3. Incoming Webhooks → 开启 → Add New Webhook to Workspace
4. 选择频道，点击 Allow，复制 URL

SLACK_WEBHOOK_URL=https://hooks.slack.com/services/TXXXXX/BXXXXX/XXXXXXXX

Enter fullscreen mode Exit fullscreen mode

Webhook 只能向一个频道发送消息，且无法读取任何内容。对于通知机器人来说，这正是合适的权限：无需 OAuth 流程、无需机器人令牌、无需审查作用域。

启用交互性

Interactivity & Shortcuts → 开启 → 设置 Request URL：

https://your-domain.example/callbacks/slack

Enter fullscreen mode Exit fullscreen mode

本地测试需要隧道：

ngrok http 8000
# → https://a1b2c3d4.ngrok.app
# Request URL: https://a1b2c3d4.ngrok.app/callbacks/slack

Enter fullscreen mode Exit fullscreen mode

免费 ngrok URL 每次重启都会变化，必须每次更新 Slack 设置。如果按钮“没反应”，请首先检查此处。

获取签名密钥

Basic Information → App Credentials → Signing Secret → 显示。

SLACK_SIGNING_SECRET=your_signing_secret_here

Enter fullscreen mode Exit fullscreen mode

这是让回调可信的关键。没有它，你的端点会为任何格式正确的 POST 删除基础设施。

---

第二部分：发送值得操作的消息

Block Kit 消息是 JSON 块数组。简单版本可以立即工作：

from slack_sdk.webhook import WebhookClient

WebhookClient(webhook_url).send(text=f"Idle volume found: {volume_id}")

Enter fullscreen mode Exit fullscreen mode

但工程师需要采取行动的警报必须在约十秒内（常在手机上）回答什么、在哪里、影响多大、点击后会发生什么。

以下是实际实现：

def send_finding_alert(self, finding: Finding, resource: Resource) -> str | None:
    webhook_url = settings.slack_webhook_url
    if not webhook_url:
        raise RuntimeError("SLACK_WEBHOOK_URL is not configured")

    remediable = is_remediable(finding.rule)
    header = (
        "*FinOps Alert: Waste Detected*"
        if remediable
        else "*FinOps Advisory: Possible Idle Resource*"
    )

    blocks: list[dict[str, Any]] = [
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": (
                    f"{header}\n\n"
                    f"*Rule:* {finding.rule}\n"
                    f"*Resource:* `{resource.resource_id}` ({resource.resource_type})\n"
                    # Region is not decoration: with several regions scanned, it is
                    # the first thing an approver needs to know where to look, and
                    # remediation runs there.
                    f"*Region:* `{resource.region}`\n"
                    f"*Cost Impact:* ${finding.est_monthly_cost_usd}/mo"
                ),
            },
        }
    ]

Enter fullscreen mode Exit fullscreen mode

Region 行是在一次真实混淆后添加的。 扫描三个区域会产生三条几乎相同的警报，仅资源 ID 不同。没有 region 信息，审批人无法判断即将修改的是哪个账户角落。

LLM 摘要放在独立块中

    if finding.llm_summary:
        # Advisor output is untrusted display copy (it summarizes
        # user-controlled tags), so it goes in its own context block and is
        # never used to build action values.
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": finding.llm_summary}],
        })

Enter fullscreen mode Exit fullscreen mode

本地 LLM 为每个发现生成通俗解释。提示包含 AWS 标签，任何拥有标签权限的人都可以写入。因此模型输出被视为不可信的显示内容：仅渲染，不解析，且绝不用于构建任何系统会执行的操作。

按钮是有条件的，这正是整个设计的关键

    if remediable:
        blocks.append({
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Approve Remediation"},
                    "style": "primary",
                    "value": f"approve_{finding.id}",
                    "action_id": "approve_remediation",
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Deny"},
                    "style": "danger",
                    "value": f"deny_{finding.id}",
                    "action_id": "deny_remediation",
                },
            ],
        })
    else:
        # Metric-inferred: no playbook is allowed to run, so offering an
        # Approve button would promise an action the domain refuses.
        blocks.append({
            "type": "context",
            "elements": [{
                "type": "mrkdwn", "text": ("_Advisory only: inferred from CloudWatch metrics. "
                         "No automated remediation is available for this rule._"),
            }],
        })

Enter fullscreen mode Exit fullscreen mode

有些发现天生就是建议性质。空闲的 EC2 实例可能是热备、批处理任务的间隔，或许可证服务器，低 CPU 只是证据，而非证明。领域模型拒绝对此类规则执行修复。

早期版本对所有警报都渲染按钮。在建议性发现上点击 Approve 会返回拒绝。一个什么都不做的按钮比没有按钮更糟：它会让用户不再信任任何按钮。因此适配器先询问领域模型，再决定是否渲染。

通知预览文本

    response = WebhookClient(webhook_url).send(
        # Notification preview text. The region belongs here too, since
        # this is all a phone lock screen shows.
        text=(f"FinOps Alert: {finding.rule} on {resource.resource_id} "
              f"in {resource.region}"),
        blocks=blocks,
    )
    if response.status_code != 200:
        raise RuntimeError(f"Slack webhook returned {response.status_code}: {response.body}")

    logger.info("Sent Slack alert for finding %s", finding.id)
    # Incoming webhooks return no message timestamp; edits happen via the
    # interaction payload's response_url instead.
    return None

Enter fullscreen mode Exit fullscreen mode

容易忽略：当存在 blocks 时，text 成为推送通知预览。省略它，手机将显示“无法显示此内容”。

注意 raise。 发送失败不会被吞没。调用方会让发现保持未通知状态，以便下次扫描重试。吞没错误会导致警报静默丢失，成本警报意味着金钱继续流失却无人知晓。

---

第三部分：安全接收点击

先验证再解析

def parse_callback(
    self, raw_body: bytes, headers: Mapping[str, str]
) -> tuple[Decision, dict[str, Any]]:
    self._verify_signature(raw_body, headers)     # ← first line, always
    ...

Enter fullscreen mode Exit fullscreen mode

签名验证是第一步操作。先解析再验证意味着你已经对敌意输入运行了 JSON 解码器。

CALLBACK_MAX_AGE_SECONDS = 60 * 5

def _verify_signature(self, raw_body: bytes, headers: Mapping[str, str]) -> None:
    secret = settings.slack_signing_secret
    if not secret:
        # No secret configured: verification bypassed (local testing).
        return

    timestamp = headers.get("x-slack-request-timestamp") or headers.get(
        "X-Slack-Request-Timestamp"
    )
    signature = headers.get("x-slack-signature") or headers.get(
        "X-Slack-Signature"
    )
    if not timestamp or not signature:
        raise PermissionError("Missing Slack signature headers")

    try:
        age = abs(time.time() - int(timestamp))
    except ValueError as exc:
        raise PermissionError("Invalid Slack timestamp header") from exc
    if age > CALLBACK_MAX_AGE_SECONDS:
        raise PermissionError("Slack request timestamp expired")

    verifier = SignatureVerifier(secret)
    if not verifier.is_valid(body=raw_body, timestamp=timestamp, signature=signature):
        raise PermissionError("Invalid Slack signature")

Enter fullscreen mode Exit fullscreen mode

三项检查，针对三种不同攻击：

检查	阻止
Header 存在	对公网端点的随意探测
时间戳在 5 分钟内	重放：捕获有效请求后稍后重发
HMAC 有效	伪造

重放检查最容易被跳过。签名永远有效，除非你限制其年龄；捕获一次合法的批准回调即可无限重放。Slack 发送时间戳正是为了让你拒绝过期请求。

SignatureVerifier 来自 slack_sdk，使用恒定时间比较，值得使用而非手动实现 HMAC 并用 == 比较。

if not secret: return 旁路是故意为本地测试提供的便利，但也是地雷。 这意味着未配置的部署会接受任何内容。在笔记本上没问题，但在任何可公开访问的地方都很危险。如果要为生产环境加固，我会让它默认拒绝，除非显式设置 ALLOW_UNSIGNED_CALLBACKS=true。

原始字节很重要

@app.post("/callbacks/{channel}")
async def notifier_callback(channel: str, request: Request) -> dict[str, Any]:
    raw_body = await request.body()          # ← bytes, not a parsed model
    decision, reply_context = notifier.parse_callback(raw_body, request.headers)

Enter fullscreen mode Exit fullscreen mode

HMAC 是针对 Slack 发送的确切字节计算的。如果先让 FastAPI 将 body 解析为 Pydantic 模型，就无法重建这些字节：键顺序、空白和编码都会改变。签名验证随后会神秘失败，浪费你一下午时间。

解析负载

Slack 以 application/x-www-form-urlencoded 发送，包含一个 payload 字段，里面是 JSON。真的。

    form = urllib.parse.parse_qs(raw_body.decode("utf-8"))
    payload_values = form.get("payload")
    if not payload_values:
        raise ValueError("No payload found")

    try:
        payload = json.loads(payload_values[0])
    except json.JSONDecodeError as exc:
        raise ValueError("Payload is not valid JSON") from exc

    actions = payload.get("actions") or []
    if not actions:
        raise ValueError("No actions in payload")

    value = str(actions[0].get("value", ""))
    action: Literal["approve", "deny"]
    if value.startswith("approve_"):
        action, finding_id = "approve", value.removeprefix("approve_")
    elif value.startswith("deny_"):
        action, finding_id = "deny", value.removeprefix("deny_")
    else:
        raise ValueError(f"Unrecognized action value: {value!r}")

    actor = payload.get("user", {}).get("username") or payload.get("user", {}).get(
        "id", "unknown"
    )

    decision = Decision(
        finding_id=finding_id,
        actor=actor,
        action=action,
        decided_at=datetime.now(UTC),
        channel=self.channel_name,
    )
    reply_context = {
        "response_url": payload.get("response_url"),
        "original_blocks": payload.get("message", {}).get("blocks", []),
    }
    return decision, reply_context

Enter fullscreen mode Exit fullscreen mode

输出是一个领域对象。所有 Slack 相关的东西（表单编码、payload 包装、response_url）都在此边界停止。接收 Decision 的领域服务不知道 Slack 的存在。

按钮值携带系统生成的发现 ID，绝不是模型输出，也绝不是用户输入的内容。解析器拒绝任何不符合预期前缀的内容。

异常到状态码的映射

    try:
        decision, reply_context = notifier.parse_callback(raw_body, request.headers)
    except PermissionError as exc:
        raise HTTPException(status_code=401, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

Enter fullscreen mode Exit fullscreen mode

端口定义了异常契约：PermissionError 用于认证失败，ValueError 用于格式错误输入，因此任何 notifier 实现都可以干净地映射到 HTTP，而路由无需知道配置的是哪一个。

---

第四部分：点击后会发生什么

路由立即将控制权交给领域：

def _decide(finding_id: str, action: str, actor: str, channel: str) -> bool:
    repo = get_repository()
    if action == "approve":
        return approve_finding(
            finding_id,
            repo,
            # The resolver, not a gateway: the service picks the endpoint for
            # the finding's own region.
            get_cloud_gateway,
            actor=actor,
            channel=channel,
            dry_run=settings.dry_run,
        )
    return deny_finding(finding_id, repo, actor=actor, channel=channel)

Enter fullscreen mode Exit fullscreen mode

在 approve_finding 内部，防护措施在批准时刻重新检查，而非信任检测时刻：

    if resource.lifecycle == ResourceLifecycle.DELETED:
        _audit(repo, "approve_blocked_resource_gone", finding.id, {...})
        return False

    if finding.protected or rules.is_protected(resource.current_tags):
        _audit(repo, "approve_blocked_protected", finding.id, {...})
        return False

    if not rules.is_remediable(finding.rule):
        _audit(repo, "approve_blocked_notify_only", finding.id, {...})
        return False

    playbook = rules.PLAYBOOK_ALLOWLIST.get(resource.resource_type)
    if playbook is None:
        _audit(repo, "approve_blocked_no_playbook", finding.id, {...})
        return False

Enter fullscreen mode Exit fullscreen mode

警报可能在 Slack 中停留数小时。在此期间，有人可能已将资源标记为 finops:protected=true，或已手动删除。较新的意图胜出，每次拒绝都有独立的审计事件名称，以便从数据库回答“为什么没有执行？”

双击问题

Slack 按钮可以轻松双击，Slack 本身也会在超时时重试。没有保护措施，你会得到两次修复。

    if not repo.transition_finding(finding.id, finding.status, FindingStatus.APPROVED):
        return False  # lost the race, someone else already decided

Enter fullscreen mode Exit fullscreen mode

由 compare-and-swap 支持：

UPDATE findings SET status = :new WHERE id = :id AND status = :expected

Enter fullscreen mode Exit fullscreen mode

仅当 rowcount == 1 时返回 True。第二次点击会发现状态已是 APPROVED，匹配失败，不更新，返回 False。最多一次修复，从结构上保证，而非靠运气。

编辑消息

def confirm_decision(self, reply_context: dict[str, Any], text: str) -> None:
    response_url = reply_context.get("response_url")
    if not response_url:
        return

    blocks: list[dict[str, Any]] = []
    original_blocks = reply_context.get("original_blocks") or []
    if original_blocks:
        blocks.append(original_blocks[0])  # keep the alert text, drop the buttons
    blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": text}})

    response = WebhookClient(response_url).send(
        text=text, blocks=blocks, replace_original=True
    )
    if response.status_code != 200:
        logger.error("Failed to update Slack message: %s", response.body)

Enter fullscreen mode Exit fullscreen mode

保留块 [0] 并丢弃其余部分，保留上下文（警报原本的内容），同时移除按钮，使已决定的发现无法再次点击。response_url 有效期 30 分钟，最多使用 5 次，足够一次编辑。

结果文本区分每种终端状态：

outcome = f"*Approved* by @{decision.actor}: remediation executed{where}."
outcome = f"*Approved* by @{decision.actor}: DRY RUN, no resources were changed."
outcome = f"*Denied* by @{decision.actor}: no action taken, finding closed."
outcome = f"*Remediation failed*{where} after approval by @{decision.actor}: ..."

Enter fullscreen mode Exit fullscreen mode

仅“Approved”无法告诉操作员实际是否发生了更改。

---

第五部分：无需工作区即可测试

整个流程无需 Slack 账户即可测试，因为 notifier 是一个端口：

class FakeNotifier(Notifier):
    """Records what was sent. Zero Slack, zero HTTP."""

    def __init__(self):
        self.alerts: list[tuple[str, str]] = []
        self.digests: list[tuple[str, list[str]]] = []

    @property
    def channel_name(self):
        return "fake"

    def send_finding_alert(self, finding, resource):
        self.alerts.append((finding.id, resource.resource_id))
        return f"msg-{len(self.alerts)}"

    def send_digest(self, title, sections):
        self.digests.append((title, sections))
        return f"digest-{len(self.digests)}"

Enter fullscreen mode Exit fullscreen mode

批准并修复流程针对 FakeNotifier + 假云网关 + 内存仓库运行。如果通过，替换 Slack 为 Telegram 不会破坏审批逻辑，因为审批逻辑从未知道 Slack 的存在。

对于适配器本身，mock webhook 客户端并断言 blocks：

def send_and_capture(finding, resource):
    """Send an alert through a mocked webhook and return the blocks sent."""
    settings.slack_webhook_url = "https://hooks.slack.test/T/B/X"
    try:
        with patch("finops_sentinel.adapters.notifications.slack.WebhookClient") as client_cls:
            client_cls.return_value.send.return_value = MagicMock(status_code=200, body="ok")
            SlackAdapter().send_finding_alert(finding, resource)
            return client_cls.return_value.send.call_args.kwargs["blocks"]
    finally:
        settings.slack_webhook_url = None


def test_advisory_findings_get_no_buttons():
    blocks = send_and_capture(make_finding("ec2_idle"), resource)
    assert not any(b["type"] == "actions" for b in blocks)

Enter fullscreen mode Exit fullscreen mode

该测试编码了一个产品决策，而非实现细节。如果有人后来为建议性警报添加按钮，测试就会失败，这正是你希望被打断的时刻。

---

故障排除

症状	原因
401 Invalid Slack signature	密钥不匹配，或在验证前解析了 body
按钮无反应	Request URL 与当前隧道不匹配；ngrok 重启后会轮换
This content can't be displayed on mobile	缺少 text= 回退与 blocks= 一起提供
超时	Slack 要求 3 秒内返回 200 OK；慢速工作应在响应后执行
消息未编辑	response_url 已过期（30 分钟 / 5 次使用）
重复修复	状态转换没有 compare-and-swap

关于 3 秒规则：本项目的状态转换是单条索引 UPDATE，因此能在预算内很好地响应。如果你的修复很慢，请立即确认，然后在后台执行，否则 Slack 会重试，而你现在依赖的幂等性可能并不存在。

---

我会做不同的改进

缺少签名密钥时默认拒绝。 旁路在本地方便，但在其他任何地方都很危险。

使用 bot token 而非 incoming webhook。 Webhook 不返回消息时间戳，因此编辑依赖 response_url 及其 30 分钟窗口。bot token 允许你在任何时间对任何消息使用 chat.update。

为高影响操作添加确认对话框。 Block Kit 支持在按钮上使用原生 confirm 对象，在误触和删除数据库之间增加一次额外点击。

---

总结

Slack 集成大约 260 行代码。大约三分之一是 Block Kit 格式化，三分之一是签名验证，三分之一是将负载转换为领域对象。

安全保障并不在 Slack 代码中。它在于按钮点击是一个请求，而非命令：领域重新检查每项防护，使用原子转换确保双击不会执行两次，并在规则不可修复时直接拒绝。

Slack 是驱动适配器。安全存在于内部。

---

完整源码：github.com/boazleleina/finops-sentinel。参见 adapters/notifications/slack.py 和 SLACK_SETUP.md。