app_access_restricted: an admin blocked the app for a user

"""Explain app_access_restricted by naming the people it happens to.

Read only. Four GET methods are used: users.info to resolve the callers,
usergroups.list and usergroups.users.list to test the group shape, and the two
Enterprise Grid admin listings when a token that can read them is available.
Nothing is written, no app is installed, approved or requested, and no email
address is read or printed.

The input is a records file: the calls your application already made, each with
the user it acted for and what came back. That is deliberate. This error is a
per-user policy, so a single instance proves nothing and the two cohorts prove
everything.
"""
import argparse
import json
import logging
import os
import sys

import requests

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

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

# Four different policies wearing four different error names. Sorting first is
# what keeps a records file from producing a cohort that is really a mixture.
AXES = {
    "app_access_restricted": (
        "app-policy", False,
        "a policy on the app and the person, not on your token; retrying cannot "
        "succeed"),
    "accesslimited": (
        "network", False,
        "a policy on where the call came from. Every user fails from this address "
        "and none of them fails from an allowed one. A different note"),
    "missing_scope": (
        "token", False,
        "the token lacks a scope, and Slack names both the one needed and the ones "
        "provided. A different note"),
    "invalid_auth": (
        "token", False,
        "the credential itself is refused, for every user equally"),
    "not_in_channel": (
        "membership", False,
        "the app is not a member of the conversation. Nothing to do with policy"),
    "ratelimited": (
        "transient", True,
        "the only answer here worth retrying"),
}

# The profile flags that make a person a guest. Guests are the group most often
# left outside an app permission grant, and they are visible without any admin
# credential at all.
GUEST_FLAGS = ("is_restricted", "is_ultra_restricted")


def refusal_axis(error):
    """Which policy is this error about? Pure.

    Returns (axis, retryable, detail). Called before the cohort is built so
    that a mixed log does not become a confident, wrong pattern.
    """
    text = str(error or "").strip()
    if not text:
        return ("none", False, "no error was recorded on this call")
    if text in AXES:
        axis, retryable, detail = AXES[text]
        return (axis, retryable, "%s: %s" % (text, detail))
    return ("other", False, "%s is not one of the policy refusals; read it against "
                            "the method reference" % text)


def split_cohorts(records):
    """Sort call records into the users refused and the users served. Pure.

    Returns (refused, served), both sorted lists of user ids. A user who
    appears in both is counted as refused, because a policy that bites
    sometimes is still the policy that bites.
    """
    refused, served = set(), set()
    for rec in records or []:
        user = str((rec or {}).get("user") or "").strip()
        if not user:
            continue
        axis = refusal_axis((rec or {}).get("error"))[0]
        if axis == "app-policy":
            refused.add(user)
        elif (rec or {}).get("ok") is True or axis == "none":
            served.add(user)
    return (sorted(refused), sorted(served - refused))


def shared_attribute(refused_profiles, served_profiles):
    """What do the refused callers share that the served ones do not? Pure.

    Takes two dicts of user id to users.info profile. Returns (code, detail).

      no-evidence         nothing was refused.
      no-control          nothing succeeded, so any attribute the refused
                          callers share might just be what your callers are.
                          The app may be restricted outright.
      guest-accounts      every refused caller is a guest and no served one is.
      one-workspace       every refusal comes from one team_id that no served
                          caller belongs to. The Grid shape.
      single-user         one person, and no rule to infer from one person.
      no-common-attribute the cohorts overlap on every attribute checked, which
                          points at an explicit per-user grant.

    An attribute is only reported when every refused caller has it and no
    served caller does. Anything weaker is a coincidence with a good story.
    """
    refused = {k: (v or {}) for k, v in (refused_profiles or {}).items()}
    served = {k: (v or {}) for k, v in (served_profiles or {}).items()}
    if not refused:
        return ("no-evidence", "no call was refused with app_access_restricted")
    if not served:
        return ("no-control", "%d refused caller(s) and no successful call to compare "
                              "against; the app may be restricted for everyone rather "
                              "than for a subset" % len(refused))

    def is_guest(p):
        return any(bool(p.get(flag)) for flag in GUEST_FLAGS)

    if all(is_guest(p) for p in refused.values()) and \
            not any(is_guest(p) for p in served.values()):
        return ("guest-accounts", "every refused caller is a guest account and no "
                                  "served caller is; the app is restricted to full "
                                  "members")
    teams = {str(p.get("team_id") or "") for p in refused.values()}
    served_teams = {str(p.get("team_id") or "") for p in served.values()}
    if len(teams) == 1 and teams and "" not in teams and not (teams & served_teams):
        team = sorted(teams)[0]
        return ("one-workspace", "every refusal comes from %s and no served caller is "
                                 "on it; the app is not approved on that workspace"
                % team)
    if len(refused) == 1:
        return ("single-user", "one refused caller, %s, and one person is not a "
                               "pattern; collect more records before naming a rule"
                % sorted(refused)[0])
    return ("no-common-attribute", "%d refused caller(s) share no attribute the %d "
                                   "served callers lack; the grant is probably "
                                   "per-user" % (len(refused), len(served)))


