Nobody has ever read the key lifecycle audit log

"""Read the key and member lifecycle events nobody has ever read.

Read only. GET requests only, against the OpenAI Audit Logs API and the
Anthropic Compliance activity feed. Nothing is created, changed or removed.

Both feeds are pull-only: there is no webhook, no email and no default alert on
either provider, which is why the control exists everywhere and has fired
nowhere. The finding is not any single event; it is that nobody is reading. So
the last thing this prints is a watermark to store for the next run.

Two honest limits are enforced in the code rather than mentioned in a comment.
An empty feed is reported as unavailable and never as clean, because audit
logging is gated to organizations that have it enabled. And the geography rule
runs on OpenAI session actors only: the Anthropic activity record carries an
email, a user id, an IP and a user agent, and no country breakdown to test.
"""
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("llm_key_lifecycle_review")

OPENAI = "https://api.openai.com/v1"
ANTHROPIC = "https://api.anthropic.com/v1"
ANTHROPIC_VERSION = "2023-06-01"

OPENAI_EVENTS = ("api_key.created", "api_key.updated", "api_key.deleted",
                 "service_account.created", "service_account.deleted",
                 "login.failed")

OFF_ROSTER = "off-roster-actor"
UNATTRIBUTABLE = "unattributable"
UNEXPECTED_COUNTRY = "unexpected-country"
OUT_OF_HOURS = "out-of-hours"
REVIEWED = "reviewed"
FINDINGS = (OFF_ROSTER, UNATTRIBUTABLE, UNEXPECTED_COUNTRY, OUT_OF_HOURS)

FEED_OK = "feed-readable"
FEED_UNAVAILABLE = "feed-unavailable"

# Highest first. An event can trip several rules at once and every reason is
# printed; the state is the one that decides how loudly it is printed.
SEVERITY = (OFF_ROSTER, UNEXPECTED_COUNTRY, UNATTRIBUTABLE, OUT_OF_HOURS)


def parse_when(value):
    """Epoch seconds from a unix integer or an RFC 3339 string. Pure.

    OpenAI dates effective_at in unix seconds; the Anthropic activity record
    uses an RFC 3339 string. Normalising here is what lets one grader read both
    feeds without either one being special-cased downstream.
    """
    if value is None or value == "" or isinstance(value, bool):
        return None
    if isinstance(value, (int, float)):
        return int(value)
    text = str(value).strip()
    if text.isdigit():
        return int(text)
    try:
        when = dt.datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        return None
    if when.tzinfo is None:
        when = when.replace(tzinfo=dt.timezone.utc)
    return int(when.timestamp())


def iso(epoch):
    """A readable UTC timestamp. Pure."""
    if epoch is None:
        return "(no timestamp)"
    return dt.datetime.fromtimestamp(int(epoch), dt.timezone.utc) \
             .strftime("%Y-%m-%dT%H:%M:%SZ")


def normalise_openai(entry):
    """One audit-log entry in the common shape. Pure.

    The actor arrives in two shapes and they keep the email in different
    places. A session actor is the forensically rich one and carries the
    address, the IP and ip_address_details. An api_key actor carries a tracking
    id and either a user email or a service account id, and frequently no email
    at all, which is a distinct outcome rather than a missing value.
    """
    row = entry or {}
    actor = row.get("actor") if isinstance(row.get("actor"), dict) else {}
    kind = str(actor.get("type") or "").strip().lower()
    email, ip, country = None, None, None
    if kind == "session":
        session = actor.get("session") if isinstance(actor.get("session"), dict) else {}
        user = session.get("user") if isinstance(session.get("user"), dict) else {}
        email = user.get("email")
        ip = session.get("ip_address")
        details = (session.get("ip_address_details")
                   if isinstance(session.get("ip_address_details"), dict) else {})
        country = details.get("country")
    elif kind == "api_key":
        api_key = actor.get("api_key") if isinstance(actor.get("api_key"), dict) else {}
        user = api_key.get("user") if isinstance(api_key.get("user"), dict) else {}
        email = user.get("email")
    project = row.get("project") if isinstance(row.get("project"), dict) else {}
    return {"source": "openai", "type": str(row.get("type") or "(untyped)"),
            "when": parse_when(row.get("effective_at")),
            "actor_kind": kind or "unknown",
            "actor_email": str(email).strip().lower() if email else None,
            "actor_ip": ip, "country": country,
            "container": project.get("name") or project.get("id")}


def normalise_anthropic(activity):
    """One compliance activity in the common shape. Pure.

    country stays None on purpose. The activity record carries email_address,
    user_id, ip_address and user_agent and no geography breakdown, and leaving
    the field absent is how the grader knows to skip the country rule here
    rather than silently passing every Anthropic event.
    """
    row = activity or {}
    actor = row.get("actor") if isinstance(row.get("actor"), dict) else {}
    email = actor.get("email_address")
    return {"source": "anthropic", "type": str(row.get("type") or "(untyped)"),
            "when": parse_when(row.get("created_at")),
            "actor_kind": "user" if email else "unknown",
            "actor_email": str(email).strip().lower() if email else None,
            "actor_ip": actor.get("ip_address"), "country": None,
            "container": row.get("organization_id")}


