🧯 システム連携の被害半径(blast radius)を広げない境界設計 — 自動連携・自動登録の事故を防ぐ

[不正アクセス] → [システム A 侵害] → [A への自動取り込み・A→B 自動連携が正規動作として実行]
                                     → [他チャネル・他商品の顧客情報まで被害範囲に入る]

BlastRadius = {DataAsset | accessible(NHI.scopes)}
              ∪ {DataAsset | reachable via IntegrationLink chain}

{
  "summary": { "title": "CRM-ERP 連携 脅威モデル", "owner": "security-team" },
  "detail": {
    "diagrams": [
      {
        "title": "DFD Level 1",
        "diagramType": "STRIDE",
        "cells": [
          { "type": "process", "label": "CRM API Gateway" },
          { "type": "process", "label": "ERP Adapter" },
          { "type": "store",   "label": "顧客 DB" },
          { "type": "actor",   "label": "営業担当者" },
          { "type": "flow",    "label": "顧客データ(氏名/メール)", "from": "CRM API Gateway", "to": "ERP Adapter" },
          { "type": "boundary","label": "インターネット境界" }
        ]
      }
    ]
  }
}

# OAuth 2.0 認可リクエスト時のスコープを連携ごとに分割する例
# 参考: https://curity.io/resources/learn/scope-best-practices/

# 連携フローに必要な最小限のスコープのみ定義する
INTEGRATION_SCOPES = {
    "crm_to_erp":         ["contacts.read"],
    "erp_to_billing":     ["invoices.read", "invoices.write"],
    "analytics_exporter": ["reports.read"],
}

def request_token(integration_id: str, auth_server_url: str) -> str:
    scopes = INTEGRATION_SCOPES.get(integration_id, [])
    if not scopes:
        raise ValueError(f"Unknown integration: {integration_id}")
    scope_str = " ".join(scopes)
    # 実際は OAuth 2.0 Client Credentials Flow を使う
    return f"{auth_server_url}/token?scope={scope_str}&grant_type=client_credentials"

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowETLPutToSpecificPrefix",
      "Effect": "Allow",
      "Action": ["s3:PutObject"],
      "Resource": "arn:aws:s3:::my-data-lake-bucket/incoming/crm/*",
      "Condition": {
        "StringEquals": { "aws:RequestedRegion": "ap-northeast-1" },
        "Bool": { "aws:SecureTransport": "true" }
      }
    }
  ]
}

# policies/integration_authz.rego
# OPA v1.0 対応: import rego.v1 を先頭に置く
# 参考: https://www.openpolicyagent.org/docs/cheatsheet
package integration.authz

import rego.v1

default allow := false

allow if {
    input.integration_id in data.approved_integrations
    all_scopes_approved
}

all_scopes_approved if {
    approved := data.integration_scopes[input.integration_id]
    every scope in input.requested_scopes {
        scope in approved
    }
}

deny_reason := "unregistered integration" if {
    not input.integration_id in data.approved_integrations
}

deny_reason := "scope not approved" if {
    input.integration_id in data.approved_integrations
    not all_scopes_approved
}

# terraform/modules/integration-segmentation/main.tf
# 参考: https://spacelift.io/blog/terraform-security-group

resource "aws_vpc" "main" {
  cidr_block           = "10.0.0.0/16"
  enable_dns_hostnames = true
  tags = { Name = "main-vpc" }
}

resource "aws_security_group" "alb" {
  name   = "sg-alb"
  vpc_id = aws_vpc.main.id
  tags   = { Name = "sg-alb" }
}

resource "aws_vpc_security_group_ingress_rule" "alb_https" {
  security_group_id = aws_security_group.alb.id
  from_port         = 443
  to_port           = 443
  ip_protocol       = "tcp"
  cidr_ipv4         = "0.0.0.0/0"
  description       = "Allow HTTPS from internet"
}

