keys still work after their owner loses project access

"""Report OpenAI API keys whose owner no longer has access to the project.

Read only. GET requests and nothing else. This one needs an ORGANIZATION ADMIN
key (sk-admin-...), because every /v1/organization/* endpoint rejects a project
key outright; an admin key provisioned read-only is enough and is what you
should give it. The repair is printed, never performed: a key on this list may
still be carrying production traffic, and revoking it before you know that is
how a cleanup becomes an outage.
"""
import argparse
import logging
import os
import sys
import time

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("openai_orphaned_key_audit")

API = "https://api.openai.com/v1"
DAY = 86400

# Worst first, so the report leads with the key that is still serving traffic
# rather than with the harmless one that has never been used.
SEVERITY = {"serving": 4, "orphaned": 3, "unknown": 2, "dormant": 1, "in-force": 0}


def owner_label(key):
    """Best identity available for whoever owns a key. Pure.

    owner.type is "user" or "service_account"; only the user branch carries an
    email, and a service account carries a name instead. Falling back to the
    type rather than to "?" keeps the row readable when neither is populated.
    """
    owner = key.get("owner") or {}
    user = owner.get("user") or {}
    account = owner.get("service_account") or {}
    return (user.get("email") or user.get("name") or account.get("name")
            or owner.get("type") or "unknown owner")


def verdict(key, now, hot_days=7):
    """Classify one organization.project.api_key object.

    Pure, so the rules can be read and tested without an admin credential and
    without a network. `now` is a unix timestamp, and so are `last_used_at` and
    `created_at` on this object; `last_used_at` is null on a key that has never
    authenticated a request.

    Returns (state, detail).
    """
    raw = key.get("owner_project_access")
    if raw is None:
        return ("unknown",
                "no owner_project_access on this object: ask for it explicitly "
                "with owner_project_access=any and re-read, rather than taking "
                "the absence for active")
    access = str(raw).strip().lower()
    if access == "active":
        return ("in-force", "owner still has access to this project")
    if access != "inactive":
        return ("unknown", "unrecognised owner_project_access %r" % (raw,))

    last = key.get("last_used_at")
    if last is None:
        return ("dormant",
                "owner has lost project access and this key has never "
                "authenticated a request. Nothing depends on it, so it is the "
                "safe one to revoke first.")
    age = (int(now) - int(last)) // DAY
    if age <= hot_days:
        return ("serving",
                "owner has lost project access and the key authenticated a "
                "request %d day(s) ago. Something in production is still "
                "holding it: re-issue before you revoke." % age)
    return ("orphaned",
            "owner has lost project access; last used %d day(s) ago" % age)


def get(session, path, **params):
    r = session.get(API + path, params=params, timeout=30)
    if r.status_code == 401:
        raise SystemExit("401 from OpenAI: /v1/organization/* needs an "
                         "organization admin key (sk-admin-...), not a project key")
    r.raise_for_status()
    return r.json()


def paged(session, path, **params):
    """Walk a cursor-paginated admin listing."""
    params.setdefault("limit", 100)
    while True:
        page = get(session, path, **params)
        data = page.get("data") or []
        for item in data:
            yield item
        if not page.get("has_more") or not data:
            return
        params["after"] = page.get("last_id") or data[-1].get("id")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--hot-days", type=int, default=7,
                    help="a key used inside this many days counts as live traffic")
    ap.add_argument("--all-keys", action="store_true",
                    help="read every key (owner_project_access=any), not only the "
                         "inactive-owner ones, for a full inventory")
    args = ap.parse_args()

    admin = os.environ.get("OPENAI_ADMIN_KEY")
    if not admin:
        log.error("set OPENAI_ADMIN_KEY to an organization admin key (sk-admin-...); "
                  "a project key cannot read /v1/organization/*")
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + admin})
    now = int(time.time())
    scope = "any" if args.all_keys else "inactive"

    rows = []
    projects = 0
    # include_archived=true, because an archived project still holds live keys
    # and is absent from the default listing.
    for project in paged(s, "/organization/projects", include_archived="true"):
        projects += 1
        path = "/organization/projects/%s/api_keys" % project["id"]
        for key in paged(s, path, owner_project_access=scope):
            state, detail = verdict(key, now, args.hot_days)
            rows.append((state, detail, project, key))

    rows.sort(key=lambda r: (-SEVERITY.get(r[0], 2), -(r[3].get("last_used_at") or 0)))

    bad = 0
    for state, detail, project, key in rows:
        line = "%-9s %s / %s  %s  %s" % (
            state, project.get("name") or project["id"], owner_label(key),
            key.get("redacted_value") or "?", detail)
        if state == "in-force":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  repair: mint a replacement under a service account, deploy "
                    "it, confirm last_used_at stops moving, then remove this one: "
                    "DELETE %s/organization/projects/%s/api_keys/%s",
                    API, project["id"], key.get("id"))

    log.info("%d key(s) read across %d project(s), %d whose owner no longer has "
             "project access", len(rows), projects, bad)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report OpenAI API keys whose owner no longer has access to the project.
 *
 * Read only. GET requests and nothing else, and it needs an ORGANIZATION ADMIN
 * key (sk-admin-...) because /v1/organization/* rejects project keys. An admin
 * key provisioned read-only is enough. The repair is printed, never performed.
 */