def resolve_actor(event, roster):
    """on-roster, off-roster or unattributable. Pure.

    The join that turns a feed into a finding. An email checked against the
    current roster is the difference between an event a reviewer can close and
    an action taken by somebody whose access has since ended.
    """
    email = (event or {}).get("actor_email")
    if not email:
        return "unattributable"
    return "on-roster" if str(email).strip().lower() in (roster or set()) \
        else "off-roster"


def hour_of(event):
    """The UTC hour of an event, or None. Pure."""
    when = (event or {}).get("when")
    if when is None:
        return None
    return dt.datetime.fromtimestamp(int(when), dt.timezone.utc).hour


def grade(event, roster, business_hours=(7, 19), operating_countries=None):
    """Classify one normalised event. Pure. Returns (state, reasons).

    Every rule that fires contributes a reason, because an event can be
    off-roster and out of hours and from an unexpected country at once and a
    reviewer wants all three. The state is the most severe reason present and
    decides how loudly the row is printed.
    """
    reasons = []
    resolution = resolve_actor(event, roster)
    if resolution == "off-roster":
        reasons.append((OFF_ROSTER, "the actor is not on the current roster"))
    elif resolution == "unattributable":
        reasons.append((UNATTRIBUTABLE,
                        "an %s actor carries no user email, so no person can "
                        "be attributed" % ((event or {}).get("actor_kind")
                                           or "unknown")))

    country = (event or {}).get("country")
    if operating_countries and country:
        if str(country).strip().upper() not in {c.upper() for c in operating_countries}:
            reasons.append((UNEXPECTED_COUNTRY,
                            "ip_address_details.country %s is outside the "
                            "operating geographies" % country))

    hour = hour_of(event)
    start, end = business_hours
    creation = str((event or {}).get("type") or "").endswith((".created", ".deleted"))
    if creation and hour is not None and not (start <= hour < end):
        reasons.append((OUT_OF_HOURS,
                        "created outside business hours (%02d:00 UTC)" % hour))

    if not reasons:
        return (REVIEWED, [])
    present = {state for state, _ in reasons}
    state = next(s for s in SEVERITY if s in present)
    return (state, [text for _, text in reasons])


def feed_state(events, reachable):
    """Whether the feed said anything at all. Pure. (state, detail).

    An empty feed is the most misreadable result on this surface. Audit logging
    is gated to organizations that have it enabled, so treating silence as "no
    findings" turns a missing control into a passing check.
    """
    if not reachable:
        return (FEED_UNAVAILABLE,
                "the feed could not be read, so nothing below is a review of "
                "anything")
    if not (events or []):
        return (FEED_UNAVAILABLE,
                "the feed returned no events at all. Audit logging is gated to "
                "organizations that have it enabled, so this is not a clean "
                "result: it is an unknown one.")
    return (FEED_OK, "%d event(s) read" % len(events))


def failed_login_bursts(events, window_seconds=600, threshold=5):
    """Clusters of login.failed inside one window. Pure.

    A single failed login is a typo. Five in ten minutes is the only pattern in
    this feed that is worth an alert on its own rather than a weekly read.
    """
    rows = sorted([e for e in (events or [])
                   if str((e or {}).get("type") or "") == "login.failed"
                   and (e or {}).get("when") is not None],
                  key=lambda e: e["when"])
    bursts = []
    for i, first in enumerate(rows):
        window = [e for e in rows[i:]
                  if e["when"] - first["when"] <= window_seconds]
        if len(window) >= threshold:
            bursts.append((first["when"], len(window),
                           first.get("actor_email") or "(no email)"))
            break
    return bursts


def watermark(events):
    """The newest timestamp seen, for the next run's cursor. Pure."""
    stamps = [e["when"] for e in (events or []) if (e or {}).get("when") is not None]
    return max(stamps) if stamps else None


def project_caveat(event):
    """Whether this entry's project field means anything. Pure.

    Admin actions taken with an Admin API key are attributed to the default
    project, so the project on those entries says nothing about where the
    action landed.
    """
    if (event or {}).get("source") != "openai":
        return None
    if (event or {}).get("actor_kind") == "api_key":
        return ("project is not meaningful here: admin actions taken with an "
                "Admin API key are attributed to the default project")
    return None


def get(session, url, params, who):
    r = session.get(url, params=params, timeout=60)
    if r.status_code == 429:
        raise SystemExit("429 from %s: this feed declares its own rate limit "
                         "with Retry-After. Back off and resume from the "
                         "stored watermark." % who)
    if r.status_code in (401, 403):
        raise SystemExit("%d from %s: the feed needs an administration "
                         "credential, and on Anthropic the "
                         "read:compliance_activities scope" % (r.status_code, who))
    r.raise_for_status()
    return r.json()


