A service account key that has never been rotated

"""Find service account keys that have never been rotated.

Read only. Four GETs against the OpenAI Administration API with an admin key:
projects, service accounts, project keys and the audit log. Nothing is created,
changed or removed, and no key value is printed.

The clock is created_at on the newest key belonging to each service account.
There is no rotated_at field anywhere on either provider, so age since minting
is the only evidence available, and the key count decides which of three
findings it is.

The audit log corroborates an absence at the PROJECT level only: the
api_key.created event carries project.id and actor and does not name a service
account. An empty or unreachable audit log is reported as unavailable rather
than as agreement, because audit logging is gated to organizations that have it
enabled and silence from it means nothing either way.
"""
import argparse
import datetime as dt
import logging
import os
import sys

import requests

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

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

SINGLE_STALE = "single-stale-key"
STALE = "stale-key"
UNFINISHED = "unfinished-rotation"
NO_KEYS = "service-account-with-no-keys"
ROTATING = "rotating"
TOO_NEW = "too-new"
FINDINGS = (SINGLE_STALE, STALE, UNFINISHED, NO_KEYS)

AUDIT_CONFIRMED = "confirmed-at-project-level"
AUDIT_ACTIVITY = "creation-activity-in-window"
AUDIT_UNAVAILABLE = "audit-unavailable"


