A CMEK key config is inert but assumed to be encrypting

"""Find Anthropic CMEK key configs that are not encrypting anything.

Read only. Two paged GETs against /v1/organizations/external_keys and
/v1/organizations/workspaces with an Admin API key. Every request is a GET.

The external keys resource offers a validate call. It is a write verb, so this
script does not use it, and the repair for an unattached config is printed for
a human to run rather than performed.

Nothing secret is printed. Provider coordinates are resource identifiers rather
than credentials, and the AWS account id inside an ARN is masked anyway.
"""
import argparse
import logging
import os
import sys

import requests

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

API = "https://api.anthropic.com/v1"
VERSION = "2023-06-01"

FINDINGS = ("unattached-and-unused", "unattached-but-referenced",
            "archived-workspaces-only", "attached-nothing-visible",
            "geo-mismatch", "attachment-unreadable")

SEVERITY = {"unattached-and-unused": 0, "geo-mismatch": 1,
            "unattached-but-referenced": 2, "attached-nothing-visible": 3,
            "archived-workspaces-only": 4, "attachment-unreadable": 5}


def attachment_type(key):
    """"attached" / "unattached" / "unknown". Pure.

    Anything else is unknown rather than assumed unattached: guessing wrong in
    that direction produces a report telling somebody to delete a live key.
    """
    kind = str(((key or {}).get("attachment") or {}).get("type") or "").strip().lower()
    return kind if kind in ("attached", "unattached") else "unknown"


def mask_arn(arn):
    """Hide the account id in an AWS ARN. Pure. Non-ARNs pass through."""
    text = str(arn or "")
    parts = text.split(":")
    if len(parts) < 6 or parts[0] != "arn":
        return text or "unknown"
    parts[4] = "****"
    return ":".join(parts)


def kms_ref(provider_config):
    """One short line naming the KMS key. Pure. No credentials, ever."""
    cfg = provider_config or {}
    kind = str(cfg.get("type") or "").strip().lower()
    if kind == "aws":
        return "aws " + mask_arn(cfg.get("kms_arn"))
    if kind == "gcp":
        return "gcp " + str(cfg.get("key_name") or "unknown")
    if kind == "azure":
        return "azure %s in %s" % (cfg.get("key_name") or "unknown",
                                   cfg.get("vault_uri") or "unknown vault")
    return "unrecognised provider %s" % (kind or "none")


def workspace_geo(workspace):
    """The workspace's storage geo, or None. Pure."""
    geo = ((workspace or {}).get("data_residency") or {}).get("workspace_geo")
    return str(geo) if geo else None


def coverage(workspaces):
    """{external_key_id: {"live": [ids], "archived": [ids]}}. Pure.

    Built from the workspaces, because the attachment object carries only its
    own type and no list of what uses it.
    """
    out = {}
    for workspace in workspaces or []:
        row = workspace or {}
        key_id = row.get("external_key_id")
        if not key_id:
            continue
        entry = out.setdefault(str(key_id), {"live": [], "archived": []})
        bucket = "archived" if row.get("archived_at") else "live"
        entry[bucket].append(str(row.get("id") or "unknown"))
    for entry in out.values():
        entry["live"].sort()
        entry["archived"].sort()
    return out


def uncovered(workspaces):
    """(live, archived) workspace ids with no external_key_id at all. Pure."""
    live, archived = [], []
    for workspace in workspaces or []:
        row = workspace or {}
        if row.get("external_key_id"):
            continue
        (archived if row.get("archived_at") else live).append(
            str(row.get("id") or "unknown"))
    return (sorted(live), sorted(archived))