def usergroup_overlap(refused, served, groups):
    """Which user group contains every refused caller and no served one? Pure.

    groups: [{"handle": ..., "users": [...]}, ...]. Returns a list of
    (handle, detail). A grant made to a group leaves no trace in a profile, so
    this is the only way to see the third shape of this policy.
    """
    refused_set, served_set = set(refused or []), set(served or [])
    out = []
    if not refused_set:
        return out
    for group in groups or []:
        members = set((group or {}).get("users") or [])
        if refused_set <= members and not (served_set & members):
            out.append((str((group or {}).get("handle") or "?"),
                        "contains all %d refused caller(s) and none of the served "
                        "ones" % len(refused_set)))
    return out


def approval_state(app_id, approved, restricted):
    """Read the app's status from the two Grid listings. Pure.

    approved / restricted are the lists those methods return, or None when the
    token could not read them. Returns (state, detail).
    """
    if approved is None and restricted is None:
        return ("unavailable", "set an Enterprise Grid user token with "
                               "admin.apps:read to read the app status per workspace")
    wanted = str(app_id or "").strip()

    def teams_for(rows):
        found = []
        for row in rows or []:
            app = ((row or {}).get("app") or {})
            if str(app.get("id") or "") == wanted:
                found.append(str((row or {}).get("scope") or
                                 (row or {}).get("team_id") or "the org"))
        return found

    blocked = teams_for(restricted)
    if blocked:
        return ("restricted", "this app is on the restricted list for %s"
                % ", ".join(blocked))
    allowed = teams_for(approved)
    if allowed:
        return ("approved", "this app is approved for %s, so the refusal is a "
                            "per-user grant rather than an approval" % ", ".join(allowed))
    return ("unlisted", "this app appears on neither list, so on a workspace that "
                        "requires approval it is unavailable to everyone there")


def get(session, method, params=None):
    """One GET against the Web API. Returns the parsed body."""
    r = session.get(API + method, params=params or {}, timeout=30)
    try:
        return r.json()
    except ValueError:
        return {"ok": False, "error": "unparseable_body"}


def resolve(session, user_ids):
    """users.info for each id. Only the flags this check compares are kept."""
    out = {}
    for user in user_ids:
        body = get(session, "users.info", {"user": user})
        if body.get("ok") is not True:
            log.warning("profile    %-14s unavailable    %s", user, body.get("error"))
            out[user] = {}
            continue
        info = body.get("user") or {}
        out[user] = {
            "name": info.get("real_name") or info.get("name") or "",
            "team_id": info.get("team_id") or "",
            "is_restricted": bool(info.get("is_restricted")),
            "is_ultra_restricted": bool(info.get("is_ultra_restricted")),
            "deleted": bool(info.get("deleted")),
        }
    return out