def age_days(stamp, now):
    """Whole days between a unix timestamp and now. Pure. None when unreadable."""
    if stamp is None or stamp == "" or isinstance(stamp, bool):
        return None
    try:
        when = dt.datetime.fromtimestamp(float(stamp), dt.timezone.utc)
    except (TypeError, ValueError, OSError, OverflowError):
        return None
    return int((now - when).total_seconds() // 86400)


def service_account_id(key):
    """The service account a key belongs to, or None. Pure."""
    owner = (key or {}).get("owner")
    if not isinstance(owner, dict):
        return None
    if str(owner.get("type") or "").strip().lower() != "service_account":
        return None
    account = owner.get("service_account")
    if not isinstance(account, dict):
        return None
    return str(account.get("id") or "") or None


def group_by_account(keys):
    """Group service-account keys by owner.service_account.id. Pure.

    Keys owned by a user are dropped here rather than counted as unattributed,
    because a personal key in a project is a finding for a different note and
    counting it towards a service account's key total would make a single-key
    account look like it has an overlap window it does not have.
    """
    out = {}
    for key in keys or []:
        account = service_account_id(key)
        if account:
            out.setdefault(account, []).append(key)
    return out


def newest_and_oldest(keys, now):
    """(newest_age, oldest_age) in days across a key group. Pure.

    The newest is the rotation clock. Using the oldest instead reports a
    service account that rotated last week as stale, because the key it
    replaced is still in the list until somebody revokes it.
    """
    ages = [a for a in (age_days((k or {}).get("created_at"), now) for k in keys or [])
            if a is not None]
    if not ages:
        return (None, None)
    return (min(ages), max(ages))


def rotation_verdict(account, keys, now, stale_after=180, min_age=30):
    """Classify one service account's rotation state. Pure. (state, detail).

    The key count is an input rather than a detail, because one stale key and
    two stale keys are different problems. With one key there has never been a
    moment when two credentials were valid at once, so every rotation is a
    synchronised cutover with no rollback, and that is the reason it keeps
    being deferred.
    """
    name = str((account or {}).get("name") or (account or {}).get("id") or "(unnamed)")
    rows = list(keys or [])
    if not rows:
        created = age_days((account or {}).get("created_at"), now)
        return (NO_KEYS,
                "service account %s has no keys at all%s"
                % (name, "" if created is None else
                   ", and was created %d day(s) ago" % created))

    newest, oldest = newest_and_oldest(rows, now)
    if newest is None:
        return (TOO_NEW, "no readable created_at on any of its %d key(s)" % len(rows))
    if newest < min_age and len(rows) == 1:
        return (TOO_NEW, "its only key is %d day(s) old" % newest)
    if newest < stale_after:
        if oldest >= stale_after and len(rows) > 1:
            return (UNFINISHED,
                    "newest key %d day(s) old, oldest %d day(s) and still live"
                    % (newest, oldest))
        return (ROTATING, "newest key %d day(s) old" % newest)
    if len(rows) == 1:
        return (SINGLE_STALE,
                "newest key %d day(s) old, and it is the only one" % newest)
    return (STALE,
            "newest key %d day(s) old across %d key(s)" % (newest, len(rows)))


def corroboration(events, project_id, audit_reachable=True, days=180):
    """What the audit log can and cannot confirm. Pure. (state, detail).

    Three outcomes and only one of them is corroboration. The api_key.created
    event names a project and an actor and never a service account, so the
    strongest available statement is about the project. An empty or unreachable
    log is reported as unavailable, because audit logging is gated and its
    silence is not evidence.
    """
    if not audit_reachable:
        return (AUDIT_UNAVAILABLE,
                "the audit log could not be read, so nothing here is "
                "corroborated. Audit logging is gated to organizations that "
                "have it enabled and its silence is not evidence.")
    rows = list(events or [])
    if not rows:
        return (AUDIT_UNAVAILABLE,
                "the audit log returned no events of any kind in %d day(s), "
                "which can mean nothing was minted or can mean nothing is "
                "being recorded. Treated as unavailable rather than clean."
                % days)
    here = [e for e in rows
            if str(((e or {}).get("project") or {}).get("id") or "") == str(project_id)]
    if here:
        return (AUDIT_ACTIVITY,
                "%d api_key.created event(s) in this project in %d day(s), so "
                "something was minted here. The event does not name a service "
                "account, so it neither confirms nor clears any one of them."
                % (len(here), days))
    return (AUDIT_CONFIRMED,
            "no api_key.created events in this project in %d day(s). That is a "
            "project-level fact: the event carries project.id and actor and "
            "not the service account, so the per-account age above remains the "
            "evidence for any single account." % days)


def rotation_plan(project_id, account_name, single_key):
    """The overlap rotation, printed and never performed. Pure."""
    steps = []
    if single_key:
        steps.append("mint a second key first. One key means every rotation is "
                     "a hard cutover with no rollback, which is the actual "
                     "reason this has not happened yet.")
    steps.extend([
        "mint the replacement with an admin POST to /v1/organization/projects/"
        "%s/service_accounts/{service_account_id}/api_keys for %s. The value "
        "is returned exactly once." % (project_id, account_name),
        "deploy the new value everywhere the old one is held, then watch the "
        "old key: its last_used_at should stop advancing within one traffic "
        "cycle. Do not skip this; it is the only rollback you get.",
        "revoke the old key with a DELETE on /v1/organization/projects/%s/"
        "api_keys/{api_key_id}, and diary the next rotation at 90 days. "
        "Project keys have no expires_at, so nothing will remind you."
        % project_id,
    ])
    return steps


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


def paged(session, path, params, limit_pages=20):
    """Walk an administration listing on has_more / last_id."""
    params = dict(params)
    for _ in range(limit_pages):
        page = get(session, path, params)
        yield page
        if not page.get("has_more") or not page.get("last_id"):
            return
        params["after"] = page["last_id"]


def collect(session, path, params):
    rows = []
    for page in paged(session, path, params):
        rows.extend(page.get("data") or [])
    return rows


def read_audit_log(session, days):
    """Read api_key.created events, tolerating an organization without them.

    Returns (events, reachable). A 4xx here is not fatal: audit logging is not
    enabled everywhere, and a script that dies on it would report nothing about
    the key ages it already has in hand.
    """
    since = int((dt.datetime.now(dt.timezone.utc)
                 - dt.timedelta(days=days)).timestamp())
    try:
        return (collect(session, "/organization/audit_logs",
                        {"limit": 100, "event_types[]": ["api_key.created"],
                         "effective_at[gte]": since}), True)
    except requests.HTTPError as err:
        status = getattr(getattr(err, "response", None), "status_code", None)
        log.warning("audit log unreadable (%s): rotation ages below stand on "
                    "created_at alone", status)
        return ([], False)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--stale-after", type=int, default=180,
                    help="days since the newest key was minted (default 180)")
    ap.add_argument("--min-age", type=int, default=30,
                    help="days before a new service account is graded at all")
    args = ap.parse_args()

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

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + admin})
    now = dt.datetime.now(dt.timezone.utc)

    events, reachable = read_audit_log(s, args.stale_after)
    if reachable:
        projects_seen = {str(((e or {}).get("project") or {}).get("id") or "")
                         for e in events}
        log.info("audit log: %d api_key.created event(s) in %d day(s) across "
                 "%d project(s)", len(events), args.stale_after,
                 len([p for p in projects_seen if p]))

    projects = collect(s, "/organization/projects",
                       {"limit": 100, "include_archived": "true"})

    accounts_seen = 0
    keys_seen = 0
    findings = 0

    for project in projects:
        pid = project.get("id")
        if not pid:
            continue
        name = project.get("name") or pid
        accounts = collect(s, "/organization/projects/%s/service_accounts" % pid,
                           {"limit": 100})
        keys = collect(s, "/organization/projects/%s/api_keys" % pid,
                       {"limit": 100, "owner_project_access": "any"})
        grouped = group_by_account(keys)
        accounts_seen += len(accounts)
        keys_seen += sum(len(v) for v in grouped.values())

        project_findings = 0
        for account in accounts:
            rows = grouped.get(str(account.get("id") or ""), [])
            state, detail = rotation_verdict(account, rows, now,
                                             args.stale_after, args.min_age)
            if state not in FINDINGS:
                continue
            findings += 1
            project_findings += 1
            log.warning("%-19s %-11s %-15s %s", state, name,
                        account.get("name") or account.get("id") or "(unnamed)",
                        detail)
            if state in (SINGLE_STALE, STALE, UNFINISHED):
                for step in rotation_plan(pid,
                                          account.get("name") or "(unnamed)",
                                          state == SINGLE_STALE):
                    log.warning("  repair: %s", step)

        if project_findings:
            audit_state, audit_detail = corroboration(events, pid, reachable,
                                                      args.stale_after)
            log.info("corroboration for %s: %s: %s", name, audit_state,
                     audit_detail)

    log.info("%d project(s), %d service account(s), %d service-account key(s)",
             len(projects), accounts_seen, keys_seen)
    log.info("%d finding(s)", findings)
    log.info("there is no rotated_at field on either provider: created_at on "
             "the newest key is the only clock available")
    return 1 if findings else 0


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

