feature_not_enabled: the admin API needs Enterprise

"""Measure, per installation, whether the admin API surface exists at all.

Read only, one probe per tenant. The admin.* families are Enterprise Grid
features, and a workspace on a smaller plan answers feature_not_enabled no
matter who calls them with what scopes. That is a property of the customer's
contract rather than of your integration, so the job here is to measure it and
store it, not to fix it.

The plan is not readable. No documented read method returns it: team.info
comes back with id, name, domain, email_domain, icon and, on Grid,
enterprise_id and enterprise_name, and nothing about tier or billing. So
capability is an experiment: one admin.teams.list?limit=1 per installation.

The rule this script exists to enforce: only feature_not_enabled means the
surface is unavailable. not_an_admin, not_allowed_token_type and missing_scope
mean the probe never reached the question, and storing any of them as an
absence creates a permanent false negative that no later run corrects, because
the feature is gated off and never probed again.
"""
import argparse
import json
import logging
import sys

import requests

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

API = "https://slack.com/api/"

# The cheapest admin read that produces the answer. Tier 2, one row, and
# feature_not_enabled is in its documented error list.
PROBE = ("admin.teams.list", {"limit": 1})

# Refusals that mean the probe never reached the question, mapped to the note
# that owns each one. Every value here becomes an unknown, never an absence.
NOT_THIS_QUESTION = {
    "not_an_admin": "/slack/not-an-admin/",
    "not_allowed_token_type": "/slack/admin-method-needs-user-token/",
    "missing_scope": "/slack/missing-scope-on-read/",
    "accesslimited": "/slack/accesslimited-ip-allowlist/",
    "ekm_access_denied": "/slack/ekm-access-denied/",
}

TRANSIENT = ("ratelimited", "internal_error", "service_unavailable", "fatal_error",
             "request_timeout", "unparseable_body")

# Admin method -> (non-admin equivalent, what the equivalent cannot see).
FALLBACKS = {
    "admin.conversations.search": (
        "conversations.list",
        "one workspace instead of the organization, and only conversations this "
        "token can already see"),
    "admin.users.list": (
        "users.list",
        "one workspace instead of every workspace in the org"),
    "admin.teams.list": (
        "team.info",
        "the workspace you are in, rather than an enumeration of the org"),
    "admin.usergroups.listChannels": (
        "usergroups.list with include_users",
        "the groups in this workspace, without the org wide channel assignment"),
}


def plan_is_readable():
    """Can the workspace's plan be read from the API? Pure.

    A function rather than a comment because it is load bearing: every other
    decision here is shaped by the fact that the answer is no, and a test
    keeps the claim attached to the code that depends on it.
    """
    return (False, "no documented read method returns the plan; team.info returns "
                   "id, name, domain, email_domain, icon and, on Grid, enterprise_id "
                   "and enterprise_name, and no billing or tier field")


def grid_signal(auth_body):
    """Does this token sit inside an Enterprise organization? Pure.

    A signal, and labelled one everywhere it is printed. It is suggestive
    about capability and it is not capability, so it never feeds the gate.
    """
    doc = auth_body or {}
    if doc.get("ok") is not True:
        return ("unknown", "auth.test did not answer: %s" % (doc.get("error") or "?"))
    if doc.get("enterprise_id"):
        return ("grid", "auth.test returned an enterprise_id, so this workspace sits "
                        "inside an organization")
    return ("not-grid", "auth.test returned no enterprise_id, which strongly suggests "
                        "the admin surface will not answer")