const API = 'https://api.openai.com/v1';
const DAY = 86400;

// Worst first, so the report leads with the key still serving traffic.
const SEVERITY = { serving: 4, orphaned: 3, unknown: 2, dormant: 1, 'in-force': 0 };

/** Best identity available for whoever owns a key. Pure. */
export function ownerLabel(key) {
  const owner = key.owner ?? {};
  const user = owner.user ?? {};
  const account = owner.service_account ?? {};
  return user.email || user.name || account.name || owner.type || 'unknown owner';
}

/**
 * Classify one organization.project.api_key object. Pure, so the rules can be
 * tested without an admin credential and without a network.
 */
export function verdict(key, now, hotDays = 7) {
  const raw = key.owner_project_access;
  if (raw === undefined || raw === null) {
    return ['unknown',
      'no owner_project_access on this object: ask for it explicitly with ' +
      'owner_project_access=any and re-read, rather than taking the absence ' +
      'for active'];
  }
  const access = String(raw).trim().toLowerCase();
  if (access === 'active') return ['in-force', 'owner still has access to this project'];
  if (access !== 'inactive') {
    return ['unknown', `unrecognised owner_project_access ${JSON.stringify(raw)}`];
  }

  const last = key.last_used_at;
  if (last === undefined || last === null) {
    return ['dormant',
      'owner has lost project access and this key has never authenticated a ' +
      'request. Nothing depends on it, so it is the safe one to revoke first.'];
  }
  const age = Math.floor((Number(now) - Number(last)) / DAY);
  if (age <= hotDays) {
    return ['serving',
      `owner has lost project access and the key authenticated a request ${age} ` +
      'day(s) ago. Something in production is still holding it: re-issue before ' +
      'you revoke.'];
  }
  return ['orphaned', `owner has lost project access; last used ${age} day(s) ago`];
}

