the installation token was narrowed below what the job needs

"""Say whether an installation access token was narrowed below what a job needs.

Read only. One paginated GET for what the token reaches, one GET per
repository the job names. The token endpoint that would mint a wider token is
a write and is not called here: the script reads the token you already hold
and prints the mint request you should be making instead.

An installation access token can be minted for fewer repositories and fewer
permissions than the installation holds, by naming them in the mint request
body. The result is a 404 on a repository the App is plainly installed on,
from one code path, while every other path using the same App works.

There is a blind spot and the report states it. A token cannot report its own
permission map; the mint response echoed it back to your own code and no read
recovers it afterwards. Pass that saved response with --grant to make the
permission half exact, or the script reports it as unseen rather than passing.

Environment:

    GITHUB_INSTALLATION_TOKEN   the token the failing job holds
"""
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("github_token_reach")

API = "https://api.github.com"
UA = "github-token-reach/1.0"

RANK = {"none": 0, "read": 1, "write": 2, "admin": 3}


def rank(level):
    """A permission level as a comparable integer. Pure."""
    return RANK.get(str(level or "none").strip().lower(), 0)


def parse_needs(spec):
    """Turn contents:read,issues:write into a map. Pure.

    A bare name with no colon is taken as read, which is the level somebody
    means when they write down a permission without thinking about it.
    """
    out = {}
    for chunk in str(spec or "").split(","):
        item = chunk.strip()
        if not item:
            continue
        name, _, level = item.partition(":")
        name = name.strip().lower()
        if name:
            out[name] = (level.strip().lower() or "read")
    return out


def parse_grant(body):
    """Read a saved mint response into the facts it carries. Pure.

    Only three fields matter and any of them can be absent. permissions is
    None rather than an empty map when it is missing, because "this token was
    granted nothing" and "we did not see the grant" are opposite findings.
    """
    body = body if isinstance(body, dict) else {}
    permissions = body.get("permissions")
    repos = body.get("repositories")
    names = []
    if isinstance(repos, list):
        for repo in repos:
            if isinstance(repo, dict) and repo.get("full_name"):
                names.append(str(repo["full_name"]))
            elif isinstance(repo, str):
                names.append(repo)
    return {"permissions": permissions if isinstance(permissions, dict) else None,
            "repository_selection": body.get("repository_selection"),
            "repositories": names}


def repo_gap(reachable, needed):
    """Needed repositories the token cannot reach. Pure.

    Case-insensitive, because GitHub treats owner/name that way and the
    configuration file that lists them does not.
    """
    have = {str(r).strip().lower() for r in (reachable or [])}
    return [r for r in (needed or []) if str(r).strip().lower() not in have]


def permission_shortfall(granted, needed):
    """Needed permissions the token holds at a lower level. Pure.

    Returns None when the grant was never seen, which the caller reports as a
    blind spot rather than as a pass.
    """
    if granted is None:
        return None
    out = []
    for name, wanted in sorted((needed or {}).items()):
        have = (granted or {}).get(name)
        if rank(have) < rank(wanted):
            out.append((name, str(wanted), str(have) if have else "absent"))
    return out


def verdict(alive, missing_repos, shortfall, selection):
    """Turn reach, grant and need into a finding. Pure.

    Order matters: an unreachable repository is reported before a permission
    shortfall, because a 404 is the symptom people arrive with and fixing the
    permission first would leave them with the same 404.
    """
    if not alive:
        return ("token-not-alive",
                "GET /installation/repositories did not return 200, so this "
                "is not a working installation access token and the "
                "narrowing question does not arise yet.")
    if missing_repos:
        return ("repos-out-of-reach",
                "%s not in this token's repository set, so every call about "
                "them answers 404 whatever the App holds. Widen the "
                "repository list in the mint request."
                % ", ".join(missing_repos))
    if shortfall is None:
        return ("narrowing-not-visible",
                "every repository the job needs is reachable. The permission "
                "half cannot be checked: a token does not report its own "
                "permission map, and no saved mint response was supplied.")
    if shortfall:
        return ("permissions-below-need",
                "%s. The mint request asked for less than the job uses, which "
                "fails as 403 rather than as 404."
                % "; ".join("%s is %s, the job needs %s" % (n, h, w)
                            for n, w, h in shortfall))
    if str(selection or "").strip().lower() == "selected":
        return ("narrowed-but-sufficient",
                "this token is narrowed to a repository subset and the subset "
                "still covers the job. Nothing to change.")
    return ("reach-covers-the-job",
            "this token reaches every repository the job needs and holds "
            "every permission at the level it asked for.")