/**
 * Find service account keys that have never been rotated.
 *
 * Read only. Four GETs against the OpenAI Administration API with an admin
 * key. Nothing is created, changed or removed, and no key value is printed.
 *
 * The clock is created_at on the newest key belonging to each service account,
 * because no rotated_at field exists on either provider. The key count decides
 * which of three findings it is.
 *
 * The audit log corroborates an absence at the PROJECT level only: the
 * api_key.created event carries project.id and actor and does not name a
 * service account. An empty or unreachable log is reported as unavailable.
 */
const API = 'https://api.openai.com/v1';

export const SINGLE_STALE = 'single-stale-key';
export const STALE = 'stale-key';
export const UNFINISHED = 'unfinished-rotation';
export const NO_KEYS = 'service-account-with-no-keys';
export const ROTATING = 'rotating';
export const TOO_NEW = 'too-new';
const FINDINGS = new Set([SINGLE_STALE, STALE, UNFINISHED, NO_KEYS]);

export const AUDIT_CONFIRMED = 'confirmed-at-project-level';
export const AUDIT_ACTIVITY = 'creation-activity-in-window';
export const AUDIT_UNAVAILABLE = 'audit-unavailable';

/** Whole days between a unix timestamp and now. Pure. null when unreadable. */
export function ageDays(stamp, now) {
  if (stamp === null || stamp === undefined || stamp === '' ||
      typeof stamp === 'boolean') return null;
  const seconds = Number(stamp);
  if (!Number.isFinite(seconds)) return null;
  return Math.floor((now.getTime() - seconds * 1000) / 86400000);
}