resource "aws_vpc_security_group_egress_rule" "alb_to_app" {
  security_group_id            = aws_security_group.alb.id
  from_port                    = 8080
  to_port                      = 8080
  ip_protocol                  = "tcp"
  referenced_security_group_id = aws_security_group.app.id
  description                  = "Allow traffic to app tier only"
}

resource "aws_security_group" "app" {
  name   = "sg-app"
  vpc_id = aws_vpc.main.id
  tags   = { Name = "sg-app" }
}

resource "aws_vpc_security_group_ingress_rule" "app_from_alb" {
  security_group_id            = aws_security_group.app.id
  from_port                    = 8080
  to_port                      = 8080
  ip_protocol                  = "tcp"
  referenced_security_group_id = aws_security_group.alb.id
  description                  = "Accept only from ALB security group"
}

resource "aws_vpc_security_group_egress_rule" "app_to_db" {
  security_group_id            = aws_security_group.app.id
  from_port                    = 5432
  to_port                      = 5432
  ip_protocol                  = "tcp"
  referenced_security_group_id = aws_security_group.db.id
  description                  = "Allow DB access to DB tier only"
}

resource "aws_security_group" "db" {
  name   = "sg-db"
  vpc_id = aws_vpc.main.id
  tags   = { Name = "sg-db" }
}

resource "aws_vpc_security_group_ingress_rule" "db_from_app" {
  security_group_id            = aws_security_group.db.id
  from_port                    = 5432
  to_port                      = 5432
  ip_protocol                  = "tcp"
  referenced_security_group_id = aws_security_group.app.id
  description                  = "Accept only from App security group"
}

# k8s/network-policy/default-deny-ingress.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: default-deny-ingress
  namespace: integration-ns
spec:
  podSelector: {}
  policyTypes:
    - Ingress
---
# k8s/network-policy/allow-from-crm-ns.yaml
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: allow-from-crm-ns
  namespace: integration-ns
spec:
  podSelector:
    matchLabels:
      app: erp-adapter
  policyTypes:
    - Ingress
  ingress:
    - from:
        - namespaceSelector:
            matchLabels:
              kubernetes.io/metadata.name: crm-ns
          podSelector:
            matchLabels:
              app: crm-gateway
      ports:
        - protocol: TCP
          port: 8080

# 連携登録テンプレート（PR ゲートとして使用する YAML フォーム、実装案）
integration:
  id: INT-003
  name: "Salesforce -> Slack 商談通知"
  owner: "sales-ops@example.com"
  data_classification: "internal"      # public / internal / confidential / secret
  data_fields:
    - deal_name                        # 商談名（個人情報なし）
    - amount                           # 金額
    # 禁止: customer_email, phone_number
  credentials:
    type: "oauth2_client_credentials"
    scope: "deals.read"
    ttl_hours: 1
  approvals:
    - role: "security-team"
      approved_at: ""                  # 承認日時を記入するまでマージ不可
    - role: "privacy-team"
      approved_at: ""
  kill_switch:
    runbook_url: "https://wiki.example.com/runbooks/int-003-kill"
    on_call_slack: "#security-incidents"

# データ最小化フィルタ（実装案）
# 参考: https://complydog.com/blog/api-data-protection-developers-gdpr-implementation-guide
from typing import Any

ALLOWED_FIELDS: dict[str, set[str]] = {
    "INT-001": {"deal_name", "amount", "stage"},   # 個人情報フィールドを除外
    "INT-002": {"invoice_id", "total", "due_date"},
}

def minimize_payload(integration_id: str, raw_payload: dict[str, Any]) -> dict[str, Any]:
    allowed = ALLOWED_FIELDS.get(integration_id)
    if allowed is None:
        raise ValueError(f"Unknown integration: {integration_id}")
    filtered = {k: v for k, v in raw_payload.items() if k in allowed}
    removed = set(raw_payload.keys()) - allowed
    if removed:
        import logging
        # 削除したフィールド名のみログに記録する（値は記録しない）
        logging.getLogger(__name__).info(
            "data_minimization removed fields",
            extra={"integration_id": integration_id, "removed_fields": sorted(removed)},
        )
    return filtered