def collect(session, url, params, who, cursor="after", limit_pages=20):
    rows = []
    params = dict(params)
    for _ in range(limit_pages):
        page = get(session, url, params, who)
        rows.extend(page.get("data") or [])
        if not page.get("has_more") or not page.get("last_id"):
            return rows
        params[cursor] = page["last_id"]
    return rows


def report(name, events, roster, args, geography):
    state, detail = feed_state(events, True)
    log.info("%s: %s (%s), roster of %d member(s); %s", name, state, detail,
             len(roster), "country and session rules available" if geography
             else "no geography on this feed")
    if state == FEED_UNAVAILABLE:
        return 0

    countries = [c.strip() for c in args.countries.split(",") if c.strip()]
    findings = 0
    for event in sorted(events, key=lambda e: e.get("when") or 0):
        verdict, reasons = grade(event, roster,
                                 (args.hours_from, args.hours_to),
                                 countries if geography else None)
        if verdict == REVIEWED:
            continue
        findings += 1
        log.warning("%-19s %-22s %s  %-18s %-15s %s", verdict, event["type"],
                    iso(event.get("when")),
                    event.get("actor_email") or "(%s actor)" % event.get("actor_kind"),
                    event.get("actor_ip") or "-", event.get("country") or "")
        for reason in reasons:
            log.warning("  reason: %s", reason)
        caveat = project_caveat(event)
        if caveat:
            log.info("  note: %s", caveat)

    for when, count, who in failed_login_bursts(events):
        findings += 1
        log.warning("login-failed-burst   %d failure(s) within 10 minutes from "
                    "%s, starting %s", count, who, iso(when))

    mark = watermark(events)
    if mark is not None:
        log.info("watermark: store the cursor %d (%s) for the next %s run",
                 mark, iso(mark), name)
    return findings


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7,
                    help="days of lifecycle events to read (default 7)")
    ap.add_argument("--hours-from", type=int, default=7,
                    help="first business hour, UTC")
    ap.add_argument("--hours-to", type=int, default=19,
                    help="first non-business hour, UTC")
    ap.add_argument("--countries", default="US,GB,DE,IE",
                    help="comma separated operating geographies for the "
                         "country rule, which runs on OpenAI events only")
    args = ap.parse_args()

    openai_key = os.environ.get("OPENAI_ADMIN_KEY")
    anthropic_key = os.environ.get("ANTHROPIC_ADMIN_KEY")
    if not openai_key and not anthropic_key:
        log.error("set OPENAI_ADMIN_KEY or ANTHROPIC_ADMIN_KEY, or both; the "
                  "Anthropic credential also needs the "
                  "read:compliance_activities scope")
        return 2

    since = int((dt.datetime.now(dt.timezone.utc)
                 - dt.timedelta(days=args.days)).timestamp())
    findings = 0

    if openai_key:
        s = requests.Session()
        s.headers.update({"Authorization": "Bearer " + openai_key})
        roster = {str(u.get("email") or "").strip().lower()
                  for u in collect(s, OPENAI + "/organization/users",
                                   {"limit": 100}, "OpenAI")
                  if u.get("email")}
        raw = collect(s, OPENAI + "/organization/audit_logs",
                      {"limit": 100, "effective_at[gte]": since,
                       "event_types[]": list(OPENAI_EVENTS)}, "OpenAI")
        findings += report("openai", [normalise_openai(e) for e in raw],
                           roster, args, geography=True)

    if anthropic_key:
        s = requests.Session()
        s.headers.update({"x-api-key": anthropic_key,
                          "anthropic-version": ANTHROPIC_VERSION})
        roster = {str(u.get("email") or "").strip().lower()
                  for u in collect(s, ANTHROPIC + "/organizations/users",
                                   {"limit": 1000}, "Anthropic",
                                   cursor="after_id")
                  if u.get("email")}
        raw = collect(s, ANTHROPIC + "/compliance/activities", {"limit": 100},
                      "Anthropic")
        findings += report("anthropic",
                           [normalise_anthropic(a) for a in raw
                            if (parse_when((a or {}).get("created_at")) or 0) >= since],
                           roster, args, geography=False)

    log.info("%d finding(s)", findings)
    log.info("the repair is a schedule, not a run: poll from the stored "
             "watermark and route these events to somewhere a person looks")
    return 1 if findings else 0


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

/**
 * Read the key and member lifecycle events nobody has ever read.
 *
 * Read only. GET requests only, against the OpenAI Audit Logs API and the
 * Anthropic Compliance activity feed.
 *
 * Both feeds are pull-only, which is why the control exists everywhere and has
 * fired nowhere. The finding is not any single event; it is that nobody is
 * reading, so the last thing printed is a watermark for the next run.
 *
 * An empty feed is reported as unavailable and never as clean, and the
 * geography rule runs on OpenAI session actors only: the Anthropic activity
 * record has no country breakdown to test.
 */
const OPENAI = 'https://api.openai.com/v1';
const ANTHROPIC = 'https://api.anthropic.com/v1';
const ANTHROPIC_VERSION = '2023-06-01';

const OPENAI_EVENTS = ['api_key.created', 'api_key.updated', 'api_key.deleted',
                       'service_account.created', 'service_account.deleted',
                       'login.failed'];