/** The service account a key belongs to, or null. Pure. */
export function serviceAccountId(key) {
  const owner = (key ?? {}).owner;
  if (!owner || typeof owner !== 'object') return null;
  if (String(owner.type ?? '').trim().toLowerCase() !== 'service_account') return null;
  const account = owner.service_account;
  if (!account || typeof account !== 'object') return null;
  return String(account.id ?? '') || null;
}

/** Group service-account keys by owner.service_account.id. Pure. */
export function groupByAccount(keys) {
  const out = {};
  for (const key of keys ?? []) {
    const account = serviceAccountId(key);
    if (!account) continue;
    out[account] = out[account] ?? [];
    out[account].push(key);
  }
  return out;
}

/** [newestAge, oldestAge] in days across a key group. Pure. */
export function newestAndOldest(keys, now) {
  const ages = (keys ?? [])
    .map((k) => ageDays((k ?? {}).created_at, now))
    .filter((a) => a !== null);
  if (!ages.length) return [null, null];
  return [Math.min(...ages), Math.max(...ages)];
}

/** Classify one service account's rotation state. Pure. [state, detail]. */
export function rotationVerdict(account, keys, now, staleAfter = 180, minAge = 30) {
  const name = String((account ?? {}).name ?? (account ?? {}).id ?? '(unnamed)');
  const rows = [...(keys ?? [])];
  if (!rows.length) {
    const created = ageDays((account ?? {}).created_at, now);
    return [NO_KEYS,
      `service account ${name} has no keys at all` +
      (created === null ? '' : `, and was created ${created} day(s) ago`)];
  }
  const [newest, oldest] = newestAndOldest(rows, now);
  if (newest === null) {
    return [TOO_NEW, `no readable created_at on any of its ${rows.length} key(s)`];
  }
  if (newest < minAge && rows.length === 1) {
    return [TOO_NEW, `its only key is ${newest} day(s) old`];
  }
  if (newest < staleAfter) {
    if (oldest >= staleAfter && rows.length > 1) {
      return [UNFINISHED,
        `newest key ${newest} day(s) old, oldest ${oldest} day(s) and still live`];
    }
    return [ROTATING, `newest key ${newest} day(s) old`];
  }
  if (rows.length === 1) {
    return [SINGLE_STALE, `newest key ${newest} day(s) old, and it is the only one`];
  }
  return [STALE, `newest key ${newest} day(s) old across ${rows.length} key(s)`];
}

/** What the audit log can and cannot confirm. Pure. [state, detail]. */
export function corroboration(events, projectId, auditReachable = true, days = 180) {
  if (!auditReachable) {
    return [AUDIT_UNAVAILABLE,
      'the audit log could not be read, so nothing here is corroborated. ' +
      'Audit logging is gated to organizations that have it enabled and its ' +
      'silence is not evidence.'];
  }
  const rows = [...(events ?? [])];
  if (!rows.length) {
    return [AUDIT_UNAVAILABLE,
      `the audit log returned no events of any kind in ${days} day(s), which ` +
      'can mean nothing was minted or can mean nothing is being recorded. ' +
      'Treated as unavailable rather than clean.'];
  }
  const here = rows.filter(
    (e) => String((e?.project ?? {}).id ?? '') === String(projectId));
  if (here.length) {
    return [AUDIT_ACTIVITY,
      `${here.length} api_key.created event(s) in this project in ${days} ` +
      'day(s), so something was minted here. The event does not name a ' +
      'service account, so it neither confirms nor clears any one of them.'];
  }
  return [AUDIT_CONFIRMED,
    `no api_key.created events in this project in ${days} day(s). That is a ` +
    'project-level fact: the event carries project.id and actor and not the ' +
    'service account, so the per-account age above remains the evidence for ' +
    'any single account.'];
}