def load_groups(session):
    """usergroups.list, then usergroups.users.list for each. Reads only."""
    body = get(session, "usergroups.list")
    if body.get("ok") is not True:
        log.info("usergroup  unavailable    %s", body.get("error"))
        return []
    groups = []
    for group in body.get("usergroups") or []:
        members = get(session, "usergroups.users.list", {"usergroup": group.get("id")})
        groups.append({"handle": group.get("handle") or group.get("name") or "?",
                       "users": list(members.get("users") or [])})
    return groups


def admin_list(session, method, token):
    """One of the two Grid listings. Returns the rows, or None if unreadable."""
    r = session.get(API + method, headers={"Authorization": "Bearer " + token},
                    params={"limit": "200"}, timeout=30)
    try:
        body = r.json()
    except ValueError:
        return None
    if body.get("ok") is not True:
        log.info("approval   %-14s %s", method.split(".")[2], body.get("error"))
        return None
    return body.get("apps") or []


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--records", required=True,
                    help="json file of call records: user, ok and error per call")
    ap.add_argument("--app-id", default="", help="the app id to look for, A...")
    ap.add_argument("--token-env", default="SLACK_BOT_TOKEN",
                    help="environment variable holding a token with users:read")
    ap.add_argument("--admin-token-env", default="SLACK_ADMIN_TOKEN",
                    help="environment variable holding a Grid token with "
                         "admin.apps:read")
    args = ap.parse_args()

    records = json.loads(open(args.records, encoding="utf-8").read())
    counts = {}
    for rec in records:
        axis, _retryable, detail = refusal_axis((rec or {}).get("error"))
        counts.setdefault(axis, [0, detail])
        counts[axis][0] += 1
    for axis in sorted(counts):
        if axis == "none":
            continue
        count, detail = counts[axis]
        log.info("axis       %-14s %d record(s): %s", axis, count, detail)

    refused, served = split_cohorts(records)
    log.info("cohort     refused        %d user(s), served %d user(s)",
             len(refused), len(served))
    if not refused:
        log.info("verdict    clean          nothing in these records was refused by "
                 "an app policy")
        return 0

    token = os.environ.get(args.token_env)
    if not token:
        log.error("set %s to a token with users:read and usergroups:read",
                  args.token_env)
        return 2
    session = requests.Session()
    session.headers.update({"Authorization": "Bearer " + token})

    refused_profiles = resolve(session, refused)
    served_profiles = resolve(session, served)
    for user in refused:
        p = refused_profiles.get(user) or {}
        flags = ",".join(f for f in GUEST_FLAGS if p.get(f)) or "full member"
        log.warning("profile    %-14s %-18s %s", user, p.get("name") or "?", flags)

    code, detail = shared_attribute(refused_profiles, served_profiles)
    log.warning("shared     %-14s %s", code, detail)

    overlaps = usergroup_overlap(refused, served, load_groups(session))
    if overlaps:
        for handle, why in overlaps:
            log.warning("usergroup  %-14s %s", handle, why)
    else:
        log.info("usergroup  none           no single user group separates the two "
                 "cohorts")

    admin_token = os.environ.get(args.admin_token_env)
    approved = restricted = None
    if admin_token and args.app_id:
        approved = admin_list(session, "admin.apps.approved.list", admin_token)
        restricted = admin_list(session, "admin.apps.restricted.list", admin_token)
    state, why = approval_state(args.app_id, approved, restricted)
    (log.info if state in ("approved", "unavailable") else log.warning)(
        "approval   %-14s %s", state, why)

    log.warning("verdict    1 finding(s)")
    log.warning("  repair: ask an admin to permit these %d account(s), or the group "
                "they belong to, under Manage apps -> this app -> permissions",
                len(refused))
    log.warning("  repair: catch app_access_restricted by name, tell that user their "
                "admin has restricted the app, and do not retry")
    return 1


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

/**
 * Explain app_access_restricted by naming the people it happens to.
 *
 * Read only. Four GET methods are used: users.info, usergroups.list,
 * usergroups.users.list, and the two Enterprise Grid admin listings when a
 * token that can read them is available. No email address is read or printed.
 *
 * The input is a records file of calls your application already made, because
 * this error is a per-user policy: one instance proves nothing and two cohorts
 * prove everything.
 */