def repair(state, missing_repos, shortfall):
    """The change to make, in the mint request rather than in the App. Pure."""
    if state == "repos-out-of-reach":
        return ("add %s to the repository list in the token request this job "
                "makes. If the installation already covers them, the App does "
                "not change at all." % ", ".join(missing_repos))
    if state == "permissions-below-need":
        return ("raise %s in the permission map of the token request. If the "
                "installation does not hold it either, that is an App "
                "permission problem instead."
                % ", ".join("%s to %s" % (n, w) for n, w, _ in shortfall))
    if state == "narrowing-not-visible":
        return ("keep the mint response your code already receives, with the "
                "token value stripped, and pass it back in. It is the only "
                "place the granted permission map is ever visible.")
    return "nothing. This token is not the constraint."


def get(session, path):
    """One GET. Returns (status, json-or-None)."""
    url = API + path if path.startswith("/") else path
    r = session.get(url, timeout=30)
    try:
        body = r.json()
    except ValueError:
        body = None
    return r.status_code, body


def reachable_repositories(session, pages=10):
    """Every repository this token reaches. Returns (alive, names, selection)."""
    names, selection, alive = [], None, False
    for page in range(1, pages + 1):
        status, body = get(session,
                           "/installation/repositories?per_page=100&page=%d" % page)
        if status != 200 or not isinstance(body, dict):
            if page == 1:
                log.error("GET /installation/repositories returned %d; only an "
                          "installation access token can answer it", status)
            break
        alive = True
        selection = body.get("repository_selection", selection)
        rows = body.get("repositories") or []
        names.extend(str(r.get("full_name")) for r in rows
                     if isinstance(r, dict) and r.get("full_name"))
        if len(rows) < 100:
            break
    return alive, names, selection


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--repo", action="append", default=[],
                    help="a repository the job needs, as owner/name; repeatable")
    ap.add_argument("--needs", default="",
                    help="permissions the job uses, as contents:read,issues:write")
    ap.add_argument("--grant", default=None,
                    help="path to the saved mint response body, token stripped")
    args = ap.parse_args()

    token = os.environ.get("GITHUB_INSTALLATION_TOKEN") or \
        os.environ.get("GITHUB_TOKEN")
    if not token:
        log.error("set GITHUB_INSTALLATION_TOKEN to the token the failing job "
                  "holds. The narrowing is a property of that token and of no "
                  "other credential")
        return 2

    session = requests.Session()
    session.headers.update({
        "Authorization": "Bearer " + token,
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
        "User-Agent": UA,
    })

    alive, reach, selection = reachable_repositories(session)
    if alive:
        log.info("token reaches %d repository(ies), repository_selection=%s",
                 len(reach), selection or "unreported")

    # Confirming per repository rather than trusting the list, which can be
    # truncated by a pagination cap the caller did not notice.
    confirmed = []
    for name in args.repo:
        if not alive:
            break
        status, _ = get(session, "/repos/%s" % name.strip())
        log.info("GET /repos/%s returned %d", name.strip(), status)
        if status == 200:
            confirmed.append(name.strip())
    reach_all = sorted({*reach, *confirmed})

    grant = {"permissions": None, "repository_selection": selection,
             "repositories": []}
    if args.grant:
        try:
            with open(args.grant, encoding="utf-8") as fh:
                grant = parse_grant(json.load(fh))
        except (OSError, ValueError) as exc:
            log.error("could not read the saved mint response: %s", exc)
        else:
            selection = grant.get("repository_selection") or selection

    missing = repo_gap(reach_all, args.repo)
    shortfall = permission_shortfall(grant["permissions"], parse_needs(args.needs))
    state, detail = verdict(alive, missing, shortfall, selection)
    log.info("%s: %s", state, detail)
    log.info("repair: %s", repair(state, missing, shortfall))

    print(json.dumps({"reachable": reach_all, "repository_selection": selection,
                      "needed_repositories": args.repo,
                      "missing_repositories": missing,
                      "permission_shortfall": shortfall,
                      "state": state}, indent=2, default=str))
    return 1 if state in ("repos-out-of-reach", "permissions-below-need") else 0


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