def classify(key, cover, geos):
    """Classify one key config. Pure. Returns (state, detail).

    cover: {"live": [ids], "archived": [ids]} from the workspace listing.
    geos:  [(workspace_id, workspace_geo)] for the workspaces that name it.
    """
    kind = attachment_type(key)
    live = list((cover or {}).get("live") or [])
    archived = list((cover or {}).get("archived") or [])

    if kind == "unknown":
        return ("attachment-unreadable",
                "attachment.type is not attached or unattached, so this audit "
                "will not say whether the config is in use")

    if kind == "unattached":
        if live or archived:
            return ("unattached-but-referenced",
                    "the config reports unattached while %d workspace(s) name it "
                    "(%s). The two listings disagree"
                    % (len(live) + len(archived), ", ".join(live + archived)))
        return ("unattached-and-unused",
                "attachment.type is unattached and no workspace names it. The API "
                "describes this state as inert: it takes part in no encryption "
                "path")

    if not live and not archived:
        return ("attached-nothing-visible",
                "reported attached, and no workspace this key can enumerate names "
                "it. An attachment you cannot see is still an attachment")
    if not live:
        return ("archived-workspaces-only",
                "attached, and the only workspaces naming it are archived (%s). "
                "Their retained data is still encrypted under this config"
                % ", ".join(archived))

    want = str((key or {}).get("geo") or "")
    mismatched = [(w, g) for w, g in (geos or []) if g and want and str(g) != want]
    if mismatched:
        return ("geo-mismatch",
                "config geo is %s and it covers %s"
                % (want, ", ".join("%s at %s" % pair for pair in mismatched)))

    return ("covered",
            "attached, covering %d live workspace(s)%s"
            % (len(live), " and %d archived" % len(archived) if archived else ""))


def repair_lines(state, key):
    """The repair for one key config. Pure. Printed, never performed.

    A delete is printed for exactly one state. external_key_id is write-once on
    a workspace, so no repair here proposes re-pointing a workspace at a
    different config: that is not something the API allows.
    """
    key_id = str((key or {}).get("id") or "unknown")
    lines = []
    if state not in FINDINGS:
        return lines
    if state == "unattached-and-unused":
        lines.append("attach it to the workspace it was made for. Attachment is "
                     "the step that makes a config live; creating it is not.")
        lines.append("if it was superseded, it can be deleted: DELETE "
                     "/v1/organizations/external_keys/%s. Nothing depends on it."
                     % key_id)
    elif state == "unattached-but-referenced":
        lines.append("do not delete this. Two listings disagree, and the safe "
                     "reading is the one that says something is using it.")
    elif state == "archived-workspaces-only":
        lines.append("do not delete this. Deleting a config an archived workspace "
                     "depends on makes that workspace's retained data "
                     "unrecoverable.")
    elif state == "attached-nothing-visible":
        lines.append("the coverage map is incomplete rather than empty. Widen the "
                     "workspace listing before concluding anything about this "
                     "config.")
    elif state == "geo-mismatch":
        lines.append("a workspace cannot be re-pointed: external_key_id is "
                     "write-once and cannot be detached or replaced. Resolve this "
                     "against the residency commitment, not by swapping keys.")
    else:
        lines.append("read this config by hand. The attachment discriminator was "
                     "not one of the two values this audit recognises.")
    lines.append("the validate call on this resource is a write verb and this "
                 "script does not use it. Run it deliberately if you need it.")
    return lines


def get(session, path, params):
    r = session.get(API + path, params=params, timeout=60)
    if r.status_code in (401,):
        raise SystemExit("401 from Anthropic: /v1/organizations/* needs an Admin "
                         "API key, not a workspace key")
    if r.status_code in (403, 404):
        return None
    r.raise_for_status()
    return r.json()


def paged_cursor(session, path, **params):
    """external_keys pagination: next_page fed back as the page parameter."""
    params = dict(params)
    while True:
        page = get(session, path, params)
        if page is None:
            return
        for item in page.get("data") or []:
            yield item
        cursor = page.get("next_page")
        if not page.get("has_more") or not cursor:
            return
        params = dict(params, page=cursor)


