one user revoked your app and only their token is dead

"""Tell an individual OAuth revocation apart from an application wide one.

Read only. One GET /user per stored user token, which is the cheapest call
that answers "is this credential still accepted, and whose is it". Nothing here
mints, refreshes or revokes anything; the repair is a URL that is printed for
the affected person to open.

The reading is a population, not a request. One refusal among many successes is
that user's revocation. Every refusal at once is the application: a rotated
client secret, a suspended app, or an organization owner removing the approval
for a whole cohort. No single response can tell those apart.

The definitive per-token check lives at /applications/{client_id}/token, which
is a write shaped call needing the client secret. This section's scripts do not
hold application secrets and do not make that call.
"""
import argparse
import json
import logging
import os
import sys
from urllib.parse import urlencode

import requests

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

API = "https://api.github.com"
UA = "github-user-token-liveness/1.0 (+https://example.com/contact)"
AUTHORIZE = "https://github.com/login/oauth/authorize"


def collect_tokens(environ, prefix):
    """Gather stored user tokens out of a mapping by prefix. Pure.

    Values leave here only to be sent. Everything the report prints is the
    variable name and the login, which are the two things a person can act on
    and neither of which is a secret.
    """
    return sorted((name, value) for name, value in environ.items()
                  if name.startswith(prefix) and value)


def token_result(status, message=None):
    """Classify one liveness probe. Pure.

    Four states, because a 403 is not a revocation: it is an account or an
    organization refusing something the credential is otherwise entitled to
    present, and it does not mean the authorization is gone.
    """
    if status == 200:
        return "alive"
    if status == 401:
        return "rejected"
    if status == 403:
        return "forbidden"
    return "error"


def population_verdict(results):
    """Read the fleet rather than the request. Pure.

    results: [(name, state), ...]. The counts are the diagnosis; no individual
    response contains this information.
    """
    if not results:
        return ("no-tokens",
                "nothing was collected, so there is nothing to read. Check the "
                "prefix the variables are named with.")
    alive = [n for n, s in results if s == "alive"]
    rejected = [n for n, s in results if s == "rejected"]
    if not rejected:
        return ("all-healthy",
                "every stored token is accepted, so no authorization has been "
                "revoked. Whatever you are chasing is somewhere else.")
    if len(results) == 1:
        return ("single-token-inconclusive",
                "one token is stored and it is refused. That is consistent "
                "with this user revoking, and equally consistent with the "
                "application being suspended or its secret rotated. With one "
                "sample the two cannot be separated.")
    if alive:
        return ("individual-revocation",
                "%d of %d stored tokens are refused while others work, so this "
                "is those people's decision rather than an application "
                "problem: %s" % (len(rejected), len(results),
                                 ", ".join(rejected)))
    return ("application-wide",
            "all %d stored tokens are refused at once. Users do not coordinate "
            "revocations. Look at the application: a rotated client secret, a "
            "suspended app, or an organization owner removing the approval for "
            "the whole cohort." % len(results))


def retry_disposition(state):
    """Say whether a state should ever be retried. Pure.

    The operationally valuable half of this note. A revoked user token does not
    recover, so a schedule that keeps trying turns one broken connection into a
    permanent stream of refusals competing with the users who still work.
    """
    if state == "rejected":
        return ("terminal",
                "a revoked or invalid user token never recovers on its own. "
                "Mark the connection broken, take it off the schedule, and ask "
                "the person to authorize again.")
    if state == "forbidden":
        return ("terminal",
                "the credential was accepted and the action was refused. "
                "Retrying changes nothing; this is an access question.")
    if state == "error":
        return ("retryable",
                "the probe itself did not complete, so nothing is known about "
                "the credential. This one is worth trying again.")
    return ("none", "nothing to retry.")


def authorize_url(client_id, scopes=(), redirect_uri=None, state=None):
    """Build the URL that starts the authorization flow again. Pure.

    This is the whole repair for an individual revocation: there is nothing to
    fix on your side, only a person to ask.
    """
    params = [("client_id", client_id)]
    if scopes:
        params.append(("scope", " ".join(scopes)))
    if redirect_uri:
        params.append(("redirect_uri", redirect_uri))
    if state:
        params.append(("state", state))
    return AUTHORIZE + "?" + urlencode(params)