/**
 * Say whether an installation access token was narrowed below what a job needs.
 *
 * Read only. One paginated GET for what the token reaches, one GET per
 * repository the job names. The token endpoint that would mint a wider token
 * is a write and is not called here: the script reads the token you already
 * hold and prints the mint request you should be making instead.
 *
 * A token cannot report its own permission map. The mint response echoed it
 * back to your own code and no read recovers it afterwards, so pass that saved
 * response as the third argument to make the permission half exact.
 *
 * Environment:
 *   GITHUB_INSTALLATION_TOKEN   the token the failing job holds
 */
import { readFile } from 'node:fs/promises';

const API = 'https://api.github.com';
const UA = 'github-token-reach/1.0';

export const RANK = { none: 0, read: 1, write: 2, admin: 3 };

/** A permission level as a comparable integer. Pure. */
export function rank(level) {
  const key = String(level ?? 'none').trim().toLowerCase();
  return Object.prototype.hasOwnProperty.call(RANK, key) ? RANK[key] : 0;
}

/** Turn contents:read,issues:write into a map. Pure. Bare names mean read. */
export function parseNeeds(spec) {
  const out = {};
  for (const chunk of String(spec ?? '').split(',')) {
    const item = chunk.trim();
    if (!item) continue;
    const at = item.indexOf(':');
    const name = (at === -1 ? item : item.slice(0, at)).trim().toLowerCase();
    const level = (at === -1 ? '' : item.slice(at + 1)).trim().toLowerCase();
    if (name) out[name] = level || 'read';
  }
  return out;
}

/** Read a saved mint response into the facts it carries. Pure. */
export function parseGrant(body) {
  const row = body && typeof body === 'object' ? body : {};
  const permissions = row.permissions && typeof row.permissions === 'object'
    ? row.permissions : null;
  const names = [];
  if (Array.isArray(row.repositories)) {
    for (const repo of row.repositories) {
      if (repo && typeof repo === 'object' && repo.full_name) names.push(String(repo.full_name));
      else if (typeof repo === 'string') names.push(repo);
    }
  }
  return {
    permissions,
    repository_selection: row.repository_selection ?? null,
    repositories: names,
  };
}

/** Needed repositories the token cannot reach. Pure. Case-insensitive. */
export function repoGap(reachable, needed) {
  const have = new Set((reachable ?? []).map((r) => String(r).trim().toLowerCase()));
  return (needed ?? []).filter((r) => !have.has(String(r).trim().toLowerCase()));
}

/**
 * Needed permissions the token holds at a lower level. Pure.
 * null when the grant was never seen, which is a blind spot and not a pass.
 */
export function permissionShortfall(granted, needed) {
  if (granted === null || granted === undefined) return null;
  const out = [];
  for (const name of Object.keys(needed ?? {}).sort()) {
    const wanted = needed[name];
    const have = granted[name];
    if (rank(have) < rank(wanted)) {
      out.push([name, String(wanted), have ? String(have) : 'absent']);
    }
  }
  return out;
}