export const OFF_ROSTER = 'off-roster-actor';
export const UNATTRIBUTABLE = 'unattributable';
export const UNEXPECTED_COUNTRY = 'unexpected-country';
export const OUT_OF_HOURS = 'out-of-hours';
export const REVIEWED = 'reviewed';

export const FEED_OK = 'feed-readable';
export const FEED_UNAVAILABLE = 'feed-unavailable';

const SEVERITY = [OFF_ROSTER, UNEXPECTED_COUNTRY, UNATTRIBUTABLE, OUT_OF_HOURS];

/** Epoch seconds from a unix integer or an RFC 3339 string. Pure. */
export function parseWhen(value) {
  if (value === null || value === undefined || value === '' ||
      typeof value === 'boolean') return null;
  if (typeof value === 'number') return Math.trunc(value);
  const text = String(value).trim();
  if (/^\d+$/.test(text)) return Number(text);
  const when = new Date(text);
  if (Number.isNaN(when.getTime())) return null;
  return Math.floor(when.getTime() / 1000);
}

/** A readable UTC timestamp. Pure. */
export function iso(epoch) {
  if (epoch === null || epoch === undefined) return '(no timestamp)';
  return `${new Date(Number(epoch) * 1000).toISOString().slice(0, 19)}Z`;
}

/** One audit-log entry in the common shape. Pure. Two actor shapes. */
export function normaliseOpenai(entry) {
  const row = entry ?? {};
  const actor = (row.actor && typeof row.actor === 'object') ? row.actor : {};
  const kind = String(actor.type ?? '').trim().toLowerCase();
  let email = null;
  let ip = null;
  let country = null;
  if (kind === 'session') {
    const session = (actor.session && typeof actor.session === 'object') ? actor.session : {};
    const user = (session.user && typeof session.user === 'object') ? session.user : {};
    email = user.email ?? null;
    ip = session.ip_address ?? null;
    const details = (session.ip_address_details &&
                     typeof session.ip_address_details === 'object')
      ? session.ip_address_details : {};
    country = details.country ?? null;
  } else if (kind === 'api_key') {
    const apiKey = (actor.api_key && typeof actor.api_key === 'object') ? actor.api_key : {};
    const user = (apiKey.user && typeof apiKey.user === 'object') ? apiKey.user : {};
    email = user.email ?? null;
  }
  const project = (row.project && typeof row.project === 'object') ? row.project : {};
  return { source: 'openai', type: String(row.type ?? '(untyped)'),
           when: parseWhen(row.effective_at), actorKind: kind || 'unknown',
           actorEmail: email ? String(email).trim().toLowerCase() : null,
           actorIp: ip, country,
           container: project.name ?? project.id ?? null };
}

/** One compliance activity in the common shape. Pure. country stays null. */
export function normaliseAnthropic(activity) {
  const row = activity ?? {};
  const actor = (row.actor && typeof row.actor === 'object') ? row.actor : {};
  const email = actor.email_address ?? null;
  return { source: 'anthropic', type: String(row.type ?? '(untyped)'),
           when: parseWhen(row.created_at),
           actorKind: email ? 'user' : 'unknown',
           actorEmail: email ? String(email).trim().toLowerCase() : null,
           actorIp: actor.ip_address ?? null, country: null,
           container: row.organization_id ?? null };
}

/** on-roster, off-roster or unattributable. Pure. */
export function resolveActor(event, roster) {
  const email = (event ?? {}).actorEmail;
  if (!email) return 'unattributable';
  return (roster ?? new Set()).has(String(email).trim().toLowerCase())
    ? 'on-roster' : 'off-roster';
}

/** The UTC hour of an event, or null. Pure. */
export function hourOf(event) {
  const when = (event ?? {}).when;
  if (when === null || when === undefined) return null;
  return new Date(Number(when) * 1000).getUTCHours();
}

/** Classify one normalised event. Pure. Returns [state, reasons]. */
export function grade(event, roster, businessHours = [7, 19], operatingCountries = null) {
  const reasons = [];
  const resolution = resolveActor(event, roster);
  if (resolution === 'off-roster') {
    reasons.push([OFF_ROSTER, 'the actor is not on the current roster']);
  } else if (resolution === 'unattributable') {
    reasons.push([UNATTRIBUTABLE,
      `an ${(event ?? {}).actorKind ?? 'unknown'} actor carries no user ` +
      'email, so no person can be attributed']);
  }

  const country = (event ?? {}).country;
  if (operatingCountries && operatingCountries.length && country) {
    const allowed = new Set(operatingCountries.map((c) => String(c).toUpperCase()));
    if (!allowed.has(String(country).trim().toUpperCase())) {
      reasons.push([UNEXPECTED_COUNTRY,
        `ip_address_details.country ${country} is outside the operating geographies`]);
    }
  }

  const hour = hourOf(event);
  const [start, end] = businessHours;
  const type = String((event ?? {}).type ?? '');
  const creation = type.endsWith('.created') || type.endsWith('.deleted');
  if (creation && hour !== null && !(hour >= start && hour < end)) {
    reasons.push([OUT_OF_HOURS,
      `created outside business hours (${String(hour).padStart(2, '0')}:00 UTC)`]);
  }

  if (!reasons.length) return [REVIEWED, []];
  const present = new Set(reasons.map(([state]) => state));
  const state = SEVERITY.find((s) => present.has(s));
  return [state, reasons.map(([, text]) => text)];
}

