an archived project still holds live API keys

"""Report live API keys sitting inside archived OpenAI projects.

Read only. GET requests and nothing else, with an ORGANIZATION ADMIN key
(sk-admin-...) because /v1/organization/* rejects project keys; read-only admin
scopes are enough. The repair is printed, never performed.

Archiving a project hides it from the default listing without revoking anything
inside it, so the parameter below is the whole point of the script.
"""
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_archived_project_keys")

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

TRUTHY = ("true", "1", "yes", "on")


def covers_archived(params):
    """True when a projects listing will actually include archived projects.

    Pure. include_archived defaults to false, so a key audit that never passes
    it is auditing a subset of the organization and reporting a clean result
    over it. Accepts the bool and the query-string spelling, because the value
    reaches the API as a string either way.
    """
    value = params.get("include_archived")
    if isinstance(value, bool):
        return value
    if value is None:
        return False
    return str(value).strip().lower() in TRUTHY


def verdict(project, keys, now):
    """Classify one project against the keys found inside it.

    Pure, so the comparison between a key's last_used_at and the project's
    archived_at is testable without an admin credential. All three timestamps
    are unix seconds; last_used_at is null on a key that has never been used.

    Returns (state, detail).
    """
    status = str(project.get("status") or "").strip().lower()
    archived_at = project.get("archived_at")
    if status != "archived" and archived_at is None:
        return ("active", "not archived; outside the scope of this check")

    keys = list(keys or [])
    if not keys:
        return ("clean", "archived, and holds no API keys")

    used_after = [k for k in keys
                  if k.get("last_used_at") and archived_at
                  and int(k["last_used_at"]) > int(archived_at)]
    if used_after:
        newest = max(int(k["last_used_at"]) for k in used_after)
        return ("still-serving",
                "%d of %d live key(s) authenticated a request after the project "
                "was archived, the most recent %d day(s) ago. This project is "
                "closed on paper and running in fact."
                % (len(used_after), len(keys), (int(now) - newest) // DAY))

    ever_used = [k for k in keys if k.get("last_used_at")]
    if ever_used:
        newest = max(int(k["last_used_at"]) for k in ever_used)
        return ("live-keys",
                "%d live key(s) inside an archived project, last used %d day(s) "
                "ago. Nothing has needed them since the archive."
                % (len(keys), (int(now) - newest) // DAY))
    return ("dormant-keys",
            "%d live key(s) inside an archived project, none of which has ever "
            "authenticated a request" % len(keys))


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):
    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("--show-active", action="store_true",
                    help="also print the projects that are not archived")
    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())

    listing = {"limit": 100, "include_archived": "true"}
    # Stated out loud, because the silent version of this mistake is what the
    # note is about: a sweep that omits the parameter reports a clean subset.
    log.info("listing covers archived projects: %s",
             "yes" if covers_archived(listing) else "NO, this audit is partial")

    projects = list(paged(s, "/organization/projects", **listing))
    archived = 0
    exposed = 0
    for project in projects:
        keys = []
        if str(project.get("status") or "").lower() == "archived" \
                or project.get("archived_at") is not None:
            archived += 1
            keys = list(paged(s, "/organization/projects/%s/api_keys" % project["id"],
                              owner_project_access="any"))
        state, detail = verdict(project, keys, now)
        line = "%-13s %s  %s" % (state, project.get("name") or project["id"], detail)
        if state in ("active", "clean"):
            if state == "clean" or args.show_active:
                log.info(line)
            continue
        exposed += len(keys)
        log.warning(line)
        for key in keys:
            log.warning("  repair: DELETE %s/organization/projects/%s/api_keys/%s  (%s)",
                        API, project["id"], key.get("id"),
                        key.get("redacted_value") or key.get("name") or "unnamed")
        log.warning("  and check the spend: GET %s/organization/costs"
                    "?start_time=<now-30d>&group_by=project_id", API)

    log.info("%d project(s), %d archived, %d live key(s) inside them",
             len(projects), archived, exposed)
    return 1 if exposed else 0


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

/**
 * Report live API keys sitting inside archived OpenAI projects.
 *
 * Read only. GET requests and nothing else, with an ORGANIZATION ADMIN key
 * (sk-admin-...) because /v1/organization/* rejects project keys; read-only
 * admin scopes are enough. The repair is printed, never performed.
 */
const API = 'https://api.openai.com/v1';
const DAY = 86400;
const TRUTHY = ['true', '1', 'yes', 'on'];

/**
 * True when a projects listing will actually include archived projects. Pure.
 * include_archived defaults to false, so an audit that never passes it reports
 * a clean result over a subset of the organization.
 */
export function coversArchived(params = {}) {
  const value = params.include_archived;
  if (typeof value === 'boolean') return value;
  if (value === undefined || value === null) return false;
  return TRUTHY.includes(String(value).trim().toLowerCase());
}

/**
 * Classify one project against the keys found inside it. Pure. All timestamps
 * are unix seconds; last_used_at is null on a key that has never been used.
 */
export function verdict(project, keys, now) {
  const status = String(project.status ?? '').trim().toLowerCase();
  const archivedAt = project.archived_at;
  if (status !== 'archived' && (archivedAt === undefined || archivedAt === null)) {
    return ['active', 'not archived; outside the scope of this check'];
  }

  const all = [...(keys ?? [])];
  if (all.length === 0) return ['clean', 'archived, and holds no API keys'];

  const usedAfter = all.filter((k) => k.last_used_at && archivedAt &&
                                      Number(k.last_used_at) > Number(archivedAt));
  if (usedAfter.length > 0) {
    const newest = Math.max(...usedAfter.map((k) => Number(k.last_used_at)));
    const days = Math.floor((Number(now) - newest) / DAY);
    return ['still-serving',
      `${usedAfter.length} of ${all.length} live key(s) authenticated a request ` +
      `after the project was archived, the most recent ${days} day(s) ago. This ` +
      'project is closed on paper and running in fact.'];
  }

  const everUsed = all.filter((k) => k.last_used_at);
  if (everUsed.length > 0) {
    const newest = Math.max(...everUsed.map((k) => Number(k.last_used_at)));
    const days = Math.floor((Number(now) - newest) / DAY);
    return ['live-keys',
      `${all.length} live key(s) inside an archived project, last used ${days} ` +
      'day(s) ago. Nothing has needed them since the archive.'];
  }
  return ['dormant-keys',
    `${all.length} live key(s) inside an archived project, none of which has ` +
    'ever authenticated a request'];
}

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 now = Math.floor(Date.now() / 1000);

  const listing = { limit: 100, include_archived: 'true' };
  console.log(`listing covers archived projects: ${
    coversArchived(listing) ? 'yes' : 'NO, this audit is partial'}`);

  const projects = [];
  for await (const p of paged(adminKey, '/organization/projects', listing)) projects.push(p);

  let archived = 0;
  let exposed = 0;
  for (const project of projects) {
    let keys = [];
    const isArchived = String(project.status ?? '').toLowerCase() === 'archived' ||
                       (project.archived_at !== undefined && project.archived_at !== null);
    if (isArchived) {
      archived += 1;
      for await (const k of paged(adminKey,
                                  `/organization/projects/${project.id}/api_keys`,
                                  { owner_project_access: 'any' })) keys.push(k);
    }
    const [state, detail] = verdict(project, keys, now);
    const line = `${state.padEnd(13)} ${project.name ?? project.id}  ${detail}`;
    if (state === 'active' || state === 'clean') {
      if (state === 'clean') console.log(line);
      continue;
    }
    exposed += keys.length;
    console.warn(line);
    for (const key of keys) {
      console.warn(`  repair: DELETE ${API}/organization/projects/${project.id}` +
                   `/api_keys/${key.id}  (${key.redacted_value ?? key.name ?? 'unnamed'})`);
    }
    console.warn(`  and check the spend: GET ${API}/organization/costs` +
                 '?start_time=<now-30d>&group_by=project_id');
  }

  console.log(`${projects.length} project(s), ${archived} archived, ${exposed} ` +
              'live key(s) inside them');
  process.exitCode = exposed ? 1 : 0;
}

if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from openai_archived_project_keys import covers_archived, verdict

NOW = 1_756_000_000
ARCHIVED_AT = NOW - 120 * 86400


def project(**over):
    p = {"id": "proj_x", "name": "prototype", "status": "archived",
         "archived_at": ARCHIVED_AT}
    p.update(over)
    return p


def key(last_used_at=None, **over):
    k = {"id": "key_1", "redacted_value": "sk-proj-...9f2c",
         "last_used_at": last_used_at}
    k.update(over)
    return k


def test_a_listing_without_the_parameter_does_not_cover_archived():
    assert covers_archived({"limit": 100}) is False


def test_the_string_false_is_not_truthy_here():
    # The quiet version of this bug: a non-empty string read as "yes".
    assert covers_archived({"include_archived": "false"}) is False
    assert covers_archived({"include_archived": False}) is False


def test_the_parameter_is_recognised_in_the_spellings_that_reach_the_api():
    assert covers_archived({"include_archived": "true"}) is True
    assert covers_archived({"include_archived": "TRUE"}) is True
    assert covers_archived({"include_archived": True}) is True
    assert covers_archived({"include_archived": "1"}) is True


def test_an_active_project_is_out_of_scope():
    state, _ = verdict(project(status="active", archived_at=None), [key(NOW)], NOW)
    assert state == "active"


def test_an_archived_project_with_no_keys_is_clean():
    assert verdict(project(), [], NOW)[0] == "clean"


def test_a_key_used_after_the_archive_is_the_urgent_case():
    state, detail = verdict(project(), [key(ARCHIVED_AT + 10 * 86400)], NOW)
    assert state == "still-serving"
    assert "closed on paper" in detail


def test_a_key_last_used_before_the_archive_is_dead_weight():
    state, detail = verdict(project(), [key(ARCHIVED_AT - 5 * 86400)], NOW)
    assert state == "live-keys"
    assert "since the archive" in detail


def test_a_never_used_key_is_still_reported():
    state, detail = verdict(project(), [key(None)], NOW)
    assert state == "dormant-keys"
    assert "has ever authenticated" in detail


def test_status_archived_without_a_timestamp_is_still_archived():
    # Nothing to compare last_used_at against, so it cannot be still-serving,
    # but it must not fall through to "active" either.
    state, _ = verdict(project(archived_at=None), [key(NOW - 86400)], NOW)
    assert state == "live-keys"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { coversArchived, verdict } from './openai-archived-project-keys.mjs';

const NOW = 1_756_000_000;
const ARCHIVED_AT = NOW - 120 * 86400;

const project = (over = {}) => ({
  id: 'proj_x', name: 'prototype', status: 'archived', archived_at: ARCHIVED_AT, ...over,
});
const key = (lastUsedAt = null, over = {}) => ({
  id: 'key_1', redacted_value: 'sk-proj-...9f2c', last_used_at: lastUsedAt, ...over,
});

test('a listing without the parameter does not cover archived', () => {
  assert.equal(coversArchived({ limit: 100 }), false);
  assert.equal(coversArchived(), false);
});

test('the string false is not truthy here', () => {
  assert.equal(coversArchived({ include_archived: 'false' }), false);
  assert.equal(coversArchived({ include_archived: false }), false);
});

test('the parameter is recognised in the spellings that reach the API', () => {
  assert.equal(coversArchived({ include_archived: 'true' }), true);
  assert.equal(coversArchived({ include_archived: 'TRUE' }), true);
  assert.equal(coversArchived({ include_archived: true }), true);
  assert.equal(coversArchived({ include_archived: '1' }), true);
});

test('an active project is out of scope', () => {
  assert.equal(
    verdict(project({ status: 'active', archived_at: null }), [key(NOW)], NOW)[0],
    'active');
});

test('an archived project with no keys is clean', () => {
  assert.equal(verdict(project(), [], NOW)[0], 'clean');
});

test('a key used after the archive is the urgent case', () => {
  const [state, detail] = verdict(project(), [key(ARCHIVED_AT + 10 * 86400)], NOW);
  assert.equal(state, 'still-serving');
  assert.match(detail, /closed on paper/);
});

test('a key last used before the archive is dead weight', () => {
  const [state, detail] = verdict(project(), [key(ARCHIVED_AT - 5 * 86400)], NOW);
  assert.equal(state, 'live-keys');
  assert.match(detail, /since the archive/);
});

test('a never used key is still reported', () => {
  const [state, detail] = verdict(project(), [key(null)], NOW);
  assert.equal(state, 'dormant-keys');
  assert.match(detail, /has ever authenticated/);
});

test('status archived without a timestamp is still archived', () => {
  assert.equal(verdict(project({ archived_at: null }), [key(NOW - 86400)], NOW)[0],
               'live-keys');
});