漏洩しても請求が膨らむ前に止める — 無人パイプラインのAPIキー被害半径設計

# key_state.py — 重なり合う2枚でキーを無停止ローテーションする状態機械
# 状態ファイル（JSON）に「いまどちらを優先して使うか」を持たせる。
import json
import os
from pathlib import Path
 
STATE_PATH = Path(os.environ.get("KEY_STATE_PATH", "/secure/key_state.json"))
 
def load_state() -> dict:
    # 初回は primary を優先。secondary は空でもよい。
    if not STATE_PATH.exists():
        return {"active": "primary", "primary_env": "ANTHROPIC_API_KEY_PRIMARY",
                "secondary_env": "ANTHROPIC_API_KEY_SECONDARY"}
    return json.loads(STATE_PATH.read_text())
 
def active_api_key() -> str:
    """いま優先すべきキーを環境変数から解決して返す。
    優先側が未設定なら、もう一方にフォールバックする（猶予期間中の保険）。"""
    st = load_state()
    primary = os.environ.get(st["primary_env"], "")
    secondary = os.environ.get(st["secondary_env"], "")
    prefer = primary if st["active"] == "primary" else secondary
    fallback = secondary if st["active"] == "primary" else primary
    key = prefer or fallback
    if not key:
        raise RuntimeError("有効なAPIキーが環境にありません（両方未設定）")
    return key
 
def promote_secondary() -> None:
    """新キーをセカンダリに入れた後に呼ぶ。優先を secondary 側へ倒す。
    この時点では primary もまだ有効なので、走行中ジョブは落ちない。"""
    st = load_state()
    st["active"] = "secondary" if st["active"] == "primary" else "primary"
    STATE_PATH.write_text(json.dumps(st, ensure_ascii=False, indent=2))
    print(f"優先キーを {st['active']} に切り替えました。旧キーは猶予後に失効してください。")
 
if __name__ == "__main__":
    # 動作確認: 環境変数を仮に与えて、解決されるキーを確かめる
    os.environ.setdefault("ANTHROPIC_API_KEY_PRIMARY", "sk-ant-PRIMARY-PLACEHOLDER")
    os.environ.setdefault("ANTHROPIC_API_KEY_SECONDARY", "sk-ant-SECONDARY-PLACEHOLDER")
    print("active:", active_api_key())   # → primary 側のプレースホルダ
    promote_secondary()
    print("active:", active_api_key())   # → secondary 側のプレースホルダ

# usage_watchdog.py — 使用量のスパイクをロバストに検知する番犬（標準ライブラリのみ）
from statistics import median
 
def robust_z_scores(series):
    """各点が中央値から何MAD離れているかを返す。
    MAD=0（全点同値）の窒息を避けるため微小な下駄をはかせる。"""
    if len(series) < 3:
        return [0.0] * len(series)
    med = median(series)
    abs_dev = [abs(x - med) for x in series]
    mad = median(abs_dev)
    scale = 1.4826 * mad if mad > 0 else 1e-9  # 正規分布でσに一致させる係数
    return [(x - med) / scale for x in series]
 
def detect_spikes(usage_by_bucket, z_threshold=6.0, min_tokens=50_000):
    """usage_by_bucket: [(bucket_label, tokens), ...] 時系列順
    返り値: 警告すべきバケットのリスト。
    min_tokens 未満の小さな値は、比率が暴れても無視する（誤報抑制）。"""
    series = [t for _, t in usage_by_bucket]
    zs = robust_z_scores(series)
    alerts = []
    for (label, tokens), z in zip(usage_by_bucket, zs):
        if z >= z_threshold and tokens >= min_tokens:
            alerts.append({"bucket": label, "tokens": tokens, "robust_z": round(z, 1)})
    return alerts
 
if __name__ == "__main__":
    # 平常は8万トークン前後。最後の1時間だけ漏洩で跳ねた、という想定データ。
    data = [
        ("06-27T00", 78_000), ("06-27T01", 81_000), ("06-27T02", 79_500),
        ("06-27T03", 77_000), ("06-27T04", 82_000), ("06-27T05", 80_000),
        ("06-27T06", 1_900_000),   # ← 漏洩した第三者の利用を模した急騰
    ]
    for a in detect_spikes(data):
        print(f"⚠️ スパイク検知: {a['bucket']} tokens={a['tokens']:,} robust_z={a['robust_z']}")
    # 期待出力:
    # ⚠️ スパイク検知: 06-27T06 tokens=1,900,000 robust_z=... （大きな正の値）

# watchdog_run.py — 1時間ごとに回す番犬の本体（擬似化したレポート取得を実装に差し替える）
import json
import urllib.request
from usage_watchdog import detect_spikes
 
def fetch_usage_buckets(admin_key: str):
    """Admin の使用量レポートを時間バケットで取得して
    [(bucket_label, total_tokens), ...] に整形する。
    ※エンドポイントとレスポンス形は最新の公式ドキュメントで確認すること。"""
    # 実装イメージ（要・公式仕様確認）:
    # req = urllib.request.Request(USAGE_REPORT_URL, headers={"x-api-key": admin_key})
    # raw = json.loads(urllib.request.urlopen(req, timeout=30).read())
    # return [(b["starting_at"], b["input_tokens"] + b["output_tokens"]) for b in raw["data"]]
    raise NotImplementedError("レポート取得を実装に差し替える")
 
def notify(message: str):
    """無人運用では「自分が即座に気づける経路」に出すのが要。"""
    print(message)  # 実運用ではチャット通知などへ
 
def main():
    import os
    admin_key = os.environ["ANTHROPIC_ADMIN_KEY"]  # 番犬専用の最小権限キー
    buckets = fetch_usage_buckets(admin_key)
    alerts = detect_spikes(buckets, z_threshold=6.0, min_tokens=50_000)
    if alerts:
        lines = [f"{a['bucket']}: {a['tokens']:,} tokens (z={a['robust_z']})" for a in alerts]
        notify("API使用量スパイク検知。漏洩疑いとしてキー失効を検討:\n" + "\n".join(lines))
    else:
        print("使用量は平常範囲です。")
 
if __name__ == "__main__":
    main()