import { readFileSync } from 'node:fs';

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

// Four policies wearing four error names. Sorting first is what keeps a
// records file from producing a cohort that is really a mixture.
export const AXES = {
  app_access_restricted: ['app-policy', false,
    'a policy on the app and the person, not on your token; retrying cannot succeed'],
  accesslimited: ['network', false,
    'a policy on where the call came from. Every user fails from this address and '
    + 'none of them fails from an allowed one. A different note'],
  missing_scope: ['token', false,
    'the token lacks a scope, and Slack names both the one needed and the ones '
    + 'provided. A different note'],
  invalid_auth: ['token', false,
    'the credential itself is refused, for every user equally'],
  not_in_channel: ['membership', false,
    'the app is not a member of the conversation. Nothing to do with policy'],
  ratelimited: ['transient', true, 'the only answer here worth retrying'],
};

// The profile flags that make a person a guest, visible without any admin
// credential at all.
export const GUEST_FLAGS = ['is_restricted', 'is_ultra_restricted'];

/** Which policy is this error about? Pure. Returns [axis, retryable, detail]. */
export function refusalAxis(error) {
  const text = String(error ?? '').trim();
  if (!text) return ['none', false, 'no error was recorded on this call'];
  if (Object.prototype.hasOwnProperty.call(AXES, text)) {
    const [axis, retryable, detail] = AXES[text];
    return [axis, retryable, `${text}: ${detail}`];
  }
  return ['other', false,
    `${text} is not one of the policy refusals; read it against the method reference`];
}

/** Sort call records into the users refused and the users served. Pure. */
export function splitCohorts(records) {
  const refused = new Set();
  const served = new Set();
  for (const rec of records ?? []) {
    const user = String((rec ?? {}).user ?? '').trim();
    if (!user) continue;
    const axis = refusalAxis((rec ?? {}).error)[0];
    if (axis === 'app-policy') refused.add(user);
    else if ((rec ?? {}).ok === true || axis === 'none') served.add(user);
  }
  return [[...refused].sort(),
    [...served].filter((u) => !refused.has(u)).sort()];
}

/**
 * What do the refused callers share that the served ones do not? Pure.
 * Returns [code, detail]; no-evidence, no-control, guest-accounts,
 * one-workspace, single-user, no-common-attribute.
 */
export function sharedAttribute(refusedProfiles, servedProfiles) {
  const refused = Object.entries(refusedProfiles ?? {})
    .map(([k, v]) => [k, v ?? {}]);
  const served = Object.entries(servedProfiles ?? {}).map(([k, v]) => [k, v ?? {}]);
  if (!refused.length) {
    return ['no-evidence', 'no call was refused with app_access_restricted'];
  }
  if (!served.length) {
    return ['no-control', `${refused.length} refused caller(s) and no successful call `
      + 'to compare against; the app may be restricted for everyone rather than for a '
      + 'subset'];
  }
  const isGuest = (p) => GUEST_FLAGS.some((flag) => Boolean(p[flag]));
  if (refused.every(([, p]) => isGuest(p)) && !served.some(([, p]) => isGuest(p))) {
    return ['guest-accounts', 'every refused caller is a guest account and no served '
      + 'caller is; the app is restricted to full members'];
  }
  const teams = new Set(refused.map(([, p]) => String(p.team_id ?? '')));
  const servedTeams = new Set(served.map(([, p]) => String(p.team_id ?? '')));
  const team = [...teams][0];
  if (teams.size === 1 && team && !servedTeams.has(team)) {
    return ['one-workspace', `every refusal comes from ${team} and no served caller is `
      + 'on it; the app is not approved on that workspace'];
  }
  if (refused.length === 1) {
    return ['single-user', `one refused caller, ${refused[0][0]}, and one person is `
      + 'not a pattern; collect more records before naming a rule'];
  }
  return ['no-common-attribute', `${refused.length} refused caller(s) share no `
    + `attribute the ${served.length} served callers lack; the grant is probably `
    + 'per-user'];
}