def capability(body):
    """Read one probe response as a capability. Pure.

    Seven states, and the verbosity of the unknown half is the point:

      available          the probe answered ok.
      unavailable-plan   feature_not_enabled; the only state that gates a feature off.
      unknown-role       not_an_admin; the caller lacks the organization role.
      unknown-class      not_allowed_token_type; a bot token was used.
      unknown-scope      missing_scope; the grant is short.
      unknown-transient  ratelimited or a 5xx shaped error; ask again.
      unknown-other      anything else, including the Grid session refusals.
    """
    doc = body or {}
    if doc.get("ok") is True:
        return ("available", "the probe answered ok, so the admin surface exists here")
    error = str(doc.get("error") or "")
    if error == "feature_not_enabled":
        return ("unavailable-plan", "feature_not_enabled; the admin API is an "
                                    "Enterprise Grid surface and no scope or role "
                                    "changes that")
    if error == "not_an_admin":
        return ("unknown-role", "not_an_admin; the caller does not hold the "
                                "organization role, so the plan was never tested")
    if error == "not_allowed_token_type":
        return ("unknown-class", "not_allowed_token_type; this is a bot token, so the "
                                 "plan was never tested")
    if error == "missing_scope":
        return ("unknown-scope", "missing_scope; the grant is short, so the plan was "
                                 "never tested")
    if error in TRANSIENT:
        return ("unknown-transient", "%s; ask again rather than concluding anything"
                                     % (error or "no error"))
    return ("unknown-other", "%s; not a refusal that answers the plan question"
                             % (error or "no error"))


def gate(state):
    """Turn a capability into the flag you are allowed to store. Pure.

    on     the feature is available for this tenant.
    off    measured unavailable, and only ever from feature_not_enabled.
    retry  inconclusive; probe again rather than writing the tenant off.

    A false off is invisible and permanent, because a gated feature is never
    probed again. A false retry costs one request.
    """
    if state == "available":
        return ("on", "measured available")
    if state == "unavailable-plan":
        return ("off", "measured unavailable; this is the only state that gates a "
                       "feature off")
    return ("retry", "inconclusive; an unknown is not an absence and must not be "
                     "stored as one")


def owner_of(error):
    """Which note owns this refusal, if not this one? Pure."""
    return NOT_THIS_QUESTION.get(str(error or "").strip())


def fallback_for(method):
    """The non-admin equivalent and what it cannot see. Pure."""
    return FALLBACKS.get(str(method or "").strip())


def matrix_verdict(rows):
    """One line for the whole matrix. Pure.

    rows: [(tenant, capability), ...]. Unknowns are reported as the finding,
    because a matrix with unknowns in it is not yet a matrix.
    """
    counts = {}
    for _tenant, state in rows or []:
        key = "unknown" if state.startswith("unknown") else state
        counts[key] = counts.get(key, 0) + 1
    if not rows:
        return ("empty", "no installations were probed")
    unknown = counts.get("unknown", 0)
    summary = "%d available, %d unavailable, %d unknown" % (
        counts.get("available", 0), counts.get("unavailable-plan", 0), unknown)
    if unknown:
        return ("%d unknown" % unknown, summary)
    return ("measured", summary)


def get(session, method, token, params=None):
    """One GET with one tenant's token. Returns the parsed body."""
    r = session.get(API + method, params=params or {},
                    headers={"Authorization": "Bearer " + token}, timeout=30)
    try:
        return r.json()
    except ValueError:
        return {"ok": False, "error": "unparseable_body"}


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--installs", required=True,
                    help="JSON list of installations, each with id and user_token")
    ap.add_argument("--methods", default="admin.conversations.search",
                    help="comma separated admin methods your feature depends on, so "
                         "their fallbacks are printed")
    args = ap.parse_args()

    with open(args.installs, encoding="utf-8") as handle:
        installs = json.load(handle) or []

    readable, why = plan_is_readable()
    if not readable:
        log.info("plan       not readable     %s", why)

    session = requests.Session()
    rows = []
    for row in installs:
        row = row or {}
        tenant = str(row.get("id") or "?")
        token = row.get("user_token") or ""
        if not token:
            log.warning("tenant     %-16s no user token stored, so nothing can be "
                        "measured", tenant)
            rows.append((tenant, "unknown-class"))
            continue
        signal, signal_why = grid_signal(get(session, "auth.test", token))
        body = get(session, PROBE[0], token, PROBE[1])
        state, detail = capability(body)
        flag, flag_why = gate(state)
        level = log.info if state in ("available", "unavailable-plan") else log.warning
        level("tenant     %-16s signal=%-9s capability=%-18s gate=%s",
              tenant, signal, state, flag)
        log.info("           signal           %s", signal_why)
        log.info("           note             %s", detail)
        if flag == "retry":
            log.warning("           gate             %s", flag_why)
        pointer = owner_of(body.get("error"))
        if pointer:
            log.info("           see also         %s", pointer)
        rows.append((tenant, state))

    for method in [m.strip() for m in args.methods.split(",") if m.strip()]:
        pair = fallback_for(method)
        if not pair:
            log.info("fallback   %-30s none named here; check the method reference",
                     method)
            continue
        log.info("fallback   %s -> %s", method, pair[0])
        log.info("           loses            %s", pair[1])

    state, summary = matrix_verdict(rows)
    if state == "measured":
        log.info("matrix     %d tenant(s)      %s", len(rows), summary)
        log.info("verdict    measured         every tenant was gated on a measurement")
        return 0
    log.warning("matrix     %d tenant(s)      %s", len(rows), summary)
    log.warning("verdict    %-16s re-probe before gating; an unknown is not an "
                "absence", state)
    log.warning("  repair: feature detect at install time and store the result, so the "
                "gate reads a measured fact rather than an assumption")
    log.warning("  repair: present a clear requires Enterprise Grid message for a "
                "measured off, and a pending state for a retry")
    log.warning("  repair: build the non-admin fallback where the narrower answer is "
                "worth having; the loss is printed above")
    return 1


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