/** The overlap rotation, printed and never performed. Pure. */
export function rotationPlan(projectId, accountName, singleKey) {
  const steps = [];
  if (singleKey) {
    steps.push('mint a second key first. One key means every rotation is a ' +
               'hard cutover with no rollback, which is the actual reason ' +
               'this has not happened yet.');
  }
  steps.push(
    'mint the replacement with an admin POST to /v1/organization/projects/' +
    `${projectId}/service_accounts/{service_account_id}/api_keys for ` +
    `${accountName}. The value is returned exactly once.`,
    'deploy the new value everywhere the old one is held, then watch the old ' +
    'key: its last_used_at should stop advancing within one traffic cycle. ' +
    'Do not skip this; it is the only rollback you get.',
    `revoke the old key with a DELETE on /v1/organization/projects/${projectId}` +
    '/api_keys/{api_key_id}, and diary the next rotation at 90 days. Project ' +
    'keys have no expires_at, so nothing will remind you.');
  return steps;
}

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

async function collect(key, path, params) {
  const rows = [];
  let after = null;
  for (let page = 0; page < 20; page += 1) {
    const query = after ? [...params, ['after', after]] : params;
    const body = await getJson(key, path, query);
    rows.push(...(body.data ?? []));
    if (!body.has_more || !body.last_id) return rows;
    after = body.last_id;
  }
  return rows;
}

async function main() {
  const admin = process.env.OPENAI_ADMIN_KEY;
  if (!admin) {
    console.error('set OPENAI_ADMIN_KEY to an admin key (sk-admin-) with read ' +
                  'scopes; a project key cannot read /v1/organization/*');
    process.exitCode = 2;
    return;
  }
  const staleAfter = Number(process.env.STALE_AFTER ?? 180);
  const minAge = Number(process.env.MIN_AGE ?? 30);
  const now = new Date();
  const since = Math.floor(now.getTime() / 1000) - staleAfter * 86400;

  let events = [];
  let reachable = true;
  try {
    events = await collect(admin, '/organization/audit_logs', [
      ['limit', 100], ['event_types[]', 'api_key.created'],
      ['effective_at[gte]', since]]);
    console.log(`audit log: ${events.length} api_key.created event(s) in ` +
                `${staleAfter} day(s)`);
  } catch (err) {
    reachable = false;
    console.warn(`audit log unreadable (${err.status ?? err.message}): ` +
                 'rotation ages below stand on created_at alone');
  }

  const projects = await collect(admin, '/organization/projects',
                                 [['limit', 100], ['include_archived', 'true']]);
  let accountsSeen = 0;
  let keysSeen = 0;
  let findings = 0;

  for (const project of projects) {
    if (!project.id) continue;
    const name = project.name ?? project.id;
    const accounts = await collect(
      admin, `/organization/projects/${project.id}/service_accounts`, [['limit', 100]]);
    const keys = await collect(
      admin, `/organization/projects/${project.id}/api_keys`,
      [['limit', 100], ['owner_project_access', 'any']]);
    const grouped = groupByAccount(keys);
    accountsSeen += accounts.length;
    keysSeen += Object.values(grouped).reduce((n, v) => n + v.length, 0);

    let projectFindings = 0;
    for (const account of accounts) {
      const rows = grouped[String(account.id ?? '')] ?? [];
      const [state, detail] = rotationVerdict(account, rows, now, staleAfter, minAge);
      if (!FINDINGS.has(state)) continue;
      findings += 1;
      projectFindings += 1;
      console.warn(`${state.padEnd(19)} ${String(name).padEnd(11)} ` +
                   `${String(account.name ?? account.id ?? '(unnamed)').padEnd(15)} ${detail}`);
      if (state === SINGLE_STALE || state === STALE || state === UNFINISHED) {
        for (const step of rotationPlan(project.id, account.name ?? '(unnamed)',
                                        state === SINGLE_STALE)) {
          console.warn(`  repair: ${step}`);
        }
      }
    }

    if (projectFindings) {
      const [auditState, auditDetail] =
        corroboration(events, project.id, reachable, staleAfter);
      console.log(`corroboration for ${name}: ${auditState}: ${auditDetail}`);
    }
  }

  console.log(`${projects.length} project(s), ${accountsSeen} service account(s), ` +
              `${keysSeen} service-account key(s)`);
  console.log(`${findings} finding(s)`);
  console.log('there is no rotated_at field on either provider: created_at on ' +
              'the newest key is the only clock available');
  process.exitCode = findings ? 1 : 0;
}

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