/** Which user group holds every refused caller and no served one? Pure. */
export function usergroupOverlap(refused, served, groups) {
  const refusedSet = new Set(refused ?? []);
  const servedSet = new Set(served ?? []);
  const out = [];
  if (!refusedSet.size) return out;
  for (const group of groups ?? []) {
    const members = new Set((group ?? {}).users ?? []);
    const holdsAll = [...refusedSet].every((u) => members.has(u));
    const holdsNone = ![...servedSet].some((u) => members.has(u));
    if (holdsAll && holdsNone) {
      out.push([String((group ?? {}).handle ?? '?'),
        `contains all ${refusedSet.size} refused caller(s) and none of the served ones`]);
    }
  }
  return out;
}

/** Read the app status from the two Grid listings. Pure. */
export function approvalState(appId, approved, restricted) {
  if (approved === null && restricted === null) {
    return ['unavailable', 'set an Enterprise Grid user token with admin.apps:read to '
      + 'read the app status per workspace'];
  }
  const wanted = String(appId ?? '').trim();
  const teamsFor = (rows) => (rows ?? [])
    .filter((row) => String(((row ?? {}).app ?? {}).id ?? '') === wanted)
    .map((row) => String((row ?? {}).scope ?? (row ?? {}).team_id ?? 'the org'));
  const blocked = teamsFor(restricted);
  if (blocked.length) {
    return ['restricted', `this app is on the restricted list for ${blocked.join(', ')}`];
  }
  const allowed = teamsFor(approved);
  if (allowed.length) {
    return ['approved', `this app is approved for ${allowed.join(', ')}, so the `
      + 'refusal is a per-user grant rather than an approval'];
  }
  return ['unlisted', 'this app appears on neither list, so on a workspace that '
    + 'requires approval it is unavailable to everyone there'];
}

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

async function get(headers, method, params = {}) {
  const query = new URLSearchParams(params);
  const r = await fetch(`${API}${method}?${query}`, { headers });
  try {
    return await r.json();
  } catch {
    return { ok: false, error: 'unparseable_body' };
  }
}

async function resolve(headers, userIds) {
  const out = {};
  for (const user of userIds) {
    const body = await get(headers, 'users.info', { user });
    if (body.ok !== true) {
      console.warn(`profile    ${user.padEnd(14)} unavailable    ${body.error}`);
      out[user] = {};
      continue;
    }
    const info = body.user ?? {};
    out[user] = {
      name: info.real_name ?? info.name ?? '',
      team_id: info.team_id ?? '',
      is_restricted: Boolean(info.is_restricted),
      is_ultra_restricted: Boolean(info.is_ultra_restricted),
      deleted: Boolean(info.deleted),
    };
  }
  return out;
}

async function loadGroups(headers) {
  const body = await get(headers, 'usergroups.list');
  if (body.ok !== true) {
    console.log(`usergroup  unavailable    ${body.error}`);
    return [];
  }
  const groups = [];
  for (const group of body.usergroups ?? []) {
    const members = await get(headers, 'usergroups.users.list',
      { usergroup: group.id });
    groups.push({
      handle: group.handle ?? group.name ?? '?',
      users: [...(members.users ?? [])],
    });
  }
  return groups;
}

async function adminList(method, token) {
  const r = await fetch(`${API}${method}?limit=200`,
    { headers: { Authorization: `Bearer ${token}` } });
  let body;
  try {
    body = await r.json();
  } catch {
    return null;
  }
  if (body.ok !== true) {
    console.log(`approval   ${method.split('.')[2].padEnd(14)} ${body.error}`);
    return null;
  }
  return body.apps ?? [];
}