/** Turn reach, grant and need into a finding. Pure. */
export function verdict(alive, missingRepos, shortfall, selection) {
  if (!alive) {
    return ['token-not-alive',
      'GET /installation/repositories did not return 200, so this is not a ' +
      'working installation access token and the narrowing question does not ' +
      'arise yet.'];
  }
  if (missingRepos && missingRepos.length) {
    return ['repos-out-of-reach',
      `${missingRepos.join(', ')} not in this token's repository set, so ` +
      'every call about them answers 404 whatever the App holds. Widen the ' +
      'repository list in the mint request.'];
  }
  if (shortfall === null || shortfall === undefined) {
    return ['narrowing-not-visible',
      'every repository the job needs is reachable. The permission half ' +
      'cannot be checked: a token does not report its own permission map, ' +
      'and no saved mint response was supplied.'];
  }
  if (shortfall.length) {
    return ['permissions-below-need',
      `${shortfall.map(([n, w, h]) => `${n} is ${h}, the job needs ${w}`).join('; ')}. ` +
      'The mint request asked for less than the job uses, which fails as 403 ' +
      'rather than as 404.'];
  }
  if (String(selection ?? '').trim().toLowerCase() === 'selected') {
    return ['narrowed-but-sufficient',
      'this token is narrowed to a repository subset and the subset still ' +
      'covers the job. Nothing to change.'];
  }
  return ['reach-covers-the-job',
    'this token reaches every repository the job needs and holds every ' +
    'permission at the level it asked for.'];
}

/** The change to make, in the mint request rather than in the App. Pure. */
export function repair(state, missingRepos, shortfall) {
  if (state === 'repos-out-of-reach') {
    return `add ${(missingRepos ?? []).join(', ')} to the repository list in ` +
      'the token request this job makes. If the installation already covers ' +
      'them, the App does not change at all.';
  }
  if (state === 'permissions-below-need') {
    return `raise ${(shortfall ?? []).map(([n, w]) => `${n} to ${w}`).join(', ')} ` +
      'in the permission map of the token request. If the installation does ' +
      'not hold it either, that is an App permission problem instead.';
  }
  if (state === 'narrowing-not-visible') {
    return 'keep the mint response your code already receives, with the token ' +
      'value stripped, and pass it back in. It is the only place the granted ' +
      'permission map is ever visible.';
  }
  return 'nothing. This token is not the constraint.';
}

async function get(token, path) {
  const url = path.startsWith('/') ? API + path : path;
  const res = await fetch(url, {
    headers: {
      Authorization: `Bearer ${token}`,
      Accept: 'application/vnd.github+json',
      'X-GitHub-Api-Version': '2022-11-28',
      'User-Agent': UA,
    },
  });
  let body = null;
  try { body = await res.json(); } catch { body = null; }
  return { status: res.status, body };
}

async function reachableRepositories(token, pages = 10) {
  const names = [];
  let selection = null;
  let alive = false;
  for (let page = 1; page <= pages; page += 1) {
    const { status, body } = await get(token,
      `/installation/repositories?per_page=100&page=${page}`);
    if (status !== 200 || !body || typeof body !== 'object') {
      if (page === 1) {
        console.error(`GET /installation/repositories returned ${status}; ` +
          'only an installation access token can answer it');
      }
      break;
    }
    alive = true;
    selection = body.repository_selection ?? selection;
    const rows = body.repositories ?? [];
    for (const r of rows) if (r && r.full_name) names.push(String(r.full_name));
    if (rows.length < 100) break;
  }
  return { alive, names, selection };
}

async function main() {
  const token = process.env.GITHUB_INSTALLATION_TOKEN ?? process.env.GITHUB_TOKEN;
  if (!token) {
    console.error('set GITHUB_INSTALLATION_TOKEN to the token the failing job ' +
      'holds. The narrowing is a property of that token and of no other credential');
    process.exitCode = 2;
    return;
  }
  const needed = (process.argv[2] ?? '').split(',').map((s) => s.trim()).filter(Boolean);
  const needs = parseNeeds(process.argv[3] ?? '');
  const grantPath = process.argv[4] ?? null;

  const { alive, names, selection: seen } = await reachableRepositories(token);
  let selection = seen;
  if (alive) {
    console.log(`token reaches ${names.length} repository(ies), ` +
      `repository_selection=${selection ?? 'unreported'}`);
  }

  // Confirming per repository rather than trusting the list, which can be
  // truncated by a pagination cap the caller did not notice.
  const confirmed = [];
  for (const name of alive ? needed : []) {
    const { status } = await get(token, `/repos/${name}`);
    console.log(`GET /repos/${name} returned ${status}`);
    if (status === 200) confirmed.push(name);
  }
  const reachAll = [...new Set([...names, ...confirmed])].sort();

  let grant = { permissions: null, repository_selection: selection, repositories: [] };
  if (grantPath) {
    try {
      grant = parseGrant(JSON.parse(await readFile(grantPath, 'utf8')));
      selection = grant.repository_selection ?? selection;
    } catch (err) {
      console.error(`could not read the saved mint response: ${err.message}`);
    }
  }

  const missing = repoGap(reachAll, needed);
  const shortfall = permissionShortfall(grant.permissions, needs);
  const [state, detail] = verdict(alive, missing, shortfall, selection);
  console.log(`${state}: ${detail}`);
  console.log(`repair: ${repair(state, missing, shortfall)}`);

  console.log(JSON.stringify({
    reachable: reachAll,
    repository_selection: selection,
    needed_repositories: needed,
    missing_repositories: missing,
    permission_shortfall: shortfall,
    state,
  }, null, 2));
  process.exitCode = ['repos-out-of-reach', 'permissions-below-need'].includes(state) ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not execute main(), fail on the missing token and set an exit code that
// fails the suite even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from github_token_reach import (
    parse_grant, parse_needs, permission_shortfall, rank, repair, repo_gap,
    verdict,
)