def probe(token):
    """One GET /user. Returns (status, login, message)."""
    response = requests.get(API + "/user", timeout=30, headers={
        "Authorization": "Bearer " + token,
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
        "User-Agent": UA,
    })
    try:
        body = response.json()
    except ValueError:
        body = None
    if isinstance(body, dict):
        return response.status_code, body.get("login"), body.get("message")
    return response.status_code, None, None


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--env-prefix", default="GH_USER_TOKEN_",
                    help="collect every environment variable with this prefix")
    ap.add_argument("--scopes", default="",
                    help="space separated scopes for the printed authorize URL")
    args = ap.parse_args()

    stored = collect_tokens(os.environ, args.env_prefix)
    if not stored:
        log.error("no variables found with the prefix %s. Store one token per "
                  "connection so the set can be read as a set", args.env_prefix)
        return 2

    client_id = os.environ.get("GITHUB_OAUTH_CLIENT_ID", "")
    results, findings = [], []
    for name, token in stored:
        status, login, message = probe(token)
        state = token_result(status, message)
        results.append((name, state))
        log.info("%-24s %-9s %s", name, state, login or "-")
        findings.append({"env": name, "state": state, "login": login,
                         "status": status})

    verdict, detail = population_verdict(results)
    log.info("%s: %s", verdict, detail)

    for name, state in results:
        if state == "alive":
            continue
        disposition, why = retry_disposition(state)
        log.info("%s: %s. %s", name, disposition, why)

    if verdict in ("individual-revocation", "single-token-inconclusive"):
        if client_id:
            url = authorize_url(client_id,
                                args.scopes.split() if args.scopes else ())
            log.info("repair: send the affected people through the flow again: "
                     "%s", url)
        else:
            log.info("repair: set GITHUB_OAUTH_CLIENT_ID to have the authorize "
                     "URL printed here.")
    if verdict == "application-wide":
        log.info("repair: this is not the users. Check whether the client "
                 "secret was rotated, whether the application is suspended, "
                 "and whether an organization owner removed its approval.")

    print(json.dumps({"verdict": verdict, "tokens": findings}, indent=2))
    return 1 if verdict != "all-healthy" else 0


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

/**
 * Tell an individual OAuth revocation apart from an application wide one.
 *
 * Read only. One GET /user per stored user token. Nothing here mints,
 * refreshes or revokes anything; the repair is a URL printed for the affected
 * person to open.
 *
 * The reading is a population, not a request. One refusal among many successes
 * is that user's decision. Every refusal at once is the application. No single
 * response can tell those apart.
 *
 * The definitive per-token check lives at /applications/{client_id}/token,
 * which is a write shaped call needing the client secret. This script does not
 * hold application secrets and does not make that call.
 */
const API = 'https://api.github.com';
const UA = 'github-user-token-liveness/1.0 (+https://example.com/contact)';
const AUTHORIZE = 'https://github.com/login/oauth/authorize';

/** Gather stored user tokens out of a mapping by prefix. Pure. */
export function collectTokens(environ, prefix) {
  return Object.entries(environ ?? {})
    .filter(([name, value]) => name.startsWith(prefix) && value)
    .sort(([a], [b]) => a.localeCompare(b));
}

/** Classify one liveness probe. Pure. A 403 is not a revocation. */
export function tokenResult(status) {
  if (status === 200) return 'alive';
  if (status === 401) return 'rejected';
  if (status === 403) return 'forbidden';
  return 'error';
}

