Block Kit、署名検証、そしてボタンクリックがインシデントにならないための設計判断。
このスクリーンショットは、EBSボリュームを削除する許可を求めるボットです。
修復を承認をクリックすると、ボリュームのスナップショットが作成され、スナップショットの完了を待ってからボリュームが削除され、メッセージが編集されて何が起こったかが表示されます。
これを機能させるのは主に配管作業です。これを安全に機能させ、古いクリック、リプレイされたリクエスト、またはその間に誰かが保護したリソースが損害を引き起こさないようにすることが、興味深い部分です。
これは、FinOps SentinelのSlackアダプターを使用して、両方を説明します。
問題の形状
Slackのインタラクティブ機能は、会話のように見える2つの別々のチャンネルです：
Your app ──── incoming webhook ────▶ Slack channel
│
user clicks
│
Your app ◀─── HTTP POST ──────────────────┘
(a completely new request, from Slack's servers)
フルスクリーンモードに入る フルスクリーンモードを終了
クリックはパブリックインターネットからの認証されていないPOSTとして登録したURLに到着します。リクエストのどの部分もそれがSlackから来たことや、人間が何かをクリックしたことを証明しません。それが一文で表したセキュリティ問題であり、以下はすべてそこから導き出されます。
パート1：Slackアプリのセットアップ
アプリを作成してWebhookを取得する
- api.slack.com/apps → 新しいアプリを作成 → 最初から
- 名前を付け、ワークスペースを選択
- Incoming Webhooks → オンに切り替え → ワークスペースに新しいWebhookを追加
- チャンネルを選択し、許可をクリックしてURLをコピー
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/TXXXXX/BXXXXX/XXXXXXXX
フルスクリーンモードに入る フルスクリーンモードを終了
Webhookは正確に1つのチャンネルに投稿し、何も読み取ることができません。通知ボットにとっては適切な権限レベルです：OAuthフローなし、ボットトークンなし、確認するスコープなし。
インタラクティブ機能を有効にする
インタラクティブ機能とショートカット → オンに切り替え → リクエストURLを設定：
https://your-domain.example/callbacks/slack
フルスクリーンモードに入る フルスクリーンモードを終了
ローカルではトンネルが必要です：
ngrok http 8000
# → https://a1b2c3d4.ngrok.app
# Request URL: https://a1b2c3d4.ngrok.app/callbacks/slack
フルスクリーンモードに入る フルスクリーンモードを終了
無料のngrok URLは再起動するたびに変更され、そのたびにSlackを更新する必要があります。混乱を避けるために：ボタンが「何もしない」場合は、まずこれを確認してください。
署名シークレットを取得する
基本情報 → アプリ認証情報 → 署名シークレット → 表示。
SLACK_SIGNING_SECRET=your_signing_secret_here
フルスクリーンモードに入る フルスクリーンモードを終了
これがコールバックを信頼できるものにするものです。これがないと、エンドポイントは適切に形成されたPOSTを送信する誰でもインフラを削除してしまいます。
パート2：アクションに値するメッセージの送信
Block KitメッセージはブロックのJSON配列です。素朴なバージョンはすぐに動作します：
from slack_sdk.webhook import WebhookClient
WebhookClient(webhook_url).send(text=f"Idle volume found: {volume_id}")
フルスクリーンモードに入る フルスクリーンモードを終了
しかし、エンジニアが対応する必要があるアラートは、何が、どこで、どれくらいのコストで、クリックしたらどうなるかを約10秒以内に、しばしば電話上で回答する必要があります。
実際の実装は次のとおりです：
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"
),
},
}
]
フルスクリーンモードに入る フルスクリーンモードを終了
リージョン行は実際の混乱の後に追加されました。 3つのリージョンをスキャンすると、リソースIDのみが異なる3つのほぼ同一のアラートが生成されます。リージョンがないと、承認者はどのアカウントのどの部分を変更しようとしているのか分かりません。
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}],
})
フルスクリーンモードに入る フルスクリーンモードを終了
ローカルのLLMは各findingの平易な説明を書きます。プロンプトには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._"),
}],
})
フルスクリーンモードに入る フルスクリーンモードを終了
一部のfindingは設計上アドバイザリです。アイドル状態のEC2インスタンスは、ウォームスタンバイ、バッチワーカーの実行間、またはライセンスサーバーである可能性があり、低いCPUは証拠であり、証明ではありません。ドメインはこれらのルールの修復を拒否します。
初期バージョンではすべてのアラートにボタンがレンダリングされていました。アドバイザリfindingで承認をクリックすると拒否が返されました。何もしないボタンはボタンがないよりも悪い：人々はすべてのボタンを疑うようになります。そのため、アダプターはまずドメインに問い合わせ、適切にレンダリングします。
通知プレビューテキスト
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
フルスクリーンモードに入る フルスクリーンモードを終了
忘れがちですが：blocksが存在する場合、textはプッシュ通知のプレビューになります。これを省略すると、電話には「このコンテンツは表示できません。」と表示されます。
raiseに注意してください。 送信の失敗は飲み込まれません。呼び出し元はfindingを未通知のままにして、次のスキャンで再試行できるようにします。飲み込むとアラートがサイレントで失われ、コストアラートの場合、お金が燃え続け、誰も知らないことになります。
パート3：クリックを安全に受信する
解析前に検証する
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
...
フルスクリーンモードに入る フルスクリーンモードを終了
署名検証は最初に起こるべきことです。解析してから検証すると、すでに敵対的な入力に対して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")
フルスクリーンモードに入る フルスクリーンモードを終了
3つのチェック、3つの異なる攻撃：
| チェック | 阻止するもの |
|---|---|
| ヘッダーの存在 | パブリックエンドポイントの偶発的な探索 |
| 5分以内のタイムスタンプ | リプレイ：後で再送信されるキャプチャされた有効なリクエスト |
| HMACが有効 | 偽造 |
リプレイチェックは人々がスキップするものです。署名は有効期限を指定しない限り永遠に有効です。有効な承認コールバックを1つキャプチャすれば、無期限にリプレイできます。Slackがタイムスタンプを送信するのは、まさに古いリクエストを拒否できるようにするためです。
SignatureVerifierはslack_sdkから来ており、定数時間比較を行います。HMACを手動で実装して==で比較するのではなく、これを使用する価値があります。
if not secret: returnバイパスは意図的なローカルテスト用の便宜であり、地雷です。 これは設定されていないデプロイメントが何でも受け入れることを意味します。ノートPCでは問題ありませんが、到達可能な場所では危険です。本番環境用に強化するなら、明示的に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)
フルスクリーンモードに入る フルスクリーンモードを終了
HMACはSlackが送信した正確なバイトに対して計算されます。FastAPIがボディをPydanticモデルに先に解析させると、それらを再構築できません：キーの順序、空白、エンコーディングがすべて変わります。署名検証は神秘的に失敗し、午後を失うことになります。
ペイロードの解析
Slackは単一のpayloadフィールドにJSONを含むapplication/x-www-form-urlencodedを送信します。本当に。
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
フルスクリーンモードに入る フルスクリーンモードを終了
出力はドメインオブジェクトです。Slack形状のすべて（フォームエンコーディング、payloadラッパー、response_url）はこの境界で停止します。このDecisionを受け取るドメインサービスは、Slackが存在することを知りません。
ボタンの値はシステム生成のfinding 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
フルスクリーンモードに入る フルスクリーンモードを終了
ポートは例外契約を定義します：PermissionErrorは認証失敗用、ValueErrorは不正な入力用なので、どのnotifier実装もルートがどちらが設定されているかを知らずにHTTPにクリーンにマッピングできます。
パート4：クリック後の処理
ルートは即座にドメインに処理を委譲します：
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)
フルスクリーンモードに入る フルスクリーンモードを終了
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
フルスクリーンモードに入る フルスクリーンモードを終了
アラートはSlackに何時間も留まることがあります。その間に誰かがリソースにfinops:protected=trueとタグ付けしたり、帯域外で削除したりする可能性があります。より新しい意図が勝ちます、そして各拒否は独自の監査イベント名を取得するので、「なぜアクションが実行されなかったのか？」がデータベースから回答可能になります。
ダブルクリックの問題
Slackのボタンは簡単にダブルクリックでき、Slack自体はタイムアウト時に再試行します。保護なしでは2回の修復が発生します。
if not repo.transition_finding(finding.id, finding.status, FindingStatus.APPROVED):
return False # lost the race, someone else already decided
フルスクリーンモードに入る フルスクリーンモードを終了
compare-and-swapでバックアップ：
UPDATE findings SET status = :new WHERE id = :id AND status = :expected
フルスクリーンモードに入る フルスクリーンモードを終了
rowcount == 1の場合のみTrueを返します。2回目のクリックはステータスがすでにAPPROVEDであることを見つけ、一致するものがなく、何も更新せず、Falseを返します。タイミングの運に頼るのではなく、構造的に最大1回の修復です。
メッセージの編集
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)
フルスクリーンモードに入る フルスクリーンモードを終了
ブロック[0]を保持して残りを削除することで、コンテキスト（アラートが何だったか）を保持しつつ、ボタンを削除して決定されたfindingが再度クリックできないようにします。response_urlは30分間、5回使用可能で、1回の編集には十分です。
結果テキストはすべての終端状態を区別します：
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}: ..."
フルスクリーンモードに入る フルスクリーンモードを終了
「Approved」だけでは、オペレーターは何かが実際に変更されたかどうかを知ることができません。
パート5：ワークスペースなしでのテスト
notifierがポートであるため、Slackアカウントなしでフロー全体をテストできます：
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)}"
フルスクリーンモードに入る フルスクリーンモードを終了
承認と修復のフローはFakeNotifier + 偽のクラウドゲートウェイ + インメモリリポジトリに対して実行されます。これが通れば、SlackをTelegramに置き換えても承認ロジックが壊れることはありません。なぜなら、承認ロジックはSlackのことを知らなかったからです。
アダプター自体については、Webhookクライアントをモックしてブロックをアサートします：
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)
フルスクリーンモードに入る フルスクリーンモードを終了
このテストは実装の詳細ではなく、製品の決定をエンコードしています。誰かが後でアドバイザリアラートにボタンを追加した場合に失敗しますが、まさにその時に中断されるべきです。
トラブルシューティング
| 症状 | 原因 |
|---|---|
401 Invalid Slack signature |
シークレットの不一致、または検証前にボディが解析された |
| ボタンが何もしない | リクエストURLが現在のトンネルと一致しない。ngrokは再起動時にローテーションする |
モバイルでこのコンテンツは表示できません |
blocks=と一緒にtext=フォールバックが欠けている |
| タイムアウト | Slackは3秒以内に200 OKを求める。応答後に遅い作業を行う |
| メッセージが編集されない |
response_urlが期限切れ（30分/5回使用） |
| 重複した修復 | ステータス遷移にcompare-and-swapがない |
3秒ルールについて：このプロジェクトの状態遷移は単一のインデックス付きUPDATEなので、予算内にうまく応答します。修復が遅い場合は、即座に確認応答し、バックグラウンドで作業を行う必要があります。そうしないとSlackが再試行し、冪等性に依存することになりますが、それは持っていないかもしれません。
違ったやり方
署名シークレットがない場合はフェイルクローズする。 バイパスはローカルでは便利ですが、他の場所では危険です。
受信Webhookの代わりにボットトークンを使用する。 Webhookはメッセージのタイムスタンプを返さないため、編集はresponse_urlとその30分のウィンドウに依存します。ボットトークンはいつでもどのメッセージでもchat.updateを提供します。
影響の大きいアクションには確認ダイアログを追加する。 Block Kitはボタンにネイティブなconfirmオブジェクトをサポートしており、誤タップとデータベースの削除の間に1回の追加タップを挟みます。
まとめ
Slack統合は約260行です。およそ3分の1がBlock Kitのフォーマット、3分の1が署名検証、3分の1がペイロードをドメインオブジェクトに変換するものです。
これを安全にするものはSlackコードにはありません。ボタンクリックはコマンドではなくリクエストであるということです：ドメインはすべてのガードレールを再チェックし、ダブルクリックが2回実行されないようにアトミックな遷移を使用し、ルールが修復可能でなかった場合は完全に拒否します。
Slackは駆動するアダプターです。安全性は内部にあります。
完全なソース：github.com/boazleleina/finops-sentinel。adapters/notifications/slack.pyとSLACK_SETUP.mdを参照してください。
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.