def paged_after_id(session, path, **params):
    """workspaces pagination: after_id, a different cursor in the same script."""
    params = dict(params)
    while True:
        page = get(session, path, params) or {}
        data = page.get("data") or []
        for item in data:
            yield item
        if not page.get("has_more") or not data:
            return
        params["after_id"] = page.get("last_id") or (data[-1] or {}).get("id")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--geo", default=None,
                    help="the storage geo your residency commitment claims")
    args = ap.parse_args()

    admin = os.environ.get("ANTHROPIC_ADMIN_KEY")
    if not admin:
        log.error("set ANTHROPIC_ADMIN_KEY to an Admin API key; a workspace key "
                  "cannot read /v1/organizations/*")
        return 2

    s = requests.Session()
    s.headers.update({"x-api-key": admin, "anthropic-version": VERSION})

    keys = list(paged_cursor(s, "/organizations/external_keys",
                             beta="true", limit=100))
    if not keys:
        probe = get(s, "/organizations/external_keys", {"beta": "true", "limit": 1})
        if probe is None:
            log.info("the external keys endpoint is not available to this "
                     "organization. CMEK is a beta enterprise feature and this is "
                     "an answer, not a finding.")
            return 0

    workspaces = list(paged_after_id(s, "/organizations/workspaces",
                                     beta="true", limit=100,
                                     include_archived="true"))
    cover = coverage(workspaces)
    by_id = {str((w or {}).get("id")): w for w in workspaces}

    findings = []
    for key in keys:
        key_id = str(key.get("id") or "")
        entry = cover.get(key_id) or {}
        geos = [(w, workspace_geo(by_id.get(w)))
                for w in (entry.get("live") or []) + (entry.get("archived") or [])]
        state, detail = classify(key, entry, geos)
        if state in FINDINGS:
            findings.append((key, state, detail))

    live_bare, archived_bare = uncovered(workspaces)

    log.info("%d external key config(s), %d workspace(s), %d finding(s)",
             len(keys), len(workspaces), len(findings))

    findings.sort(key=lambda r: (SEVERITY.get(r[1], 9), str(r[0].get("id") or "")))
    for key, state, detail in findings:
        log.warning("%-26s %-12s %s", state, key.get("id"),
                    key.get("display_name") or "(unnamed)")
        log.warning("  %s", detail)
        log.warning("  provider: %s", kms_ref(key.get("provider_config")))
        for line in repair_lines(state, key):
            log.warning("  repair: %s", line)

    if live_bare:
        log.warning("uncovered: %d of %d workspace(s) have no external_key_id at "
                    "all (%s)", len(live_bare), len(workspaces),
                    ", ".join(live_bare))
    if archived_bare:
        log.info("uncovered and archived: %d workspace(s) (%s)",
                 len(archived_bare), ", ".join(archived_bare))
    if args.geo:
        log.info("claimed storage geo: %s", args.geo)
        for workspace in workspaces:
            got = workspace_geo(workspace)
            if got and got != args.geo:
                log.warning("residency  %-12s workspace_geo is %s, and %s was "
                            "claimed", workspace.get("id"), got, args.geo)

    return 1 if (findings or live_bare) else 0


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

/**
 * Find Anthropic CMEK key configs that are not encrypting anything.
 *
 * Read only. Two paged GETs against /v1/organizations/external_keys and
 * /v1/organizations/workspaces with an Admin API key.
 *
 * The external keys resource offers a validate call. It is a write verb, so
 * this script does not use it. Provider coordinates are resource identifiers
 * rather than credentials, and the AWS account id inside an ARN is masked.
 */
const API = 'https://api.anthropic.com/v1';
const VERSION = '2023-06-01';

const FINDINGS = new Set(['unattached-and-unused', 'unattached-but-referenced',
                          'archived-workspaces-only', 'attached-nothing-visible',
                          'geo-mismatch', 'attachment-unreadable']);

const SEVERITY = { 'unattached-and-unused': 0, 'geo-mismatch': 1,
                   'unattached-but-referenced': 2, 'attached-nothing-visible': 3,
                   'archived-workspaces-only': 4, 'attachment-unreadable': 5 };

/** "attached" / "unattached" / "unknown". Pure. Never guesses. */
export function attachmentType(key) {
  const kind = String(key?.attachment?.type ?? '').trim().toLowerCase();
  return (kind === 'attached' || kind === 'unattached') ? kind : 'unknown';
}

/** Hide the account id in an AWS ARN. Pure. Non-ARNs pass through. */
export function maskArn(arn) {
  const text = String(arn ?? '');
  const parts = text.split(':');
  if (parts.length < 6 || parts[0] !== 'arn') return text || 'unknown';
  parts[4] = '****';
  return parts.join(':');
}