import datetime as dt

from openai_key_rotation_clock import (age_days, corroboration,
                                       group_by_account, newest_and_oldest,
                                       rotation_plan, rotation_verdict,
                                       service_account_id)

NOW = dt.datetime(2026, 8, 31, 12, 0, 0, tzinfo=dt.timezone.utc)
ACCOUNT = {"id": "svc_1", "name": "ingest-worker", "created_at": 0}


def unix(days_ago):
    return int((NOW - dt.timedelta(days=days_ago)).timestamp())


def sa_key(key_id, days_ago, account="svc_1"):
    return {"id": key_id, "created_at": unix(days_ago),
            "owner": {"type": "service_account",
                      "service_account": {"id": account, "name": "ingest-worker"}}}


def test_the_newest_key_is_the_clock_and_the_oldest_would_lie():
    # An account rotated last week still holds the retired key until somebody
    # revokes it. A reader that takes the oldest created_at calls that account
    # two years stale, which is the single mistake that would make this note
    # useless. Reading the newest calls it a rotation that has not been
    # finished, which is what it is.
    rotated = [sa_key("key_new", 45), sa_key("key_old", 731)]
    newest, oldest = newest_and_oldest(rotated, NOW)
    assert (newest, oldest) == (45, 731)
    state, detail = rotation_verdict(ACCOUNT, rotated, NOW)
    assert state != "single-stale-key" and state != "stale-key"
    assert state == "unfinished-rotation"
    assert "newest key 45 day(s) old" in detail

    # And once the retired key is revoked, there is nothing left to report.
    finished = rotation_verdict(ACCOUNT, [sa_key("key_new", 45)], NOW)
    assert finished[0] == "rotating"
    assert "45 day(s) old" in finished[1]


def test_the_key_count_produces_three_different_findings():
    single = rotation_verdict(ACCOUNT, [sa_key("key_a", 731)], NOW)
    assert single[0] == "single-stale-key"
    assert "it is the only one" in single[1]
    assert any("mint a second key first" in step
               for step in rotation_plan("proj_1", "ingest-worker", True))

    both_old = rotation_verdict(ACCOUNT, [sa_key("key_a", 402),
                                          sa_key("key_b", 500)], NOW)
    assert both_old[0] == "stale-key"
    assert "across 2 key(s)" in both_old[1]
    assert not any("mint a second key first" in step
                   for step in rotation_plan("proj_1", "ingest-worker", False))

    halfway = rotation_verdict(ACCOUNT, [sa_key("key_a", 12),
                                         sa_key("key_b", 588)], NOW)
    assert halfway[0] == "unfinished-rotation"
    assert "still live" in halfway[1]


def test_an_empty_or_unreachable_audit_log_is_never_corroboration():
    unreachable = corroboration([], "proj_1", audit_reachable=False)
    assert unreachable[0] == "audit-unavailable"
    assert "silence is not evidence" in unreachable[1]

    empty = corroboration([], "proj_1", audit_reachable=True)
    assert empty[0] == "audit-unavailable"
    assert "nothing is being recorded" in empty[1]