/** Whether the feed said anything at all. Pure. [state, detail]. */
export function feedState(events, reachable) {
  if (!reachable) {
    return [FEED_UNAVAILABLE,
      'the feed could not be read, so nothing below is a review of anything'];
  }
  if (!(events ?? []).length) {
    return [FEED_UNAVAILABLE,
      'the feed returned no events at all. Audit logging is gated to ' +
      'organizations that have it enabled, so this is not a clean result: ' +
      'it is an unknown one.'];
  }
  return [FEED_OK, `${events.length} event(s) read`];
}

/** Clusters of login.failed inside one window. Pure. */
export function failedLoginBursts(events, windowSeconds = 600, threshold = 5) {
  const rows = (events ?? [])
    .filter((e) => String(e?.type ?? '') === 'login.failed' && e?.when !== null &&
                   e?.when !== undefined)
    .sort((a, b) => a.when - b.when);
  for (let i = 0; i < rows.length; i += 1) {
    const window = rows.slice(i).filter((e) => e.when - rows[i].when <= windowSeconds);
    if (window.length >= threshold) {
      return [[rows[i].when, window.length, rows[i].actorEmail ?? '(no email)']];
    }
  }
  return [];
}

/** The newest timestamp seen, for the next run's cursor. Pure. */
export function watermark(events) {
  const stamps = (events ?? [])
    .map((e) => e?.when)
    .filter((w) => w !== null && w !== undefined);
  return stamps.length ? Math.max(...stamps) : null;
}

/** Whether this entry's project field means anything. Pure. */
export function projectCaveat(event) {
  if ((event ?? {}).source !== 'openai') return null;
  if ((event ?? {}).actorKind === 'api_key') {
    return 'project is not meaningful here: admin actions taken with an ' +
           'Admin API key are attributed to the default project';
  }
  return null;
}

