a new App permission that installers never accepted

"""Find GitHub App installations still living on an older permission grant.

Read only. Two GETs with the App's JWT: the App's own declaration and the list
of installations. Nothing is granted, accepted, upgraded or changed. Accepting
a permission upgrade is a human act performed by an account owner, so the
script prints who has to be asked and for what.

Editing an App's permissions does not apply the change to installations that
already exist. Each installation keeps the grant it accepted until an owner
accepts the new one, so the App declaration and the installation grant are two
different objects that drift apart the moment the App is edited.

Environment:

    GITHUB_APP_JWT   the JWT your own signing code produced
"""
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_permission_upgrade_lag")

API = "https://api.github.com"
UA = "github-permission-upgrade-lag/1.0"

# Permission levels are ordered, not a set. An installation that granted read
# where the App declares write is behind, and a check that only compares keys
# calls it current and then wonders why the writes keep failing.
RANK = {"none": 0, "read": 1, "write": 2, "admin": 3}


def rank(level):
    """A permission level as a comparable integer. Pure.

    Anything unrecognised sorts as none rather than raising, because a level
    this script has not heard of is not evidence that access was granted.
    """
    return RANK.get(str(level or "none").strip().lower(), 0)


def permission_gap(declared, granted):
    """Declared permissions an installation holds at a lower level. Pure.

    Returns [(permission, declared_level, granted_level), ...] sorted by name.
    granted_level is the literal string "absent" where the key is missing, so
    the report can say which of the two shapes it found.
    """
    out = []
    for name, wanted in sorted((declared 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 permission_surplus(declared, granted):
    """Permissions an installation holds beyond what the App declares. Pure."""
    out = []
    for name, have in sorted((granted or {}).items()):
        wanted = (declared or {}).get(name)
        if rank(have) > rank(wanted):
            out.append((name, str(wanted) if wanted else "not declared", str(have)))
    return out


def event_gap(declared_events, granted_events):
    """Declared events an installation has not accepted. Pure."""
    have = {str(e).strip().lower() for e in (granted_events or [])}
    return sorted({str(e).strip().lower() for e in (declared_events or [])} - have)


def classify(declared_permissions, declared_events, inst):
    """Sort one installation against the App declaration. Pure.

    Returns a row: account, id, state, and the three diffs. The state is
    upgrade-pending when anything declared is not held, grant-ahead when the
    installation holds more than the App declares and nothing less, and
    current when the two maps agree.
    """
    inst = inst if isinstance(inst, dict) else {}
    account = inst.get("account") or {}
    gaps = permission_gap(declared_permissions, inst.get("permissions"))
    extra = permission_surplus(declared_permissions, inst.get("permissions"))
    events = event_gap(declared_events, inst.get("events"))
    if gaps or events:
        state = "upgrade-pending"
    elif extra:
        state = "grant-ahead"
    else:
        state = "current"
    return {"installation_id": inst.get("id"),
            "account": account.get("login") if isinstance(account, dict) else None,
            "state": state, "permission_gap": gaps,
            "permission_surplus": extra, "event_gap": events}


def verdict(rows):
    """Turn the per-installation rows into one finding. Pure."""
    rows = rows or []
    if not rows:
        return ("no-installations",
                "this App has no installations, so there is nothing to be "
                "behind. Nothing here is evidence about permissions.")
    behind = [r for r in rows if r["state"] == "upgrade-pending"]
    if behind:
        return ("upgrades-pending",
                "%d of %d installation(s) are behind the App declaration. "
                "Their tokens carry the permission map they accepted, not the "
                "one the App settings page shows."
                % (len(behind), len(rows)))
    ahead = [r for r in rows if r["state"] == "grant-ahead"]
    if ahead:
        return ("grants-ahead",
                "%d of %d installation(s) hold more than the App declares, "
                "which happens after a permission is removed rather than "
                "added. Nothing is failing; the access is simply unused."
                % (len(ahead), len(rows)))
    return ("all-current",
            "all %d installation(s) have accepted what the App declares."
            % len(rows))


def cohorts(rows):
    """Group the laggards by exactly what they are missing. Pure.

    A hundred rows usually collapse to two or three, and the cohorts are the
    versions of the App people accepted. That is the pattern which looked
    random from inside a single 403.
    """
    out = {}
    for row in rows or []:
        if row["state"] != "upgrade-pending":
            continue
        key = ", ".join("%s %s (declared %s)" % (n, g, d)
                        for n, d, g in row["permission_gap"]) or "events only"
        out.setdefault(key, []).append(row["account"] or row["installation_id"])
    return {k: sorted(map(str, v)) for k, v in sorted(out.items())}


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 list_installations(session, pages=10):
    """Every installation this App can see, paginated. Read only."""
    out = []
    for page in range(1, pages + 1):
        status, body = get(session, "/app/installations?per_page=100&page=%d" % page)
        if status != 200 or not isinstance(body, list):
            if page == 1:
                log.error("GET /app/installations returned %d; this endpoint "
                          "wants the App JWT", status)
            break
        out.extend(body)
        if len(body) < 100:
            break
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--account", default=None,
                    help="report on one account login only")
    args = ap.parse_args()

    jwt = os.environ.get("GITHUB_APP_JWT")
    if not jwt:
        log.error("set GITHUB_APP_JWT to the JWT your own signing code "
                  "produced. Both reads here are App-level and neither one "
                  "accepts an installation token")
        return 2

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

    status, app = get(session, "/app")
    if status != 200 or not isinstance(app, dict):
        log.error("GET /app returned %d, so there is no declaration to "
                  "compare against", status)
        return 2
    declared_permissions = app.get("permissions") or {}
    declared_events = app.get("events") or []
    log.info("app declares %d permission(s) and %d event(s)",
             len(declared_permissions), len(declared_events))

    installations = list_installations(session)
    rows = [classify(declared_permissions, declared_events, i) for i in installations]
    if args.account:
        rows = [r for r in rows if r["account"] == args.account]

    state, detail = verdict(rows)
    log.info("%s: %s", state, detail)
    for row in rows:
        if row["state"] == "upgrade-pending":
            for name, want, have in row["permission_gap"]:
                log.info("  %s %s: %s %s, declared %s",
                         row["installation_id"], row["account"], name, have, want)
            if row["event_gap"]:
                log.info("  %s %s: events not accepted: %s",
                         row["installation_id"], row["account"],
                         ", ".join(row["event_gap"]))
        if row["state"] == "grant-ahead":
            for name, want, have in row["permission_surplus"]:
                log.info("  %s %s holds %s %s, %s",
                         row["installation_id"], row["account"], name, have, want)

    if state == "upgrades-pending":
        log.info("repair: an owner on each account accepts the pending "
                 "permission request from that org's Installed GitHub Apps "
                 "page. Until then, branch on the installation's own "
                 "permission map rather than on the App declaration")
    elif state == "grants-ahead":
        log.info("repair: nothing urgent. Those installations carry access "
                 "the App no longer declares, which is a tidy-up rather than "
                 "an outage")

    print(json.dumps({"declared_permissions": declared_permissions,
                      "declared_events": sorted(str(e) for e in declared_events),
                      "state": state, "cohorts": cohorts(rows),
                      "installations": rows}, indent=2, default=str))
    return 1 if state == "upgrades-pending" else 0


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

/**
 * Find GitHub App installations still living on an older permission grant.
 *
 * Read only. Two GETs with the App's JWT: the App's own declaration and the
 * list of installations. Accepting a permission upgrade is a human act
 * performed by an account owner, so the script prints who has to be asked and
 * for what rather than doing anything itself.
 *
 * Environment:
 *   GITHUB_APP_JWT   the JWT your own signing code produced
 */
const API = 'https://api.github.com';
const UA = 'github-permission-upgrade-lag/1.0';

/** Permission levels are ordered, not a set. */
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;
}