/**
 * Measure, per installation, whether the admin API surface exists at all.
 *
 * Read only, one probe per tenant. The admin.* families are Enterprise Grid
 * features and a workspace on a smaller plan answers feature_not_enabled
 * whoever calls them with whatever scopes.
 *
 * The plan is not readable: no documented read method returns it, so
 * capability is an experiment rather than a lookup. The rule this module
 * exists to enforce is that only feature_not_enabled means unavailable.
 * not_an_admin, not_allowed_token_type and missing_scope mean the probe never
 * reached the question, and storing one of those as an absence creates a
 * permanent false negative that no later run corrects.
 */
import { readFile } from 'node:fs/promises';

const API = 'https://slack.com/api/';

export const PROBE = ['admin.teams.list', { limit: 1 }];

export const NOT_THIS_QUESTION = {
  not_an_admin: '/slack/not-an-admin/',
  not_allowed_token_type: '/slack/admin-method-needs-user-token/',
  missing_scope: '/slack/missing-scope-on-read/',
  accesslimited: '/slack/accesslimited-ip-allowlist/',
  ekm_access_denied: '/slack/ekm-access-denied/',
};

export const TRANSIENT = ['ratelimited', 'internal_error', 'service_unavailable',
  'fatal_error', 'request_timeout', 'unparseable_body'];

export const FALLBACKS = {
  'admin.conversations.search': ['conversations.list',
    'one workspace instead of the organization, and only conversations this token '
    + 'can already see'],
  'admin.users.list': ['users.list',
    'one workspace instead of every workspace in the org'],
  'admin.teams.list': ['team.info',
    'the workspace you are in, rather than an enumeration of the org'],
  'admin.usergroups.listChannels': ['usergroups.list with include_users',
    'the groups in this workspace, without the org wide channel assignment'],
};

// A function rather than a comment because it is load bearing.
export function planIsReadable() {
  return [false, 'no documented read method returns the plan; team.info returns id, '
    + 'name, domain, email_domain, icon and, on Grid, enterprise_id and '
    + 'enterprise_name, and no billing or tier field'];
}

// Suggestive about capability, and not capability. Never feeds the gate.
export function gridSignal(authBody) {
  const doc = authBody ?? {};
  if (doc.ok !== true) return ['unknown', `auth.test did not answer: ${doc.error ?? '?'}`];
  if (doc.enterprise_id) {
    return ['grid', 'auth.test returned an enterprise_id, so this workspace sits '
      + 'inside an organization'];
  }
  return ['not-grid', 'auth.test returned no enterprise_id, which strongly suggests '
    + 'the admin surface will not answer'];
}