async function main() {
  const args = process.argv.slice(2);
  const records = JSON.parse(readFileSync(arg(args, '--records'), 'utf-8'));
  const appId = arg(args, '--app-id');

  const counts = new Map();
  for (const rec of records) {
    const [axis, , detail] = refusalAxis((rec ?? {}).error);
    if (!counts.has(axis)) counts.set(axis, [0, detail]);
    counts.get(axis)[0] += 1;
  }
  for (const axis of [...counts.keys()].sort()) {
    if (axis === 'none') continue;
    const [count, detail] = counts.get(axis);
    console.log(`axis       ${axis.padEnd(14)} ${count} record(s): ${detail}`);
  }

  const [refused, served] = splitCohorts(records);
  console.log(`cohort     refused        ${refused.length} user(s), served `
    + `${served.length} user(s)`);
  if (!refused.length) {
    console.log('verdict    clean          nothing in these records was refused by an '
      + 'app policy');
    return;
  }

  const tokenEnv = arg(args, '--token-env', 'SLACK_BOT_TOKEN');
  const token = process.env[tokenEnv];
  if (!token) {
    console.error(`set ${tokenEnv} to a token with users:read and usergroups:read`);
    process.exitCode = 2;
    return;
  }
  const headers = { Authorization: `Bearer ${token}` };

  const refusedProfiles = await resolve(headers, refused);
  const servedProfiles = await resolve(headers, served);
  for (const user of refused) {
    const p = refusedProfiles[user] ?? {};
    const flags = GUEST_FLAGS.filter((f) => p[f]).join(',') || 'full member';
    console.warn(`profile    ${user.padEnd(14)} ${String(p.name || '?').padEnd(18)} `
      + `${flags}`);
  }

  const [code, detail] = sharedAttribute(refusedProfiles, servedProfiles);
  console.warn(`shared     ${code.padEnd(14)} ${detail}`);

  const overlaps = usergroupOverlap(refused, served, await loadGroups(headers));
  if (overlaps.length) {
    for (const [handle, why] of overlaps) {
      console.warn(`usergroup  ${handle.padEnd(14)} ${why}`);
    }
  } else {
    console.log('usergroup  none           no single user group separates the two '
      + 'cohorts');
  }

  const adminToken = process.env[arg(args, '--admin-token-env', 'SLACK_ADMIN_TOKEN')];
  let approved = null;
  let restricted = null;
  if (adminToken && appId) {
    approved = await adminList('admin.apps.approved.list', adminToken);
    restricted = await adminList('admin.apps.restricted.list', adminToken);
  }
  const [state, why] = approvalState(appId, approved, restricted);
  console.log(`approval   ${state.padEnd(14)} ${why}`);

  console.warn('verdict    1 finding(s)');
  console.warn(`  repair: ask an admin to permit these ${refused.length} account(s), `
    + 'or the group they belong to, under Manage apps -> this app -> permissions');
  console.warn('  repair: catch app_access_restricted by name, tell that user their '
    + 'admin has restricted the app, and do not retry');
  process.exitCode = 1;
}

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

from slack_app_restriction import (
    approval_state, refusal_axis, shared_attribute, split_cohorts, usergroup_overlap,
)

GUEST = {"name": "Dana", "team_id": "T1", "is_ultra_restricted": True}
MEMBER = {"name": "Sam", "team_id": "T1"}
OTHER_TEAM = {"name": "Wei", "team_id": "T2"}


def test_the_app_policy_error_is_on_its_own_axis():
    axis, retryable, detail = refusal_axis("app_access_restricted")
    assert axis == "app-policy"
    assert retryable is False
    assert "not on your token" in detail


def test_a_network_policy_is_a_different_axis_and_says_so():
    axis, _retryable, detail = refusal_axis("accesslimited")
    assert axis == "network"
    assert "where the call came from" in detail


def test_a_scope_error_is_not_swept_into_the_cohort():
    assert refusal_axis("missing_scope")[0] == "token"
    assert refusal_axis("not_in_channel")[0] == "membership"


def test_only_ratelimited_is_worth_retrying():
    assert refusal_axis("ratelimited")[1] is True
    for err in ("app_access_restricted", "accesslimited", "missing_scope",
                "invalid_auth", "not_in_channel"):
        assert refusal_axis(err)[1] is False


def test_the_cohorts_come_only_from_the_app_policy_error():
    records = [{"user": "U1", "error": "app_access_restricted"},
               {"user": "U2", "error": "accesslimited"},
               {"user": "U3", "ok": True}]
    refused, served = split_cohorts(records)
    assert refused == ["U1"]
    assert served == ["U3"]