def test_the_audit_log_confirms_a_project_and_never_an_account():
    elsewhere = [{"type": "api_key.created", "project": {"id": "proj_other"}}]
    state, detail = corroboration(elsewhere, "proj_1", True, 180)
    assert state == "confirmed-at-project-level"
    assert "project-level fact" in detail
    assert "does not name" in detail or "not the service account" in detail

    here = [{"type": "api_key.created", "project": {"id": "proj_1"}},
            {"type": "api_key.created", "project": {"id": "proj_1"}}]
    state, detail = corroboration(here, "proj_1", True, 180)
    assert state == "creation-activity-in-window"
    assert "neither confirms nor clears" in detail


def test_a_personal_key_is_not_counted_towards_a_service_account():
    keys = [sa_key("key_a", 731),
            {"id": "key_user", "created_at": unix(2),
             "owner": {"type": "user", "user": {"email": "dev@example.test"}}},
            {"id": "key_odd", "created_at": unix(2), "owner": None}]
    grouped = group_by_account(keys)
    assert list(grouped) == ["svc_1"]
    assert len(grouped["svc_1"]) == 1
    # Counting the personal key here would turn a single-key account into a
    # two-key one and hide the fact that no overlap window has ever existed.
    assert rotation_verdict(ACCOUNT, grouped["svc_1"], NOW)[0] == "single-stale-key"
    assert service_account_id({"owner": {"type": "service_account"}}) is None
    assert service_account_id(None) is None
    assert group_by_account([]) == {}


def test_a_service_account_with_no_keys_and_one_too_new_to_judge():
    empty = rotation_verdict({"id": "svc_2", "name": "search-indexer",
                              "created_at": unix(300)}, [], NOW)
    assert empty[0] == "service-account-with-no-keys"
    assert "300 day(s) ago" in empty[1]
    fresh = rotation_verdict(ACCOUNT, [sa_key("key_a", 4)], NOW)
    assert fresh[0] == "too-new"
    unreadable = rotation_verdict(ACCOUNT, [{"id": "key_a", "created_at": None,
                                             "owner": None}], NOW)
    assert unreadable[0] == "too-new"


def test_ages_are_read_from_unix_seconds_only():
    assert age_days(unix(180), NOW) == 180
    assert age_days(None, NOW) is None
    assert age_days("", NOW) is None
    assert age_days(True, NOW) is None
    assert age_days("not a number", NOW) is None


def test_the_rotation_plan_revokes_last_and_names_the_missing_field():
    steps = rotation_plan("proj_prod", "ingest-worker", False)
    assert len(steps) == 3
    assert "returned exactly once" in steps[0]
    assert "last_used_at should stop advancing" in steps[1]
    assert steps[2].startswith("revoke the old key")
    assert "no expires_at" in steps[2]

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { ageDays, corroboration, groupByAccount, newestAndOldest, rotationPlan,
         rotationVerdict, serviceAccountId }
  from './openai-key-rotation-clock.mjs';

const NOW = new Date('2026-08-31T12:00:00Z');
const ACCOUNT = { id: 'svc_1', name: 'ingest-worker', created_at: 0 };
const unix = (daysAgo) => Math.floor(NOW.getTime() / 1000) - daysAgo * 86400;

const saKey = (id, daysAgo, account = 'svc_1') => ({
  id, created_at: unix(daysAgo),
  owner: { type: 'service_account',
           service_account: { id: account, name: 'ingest-worker' } },
});

test('the newest key is the clock and the oldest would lie', () => {
  const rotated = [saKey('key_new', 45), saKey('key_old', 731)];
  assert.deepEqual(newestAndOldest(rotated, NOW), [45, 731]);
  const [state, detail] = rotationVerdict(ACCOUNT, rotated, NOW);
  assert.ok(state !== 'single-stale-key' && state !== 'stale-key');
  assert.equal(state, 'unfinished-rotation');
  assert.match(detail, /newest key 45 day\(s\) old/);

  const finished = rotationVerdict(ACCOUNT, [saKey('key_new', 45)], NOW);
  assert.equal(finished[0], 'rotating');
});