/** One short line naming the KMS key. Pure. No credentials, ever. */
export function kmsRef(providerConfig) {
  const cfg = providerConfig ?? {};
  const kind = String(cfg.type ?? '').trim().toLowerCase();
  if (kind === 'aws') return `aws ${maskArn(cfg.kms_arn)}`;
  if (kind === 'gcp') return `gcp ${cfg.key_name ?? 'unknown'}`;
  if (kind === 'azure') {
    return `azure ${cfg.key_name ?? 'unknown'} in ${cfg.vault_uri ?? 'unknown vault'}`;
  }
  return `unrecognised provider ${kind || 'none'}`;
}

/** The workspace's storage geo, or null. Pure. */
export function workspaceGeo(workspace) {
  const geo = workspace?.data_residency?.workspace_geo;
  return geo ? String(geo) : null;
}

/** {external_key_id: {live: [], archived: []}}. Pure. Built from workspaces. */
export function coverage(workspaces) {
  const out = {};
  for (const workspace of workspaces ?? []) {
    const keyId = workspace?.external_key_id;
    if (!keyId) continue;
    const entry = (out[String(keyId)] ??= { live: [], archived: [] });
    entry[workspace?.archived_at ? 'archived' : 'live']
      .push(String(workspace?.id ?? 'unknown'));
  }
  for (const entry of Object.values(out)) {
    entry.live.sort();
    entry.archived.sort();
  }
  return out;
}

/** [live, archived] workspace ids with no external_key_id at all. Pure. */
export function uncovered(workspaces) {
  const live = [];
  const archived = [];
  for (const workspace of workspaces ?? []) {
    if (workspace?.external_key_id) continue;
    (workspace?.archived_at ? archived : live).push(String(workspace?.id ?? 'unknown'));
  }
  return [live.sort(), archived.sort()];
}

/** Classify one key config. Pure. Returns [state, detail]. */
export function classify(key, cover, geos) {
  const kind = attachmentType(key);
  const live = [...(cover?.live ?? [])];
  const archived = [...(cover?.archived ?? [])];

  if (kind === 'unknown') {
    return ['attachment-unreadable',
            'attachment.type is not attached or unattached, so this audit will '
            + 'not say whether the config is in use'];
  }
  if (kind === 'unattached') {
    if (live.length || archived.length) {
      return ['unattached-but-referenced',
              `the config reports unattached while ${live.length + archived.length} `
              + `workspace(s) name it (${[...live, ...archived].join(', ')}). The two `
              + 'listings disagree'];
    }
    return ['unattached-and-unused',
            'attachment.type is unattached and no workspace names it. The API '
            + 'describes this state as inert: it takes part in no encryption path'];
  }
  if (!live.length && !archived.length) {
    return ['attached-nothing-visible',
            'reported attached, and no workspace this key can enumerate names it. '
            + 'An attachment you cannot see is still an attachment'];
  }
  if (!live.length) {
    return ['archived-workspaces-only',
            `attached, and the only workspaces naming it are archived `
            + `(${archived.join(', ')}). Their retained data is still encrypted `
            + 'under this config'];
  }
  const want = String(key?.geo ?? '');
  const mismatched = (geos ?? []).filter(([, g]) => g && want && String(g) !== want);
  if (mismatched.length) {
    return ['geo-mismatch',
            `config geo is ${want} and it covers `
            + mismatched.map(([w, g]) => `${w} at ${g}`).join(', ')];
  }
  return ['covered',
          `attached, covering ${live.length} live workspace(s)`
          + (archived.length ? ` and ${archived.length} archived` : '')];
}

/** The repair for one key config. Pure. Printed, never performed. */
export function repairLines(state, key) {
  const keyId = String(key?.id ?? 'unknown');
  const lines = [];
  if (!FINDINGS.has(state)) return lines;
  if (state === 'unattached-and-unused') {
    lines.push('attach it to the workspace it was made for. Attachment is the step '
      + 'that makes a config live; creating it is not.');
    lines.push('if it was superseded, it can be deleted: DELETE '
      + `/v1/organizations/external_keys/${keyId}. Nothing depends on it.`);
  } else if (state === 'unattached-but-referenced') {
    lines.push('do not delete this. Two listings disagree, and the safe reading is '
      + 'the one that says something is using it.');
  } else if (state === 'archived-workspaces-only') {
    lines.push('do not delete this. Deleting a config an archived workspace depends '
      + "on makes that workspace's retained data unrecoverable.");
  } else if (state === 'attached-nothing-visible') {
    lines.push('the coverage map is incomplete rather than empty. Widen the '
      + 'workspace listing before concluding anything about this config.');
  } else if (state === 'geo-mismatch') {
    lines.push('a workspace cannot be re-pointed: external_key_id is write-once and '
      + 'cannot be detached or replaced. Resolve this against the residency '
      + 'commitment, not by swapping keys.');
  } else {
    lines.push('read this config by hand. The attachment discriminator was not one '
      + 'of the two values this audit recognises.');
  }
  lines.push('the validate call on this resource is a write verb and this script '
    + 'does not use it. Run it deliberately if you need it.');
  return lines;
}