/** Read the fleet rather than the request. Pure. */
export function populationVerdict(results) {
  if (!results || !results.length) {
    return ['no-tokens',
      'nothing was collected, so there is nothing to read. Check the prefix ' +
      'the variables are named with.'];
  }
  const alive = results.filter(([, s]) => s === 'alive').map(([n]) => n);
  const rejected = results.filter(([, s]) => s === 'rejected').map(([n]) => n);
  if (!rejected.length) {
    return ['all-healthy',
      'every stored token is accepted, so no authorization has been revoked. ' +
      'Whatever you are chasing is somewhere else.'];
  }
  if (results.length === 1) {
    return ['single-token-inconclusive',
      'one token is stored and it is refused. That is consistent with this ' +
      'user revoking, and equally consistent with the application being ' +
      'suspended or its secret rotated. With one sample the two cannot be ' +
      'separated.'];
  }
  if (alive.length) {
    return ['individual-revocation',
      `${rejected.length} of ${results.length} stored tokens are refused ` +
      'while others work, so this is those people\'s decision rather than an ' +
      `application problem: ${rejected.join(', ')}`];
  }
  return ['application-wide',
    `all ${results.length} stored tokens are refused at once. Users do not ` +
    'coordinate revocations. Look at the application: a rotated client ' +
    'secret, a suspended app, or an organization owner removing the approval ' +
    'for the whole cohort.'];
}

/** Say whether a state should ever be retried. Pure. */
export function retryDisposition(state) {
  if (state === 'rejected') {
    return ['terminal',
      'a revoked or invalid user token never recovers on its own. Mark the ' +
      'connection broken, take it off the schedule, and ask the person to ' +
      'authorize again.'];
  }
  if (state === 'forbidden') {
    return ['terminal',
      'the credential was accepted and the action was refused. Retrying ' +
      'changes nothing; this is an access question.'];
  }
  if (state === 'error') {
    return ['retryable',
      'the probe itself did not complete, so nothing is known about the ' +
      'credential. This one is worth trying again.'];
  }
  return ['none', 'nothing to retry.'];
}

/** Build the URL that starts the authorization flow again. Pure. */
export function authorizeUrl(clientId, scopes = [], redirectUri = null, state = null) {
  const params = new URLSearchParams([['client_id', clientId]]);
  if (scopes && scopes.length) params.append('scope', scopes.join(' '));
  if (redirectUri) params.append('redirect_uri', redirectUri);
  if (state) params.append('state', state);
  return `${AUTHORIZE}?${params.toString()}`;
}

async function probe(token) {
  const res = await fetch(`${API}/user`, {
    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; }
  const login = body && typeof body === 'object' ? body.login ?? null : null;
  return [res.status, login];
}

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

async function main() {
  const prefix = arg('--env-prefix', 'GH_USER_TOKEN_');
  const scopes = arg('--scopes', '');

  const stored = collectTokens(process.env, prefix);
  if (!stored.length) {
    console.error(`no variables found with the prefix ${prefix}. Store one ` +
      'token per connection so the set can be read as a set');
    process.exitCode = 2;
    return;
  }

  const clientId = process.env.GITHUB_OAUTH_CLIENT_ID ?? '';
  const results = [];
  const findings = [];
  for (const [name, token] of stored) {
    const [status, login] = await probe(token);
    const state = tokenResult(status);
    results.push([name, state]);
    console.log(`${name.padEnd(24)} ${state.padEnd(9)} ${login ?? '-'}`);
    findings.push({ env: name, state, login, status });
  }

  const [verdict, detail] = populationVerdict(results);
  console.log(`${verdict}: ${detail}`);

  for (const [name, state] of results) {
    if (state === 'alive') continue;
    const [disposition, why] = retryDisposition(state);
    console.log(`${name}: ${disposition}. ${why}`);
  }

  if (verdict === 'individual-revocation' || verdict === 'single-token-inconclusive') {
    if (clientId) {
      const url = authorizeUrl(clientId, scopes ? scopes.split(' ').filter(Boolean) : []);
      console.log(`repair: send the affected people through the flow again: ${url}`);
    } else {
      console.log('repair: set GITHUB_OAUTH_CLIENT_ID to have the authorize ' +
        'URL printed here.');
    }
  }
  if (verdict === 'application-wide') {
    console.log('repair: this is not the users. Check whether the client ' +
      'secret was rotated, whether the application is suspended, and whether ' +
      'an organization owner removed its approval.');
  }

  console.log(JSON.stringify({ verdict, tokens: findings }, null, 2));
  process.exitCode = verdict === 'all-healthy' ? 0 : 1;
}

// Only run when invoked directly, so importing this module from the test file
// does not fire live requests and set an exit code the suite then inherits.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from github_user_token_liveness import (
    authorize_url, collect_tokens, population_verdict, retry_disposition,
    token_result,
)