def test_a_user_refused_once_counts_as_refused_even_if_served_elsewhere():
    records = [{"user": "U1", "ok": True},
               {"user": "U1", "error": "app_access_restricted"}]
    refused, served = split_cohorts(records)
    assert refused == ["U1"]
    assert served == []


def test_guests_refused_and_members_served_names_the_rule():
    code, detail = shared_attribute({"U1": GUEST, "U2": GUEST}, {"U3": MEMBER})
    assert code == "guest-accounts"
    assert "full members" in detail


def test_a_served_guest_disproves_the_guest_rule():
    code, _detail = shared_attribute({"U1": GUEST, "U2": GUEST},
                                     {"U3": GUEST, "U4": MEMBER})
    assert code != "guest-accounts"


def test_one_workspace_is_the_grid_shape():
    code, detail = shared_attribute({"U1": OTHER_TEAM, "U2": OTHER_TEAM},
                                    {"U3": MEMBER})
    assert code == "one-workspace"
    assert "T2" in detail


def test_no_successful_call_means_there_is_no_control_group():
    code, detail = shared_attribute({"U1": GUEST}, {})
    assert code == "no-control"
    assert "restricted for everyone" in detail


def test_one_refused_user_is_not_a_pattern():
    assert shared_attribute({"U1": MEMBER}, {"U3": MEMBER})[0] == "single-user"


def test_cohorts_that_look_alike_report_a_per_user_grant():
    code, _detail = shared_attribute({"U1": MEMBER, "U2": MEMBER}, {"U3": MEMBER})
    assert code == "no-common-attribute"


def test_nothing_refused_is_no_evidence_rather_than_a_clean_bill():
    assert shared_attribute({}, {"U3": MEMBER})[0] == "no-evidence"


def test_a_group_holding_every_refused_caller_and_no_other_is_reported():
    groups = [{"handle": "support", "users": ["U1", "U2"]},
              {"handle": "everyone", "users": ["U1", "U2", "U3"]}]
    found = usergroup_overlap(["U1", "U2"], ["U3"], groups)
    assert [h for h, _w in found] == ["support"]


def test_a_group_with_no_refused_callers_is_not_reported():
    assert usergroup_overlap(["U1"], ["U3"], [{"handle": "x", "users": ["U9"]}]) == []


def test_the_restricted_listing_settles_it_when_it_is_available():
    state, detail = approval_state("A1", [], [{"app": {"id": "A1"}, "scope": "T2"}])
    assert state == "restricted"
    assert "T2" in detail


def test_an_approved_app_points_back_at_a_per_user_grant():
    state, detail = approval_state("A1", [{"app": {"id": "A1"}, "scope": "T1"}], [])
    assert state == "approved"
    assert "per-user grant" in detail


def test_an_app_on_neither_list_is_unlisted_not_approved():
    assert approval_state("A1", [], [])[0] == "unlisted"


def test_without_an_admin_token_the_check_says_so_rather_than_guessing():
    state, detail = approval_state("A1", None, None)
    assert state == "unavailable"
    assert "admin.apps:read" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  approvalState, refusalAxis, sharedAttribute, splitCohorts, usergroupOverlap,
} from './slack-app-restriction.mjs';

const GUEST = { name: 'Dana', team_id: 'T1', is_ultra_restricted: true };
const MEMBER = { name: 'Sam', team_id: 'T1' };
const OTHER_TEAM = { name: 'Wei', team_id: 'T2' };

test('the app policy error is on its own axis', () => {
  const [axis, retryable, detail] = refusalAxis('app_access_restricted');
  assert.equal(axis, 'app-policy');
  assert.equal(retryable, false);
  assert.match(detail, /not on your token/);
});

test('a network policy is a different axis and says so', () => {
  const [axis, , detail] = refusalAxis('accesslimited');
  assert.equal(axis, 'network');
  assert.match(detail, /where the call came from/);
});