async function read(key, path, params) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, String(v));
  const r = await fetch(url, {
    headers: { 'x-api-key': key, 'anthropic-version': VERSION },
  });
  if (r.status === 401) {
    throw new Error('401 from Anthropic: /v1/organizations/* needs an Admin API '
                    + 'key, not a workspace key');
  }
  if (r.status === 403 || r.status === 404) return null;
  if (!r.ok) throw new Error(`${r.status} ${await r.text()}`);
  return r.json();
}

async function pagedCursor(key, path, params) {
  const out = [];
  let q = { ...params };
  for (;;) {
    const page = await read(key, path, q);
    if (page === null) return out;
    out.push(...(page.data ?? []));
    if (!page.has_more || !page.next_page) return out;
    q = { ...q, page: page.next_page };
  }
}

async function pagedAfterId(key, path, params) {
  const out = [];
  const q = { ...params };
  for (;;) {
    const page = (await read(key, path, q)) ?? {};
    const data = page.data ?? [];
    out.push(...data);
    if (!page.has_more || data.length === 0) return out;
    q.after_id = page.last_id ?? data[data.length - 1]?.id;
  }
}

async function main() {
  const admin = process.env.ANTHROPIC_ADMIN_KEY;
  if (!admin) {
    console.error('set ANTHROPIC_ADMIN_KEY to an Admin API key; a workspace key '
                  + 'cannot read /v1/organizations/*');
    process.exitCode = 2;
    return;
  }
  const wantGeo = process.env.GEO ?? null;

  const keys = await pagedCursor(admin, '/organizations/external_keys',
                                 { beta: 'true', limit: 100 });
  if (!keys.length) {
    const probe = await read(admin, '/organizations/external_keys',
                             { beta: 'true', limit: 1 });
    if (probe === null) {
      console.log('the external keys endpoint is not available to this '
        + 'organization. CMEK is a beta enterprise feature and this is an answer, '
        + 'not a finding.');
      return;
    }
  }

  const workspaces = await pagedAfterId(admin, '/organizations/workspaces',
    { beta: 'true', limit: 100, include_archived: 'true' });
  const cover = coverage(workspaces);
  const byId = Object.fromEntries(workspaces.map((w) => [String(w?.id), w]));

  const findings = [];
  for (const key of keys) {
    const entry = cover[String(key.id ?? '')] ?? {};
    const geos = [...(entry.live ?? []), ...(entry.archived ?? [])]
      .map((w) => [w, workspaceGeo(byId[w])]);
    const [state, detail] = classify(key, entry, geos);
    if (FINDINGS.has(state)) findings.push([key, state, detail]);
  }

  const [liveBare, archivedBare] = uncovered(workspaces);

  console.log(`${keys.length} external key config(s), ${workspaces.length} `
              + `workspace(s), ${findings.length} finding(s)`);

  findings.sort(([ka, sa], [kb, sb]) =>
    (SEVERITY[sa] ?? 9) - (SEVERITY[sb] ?? 9)
    || String(ka.id ?? '').localeCompare(String(kb.id ?? '')));

  for (const [key, state, detail] of findings) {
    console.warn(`${state.padEnd(26)} ${String(key.id).padEnd(12)} `
                 + `${key.display_name ?? '(unnamed)'}`);
    console.warn(`  ${detail}`);
    console.warn(`  provider: ${kmsRef(key.provider_config)}`);
    for (const line of repairLines(state, key)) console.warn(`  repair: ${line}`);
  }

  if (liveBare.length) {
    console.warn(`uncovered: ${liveBare.length} of ${workspaces.length} workspace(s) `
                 + `have no external_key_id at all (${liveBare.join(', ')})`);
  }
  if (archivedBare.length) {
    console.log(`uncovered and archived: ${archivedBare.length} workspace(s) `
                + `(${archivedBare.join(', ')})`);
  }
  if (wantGeo) {
    console.log(`claimed storage geo: ${wantGeo}`);
    for (const workspace of workspaces) {
      const got = workspaceGeo(workspace);
      if (got && got !== wantGeo) {
        console.warn(`residency  ${String(workspace.id).padEnd(12)} workspace_geo `
                     + `is ${got}, and ${wantGeo} was claimed`);
      }
    }
  }
  process.exitCode = (findings.length || liveBare.length) ? 1 : 0;
}