export function capability(body) {
  const doc = body ?? {};
  if (doc.ok === true) {
    return ['available', 'the probe answered ok, so the admin surface exists here'];
  }
  const error = String(doc.error ?? '');
  if (error === 'feature_not_enabled') {
    return ['unavailable-plan', 'feature_not_enabled; the admin API is an Enterprise '
      + 'Grid surface and no scope or role changes that'];
  }
  if (error === 'not_an_admin') {
    return ['unknown-role', 'not_an_admin; the caller does not hold the organization '
      + 'role, so the plan was never tested'];
  }
  if (error === 'not_allowed_token_type') {
    return ['unknown-class', 'not_allowed_token_type; this is a bot token, so the plan '
      + 'was never tested'];
  }
  if (error === 'missing_scope') {
    return ['unknown-scope', 'missing_scope; the grant is short, so the plan was never '
      + 'tested'];
  }
  if (TRANSIENT.includes(error)) {
    return ['unknown-transient', `${error || 'no error'}; ask again rather than `
      + 'concluding anything'];
  }
  return ['unknown-other', `${error || 'no error'}; not a refusal that answers the `
    + 'plan question'];
}

// A false off is invisible and permanent. A false retry costs one request.
export function gate(state) {
  if (state === 'available') return ['on', 'measured available'];
  if (state === 'unavailable-plan') {
    return ['off', 'measured unavailable; this is the only state that gates a feature off'];
  }
  return ['retry', 'inconclusive; an unknown is not an absence and must not be stored '
    + 'as one'];
}

export function ownerOf(error) {
  return NOT_THIS_QUESTION[String(error ?? '').trim()] ?? null;
}

export function fallbackFor(method) {
  return FALLBACKS[String(method ?? '').trim()] ?? null;
}

export function matrixVerdict(rows) {
  const counts = {};
  for (const [, state] of rows ?? []) {
    const key = state.startsWith('unknown') ? 'unknown' : state;
    counts[key] = (counts[key] ?? 0) + 1;
  }
  if (!(rows ?? []).length) return ['empty', 'no installations were probed'];
  const unknown = counts.unknown ?? 0;
  const summary = `${counts.available ?? 0} available, `
    + `${counts['unavailable-plan'] ?? 0} unavailable, ${unknown} unknown`;
  if (unknown) return [`${unknown} unknown`, summary];
  return ['measured', summary];
}

function arg(args, name, fallback = '') {
  const i = args.indexOf(name);
  return i === -1 ? fallback : args[i + 1];
}

async function read(token, method, params = {}) {
  const url = new URL(API + method);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, String(v));
  const r = await fetch(url, { headers: { Authorization: `Bearer ${token}` } });
  try {
    return await r.json();
  } catch {
    return { ok: false, error: 'unparseable_body' };
  }
}

async function main() {
  const args = process.argv.slice(2);
  const installsPath = arg(args, '--installs');
  if (!installsPath) {
    console.error('pass --installs with a JSON list of installations');
    process.exitCode = 2;
    return;
  }
  const installs = JSON.parse(await readFile(installsPath, 'utf8')) ?? [];
  const methods = arg(args, '--methods', 'admin.conversations.search')
    .split(',').map((s) => s.trim()).filter(Boolean);

  const [readable, why] = planIsReadable();
  if (!readable) console.log(`plan       not readable     ${why}`);

  const rows = [];
  for (const raw of installs) {
    const row = raw ?? {};
    const tenant = String(row.id ?? '?');
    const token = row.user_token ?? '';
    if (!token) {
      console.warn(`tenant     ${tenant.padEnd(16)} no user token stored, so nothing `
        + 'can be measured');
      rows.push([tenant, 'unknown-class']);
      continue;
    }
    // eslint-disable-next-line no-await-in-loop
    const [signal, signalWhy] = gridSignal(await read(token, 'auth.test'));
    // eslint-disable-next-line no-await-in-loop
    const body = await read(token, PROBE[0], PROBE[1]);
    const [state, detail] = capability(body);
    const [flag, flagWhy] = gate(state);
    const line = `tenant     ${tenant.padEnd(16)} signal=${signal.padEnd(9)} `
      + `capability=${state.padEnd(18)} gate=${flag}`;
    if (state === 'available' || state === 'unavailable-plan') console.log(line);
    else console.warn(line);
    console.log(`           signal           ${signalWhy}`);
    console.log(`           note             ${detail}`);
    if (flag === 'retry') console.warn(`           gate             ${flagWhy}`);
    const pointer = ownerOf(body.error);
    if (pointer) console.log(`           see also         ${pointer}`);
    rows.push([tenant, state]);
  }

  for (const method of methods) {
    const pair = fallbackFor(method);
    if (!pair) {
      console.log(`fallback   ${method.padEnd(30)} none named here; check the method `
        + 'reference');
      continue;
    }
    console.log(`fallback   ${method} -> ${pair[0]}`);
    console.log(`           loses            ${pair[1]}`);
  }

  const [state, summary] = matrixVerdict(rows);
  if (state === 'measured') {
    console.log(`matrix     ${rows.length} tenant(s)      ${summary}`);
    console.log('verdict    measured         every tenant was gated on a measurement');
    return;
  }
  console.warn(`matrix     ${rows.length} tenant(s)      ${summary}`);
  console.warn(`verdict    ${state.padEnd(16)} re-probe before gating; an unknown is `
    + 'not an absence');
  console.warn('  repair: feature detect at install time and store the result, so the '
    + 'gate reads a measured fact rather than an assumption');
  console.warn('  repair: present a clear requires Enterprise Grid message for a '
    + 'measured off, and a pending state for a retry');
  console.warn('  repair: build the non-admin fallback where the narrower answer is '
    + 'worth having; the loss is printed above');
  process.exitCode = 1;
}