ENV = {
    "GH_USER_TOKEN_BEN": "gho_fake2",
    "GH_USER_TOKEN_ALICE": "gho_fake1",
    "GITHUB_TOKEN": "ghp_fake",
    "GH_USER_TOKEN_EMPTY": "",
}


def test_tokens_are_collected_by_prefix_and_sorted():
    found = collect_tokens(ENV, "GH_USER_TOKEN_")
    assert [name for name, _ in found] == ["GH_USER_TOKEN_ALICE",
                                           "GH_USER_TOKEN_BEN"]


def test_an_unrelated_variable_is_not_collected():
    assert all(name != "GITHUB_TOKEN" for name, _ in
               collect_tokens(ENV, "GH_USER_TOKEN_"))


def test_an_empty_value_is_not_a_stored_token():
    assert all(name != "GH_USER_TOKEN_EMPTY" for name, _ in
               collect_tokens(ENV, "GH_USER_TOKEN_"))


def test_a_403_is_not_a_revocation():
    assert token_result(200) == "alive"
    assert token_result(401) == "rejected"
    assert token_result(403) == "forbidden"
    assert token_result(500) == "error"


def test_one_refusal_among_successes_is_that_person():
    state, detail = population_verdict([("a", "alive"), ("b", "rejected"),
                                        ("c", "alive")])
    assert state == "individual-revocation"
    assert "1 of 3" in detail
    assert "b" in detail


def test_every_token_refused_at_once_is_the_application():
    state, detail = population_verdict([("a", "rejected"), ("b", "rejected"),
                                        ("c", "rejected")])
    assert state == "application-wide"
    assert "do not coordinate" in detail


def test_one_stored_token_cannot_separate_the_two_causes():
    state, detail = population_verdict([("a", "rejected")])
    assert state == "single-token-inconclusive"
    assert "cannot be separated" in detail


def test_a_healthy_fleet_says_look_elsewhere():
    assert population_verdict([("a", "alive"), ("b", "alive")])[0] == "all-healthy"


def test_an_empty_fleet_is_not_a_verdict_about_users():
    assert population_verdict([])[0] == "no-tokens"


def test_errors_alone_are_not_read_as_an_application_failure():
    assert population_verdict([("a", "error"), ("b", "error")])[0] == "all-healthy"


def test_a_revoked_token_is_terminal_rather_than_retryable():
    disposition, detail = retry_disposition("rejected")
    assert disposition == "terminal"
    assert "never recovers" in detail


def test_a_failed_probe_is_the_only_retryable_state():
    assert retry_disposition("error")[0] == "retryable"
    assert retry_disposition("forbidden")[0] == "terminal"
    assert retry_disposition("alive")[0] == "none"


def test_the_authorize_url_carries_the_client_id_and_the_scopes():
    url = authorize_url("Iv1.example", ["repo", "read:org"])
    assert url.startswith("https://github.com/login/oauth/authorize?")
    assert "client_id=Iv1.example" in url
    assert "scope=repo+read%3Aorg" in url


def test_optional_parameters_are_omitted_rather_than_sent_empty():
    url = authorize_url("Iv1.example")
    assert "scope=" not in url
    assert "redirect_uri=" not in url
    assert "state=" not in url


def test_a_redirect_and_a_state_are_encoded():
    url = authorize_url("Iv1.example", ["repo"], "https://app.example/cb", "xyz")
    assert "redirect_uri=https%3A%2F%2Fapp.example%2Fcb" in url
    assert "state=xyz" in url

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  authorizeUrl, collectTokens, populationVerdict, retryDisposition, tokenResult,
} from './github-user-token-liveness.mjs';