async function get(adminKey, path, params = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
  const res = await fetch(url, { headers: { Authorization: `Bearer ${adminKey}` } });
  if (res.status === 401) {
    throw new Error('401 from OpenAI: /v1/organization/* needs an organization ' +
                    'admin key (sk-admin-...), not a project key');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

async function* paged(adminKey, path, params = {}) {
  const q = { limit: 100, ...params };
  for (;;) {
    const page = await get(adminKey, path, q);
    const data = page.data ?? [];
    for (const item of data) yield item;
    if (!page.has_more || data.length === 0) return;
    q.after = page.last_id ?? data[data.length - 1].id;
  }
}

async function main() {
  const adminKey = process.env.OPENAI_ADMIN_KEY;
  if (!adminKey) {
    console.error('set OPENAI_ADMIN_KEY to an organization admin key (sk-admin-...); ' +
                  'a project key cannot read /v1/organization/*');
    process.exitCode = 2;
    return;
  }

  const hotDays = Number(process.env.HOT_DAYS ?? 7);
  const scope = process.env.ALL_KEYS ? 'any' : 'inactive';
  const now = Math.floor(Date.now() / 1000);

  const rows = [];
  let projects = 0;
  // include_archived=true: an archived project still holds live keys and is
  // absent from the default listing.
  for await (const project of paged(adminKey, '/organization/projects',
                                    { include_archived: 'true' })) {
    projects += 1;
    const path = `/organization/projects/${project.id}/api_keys`;
    for await (const key of paged(adminKey, path, { owner_project_access: scope })) {
      const [state, detail] = verdict(key, now, hotDays);
      rows.push({ state, detail, project, key });
    }
  }

  rows.sort((a, b) =>
    (SEVERITY[b.state] ?? 2) - (SEVERITY[a.state] ?? 2) ||
    (b.key.last_used_at ?? 0) - (a.key.last_used_at ?? 0));

  let bad = 0;
  for (const { state, detail, project, key } of rows) {
    const line = `${state.padEnd(9)} ${project.name ?? project.id} / ` +
                 `${ownerLabel(key)}  ${key.redacted_value ?? '?'}  ${detail}`;
    if (state === 'in-force') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn('  repair: mint a replacement under a service account, deploy it, ' +
                 'confirm last_used_at stops moving, then remove this one: ' +
                 `DELETE ${API}/organization/projects/${project.id}/api_keys/${key.id}`);
  }

  console.log(`${rows.length} key(s) read across ${projects} project(s), ${bad} ` +
              'whose owner no longer has project access');
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not fire main(), fail on the missing key, and set a non-zero exit code
// that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from openai_orphaned_key_audit import owner_label, verdict

NOW = 1_756_000_000  # a fixed clock, so these never age out


def make(**over):
    key = {
        "id": "key_abc",
        "redacted_value": "sk-proj-...aB3d",
        "owner_project_access": "active",
        "last_used_at": NOW - 3600,
        "owner": {"type": "user", "user": {"email": "dev@example.com"}},
    }
    key.update(over)
    return key


def test_active_owner_is_not_a_finding():
    state, _ = verdict(make(), NOW)
    assert state == "in-force"


def test_inactive_owner_used_today_is_production_traffic():
    state, detail = verdict(make(owner_project_access="inactive"), NOW)
    assert state == "serving"
    assert "re-issue" in detail


def test_inactive_owner_long_idle_is_orphaned_not_serving():
    state, detail = verdict(
        make(owner_project_access="inactive", last_used_at=NOW - 90 * 86400), NOW)
    assert state == "orphaned"
    assert "90 day(s)" in detail


def test_inactive_owner_never_used_is_the_safe_one():
    state, detail = verdict(
        make(owner_project_access="inactive", last_used_at=None), NOW)
    assert state == "dormant"
    assert "revoke first" in detail


def test_missing_access_field_is_never_read_as_active():
    # The whole point: an absent field is an unanswered question, not a clean org.
    key = make()
    del key["owner_project_access"]
    state, detail = verdict(key, NOW)
    assert state == "unknown"
    assert "owner_project_access=any" in detail


def test_unrecognised_access_value_is_not_silently_fine():
    assert verdict(make(owner_project_access="pending"), NOW)[0] == "unknown"


def test_a_service_account_key_is_judged_on_the_same_field():
    key = make(owner_project_access="inactive",
               owner={"type": "service_account",
                      "service_account": {"name": "batch-runner"}})
    assert verdict(key, NOW)[0] == "serving"
    assert owner_label(key) == "batch-runner"


def test_owner_label_prefers_the_email():
    assert owner_label(make()) == "dev@example.com"
    assert owner_label({"owner": {"type": "user"}}) == "user"
    assert owner_label({}) == "unknown owner"


def test_the_hot_window_is_a_parameter_not_a_constant():
    key = make(owner_project_access="inactive", last_used_at=NOW - 20 * 86400)
    assert verdict(key, NOW, hot_days=7)[0] == "orphaned"
    assert verdict(key, NOW, hot_days=30)[0] == "serving"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { ownerLabel, verdict } from './openai-orphaned-key-audit.mjs';

const NOW = 1_756_000_000;

const make = (over = {}) => ({
  id: 'key_abc',
  redacted_value: 'sk-proj-...aB3d',
  owner_project_access: 'active',
  last_used_at: NOW - 3600,
  owner: { type: 'user', user: { email: 'dev@example.com' } },
  ...over,
});

test('active owner is not a finding', () => {
  assert.equal(verdict(make(), NOW)[0], 'in-force');
});

test('inactive owner used today is production traffic', () => {
  const [state, detail] = verdict(make({ owner_project_access: 'inactive' }), NOW);
  assert.equal(state, 'serving');
  assert.match(detail, /re-issue/);
});

test('inactive owner long idle is orphaned not serving', () => {
  const [state, detail] = verdict(
    make({ owner_project_access: 'inactive', last_used_at: NOW - 90 * 86400 }), NOW);
  assert.equal(state, 'orphaned');
  assert.match(detail, /90 day\(s\)/);
});

test('inactive owner never used is the safe one', () => {
  const [state, detail] = verdict(
    make({ owner_project_access: 'inactive', last_used_at: null }), NOW);
  assert.equal(state, 'dormant');
  assert.match(detail, /revoke first/);
});

test('missing access field is never read as active', () => {
  const key = make();
  delete key.owner_project_access;
  const [state, detail] = verdict(key, NOW);
  assert.equal(state, 'unknown');
  assert.match(detail, /owner_project_access=any/);
});

test('unrecognised access value is not silently fine', () => {
  assert.equal(verdict(make({ owner_project_access: 'pending' }), NOW)[0], 'unknown');
});

test('a service account key is judged on the same field', () => {
  const key = make({
    owner_project_access: 'inactive',
    owner: { type: 'service_account', service_account: { name: 'batch-runner' } },
  });
  assert.equal(verdict(key, NOW)[0], 'serving');
  assert.equal(ownerLabel(key), 'batch-runner');
});

test('owner label prefers the email', () => {
  assert.equal(ownerLabel(make()), 'dev@example.com');
  assert.equal(ownerLabel({ owner: { type: 'user' } }), 'user');
  assert.equal(ownerLabel({}), 'unknown owner');
});

test('the hot window is a parameter not a constant', () => {
  const key = make({ owner_project_access: 'inactive', last_used_at: NOW - 20 * 86400 });
  assert.equal(verdict(key, NOW, 7)[0], 'orphaned');
  assert.equal(verdict(key, NOW, 30)[0], 'serving');
});