MINT = {"expires_at": "2026-08-30T13:00:00Z",
        "permissions": {"contents": "read", "metadata": "read"},
        "repository_selection": "selected",
        "repositories": [{"full_name": "acme/api"}]}


def test_a_bare_permission_name_means_read():
    assert parse_needs("contents, issues:write") == {"contents": "read",
                                                     "issues": "write"}
    assert parse_needs("") == {}
    assert parse_needs("  ,  ") == {}


def test_levels_are_ranked_rather_than_compared_as_strings():
    assert rank("write") > rank("read") > rank(None)
    assert rank("nonsense") == 0


def test_a_mint_response_is_read_for_the_three_fields_that_matter():
    grant = parse_grant(MINT)
    assert grant["permissions"] == {"contents": "read", "metadata": "read"}
    assert grant["repository_selection"] == "selected"
    assert grant["repositories"] == ["acme/api"]


def test_a_mint_response_with_no_permissions_block_is_unseen_not_empty():
    assert parse_grant({"repository_selection": "all"})["permissions"] is None
    assert parse_grant(None)["permissions"] is None


def test_repository_names_match_regardless_of_case():
    assert repo_gap(["acme/API"], ["Acme/api"]) == []
    assert repo_gap(["acme/api"], ["acme/api", "acme/docs"]) == ["acme/docs"]


def test_an_unseen_grant_is_not_a_permission_pass():
    assert permission_shortfall(None, {"issues": "write"}) is None
    assert permission_shortfall({}, {"issues": "write"}) == [("issues", "write", "absent")]


def test_read_where_write_is_needed_is_a_shortfall():
    assert permission_shortfall({"issues": "read"}, {"issues": "write"}) == \
        [("issues", "write", "read")]
    assert permission_shortfall({"issues": "write"}, {"issues": "read"}) == []


def test_an_unreachable_repository_is_reported_before_a_permission_shortfall():
    state, detail = verdict(True, ["acme/docs"], [("issues", "write", "read")],
                            "selected")
    assert state == "repos-out-of-reach"
    assert "acme/docs" in detail


def test_a_permission_shortfall_names_both_levels():
    state, detail = verdict(True, [], [("issues", "write", "read")], "all")
    assert state == "permissions-below-need"
    assert "issues is read, the job needs write" in detail


def test_an_unseen_grant_gets_its_own_state_rather_than_a_clean_bill():
    state, detail = verdict(True, [], None, "all")
    assert state == "narrowing-not-visible"
    assert "does not report its own permission map" in detail


def test_a_narrowed_token_that_still_covers_the_job_is_not_a_fault():
    assert verdict(True, [], [], "selected")[0] == "narrowed-but-sufficient"


def test_a_wide_token_that_covers_the_job_is_clean():
    assert verdict(True, [], [], "all")[0] == "reach-covers-the-job"


def test_a_dead_token_is_never_reported_as_a_narrowing():
    state, detail = verdict(False, ["acme/docs"], None, None)
    assert state == "token-not-alive"
    assert "does not arise yet" in detail