const ENV = {
  GH_USER_TOKEN_BEN: 'gho_fake2',
  GH_USER_TOKEN_ALICE: 'gho_fake1',
  GITHUB_TOKEN: 'ghp_fake',
  GH_USER_TOKEN_EMPTY: '',
};

test('tokens are collected by prefix and sorted', () => {
  assert.deepEqual(collectTokens(ENV, 'GH_USER_TOKEN_').map(([n]) => n),
    ['GH_USER_TOKEN_ALICE', 'GH_USER_TOKEN_BEN']);
});

test('an unrelated variable is not collected', () => {
  assert.ok(collectTokens(ENV, 'GH_USER_TOKEN_').every(([n]) => n !== 'GITHUB_TOKEN'));
});

test('an empty value is not a stored token', () => {
  assert.ok(collectTokens(ENV, 'GH_USER_TOKEN_')
    .every(([n]) => n !== 'GH_USER_TOKEN_EMPTY'));
});

test('a 403 is not a revocation', () => {
  assert.equal(tokenResult(200), 'alive');
  assert.equal(tokenResult(401), 'rejected');
  assert.equal(tokenResult(403), 'forbidden');
  assert.equal(tokenResult(500), 'error');
});

test('one refusal among successes is that person', () => {
  const [state, detail] = populationVerdict([['a', 'alive'], ['b', 'rejected'], ['c', 'alive']]);
  assert.equal(state, 'individual-revocation');
  assert.ok(detail.includes('1 of 3'));
  assert.ok(detail.includes('b'));
});

test('every token refused at once is the application', () => {
  const [state, detail] = populationVerdict([['a', 'rejected'], ['b', 'rejected'], ['c', 'rejected']]);
  assert.equal(state, 'application-wide');
  assert.ok(detail.includes('do not coordinate'));
});

test('one stored token cannot separate the two causes', () => {
  const [state, detail] = populationVerdict([['a', 'rejected']]);
  assert.equal(state, 'single-token-inconclusive');
  assert.ok(detail.includes('cannot be separated'));
});

test('a healthy fleet says look elsewhere', () => {
  assert.equal(populationVerdict([['a', 'alive'], ['b', 'alive']])[0], 'all-healthy');
});

test('an empty fleet is not a verdict about users', () => {
  assert.equal(populationVerdict([])[0], 'no-tokens');
});

test('errors alone are not read as an application failure', () => {
  assert.equal(populationVerdict([['a', 'error'], ['b', 'error']])[0], 'all-healthy');
});

test('a revoked token is terminal rather than retryable', () => {
  const [disposition, detail] = retryDisposition('rejected');
  assert.equal(disposition, 'terminal');
  assert.ok(detail.includes('never recovers'));
});

test('a failed probe is the only retryable state', () => {
  assert.equal(retryDisposition('error')[0], 'retryable');
  assert.equal(retryDisposition('forbidden')[0], 'terminal');
  assert.equal(retryDisposition('alive')[0], 'none');
});

test('the authorize url carries the client id and the scopes', () => {
  const url = authorizeUrl('Iv1.example', ['repo', 'read:org']);
  assert.ok(url.startsWith('https://github.com/login/oauth/authorize?'));
  assert.ok(url.includes('client_id=Iv1.example'));
  assert.ok(url.includes('scope=repo+read%3Aorg'));
});

test('optional parameters are omitted rather than sent empty', () => {
  const url = authorizeUrl('Iv1.example');
  assert.ok(!url.includes('scope='));
  assert.ok(!url.includes('redirect_uri='));
  assert.ok(!url.includes('state='));
});

test('a redirect and a state are encoded', () => {
  const url = authorizeUrl('Iv1.example', ['repo'], 'https://app.example/cb', 'xyz');
  assert.ok(url.includes('redirect_uri=https%3A%2F%2Fapp.example%2Fcb'));
  assert.ok(url.includes('state=xyz'));
});