Всплеск использования API-ключа (возможная утечка)

Утекший ключ провайдера — это не абстрактный риск: sk-... попадает в публичный коммит, в дамп логов, в собранный фронтенд-бандл — и через час кто-то майнит на нём эмбеддинги для своего стартапа за ваш счёт. Первый внятный сигнал обычно — счёт в конце месяца. Push при резком всплеске ловит это в первые минуты.

Три независимых сигнала, каждый из которых стоит алёрта:

• всплеск объёма — ключ делает кратно больше запросов/токенов, чем в типичный час;
• новый IP / страна — запросы пошли с адреса, которого вы раньше не видели;
• всплеск расходов — provider usage API показывает spend выше дневного бюджета.

Вариант 1: считаем rate по ключу в своём gateway

Если весь трафик к провайдеру идёт через ваш прокси/шлюз, вы сами видите, кто сколько жжёт. Скользящее окно на Redis + порог по ключу:

import os, time, requests, redis

R = redis.from_url(os.environ["REDIS_URL"])
NOTIFLY_URL   = os.environ["NOTIFLY_URL"]
NOTIFLY_TOKEN = os.environ["NOTIFLY_TOKEN"]

WINDOW   = 60          # окно в секундах
LIMIT    = 300         # запросов на ключ за окно, выше которого — подозрительно

def notify(title, msg, prio):
    requests.post(f"{NOTIFLY_URL}/message",
                  params={"token": NOTIFLY_TOKEN},
                  json={"title": title, "message": msg, "priority": prio},
                  timeout=5)

def on_request(key_id: str, ip: str):
    now  = int(time.time())
    rkey = f"rate:{key_id}:{now // WINDOW}"
    n    = R.incr(rkey)
    R.expire(rkey, WINDOW * 2)

    # запоминаем множество IP, с которых ходит ключ
    seen = f"ips:{key_id}"
    is_new_ip = R.sadd(seen, ip) == 1
    R.expire(seen, 86400)

    flag = f"alerted:{key_id}:{now // WINDOW}"
    if n > LIMIT and R.set(flag, 1, nx=True, ex=WINDOW):
        notify("🔑 Всплеск по API-ключу",
               f"Ключ {key_id}: {n} запросов за {WINDOW}с (порог {LIMIT}).\n"
               f"Последний IP: {ip}\nВозможна утечка — отозвать ключ?",
               prio=9)

    if is_new_ip and R.scard(seen) > 5:
        notify("🔑 Ключ ходит с новых IP",
               f"Ключ {key_id}: за сутки уже {R.scard(seen)} разных IP, "
               f"новый: {ip}.",
               prio=8)

priority=9 — потому что цена ошибки высокая (утечка = деньги + доступ к данным), и такой push должен быть громким, даже если это ложная тревога.

Вариант 2: опрос usage API провайдера по таймеру

Ключа-прокси нет, есть только сам провайдер? Многие отдают usage/costs через API. Скользящее сравнение с базовой линией в state-файле, чтобы не слать алёрт на каждом тике. Кладётся в scheduled-функцию YC:

import os, json, datetime, requests

STATE = "/tmp/apikey-spend.json"
DAILY_BUDGET = float(os.environ.get("DAILY_BUDGET_USD", "20"))

def handler(event, context):
    today = datetime.date.today().isoformat()
    # у OpenAI, например, есть /v1/usage?date=YYYY-MM-DD; поля зависят от провайдера
    r = requests.get("https://api.openai.com/v1/usage",
                     params={"date": today},
                     headers={"Authorization": f"Bearer {os.environ['PROVIDER_KEY']}"},
                     timeout=15)
    spent = r.json().get("total_usage", 0) / 100.0    # центы → доллары

    st = json.load(open(STATE)) if os.path.exists(STATE) else {}
    already = st.get(today, {}).get("alerted", False)

    if spent > DAILY_BUDGET and not already:
        push("💸 Расход по ключу выше бюджета",
             f"Сегодня потрачено ${spent:.2f} при бюджете ${DAILY_BUDGET:.2f}.\n"
             f"Если это не вы — ключ мог утечь.",
             prio=9)
        st[today] = {"alerted": True}
        json.dump(st, open(STATE, "w"))
    return {"statusCode": 200}

def push(t, m, prio):
    requests.post(f"{os.environ['NOTIFLY_URL']}/message",
                  params={"token": os.environ["NOTIFLY_TOKEN"]},
                  json={"title": t, "message": m, "priority": prio}, timeout=5)

Timer */10 * * * ? * — раз в 10 минут более чем достаточно, чтобы отреагировать до того, как счёт вырастет на порядок.

Что делать по алёрту

1. Отозвать/ротировать ключ в консоли провайдера — это первое действие.
2. Проверить, где он мог утечь: git log -p | grep -i sk-, публичный бандл, переменные окружения в CI.
3. Выпустить новый ключ уже с ограничением по IP/бюджету, если провайдер это умеет.

Что положить в текст алёрта

• id ключа (не сам секрет!) и его назначение;
• метрика, которая сработала: RPS / spend / новый IP;
• значение и порог, чтобы сразу понять масштаб;
• ссылка на страницу отзыва ключа в консоли провайдера.