def test_the_repair_points_at_the_mint_request_and_not_at_the_app():
    text = repair("repos-out-of-reach", ["acme/docs"], None)
    assert "token request" in text
    assert "the App does not change" in text
    assert "mint response" in repair("narrowing-not-visible", [], None)
    assert repair("reach-covers-the-job", [], []) == "nothing. This token is not the constraint."

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  parseGrant, parseNeeds, permissionShortfall, rank, repair, repoGap, verdict,
} from './github-token-reach.mjs';

const MINT = {
  expires_at: '2026-08-30T13:00:00Z',
  permissions: { contents: 'read', metadata: 'read' },
  repository_selection: 'selected',
  repositories: [{ full_name: 'acme/api' }],
};

test('a bare permission name means read', () => {
  assert.deepEqual(parseNeeds('contents, issues:write'),
    { contents: 'read', issues: 'write' });
  assert.deepEqual(parseNeeds(''), {});
  assert.deepEqual(parseNeeds('  ,  '), {});
});

test('levels are ranked rather than compared as strings', () => {
  assert.ok(rank('write') > rank('read'));
  assert.ok(rank('read') > rank(null));
  assert.equal(rank('nonsense'), 0);
});

test('a mint response is read for the three fields that matter', () => {
  const grant = parseGrant(MINT);
  assert.deepEqual(grant.permissions, { contents: 'read', metadata: 'read' });
  assert.equal(grant.repository_selection, 'selected');
  assert.deepEqual(grant.repositories, ['acme/api']);
});

test('a mint response with no permissions block is unseen not empty', () => {
  assert.equal(parseGrant({ repository_selection: 'all' }).permissions, null);
  assert.equal(parseGrant(null).permissions, null);
});

test('repository names match regardless of case', () => {
  assert.deepEqual(repoGap(['acme/API'], ['Acme/api']), []);
  assert.deepEqual(repoGap(['acme/api'], ['acme/api', 'acme/docs']), ['acme/docs']);
});

test('an unseen grant is not a permission pass', () => {
  assert.equal(permissionShortfall(null, { issues: 'write' }), null);
  assert.deepEqual(permissionShortfall({}, { issues: 'write' }),
    [['issues', 'write', 'absent']]);
});

test('read where write is needed is a shortfall', () => {
  assert.deepEqual(permissionShortfall({ issues: 'read' }, { issues: 'write' }),
    [['issues', 'write', 'read']]);
  assert.deepEqual(permissionShortfall({ issues: 'write' }, { issues: 'read' }), []);
});

test('an unreachable repository is reported before a permission shortfall', () => {
  const [state, detail] = verdict(true, ['acme/docs'], [['issues', 'write', 'read']], 'selected');
  assert.equal(state, 'repos-out-of-reach');
  assert.match(detail, /acme\/docs/);
});

test('a permission shortfall names both levels', () => {
  const [state, detail] = verdict(true, [], [['issues', 'write', 'read']], 'all');
  assert.equal(state, 'permissions-below-need');
  assert.match(detail, /issues is read, the job needs write/);
});

test('an unseen grant gets its own state rather than a clean bill', () => {
  const [state, detail] = verdict(true, [], null, 'all');
  assert.equal(state, 'narrowing-not-visible');
  assert.match(detail, /does not report its own permission map/);
});

test('a narrowed token that still covers the job is not a fault', () => {
  assert.equal(verdict(true, [], [], 'selected')[0], 'narrowed-but-sufficient');
});

test('a wide token that covers the job is clean', () => {
  assert.equal(verdict(true, [], [], 'all')[0], 'reach-covers-the-job');
});

test('a dead token is never reported as a narrowing', () => {
  const [state, detail] = verdict(false, ['acme/docs'], null, null);
  assert.equal(state, 'token-not-alive');
  assert.match(detail, /does not arise yet/);
});

test('the repair points at the mint request and not at the app', () => {
  const text = repair('repos-out-of-reach', ['acme/docs'], null);
  assert.match(text, /token request/);
  assert.match(text, /the App does not change/);
  assert.match(repair('narrowing-not-visible', [], null), /mint response/);
  assert.equal(repair('reach-covers-the-job', [], []),
    'nothing. This token is not the constraint.');
});