/**
 * Declared permissions an installation holds at a lower level. Pure.
 * Returns [[permission, declaredLevel, grantedLevel], ...] sorted by name.
 */
export function permissionGap(declared, granted) {
  const out = [];
  for (const name of Object.keys(declared ?? {}).sort()) {
    const wanted = declared[name];
    const have = (granted ?? {})[name];
    if (rank(have) < rank(wanted)) {
      out.push([name, String(wanted), have ? String(have) : 'absent']);
    }
  }
  return out;
}

/** Permissions an installation holds beyond what the App declares. Pure. */
export function permissionSurplus(declared, granted) {
  const out = [];
  for (const name of Object.keys(granted ?? {}).sort()) {
    const have = granted[name];
    const wanted = (declared ?? {})[name];
    if (rank(have) > rank(wanted)) {
      out.push([name, wanted ? String(wanted) : 'not declared', String(have)]);
    }
  }
  return out;
}

/** Declared events an installation has not accepted. Pure. */
export function eventGap(declaredEvents, grantedEvents) {
  const have = new Set((grantedEvents ?? []).map((e) => String(e).trim().toLowerCase()));
  return [...new Set((declaredEvents ?? []).map((e) => String(e).trim().toLowerCase()))]
    .filter((e) => !have.has(e)).sort();
}