// Only run when invoked directly, so importing this module in the tests does
// not execute main().
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from slack_admin_capability import (
    capability, fallback_for, gate, grid_signal, matrix_verdict, owner_of,
    plan_is_readable,
)


def test_the_plan_is_not_readable_and_the_code_says_so():
    readable, why = plan_is_readable()
    assert readable is False
    assert "team.info" in why


def test_an_ok_probe_is_available():
    assert capability({"ok": True, "teams": []})[0] == "available"


def test_feature_not_enabled_is_the_only_state_that_gates_a_feature_off():
    state, why = capability({"ok": False, "error": "feature_not_enabled"})
    assert state == "unavailable-plan"
    assert gate(state)[0] == "off"
    assert "Enterprise Grid surface" in why


def test_a_role_refusal_never_becomes_an_absence():
    state, _why = capability({"ok": False, "error": "not_an_admin"})
    assert state == "unknown-role"
    assert gate(state)[0] == "retry"


def test_a_class_refusal_never_becomes_an_absence():
    assert gate(capability({"ok": False, "error": "not_allowed_token_type"})[0])[0] \
        == "retry"


def test_a_scope_refusal_never_becomes_an_absence():
    assert gate(capability({"ok": False, "error": "missing_scope"})[0])[0] == "retry"


def test_a_transient_refusal_is_asked_again():
    assert capability({"ok": False, "error": "ratelimited"})[0] == "unknown-transient"
    assert gate("unknown-transient")[0] == "retry"


def test_an_unrecognised_refusal_is_unknown_rather_than_unavailable():
    state, why = capability({"ok": False, "error": "org_login_required"})
    assert state == "unknown-other"
    assert "org_login_required" in why


def test_every_unknown_state_gates_to_retry():
    for state in ("unknown-role", "unknown-class", "unknown-scope",
                  "unknown-transient", "unknown-other"):
        assert gate(state)[0] == "retry"


def test_the_grid_signal_is_a_signal_and_is_labelled_one():
    assert grid_signal({"ok": True, "enterprise_id": "E1"})[0] == "grid"
    state, why = grid_signal({"ok": True, "team_id": "T1"})
    assert state == "not-grid"
    assert "suggests" in why


def test_an_unanswered_auth_test_leaves_the_signal_unknown():
    assert grid_signal({"ok": False, "error": "invalid_auth"})[0] == "unknown"


def test_refusals_that_belong_elsewhere_are_handed_over():
    assert owner_of("not_an_admin") == "/slack/not-an-admin/"
    assert owner_of("feature_not_enabled") is None


def test_each_fallback_names_the_method_and_the_loss():
    replacement, loses = fallback_for("admin.conversations.search")
    assert replacement == "conversations.list"
    assert "one workspace" in loses
    assert fallback_for("admin.somethingNew.list") is None