if (import.meta.url === `file://${process.argv[1]}`) await main();

from anthropic_cmek_external_key_audit import (attachment_type, classify,
                                                coverage, kms_ref, mask_arn,
                                                repair_lines, uncovered,
                                                workspace_geo)

ARN = "arn:aws:kms:eu-west-1:210987654321:key/9f2c"


def key(kid, kind="unattached", geo="eu", name="EU customer key"):
    return {"id": kid, "type": "external_key", "display_name": name, "geo": geo,
            "attachment": {"type": kind},
            "provider_config": {"type": "aws", "kms_arn": ARN}}


def workspace(wid, key_id=None, geo="eu", archived=None):
    return {"id": wid, "type": "workspace", "name": wid,
            "external_key_id": key_id, "archived_at": archived,
            "data_residency": {"workspace_geo": geo,
                               "default_inference_geo": geo,
                               "allowed_inference_geos": "unrestricted"}}


def test_two_configs_with_no_live_workspace_and_only_one_may_be_deleted():
    # The note, and the most dangerous thing in it. Neither config covers a live
    # workspace. One is inert; deleting the other destroys retained data.
    inert = key("ekey_01hq", "unattached")
    holding = key("ekey_01gd", "attached", name="Legacy tenant key")
    cover = coverage([workspace("wrk_04", "ekey_01gd", archived=1_700_000_000)])

    state_a, detail_a = classify(inert, cover.get("ekey_01hq"), [])
    assert state_a == "unattached-and-unused"
    assert "inert" in detail_a

    state_b, detail_b = classify(holding, cover.get("ekey_01gd"),
                                 [("wrk_04", "eu")])
    assert state_b == "archived-workspaces-only"
    assert "still encrypted under this config" in detail_b

    lines_a = repair_lines(state_a, inert)
    lines_b = repair_lines(state_b, holding)
    assert any("can be deleted" in line for line in lines_a)
    assert not any("can be deleted" in line for line in lines_b)
    assert any("unrecoverable" in line for line in lines_b)


def test_when_the_two_listings_disagree_the_safe_reading_wins():
    stale = key("ekey_01zz", "unattached")
    cover = coverage([workspace("wrk_09", "ekey_01zz")])
    state, detail = classify(stale, cover.get("ekey_01zz"), [("wrk_09", "eu")])
    assert state == "unattached-but-referenced"
    assert "The two listings disagree" in detail
    lines = repair_lines(state, stale)
    assert any("do not delete this" in line for line in lines)
    assert not any("can be deleted" in line for line in lines)


def test_an_unrecognised_attachment_is_never_assumed_unattached():
    assert attachment_type(key("e", "attached")) == "attached"
    assert attachment_type(key("e", "UNATTACHED")) == "unattached"
    assert attachment_type({"id": "e", "attachment": {"type": "pending"}}) == "unknown"
    assert attachment_type({"id": "e"}) == "unknown"
    assert attachment_type(None) == "unknown"
    state, detail = classify({"id": "e", "attachment": {"type": "pending"}}, {}, [])
    assert state == "attachment-unreadable"
    assert "will not say whether" in detail