/** Sort one installation against the App declaration. Pure. */
export function classify(declaredPermissions, declaredEvents, inst) {
  const row = inst && typeof inst === 'object' ? inst : {};
  const account = row.account && typeof row.account === 'object' ? row.account : {};
  const gaps = permissionGap(declaredPermissions, row.permissions);
  const extra = permissionSurplus(declaredPermissions, row.permissions);
  const events = eventGap(declaredEvents, row.events);
  let state = 'current';
  if (gaps.length || events.length) state = 'upgrade-pending';
  else if (extra.length) state = 'grant-ahead';
  return {
    installation_id: row.id ?? null,
    account: account.login ?? null,
    state,
    permission_gap: gaps,
    permission_surplus: extra,
    event_gap: events,
  };
}

/** Turn the per-installation rows into one finding. Pure. */
export function verdict(rows) {
  const all = rows ?? [];
  if (!all.length) {
    return ['no-installations',
      'this App has no installations, so there is nothing to be behind. ' +
      'Nothing here is evidence about permissions.'];
  }
  const behind = all.filter((r) => r.state === 'upgrade-pending');
  if (behind.length) {
    return ['upgrades-pending',
      `${behind.length} of ${all.length} installation(s) are behind the App ` +
      'declaration. Their tokens carry the permission map they accepted, not ' +
      'the one the App settings page shows.'];
  }
  const ahead = all.filter((r) => r.state === 'grant-ahead');
  if (ahead.length) {
    return ['grants-ahead',
      `${ahead.length} of ${all.length} installation(s) hold more than the ` +
      'App declares, which happens after a permission is removed rather than ' +
      'added. Nothing is failing; the access is simply unused.'];
  }
  return ['all-current',
    `all ${all.length} installation(s) have accepted what the App declares.`];
}

/** Group the laggards by exactly what they are missing. Pure. */
export function cohorts(rows) {
  const out = new Map();
  for (const row of rows ?? []) {
    if (row.state !== 'upgrade-pending') continue;
    const key = row.permission_gap
      .map(([n, d, g]) => `${n} ${g} (declared ${d})`).join(', ') || 'events only';
    if (!out.has(key)) out.set(key, []);
    out.get(key).push(String(row.account ?? row.installation_id));
  }
  return Object.fromEntries([...out.entries()]
    .sort(([a], [b]) => a.localeCompare(b))
    .map(([k, v]) => [k, v.sort()]));
}