test('a scope error is not swept into the cohort', () => {
  assert.equal(refusalAxis('missing_scope')[0], 'token');
  assert.equal(refusalAxis('not_in_channel')[0], 'membership');
});

test('only ratelimited is worth retrying', () => {
  assert.equal(refusalAxis('ratelimited')[1], true);
  for (const err of ['app_access_restricted', 'accesslimited', 'missing_scope',
    'invalid_auth', 'not_in_channel']) {
    assert.equal(refusalAxis(err)[1], false);
  }
});

test('the cohorts come only from the app policy error', () => {
  const records = [{ user: 'U1', error: 'app_access_restricted' },
    { user: 'U2', error: 'accesslimited' },
    { user: 'U3', ok: true }];
  const [refused, served] = splitCohorts(records);
  assert.deepEqual(refused, ['U1']);
  assert.deepEqual(served, ['U3']);
});

test('a user refused once counts as refused even if served elsewhere', () => {
  const records = [{ user: 'U1', ok: true },
    { user: 'U1', error: 'app_access_restricted' }];
  const [refused, served] = splitCohorts(records);
  assert.deepEqual(refused, ['U1']);
  assert.deepEqual(served, []);
});

test('guests refused and members served names the rule', () => {
  const [code, detail] = sharedAttribute({ U1: GUEST, U2: GUEST }, { U3: MEMBER });
  assert.equal(code, 'guest-accounts');
  assert.match(detail, /full members/);
});

test('a served guest disproves the guest rule', () => {
  const [code] = sharedAttribute({ U1: GUEST, U2: GUEST },
    { U3: GUEST, U4: MEMBER });
  assert.notEqual(code, 'guest-accounts');
});

test('one workspace is the grid shape', () => {
  const [code, detail] = sharedAttribute({ U1: OTHER_TEAM, U2: OTHER_TEAM },
    { U3: MEMBER });
  assert.equal(code, 'one-workspace');
  assert.match(detail, /T2/);
});

test('no successful call means there is no control group', () => {
  const [code, detail] = sharedAttribute({ U1: GUEST }, {});
  assert.equal(code, 'no-control');
  assert.match(detail, /restricted for everyone/);
});

test('one refused user is not a pattern', () => {
  assert.equal(sharedAttribute({ U1: MEMBER }, { U3: MEMBER })[0], 'single-user');
});

test('cohorts that look alike report a per user grant', () => {
  const [code] = sharedAttribute({ U1: MEMBER, U2: MEMBER }, { U3: MEMBER });
  assert.equal(code, 'no-common-attribute');
});

test('nothing refused is no evidence rather than a clean bill', () => {
  assert.equal(sharedAttribute({}, { U3: MEMBER })[0], 'no-evidence');
});

test('a group holding every refused caller and no other is reported', () => {
  const groups = [{ handle: 'support', users: ['U1', 'U2'] },
    { handle: 'everyone', users: ['U1', 'U2', 'U3'] }];
  const found = usergroupOverlap(['U1', 'U2'], ['U3'], groups);
  assert.deepEqual(found.map(([h]) => h), ['support']);
});

test('a group with no refused callers is not reported', () => {
  assert.deepEqual(usergroupOverlap(['U1'], ['U3'],
    [{ handle: 'x', users: ['U9'] }]), []);
});

test('the restricted listing settles it when it is available', () => {
  const [state, detail] = approvalState('A1', [],
    [{ app: { id: 'A1' }, scope: 'T2' }]);
  assert.equal(state, 'restricted');
  assert.match(detail, /T2/);
});

test('an approved app points back at a per user grant', () => {
  const [state, detail] = approvalState('A1', [{ app: { id: 'A1' }, scope: 'T1' }], []);
  assert.equal(state, 'approved');
  assert.match(detail, /per-user grant/);
});

test('an app on neither list is unlisted not approved', () => {
  assert.equal(approvalState('A1', [], [])[0], 'unlisted');
});

test('without an admin token the check says so rather than guessing', () => {
  const [state, detail] = approvalState('A1', null, null);
  assert.equal(state, 'unavailable');
  assert.match(detail, /admin\.apps:read/);
});