def test_a_geo_mismatch_is_read_across_the_workspaces_it_covers():
    eu_key = key("ekey_01eu", "attached", geo="eu")
    cover = coverage([workspace("wrk_01", "ekey_01eu", geo="eu"),
                      workspace("wrk_02", "ekey_01eu", geo="us")])
    geos = [("wrk_01", "eu"), ("wrk_02", "us")]
    state, detail = classify(eu_key, cover.get("ekey_01eu"), geos)
    assert state == "geo-mismatch"
    assert "wrk_02 at us" in detail
    assert "wrk_01" not in detail
    assert any("write-once" in line for line in repair_lines(state, eu_key))
    # Matching geos are simply covered.
    assert classify(eu_key, cover.get("ekey_01eu"),
                    [("wrk_01", "eu")])[0] == "covered"
    assert repair_lines("covered", eu_key) == []


def test_the_coverage_map_and_the_uncovered_split():
    rows = [workspace("wrk_01"), workspace("wrk_02", None),
            workspace("wrk_03", "ekey_01hq"),
            workspace("wrk_04", "ekey_01hq", archived=1_700_000_000),
            workspace("wrk_05", None, archived=1_700_000_001)]
    cover = coverage(rows)
    assert cover == {"ekey_01hq": {"live": ["wrk_03"], "archived": ["wrk_04"]}}
    # A null external_key_id must never become a key id.
    assert "None" not in cover and None not in cover
    assert uncovered(rows) == (["wrk_01", "wrk_02"], ["wrk_05"])
    assert coverage(None) == {}
    assert uncovered(None) == ([], [])
    assert workspace_geo(rows[0]) == "eu"
    assert workspace_geo({"id": "w"}) is None


def test_the_provider_line_names_the_key_and_masks_the_account():
    assert mask_arn(ARN) == "arn:aws:kms:eu-west-1:****:key/9f2c"
    assert mask_arn("not-an-arn") == "not-an-arn"
    assert mask_arn(None) == "unknown"
    assert kms_ref({"type": "aws", "kms_arn": ARN}).startswith("aws arn:aws:kms:")
    assert "210987654321" not in kms_ref({"type": "aws", "kms_arn": ARN})
    assert kms_ref({"type": "gcp", "key_name": "projects/p/locations/eu/x"}) == \
        "gcp projects/p/locations/eu/x"
    assert "vault.azure.net" in kms_ref(
        {"type": "azure", "key_name": "k", "vault_uri": "https://v.vault.azure.net"})
    assert kms_ref({"type": "quantum"}) == "unrecognised provider quantum"
    assert kms_ref(None) == "unrecognised provider none"
    # Every finding says the validate call was deliberately not made.
    assert any("write verb" in line
               for line in repair_lines("unattached-and-unused", key("ekey_1")))

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { attachmentType, classify, coverage, kmsRef, maskArn, repairLines,
         uncovered, workspaceGeo } from './anthropic-cmek-external-key-audit.mjs';

const ARN = 'arn:aws:kms:eu-west-1:210987654321:key/9f2c';

const key = (id, kind = 'unattached', geo = 'eu', name = 'EU customer key') =>
  ({ id, type: 'external_key', display_name: name, geo,
     attachment: { type: kind },
     provider_config: { type: 'aws', kms_arn: ARN } });

const workspace = (id, keyId = null, geo = 'eu', archived = null) =>
  ({ id, type: 'workspace', name: id, external_key_id: keyId,
     archived_at: archived,
     data_residency: { workspace_geo: geo, default_inference_geo: geo,
                       allowed_inference_geos: 'unrestricted' } });

test('two configs with no live workspace, and only one may be deleted', () => {
  const inert = key('ekey_01hq', 'unattached');
  const holding = key('ekey_01gd', 'attached', 'eu', 'Legacy tenant key');
  const cover = coverage([workspace('wrk_04', 'ekey_01gd', 'eu', 1700000000)]);

  const [stateA, detailA] = classify(inert, cover.ekey_01hq, []);
  assert.equal(stateA, 'unattached-and-unused');
  assert.ok(detailA.includes('inert'));

  const [stateB, detailB] = classify(holding, cover.ekey_01gd, [['wrk_04', 'eu']]);
  assert.equal(stateB, 'archived-workspaces-only');
  assert.ok(detailB.includes('still encrypted under this config'));

  const linesA = repairLines(stateA, inert);
  const linesB = repairLines(stateB, holding);
  assert.ok(linesA.some((l) => l.includes('can be deleted')));
  assert.ok(!linesB.some((l) => l.includes('can be deleted')));
  assert.ok(linesB.some((l) => l.includes('unrecoverable')));
});