async function getJson(headers, url, who) {
  const res = await fetch(url, { headers });
  if (res.status === 429) {
    throw new Error(`429 from ${who}: this feed declares its own rate limit ` +
                    'with Retry-After. Back off and resume from the stored watermark.');
  }
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from ${who}: the feed needs an ` +
                    'administration credential, and on Anthropic the ' +
                    'read:compliance_activities scope');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

async function collect(headers, base, path, params, who, cursor = 'after') {
  const rows = [];
  let after = null;
  for (let page = 0; page < 20; page += 1) {
    const url = new URL(base + path);
    for (const [k, v] of params) url.searchParams.append(k, String(v));
    if (after) url.searchParams.set(cursor, after);
    const body = await getJson(headers, url, who);
    rows.push(...(body.data ?? []));
    if (!body.has_more || !body.last_id) return rows;
    after = body.last_id;
  }
  return rows;
}

function report(name, events, roster, options, geography) {
  const [state, detail] = feedState(events, true);
  console.log(`${name}: ${state} (${detail}), roster of ${roster.size} member(s); ` +
              `${geography ? 'country and session rules available'
                           : 'no geography on this feed'}`);
  if (state === FEED_UNAVAILABLE) return 0;

  let findings = 0;
  for (const event of [...events].sort((a, b) => (a.when ?? 0) - (b.when ?? 0))) {
    const [verdict, reasons] = grade(event, roster, options.businessHours,
                                     geography ? options.countries : null);
    if (verdict === REVIEWED) continue;
    findings += 1;
    console.warn(`${verdict.padEnd(19)} ${event.type.padEnd(22)} ` +
                 `${iso(event.when)}  ` +
                 `${(event.actorEmail ?? `(${event.actorKind} actor)`).padEnd(18)} ` +
                 `${(event.actorIp ?? '-').padEnd(15)} ${event.country ?? ''}`);
    for (const reason of reasons) console.warn(`  reason: ${reason}`);
    const caveat = projectCaveat(event);
    if (caveat) console.log(`  note: ${caveat}`);
  }

  for (const [when, count, who] of failedLoginBursts(events)) {
    findings += 1;
    console.warn(`login-failed-burst   ${count} failure(s) within 10 minutes ` +
                 `from ${who}, starting ${iso(when)}`);
  }

  const mark = watermark(events);
  if (mark !== null) {
    console.log(`watermark: store the cursor ${mark} (${iso(mark)}) for the ` +
                `next ${name} run`);
  }
  return findings;
}

async function main() {
  const openaiKey = process.env.OPENAI_ADMIN_KEY;
  const anthropicKey = process.env.ANTHROPIC_ADMIN_KEY;
  if (!openaiKey && !anthropicKey) {
    console.error('set OPENAI_ADMIN_KEY or ANTHROPIC_ADMIN_KEY, or both; the ' +
                  'Anthropic credential also needs the ' +
                  'read:compliance_activities scope');
    process.exitCode = 2;
    return;
  }
  const days = Number(process.env.DAYS ?? 7);
  const options = {
    businessHours: [Number(process.env.HOURS_FROM ?? 7),
                    Number(process.env.HOURS_TO ?? 19)],
    countries: String(process.env.COUNTRIES ?? 'US,GB,DE,IE')
      .split(',').map((c) => c.trim()).filter(Boolean),
  };
  const since = Math.floor(Date.now() / 1000) - days * 86400;
  let findings = 0;

  if (openaiKey) {
    const headers = { Authorization: `Bearer ${openaiKey}` };
    const users = await collect(headers, OPENAI, '/organization/users',
                                [['limit', 100]], 'OpenAI');
    const roster = new Set(users.filter((u) => u.email)
      .map((u) => String(u.email).trim().toLowerCase()));
    const raw = await collect(headers, OPENAI, '/organization/audit_logs',
      [['limit', 100], ['effective_at[gte]', since],
       ...OPENAI_EVENTS.map((t) => ['event_types[]', t])], 'OpenAI');
    findings += report('openai', raw.map(normaliseOpenai), roster, options, true);
  }

  if (anthropicKey) {
    const headers = { 'x-api-key': anthropicKey, 'anthropic-version': ANTHROPIC_VERSION };
    const users = await collect(headers, ANTHROPIC, '/organizations/users',
                                [['limit', 1000]], 'Anthropic', 'after_id');
    const roster = new Set(users.filter((u) => u.email)
      .map((u) => String(u.email).trim().toLowerCase()));
    const raw = await collect(headers, ANTHROPIC, '/compliance/activities',
                              [['limit', 100]], 'Anthropic');
    const events = raw.map(normaliseAnthropic).filter((e) => (e.when ?? 0) >= since);
    findings += report('anthropic', events, roster, options, false);
  }

  console.log(`${findings} finding(s)`);
  console.log('the repair is a schedule, not a run: poll from the stored ' +
              'watermark and route these events to somewhere a person looks');
  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 llm_key_lifecycle_review import (failed_login_bursts, feed_state, grade,
                                      hour_of, iso, normalise_anthropic,
                                      normalise_openai, parse_when,
                                      project_caveat, resolve_actor, watermark)

ROSTER = {"dana@example.test", "marco@example.test"}
COUNTRIES = ["US", "GB"]


def at(text):
    return int(dt.datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp())


def session_entry(event_type, when, email, ip="198.51.100.24", country="US"):
    return {"id": "audit_1", "type": event_type, "effective_at": when,
            "project": {"id": "proj_1", "name": "prod"},
            "actor": {"type": "session",
                      "session": {"user": {"email": email}, "ip_address": ip,
                                  "ip_address_details": {"country": country}}}}


def test_a_key_minted_at_2am_by_somebody_who_has_left_trips_three_rules():
    # The note in one assertion. Every reason is reported, and the state is the
    # most severe of them, because a reviewer wants all three and triage wants
    # one.
    event = normalise_openai(session_entry(
        "api_key.created", at("2026-03-17T02:14:08Z"), "ada@example.test",
        country="NL"))
    state, reasons = grade(event, ROSTER, (7, 19), COUNTRIES)
    assert state == "off-roster-actor"
    assert len(reasons) == 3
    assert any("not on the current roster" in r for r in reasons)
    assert any("outside the operating geographies" in r for r in reasons)
    assert any("02:00 UTC" in r for r in reasons)
    assert iso(event["when"]) == "2026-03-17T02:14:08Z"


def test_an_empty_feed_is_unavailable_and_never_clean():
    # Audit logging is gated. Reading silence as "no findings" turns a missing
    # control into a passing check, which is the worst outcome available here.
    empty_state, empty_detail = feed_state([], True)
    assert empty_state == "feed-unavailable"
    assert "not a clean result" in empty_detail
    unreachable_state, unreachable_detail = feed_state([], False)
    assert unreachable_state == "feed-unavailable"
    assert "could not be read" in unreachable_detail
    ok_state, ok_detail = feed_state([{"type": "api_key.created"}], True)
    assert ok_state == "feed-readable"
    assert "1 event(s)" in ok_detail


def test_the_two_openai_actor_shapes_keep_their_email_in_different_places():
    session = normalise_openai(session_entry(
        "api_key.created", at("2026-08-11T10:02:00Z"), "Dana@Example.test"))
    assert session["actor_kind"] == "session"
    assert session["actor_email"] == "dana@example.test"
    assert session["country"] == "US"
    assert grade(session, ROSTER, (7, 19), COUNTRIES)[0] == "reviewed"

    by_key = normalise_openai({
        "type": "api_key.deleted", "effective_at": at("2026-08-02T11:40:55Z"),
        "project": {"id": "proj_default"},
        "actor": {"type": "api_key", "api_key": {"id": "key_track",
                                                 "service_account": {"id": "svc_1"}}}})
    assert by_key["actor_kind"] == "api_key"
    assert by_key["actor_email"] is None
    assert resolve_actor(by_key, ROSTER) == "unattributable"
    state, reasons = grade(by_key, ROSTER, (7, 19), COUNTRIES)
    assert state == "unattributable"
    assert any("no user email" in r for r in reasons)
    assert "default project" in project_caveat(by_key)
    assert project_caveat(session) is None


def test_an_anthropic_activity_has_no_country_so_the_rule_is_skipped():
    event = normalise_anthropic({
        "type": "api_key.created", "created_at": "2026-08-14T09:31:00Z",
        "organization_id": "org_1",
        "actor": {"email_address": "MARCO@example.test", "user_id": "u_1",
                  "ip_address": "203.0.113.9", "user_agent": "curl/8"}})
    assert event["source"] == "anthropic"
    assert event["country"] is None
    assert event["actor_email"] == "marco@example.test"
    # The country rule cannot run, so an on-roster in-hours event is clean
    # rather than being failed for a field the feed does not have.
    assert grade(event, ROSTER, (7, 19), COUNTRIES)[0] == "reviewed"
    assert project_caveat(event) is None
    anonymous = normalise_anthropic({"type": "api_key.deleted",
                                     "created_at": "2026-08-14T09:31:00Z"})
    assert anonymous["actor_email"] is None
    assert resolve_actor(anonymous, ROSTER) == "unattributable"


def test_timestamps_arrive_in_two_shapes_and_the_hour_is_utc():
    assert parse_when(1_772_000_000) == 1_772_000_000
    assert parse_when("2026-03-17T02:14:08Z") == at("2026-03-17T02:14:08Z")
    assert parse_when("1772000000") == 1_772_000_000
    assert parse_when(None) is None
    assert parse_when(True) is None
    assert parse_when("whenever") is None
    assert hour_of({"when": at("2026-03-17T02:14:08Z")}) == 2
    assert hour_of({}) is None
    assert iso(None) == "(no timestamp)"


def test_a_burst_of_failed_logins_and_the_watermark_for_the_next_run():
    base = at("2026-08-20T09:00:00Z")
    events = [{"type": "login.failed", "when": base + i * 60,
               "actor_email": "ada@example.test"} for i in range(6)]
    events.append({"type": "api_key.created", "when": base + 4000,
                   "actor_email": "dana@example.test"})
    bursts = failed_login_bursts(events)
    assert len(bursts) == 1
    when, count, who = bursts[0]
    assert when == base and count >= 5 and who == "ada@example.test"
    # One failure is a typo, not a burst.
    assert failed_login_bursts([events[0]]) == []
    assert failed_login_bursts([]) == []
    assert watermark(events) == base + 4000
    assert watermark([]) is None
    assert watermark([{"type": "x"}]) is None


def test_an_out_of_hours_read_event_is_not_a_creation():
    # The out-of-hours rule fires on lifecycle events, not on everything that
    # happens to be timestamped at night.
    updated = {"source": "openai", "type": "api_key.updated",
               "when": at("2026-08-20T03:00:00Z"), "actor_kind": "session",
               "actor_email": "dana@example.test", "actor_ip": "203.0.113.1",
               "country": "US"}
    assert grade(updated, ROSTER, (7, 19), COUNTRIES)[0] == "reviewed"
    created = dict(updated, type="service_account.created")
    state, reasons = grade(created, ROSTER, (7, 19), COUNTRIES)
    assert state == "out-of-hours"
    assert reasons == ["created outside business hours (03:00 UTC)"]

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { failedLoginBursts, feedState, grade, hourOf, iso, normaliseAnthropic,
         normaliseOpenai, parseWhen, projectCaveat, resolveActor, watermark }
  from './llm-key-lifecycle-review.mjs';

const ROSTER = new Set(['dana@example.test', 'marco@example.test']);
const COUNTRIES = ['US', 'GB'];
const at = (text) => Math.floor(new Date(text).getTime() / 1000);

const sessionEntry = (type, when, email, ip = '198.51.100.24', country = 'US') => ({
  id: 'audit_1', type, effective_at: when,
  project: { id: 'proj_1', name: 'prod' },
  actor: { type: 'session',
           session: { user: { email }, ip_address: ip,
                      ip_address_details: { country } } },
});

test('a key minted at 2am by somebody who has left trips three rules', () => {
  const event = normaliseOpenai(sessionEntry(
    'api_key.created', at('2026-03-17T02:14:08Z'), 'ada@example.test',
    '198.51.100.24', 'NL'));
  const [state, reasons] = grade(event, ROSTER, [7, 19], COUNTRIES);
  assert.equal(state, 'off-roster-actor');
  assert.equal(reasons.length, 3);
  assert.ok(reasons.some((r) => r.includes('not on the current roster')));
  assert.ok(reasons.some((r) => r.includes('outside the operating geographies')));
  assert.ok(reasons.some((r) => r.includes('02:00 UTC')));
  assert.equal(iso(event.when), '2026-03-17T02:14:08Z');
});

test('an empty feed is unavailable and never clean', () => {
  const [emptyState, emptyDetail] = feedState([], true);
  assert.equal(emptyState, 'feed-unavailable');
  assert.match(emptyDetail, /not a clean result/);
  const [unreachableState, unreachableDetail] = feedState([], false);
  assert.equal(unreachableState, 'feed-unavailable');
  assert.match(unreachableDetail, /could not be read/);
  const [okState, okDetail] = feedState([{ type: 'api_key.created' }], true);
  assert.equal(okState, 'feed-readable');
  assert.match(okDetail, /1 event\(s\)/);
});

test('the two openai actor shapes keep their email in different places', () => {
  const session = normaliseOpenai(sessionEntry(
    'api_key.created', at('2026-08-11T10:02:00Z'), 'Dana@Example.test'));
  assert.equal(session.actorKind, 'session');
  assert.equal(session.actorEmail, 'dana@example.test');
  assert.equal(session.country, 'US');
  assert.equal(grade(session, ROSTER, [7, 19], COUNTRIES)[0], 'reviewed');

  const byKey = normaliseOpenai({
    type: 'api_key.deleted', effective_at: at('2026-08-02T11:40:55Z'),
    project: { id: 'proj_default' },
    actor: { type: 'api_key',
             api_key: { id: 'key_track', service_account: { id: 'svc_1' } } } });
  assert.equal(byKey.actorKind, 'api_key');
  assert.equal(byKey.actorEmail, null);
  assert.equal(resolveActor(byKey, ROSTER), 'unattributable');
  const [state, reasons] = grade(byKey, ROSTER, [7, 19], COUNTRIES);
  assert.equal(state, 'unattributable');
  assert.ok(reasons.some((r) => r.includes('no user email')));
  assert.match(projectCaveat(byKey), /default project/);
  assert.equal(projectCaveat(session), null);
});

test('an anthropic activity has no country so the rule is skipped', () => {
  const event = normaliseAnthropic({
    type: 'api_key.created', created_at: '2026-08-14T09:31:00Z',
    organization_id: 'org_1',
    actor: { email_address: 'MARCO@example.test', user_id: 'u_1',
             ip_address: '203.0.113.9', user_agent: 'curl/8' } });
  assert.equal(event.source, 'anthropic');
  assert.equal(event.country, null);
  assert.equal(event.actorEmail, 'marco@example.test');
  assert.equal(grade(event, ROSTER, [7, 19], COUNTRIES)[0], 'reviewed');
  assert.equal(projectCaveat(event), null);
  const anonymous = normaliseAnthropic({ type: 'api_key.deleted',
                                         created_at: '2026-08-14T09:31:00Z' });
  assert.equal(anonymous.actorEmail, null);
  assert.equal(resolveActor(anonymous, ROSTER), 'unattributable');
});

test('timestamps arrive in two shapes and the hour is utc', () => {
  assert.equal(parseWhen(1772000000), 1772000000);
  assert.equal(parseWhen('2026-03-17T02:14:08Z'), at('2026-03-17T02:14:08Z'));
  assert.equal(parseWhen('1772000000'), 1772000000);
  assert.equal(parseWhen(null), null);
  assert.equal(parseWhen(true), null);
  assert.equal(parseWhen('whenever'), null);
  assert.equal(hourOf({ when: at('2026-03-17T02:14:08Z') }), 2);
  assert.equal(hourOf({}), null);
  assert.equal(iso(null), '(no timestamp)');
});

test('a burst of failed logins and the watermark for the next run', () => {
  const base = at('2026-08-20T09:00:00Z');
  const events = Array.from({ length: 6 }, (_, i) => (
    { type: 'login.failed', when: base + i * 60, actorEmail: 'ada@example.test' }));
  events.push({ type: 'api_key.created', when: base + 4000,
                actorEmail: 'dana@example.test' });
  const bursts = failedLoginBursts(events);
  assert.equal(bursts.length, 1);
  assert.equal(bursts[0][0], base);
  assert.ok(bursts[0][1] >= 5);
  assert.equal(bursts[0][2], 'ada@example.test');
  assert.deepEqual(failedLoginBursts([events[0]]), []);
  assert.deepEqual(failedLoginBursts([]), []);
  assert.equal(watermark(events), base + 4000);
  assert.equal(watermark([]), null);
  assert.equal(watermark([{ type: 'x' }]), null);
});

test('an out of hours read event is not a creation', () => {
  const updated = { source: 'openai', type: 'api_key.updated',
                    when: at('2026-08-20T03:00:00Z'), actorKind: 'session',
                    actorEmail: 'dana@example.test', actorIp: '203.0.113.1',
                    country: 'US' };
  assert.equal(grade(updated, ROSTER, [7, 19], COUNTRIES)[0], 'reviewed');
  const created = { ...updated, type: 'service_account.created' };
  const [state, reasons] = grade(created, ROSTER, [7, 19], COUNTRIES);
  assert.equal(state, 'out-of-hours');
  assert.deepEqual(reasons, ['created outside business hours (03:00 UTC)']);
});