test('the key count produces three different findings', () => {
  const single = rotationVerdict(ACCOUNT, [saKey('key_a', 731)], NOW);
  assert.equal(single[0], 'single-stale-key');
  assert.match(single[1], /it is the only one/);
  assert.ok(rotationPlan('proj_1', 'ingest-worker', true)
    .some((s) => s.includes('mint a second key first')));

  const bothOld = rotationVerdict(ACCOUNT, [saKey('key_a', 402), saKey('key_b', 500)], NOW);
  assert.equal(bothOld[0], 'stale-key');
  assert.match(bothOld[1], /across 2 key\(s\)/);
  assert.ok(!rotationPlan('proj_1', 'ingest-worker', false)
    .some((s) => s.includes('mint a second key first')));

  const halfway = rotationVerdict(ACCOUNT, [saKey('key_a', 12), saKey('key_b', 588)], NOW);
  assert.equal(halfway[0], 'unfinished-rotation');
  assert.match(halfway[1], /still live/);
});

test('an empty or unreachable audit log is never corroboration', () => {
  const unreachable = corroboration([], 'proj_1', false);
  assert.equal(unreachable[0], 'audit-unavailable');
  assert.match(unreachable[1], /silence is not evidence/);
  const empty = corroboration([], 'proj_1', true);
  assert.equal(empty[0], 'audit-unavailable');
  assert.match(empty[1], /nothing is being recorded/);
});

test('the audit log confirms a project and never an account', () => {
  const [state, detail] = corroboration(
    [{ type: 'api_key.created', project: { id: 'proj_other' } }], 'proj_1', true, 180);
  assert.equal(state, 'confirmed-at-project-level');
  assert.match(detail, /project-level fact/);
  assert.match(detail, /not the service account/);

  const [state2, detail2] = corroboration([
    { type: 'api_key.created', project: { id: 'proj_1' } },
    { type: 'api_key.created', project: { id: 'proj_1' } }], 'proj_1', true, 180);
  assert.equal(state2, 'creation-activity-in-window');
  assert.match(detail2, /neither confirms nor clears/);
});

test('a personal key is not counted towards a service account', () => {
  const grouped = groupByAccount([
    saKey('key_a', 731),
    { id: 'key_user', created_at: unix(2),
      owner: { type: 'user', user: { email: 'dev@example.test' } } },
    { id: 'key_odd', created_at: unix(2), owner: null },
  ]);
  assert.deepEqual(Object.keys(grouped), ['svc_1']);
  assert.equal(grouped.svc_1.length, 1);
  assert.equal(rotationVerdict(ACCOUNT, grouped.svc_1, NOW)[0], 'single-stale-key');
  assert.equal(serviceAccountId({ owner: { type: 'service_account' } }), null);
  assert.equal(serviceAccountId(null), null);
  assert.deepEqual(groupByAccount([]), {});
});

test('a service account with no keys and one too new to judge', () => {
  const empty = rotationVerdict(
    { id: 'svc_2', name: 'search-indexer', created_at: unix(300) }, [], NOW);
  assert.equal(empty[0], 'service-account-with-no-keys');
  assert.match(empty[1], /300 day\(s\) ago/);
  assert.equal(rotationVerdict(ACCOUNT, [saKey('key_a', 4)], NOW)[0], 'too-new');
  assert.equal(rotationVerdict(
    ACCOUNT, [{ id: 'key_a', created_at: null, owner: null }], NOW)[0], 'too-new');
});

test('ages are read from unix seconds only', () => {
  assert.equal(ageDays(unix(180), NOW), 180);
  assert.equal(ageDays(null, NOW), null);
  assert.equal(ageDays('', NOW), null);
  assert.equal(ageDays(true, NOW), null);
  assert.equal(ageDays('not a number', NOW), null);
});

test('the rotation plan revokes last and names the missing field', () => {
  const steps = rotationPlan('proj_prod', 'ingest-worker', false);
  assert.equal(steps.length, 3);
  assert.match(steps[0], /returned exactly once/);
  assert.match(steps[1], /last_used_at should stop advancing/);
  assert.ok(steps[2].startsWith('revoke the old key'));
  assert.match(steps[2], /no expires_at/);
});