test('when the two listings disagree the safe reading wins', () => {
  const stale = key('ekey_01zz', 'unattached');
  const cover = coverage([workspace('wrk_09', 'ekey_01zz')]);
  const [state, detail] = classify(stale, cover.ekey_01zz, [['wrk_09', 'eu']]);
  assert.equal(state, 'unattached-but-referenced');
  assert.ok(detail.includes('The two listings disagree'));
  const lines = repairLines(state, stale);
  assert.ok(lines.some((l) => l.includes('do not delete this')));
  assert.ok(!lines.some((l) => l.includes('can be deleted')));
});

test('an unrecognised attachment is never assumed unattached', () => {
  assert.equal(attachmentType(key('e', 'attached')), 'attached');
  assert.equal(attachmentType(key('e', 'UNATTACHED')), 'unattached');
  assert.equal(attachmentType({ id: 'e', attachment: { type: 'pending' } }), 'unknown');
  assert.equal(attachmentType({ id: 'e' }), 'unknown');
  assert.equal(attachmentType(null), 'unknown');
  const [state, detail] = classify({ id: 'e', attachment: { type: 'pending' } }, {}, []);
  assert.equal(state, 'attachment-unreadable');
  assert.ok(detail.includes('will not say whether'));
});

test('a geo mismatch is read across the workspaces it covers', () => {
  const euKey = key('ekey_01eu', 'attached', 'eu');
  const cover = coverage([workspace('wrk_01', 'ekey_01eu', 'eu'),
                          workspace('wrk_02', 'ekey_01eu', 'us')]);
  const geos = [['wrk_01', 'eu'], ['wrk_02', 'us']];
  const [state, detail] = classify(euKey, cover.ekey_01eu, geos);
  assert.equal(state, 'geo-mismatch');
  assert.ok(detail.includes('wrk_02 at us'));
  assert.ok(!detail.includes('wrk_01'));
  assert.ok(repairLines(state, euKey).some((l) => l.includes('write-once')));
  assert.equal(classify(euKey, cover.ekey_01eu, [['wrk_01', 'eu']])[0], 'covered');
  assert.deepEqual(repairLines('covered', euKey), []);
});

test('the coverage map and the uncovered split', () => {
  const rows = [workspace('wrk_01'), workspace('wrk_02', null),
                workspace('wrk_03', 'ekey_01hq'),
                workspace('wrk_04', 'ekey_01hq', 'eu', 1700000000),
                workspace('wrk_05', null, 'eu', 1700000001)];
  const cover = coverage(rows);
  assert.deepEqual(cover, { ekey_01hq: { live: ['wrk_03'], archived: ['wrk_04'] } });
  assert.equal(Object.keys(cover).length, 1);
  assert.deepEqual(uncovered(rows), [['wrk_01', 'wrk_02'], ['wrk_05']]);
  assert.deepEqual(coverage(null), {});
  assert.deepEqual(uncovered(null), [[], []]);
  assert.equal(workspaceGeo(rows[0]), 'eu');
  assert.equal(workspaceGeo({ id: 'w' }), null);
});

test('the provider line names the key and masks the account', () => {
  assert.equal(maskArn(ARN), 'arn:aws:kms:eu-west-1:****:key/9f2c');
  assert.equal(maskArn('not-an-arn'), 'not-an-arn');
  assert.equal(maskArn(null), 'unknown');
  assert.ok(kmsRef({ type: 'aws', kms_arn: ARN }).startsWith('aws arn:aws:kms:'));
  assert.ok(!kmsRef({ type: 'aws', kms_arn: ARN }).includes('210987654321'));
  assert.equal(kmsRef({ type: 'gcp', key_name: 'projects/p/locations/eu/x' }),
               'gcp projects/p/locations/eu/x');
  assert.ok(kmsRef({ type: 'azure', key_name: 'k',
                     vault_uri: 'https://v.vault.azure.net' })
    .includes('vault.azure.net'));
  assert.equal(kmsRef({ type: 'quantum' }), 'unrecognised provider quantum');
  assert.equal(kmsRef(null), 'unrecognised provider none');
  assert.ok(repairLines('unattached-and-unused', key('ekey_1'))
    .some((l) => l.includes('write verb')));
});