def test_the_matrix_reports_unknowns_as_the_finding():
    state, summary = matrix_verdict([("a", "available"), ("b", "unavailable-plan"),
                                     ("c", "unknown-role")])
    assert state == "1 unknown"
    assert summary == "1 available, 1 unavailable, 1 unknown"


def test_a_fully_measured_matrix_says_measured():
    assert matrix_verdict([("a", "available"), ("b", "unavailable-plan")])[0] == "measured"


def test_an_empty_matrix_is_not_a_measurement():
    assert matrix_verdict([])[0] == "empty"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  capability, fallbackFor, gate, gridSignal, matrixVerdict, ownerOf, planIsReadable,
} from './slack-admin-capability.mjs';

test('the plan is not readable and the code says so', () => {
  const [readable, why] = planIsReadable();
  assert.equal(readable, false);
  assert.match(why, /team\.info/);
});

test('an ok probe is available', () => {
  assert.equal(capability({ ok: true, teams: [] })[0], 'available');
});

test('feature_not_enabled is the only state that gates a feature off', () => {
  const [state, why] = capability({ ok: false, error: 'feature_not_enabled' });
  assert.equal(state, 'unavailable-plan');
  assert.equal(gate(state)[0], 'off');
  assert.match(why, /Enterprise Grid surface/);
});

test('a role refusal never becomes an absence', () => {
  const [state] = capability({ ok: false, error: 'not_an_admin' });
  assert.equal(state, 'unknown-role');
  assert.equal(gate(state)[0], 'retry');
});

test('a class refusal never becomes an absence', () => {
  assert.equal(gate(capability({ ok: false, error: 'not_allowed_token_type' })[0])[0],
    'retry');
});

test('a scope refusal never becomes an absence', () => {
  assert.equal(gate(capability({ ok: false, error: 'missing_scope' })[0])[0], 'retry');
});

test('a transient refusal is asked again', () => {
  assert.equal(capability({ ok: false, error: 'ratelimited' })[0], 'unknown-transient');
  assert.equal(gate('unknown-transient')[0], 'retry');
});

test('an unrecognised refusal is unknown rather than unavailable', () => {
  const [state, why] = capability({ ok: false, error: 'org_login_required' });
  assert.equal(state, 'unknown-other');
  assert.match(why, /org_login_required/);
});

test('every unknown state gates to retry', () => {
  for (const state of ['unknown-role', 'unknown-class', 'unknown-scope',
    'unknown-transient', 'unknown-other']) {
    assert.equal(gate(state)[0], 'retry');
  }
});

test('the grid signal is a signal and is labelled one', () => {
  assert.equal(gridSignal({ ok: true, enterprise_id: 'E1' })[0], 'grid');
  const [state, why] = gridSignal({ ok: true, team_id: 'T1' });
  assert.equal(state, 'not-grid');
  assert.match(why, /suggests/);
});

test('an unanswered auth.test leaves the signal unknown', () => {
  assert.equal(gridSignal({ ok: false, error: 'invalid_auth' })[0], 'unknown');
});

test('refusals that belong elsewhere are handed over', () => {
  assert.equal(ownerOf('not_an_admin'), '/slack/not-an-admin/');
  assert.equal(ownerOf('feature_not_enabled'), null);
});

test('each fallback names the method and the loss', () => {
  const [replacement, loses] = fallbackFor('admin.conversations.search');
  assert.equal(replacement, 'conversations.list');
  assert.match(loses, /one workspace/);
  assert.equal(fallbackFor('admin.somethingNew.list'), null);
});

test('the matrix reports unknowns as the finding', () => {
  const [state, summary] = matrixVerdict([['a', 'available'], ['b', 'unavailable-plan'],
    ['c', 'unknown-role']]);
  assert.equal(state, '1 unknown');
  assert.equal(summary, '1 available, 1 unavailable, 1 unknown');
});

test('a fully measured matrix says measured', () => {
  assert.equal(matrixVerdict([['a', 'available'], ['b', 'unavailable-plan']])[0],
    'measured');
});

test('an empty matrix is not a measurement', () => {
  assert.equal(matrixVerdict([])[0], 'empty');
});