async function get(jwt, path) {
  const url = path.startsWith('/') ? API + path : path;
  const res = await fetch(url, {
    headers: {
      Authorization: `Bearer ${jwt}`,
      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 listInstallations(jwt, pages = 10) {
  const out = [];
  for (let page = 1; page <= pages; page += 1) {
    const { status, body } = await get(jwt, `/app/installations?per_page=100&page=${page}`);
    if (status !== 200 || !Array.isArray(body)) {
      if (page === 1) {
        console.error(`GET /app/installations returned ${status}; this ` +
          'endpoint wants the App JWT');
      }
      break;
    }
    out.push(...body);
    if (body.length < 100) break;
  }
  return out;
}

async function main() {
  const jwt = process.env.GITHUB_APP_JWT;
  if (!jwt) {
    console.error('set GITHUB_APP_JWT to the JWT your own signing code ' +
      'produced. Both reads here are App-level and neither one accepts an ' +
      'installation token');
    process.exitCode = 2;
    return;
  }
  const only = process.argv[2] ?? null;

  const app = await get(jwt, '/app');
  if (app.status !== 200 || !app.body || typeof app.body !== 'object') {
    console.error(`GET /app returned ${app.status}, so there is no ` +
      'declaration to compare against');
    process.exitCode = 2;
    return;
  }
  const declaredPermissions = app.body.permissions ?? {};
  const declaredEvents = app.body.events ?? [];
  console.log(`app declares ${Object.keys(declaredPermissions).length} ` +
    `permission(s) and ${declaredEvents.length} event(s)`);

  const installations = await listInstallations(jwt);
  let rows = installations.map((i) => classify(declaredPermissions, declaredEvents, i));
  if (only) rows = rows.filter((r) => r.account === only);

  const [state, detail] = verdict(rows);
  console.log(`${state}: ${detail}`);
  for (const row of rows) {
    if (row.state === 'upgrade-pending') {
      for (const [name, want, have] of row.permission_gap) {
        console.log(`  ${row.installation_id} ${row.account}: ${name} ${have}, declared ${want}`);
      }
      if (row.event_gap.length) {
        console.log(`  ${row.installation_id} ${row.account}: events not ` +
          `accepted: ${row.event_gap.join(', ')}`);
      }
    }
    if (row.state === 'grant-ahead') {
      for (const [name, want, have] of row.permission_surplus) {
        console.log(`  ${row.installation_id} ${row.account} holds ${name} ${have}, ${want}`);
      }
    }
  }

  if (state === 'upgrades-pending') {
    console.log('repair: an owner on each account accepts the pending ' +
      "permission request from that org's Installed GitHub Apps page. Until " +
      "then, branch on the installation's own permission map rather than on " +
      'the App declaration');
  } else if (state === 'grants-ahead') {
    console.log('repair: nothing urgent. Those installations carry access the ' +
      'App no longer declares, which is a tidy-up rather than an outage');
  }

  console.log(JSON.stringify({
    declared_permissions: declaredPermissions,
    declared_events: [...declaredEvents].map(String).sort(),
    state,
    cohorts: cohorts(rows),
    installations: rows,
  }, null, 2));
  process.exitCode = state === 'upgrades-pending' ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not execute main(), fail on the missing JWT 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_permission_upgrade_lag import (
    classify, cohorts, event_gap, permission_gap, permission_surplus, rank,
    verdict,
)

DECLARED = {"contents": "read", "issues": "write", "checks": "write"}
EVENTS = ["push", "issues", "check_run"]


def install(ident, login, permissions, events=None):
    return {"id": ident, "account": {"login": login},
            "permissions": permissions, "events": events or EVENTS}


def test_levels_are_ordered_and_absent_is_the_bottom():
    assert rank("admin") > rank("write") > rank("read") > rank(None)
    assert rank("READ ") == rank("read")


def test_an_unrecognised_level_is_treated_as_no_access():
    assert rank("superuser") == 0


def test_a_present_key_at_too_low_a_level_is_still_a_gap():
    gaps = permission_gap(DECLARED, {"contents": "read", "issues": "read",
                                     "checks": "write"})
    assert gaps == [("issues", "write", "read")]


def test_a_missing_key_reports_as_absent_rather_than_as_a_level():
    gaps = permission_gap(DECLARED, {"contents": "read", "checks": "write"})
    assert gaps == [("issues", "write", "absent")]


def test_an_installation_that_matches_has_no_gap():
    assert permission_gap(DECLARED, dict(DECLARED)) == []


def test_holding_more_than_the_app_declares_is_its_own_finding():
    extra = permission_surplus(DECLARED, {"contents": "write", "issues": "write",
                                          "checks": "write"})
    assert extra == [("contents", "read", "write")]
    undeclared = permission_surplus(DECLARED, dict(DECLARED, members="read"))
    assert undeclared == [("members", "not declared", "read")]


def test_events_are_compared_case_and_space_insensitively():
    assert event_gap(EVENTS, [" Push ", "issues", "check_run"]) == []
    assert event_gap(EVENTS, ["push"]) == ["check_run", "issues"]


def test_an_installation_behind_on_anything_is_upgrade_pending():
    row = classify(DECLARED, EVENTS,
                   install(1, "beta-inc", {"contents": "read", "checks": "write"}))
    assert row["state"] == "upgrade-pending"
    assert row["account"] == "beta-inc"
    assert row["permission_gap"] == [("issues", "write", "absent")]


def test_an_installation_behind_only_on_events_is_still_pending():
    row = classify(DECLARED, EVENTS, install(2, "acme", dict(DECLARED), ["push"]))
    assert row["state"] == "upgrade-pending"
    assert row["event_gap"] == ["check_run", "issues"]


def test_an_installation_that_agrees_is_current():
    assert classify(DECLARED, EVENTS, install(3, "acme", dict(DECLARED)))["state"] == "current"


def test_the_verdict_reports_pending_before_anything_else():
    rows = [classify(DECLARED, EVENTS, install(1, "a", dict(DECLARED))),
            classify(DECLARED, EVENTS, install(2, "b", {"contents": "read"}))]
    state, detail = verdict(rows)
    assert state == "upgrades-pending"
    assert "1 of 2" in detail


def test_a_fleet_that_is_only_ahead_is_not_an_outage():
    rows = [classify(DECLARED, EVENTS,
                     install(1, "a", dict(DECLARED, contents="write")))]
    state, detail = verdict(rows)
    assert state == "grants-ahead"
    assert "Nothing is failing" in detail


def test_an_app_with_no_installations_says_so_rather_than_all_current():
    assert verdict([])[0] == "no-installations"


def test_all_current_when_every_map_agrees():
    rows = [classify(DECLARED, EVENTS, install(i, str(i), dict(DECLARED)))
            for i in range(3)]
    assert verdict(rows)[0] == "all-current"


def test_accounts_missing_the_same_thing_collapse_into_one_cohort():
    rows = [classify(DECLARED, EVENTS, install(1, "beta", {"contents": "read", "checks": "write"})),
            classify(DECLARED, EVENTS, install(2, "gamma", {"contents": "read", "checks": "write"})),
            classify(DECLARED, EVENTS, install(3, "delta", dict(DECLARED)))]
    grouped = cohorts(rows)
    assert len(grouped) == 1
    assert list(grouped.values())[0] == ["beta", "gamma"]

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  classify, cohorts, eventGap, permissionGap, permissionSurplus, rank, verdict,
} from './github-permission-upgrade-lag.mjs';

const DECLARED = { contents: 'read', issues: 'write', checks: 'write' };
const EVENTS = ['push', 'issues', 'check_run'];

function install(id, login, permissions, events = null) {
  return { id, account: { login }, permissions, events: events ?? EVENTS };
}

test('levels are ordered and absent is the bottom', () => {
  assert.ok(rank('admin') > rank('write'));
  assert.ok(rank('write') > rank('read'));
  assert.ok(rank('read') > rank(null));
  assert.equal(rank('READ '), rank('read'));
});

test('an unrecognised level is treated as no access', () => {
  assert.equal(rank('superuser'), 0);
});

test('a present key at too low a level is still a gap', () => {
  const gaps = permissionGap(DECLARED, { contents: 'read', issues: 'read', checks: 'write' });
  assert.deepEqual(gaps, [['issues', 'write', 'read']]);
});

test('a missing key reports as absent rather than as a level', () => {
  const gaps = permissionGap(DECLARED, { contents: 'read', checks: 'write' });
  assert.deepEqual(gaps, [['issues', 'write', 'absent']]);
});

test('an installation that matches has no gap', () => {
  assert.deepEqual(permissionGap(DECLARED, { ...DECLARED }), []);
});

test('holding more than the app declares is its own finding', () => {
  assert.deepEqual(
    permissionSurplus(DECLARED, { contents: 'write', issues: 'write', checks: 'write' }),
    [['contents', 'read', 'write']],
  );
  assert.deepEqual(
    permissionSurplus(DECLARED, { ...DECLARED, members: 'read' }),
    [['members', 'not declared', 'read']],
  );
});

test('events are compared case and space insensitively', () => {
  assert.deepEqual(eventGap(EVENTS, [' Push ', 'issues', 'check_run']), []);
  assert.deepEqual(eventGap(EVENTS, ['push']), ['check_run', 'issues']);
});

test('an installation behind on anything is upgrade pending', () => {
  const row = classify(DECLARED, EVENTS, install(1, 'beta-inc', { contents: 'read', checks: 'write' }));
  assert.equal(row.state, 'upgrade-pending');
  assert.equal(row.account, 'beta-inc');
  assert.deepEqual(row.permission_gap, [['issues', 'write', 'absent']]);
});

test('an installation behind only on events is still pending', () => {
  const row = classify(DECLARED, EVENTS, install(2, 'acme', { ...DECLARED }, ['push']));
  assert.equal(row.state, 'upgrade-pending');
  assert.deepEqual(row.event_gap, ['check_run', 'issues']);
});

test('an installation that agrees is current', () => {
  assert.equal(classify(DECLARED, EVENTS, install(3, 'acme', { ...DECLARED })).state, 'current');
});

test('the verdict reports pending before anything else', () => {
  const rows = [
    classify(DECLARED, EVENTS, install(1, 'a', { ...DECLARED })),
    classify(DECLARED, EVENTS, install(2, 'b', { contents: 'read' })),
  ];
  const [state, detail] = verdict(rows);
  assert.equal(state, 'upgrades-pending');
  assert.match(detail, /1 of 2/);
});

test('a fleet that is only ahead is not an outage', () => {
  const rows = [classify(DECLARED, EVENTS, install(1, 'a', { ...DECLARED, contents: 'write' }))];
  const [state, detail] = verdict(rows);
  assert.equal(state, 'grants-ahead');
  assert.match(detail, /Nothing is failing/);
});

test('an app with no installations says so rather than all current', () => {
  assert.equal(verdict([])[0], 'no-installations');
});

test('all current when every map agrees', () => {
  const rows = [0, 1, 2].map((i) => classify(DECLARED, EVENTS, install(i, String(i), { ...DECLARED })));
  assert.equal(verdict(rows)[0], 'all-current');
});

test('accounts missing the same thing collapse into one cohort', () => {
  const rows = [
    classify(DECLARED, EVENTS, install(1, 'beta', { contents: 'read', checks: 'write' })),
    classify(DECLARED, EVENTS, install(2, 'gamma', { contents: 'read', checks: 'write' })),
    classify(DECLARED, EVENTS, install(3, 'delta', { ...DECLARED })),
  ];
  const grouped = cohorts(rows);
  assert.equal(Object.keys(grouped).length, 1);
  assert.deepEqual(Object.values(grouped)[0], ['beta', 'gamma']);
});