# kill switch（環境変数による緊急停止 + フラグストア、実装案）
# 参考: https://launchdarkly.com/docs/home/flags/killswitch
import os

def is_integration_enabled(integration_id: str) -> bool:
    # 1. 環境変数による緊急 kill switch（デプロイ不要・即時反映）
    env_key = f"INT_{integration_id.replace('-', '_')}_DISABLED"
    if os.getenv(env_key, "").lower() == "true":
        return False
    # 2. フラグストア（LaunchDarkly / Unleash / Flagsmith 等に差し替え可能）
    return _load_flag(integration_id)

def _load_flag(integration_id: str) -> bool:
    flags_file = os.getenv("FLAGS_CONFIG_PATH", "/etc/integration-flags/flags.json")
    try:
        import json
        with open(flags_file) as f:
            flags = json.load(f)
        return flags.get(integration_id, True)   # 未設定時はデフォルト有効
    except FileNotFoundError:
        return True

# 参考: https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html
import boto3

def get_temp_credentials(role_arn: str, session_name: str, duration_seconds: int = 900) -> dict:
    sts = boto3.client("sts", region_name="ap-northeast-1")
    response = sts.assume_role(
        RoleArn=role_arn,
        RoleSessionName=session_name,
        DurationSeconds=duration_seconds,
        # セッションポリシーでロール権限をさらに絞る
        Policy='{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:PutObject"],"Resource":"arn:aws:s3:::my-data-lake-bucket/incoming/crm/*"}]}',
    )
    creds = response["Credentials"]
    return {
        "access_key_id":     creds["AccessKeyId"],
        "secret_access_key": creds["SecretAccessKey"],
        "session_token":     creds["SessionToken"],
        "expiration":        creds["Expiration"].isoformat(),
    }

# .github/workflows/deploy.yml
permissions:
  id-token: write   # OIDC トークン要求に必須
  contents: read
jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS credentials via OIDC
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::123456789012:role/GithubActionsDeployRole
          aws-region: ap-northeast-1
          role-duration-seconds: 900
      - name: Deploy
        run: aws s3 sync ./dist s3://my-static-bucket/

{"event_id":"evt-001","ts":"2026-06-22T10:00:00Z","integration_id":"INT-001","actor_nhi_id":"svc-crm-exporter","operation":"GET","resource":"contacts","record_count":150,"outcome":"success","data_classification":"internal"}

P1（即時対応）: 廃止済みトークン利用 / 退職者IDによるアクセス / 境界外データ移動
P2（1時間以内）: 通常の10倍超のAPI呼び出し / 未知User-Agent / 地理的異常
P3（24時間以内）: スコープ外アクセス試行 / 設定変更の未承認実施

# IAM クレデンシャルの即時無効化
aws iam update-access-key --access-key-id <KEY_ID> --status Inactive

# EC2 インスタンスをネットワークから隔離（セキュリティグループ差し替え）
aws ec2 modify-instance-attribute --instance-id <INSTANCE_ID> --groups <ISOLATION_SG_ID>

# Lambda のリトライを止める（再実行による二次拡大を抑止）
aws lambda put-function-event-invoke-config --function-name <FUNCTION_NAME> --maximum-retry-attempts 0

# Circuit Breaker（実装案）。固定ウィンドウのカウンタで呼出回数を評価する
import redis, time

def is_tripped(integration_id: str, limit: int = 5, window_sec: int = 600) -> bool:
    r = redis.Redis()
    # ウィンドウ境界をキーに含めることで固定時間窓のカウントになる
    key = f"cb:{integration_id}:{int(time.time()) // window_sec}"
    count = r.incr(key)
    r.expire(key, window_sec)
    return count > limit   # True なら遮断（クールダウンへ）

しきい値例:
- 10分以内に同一ツールを5回連続呼び出し → セッション一時停止（15分クールダウン）
- 1時間のAPI呼び出しコストが閾値超過 → 自動停止 + アラート
- 通常の100倍超のデータ読み取り → 即時停止 + P1アラート