a webhook with no secret sends no signature to verify

"""Find GitHub webhooks with no secret, and hooks whose secret is being rejected.

Read only. Every request is a GET. The script can prove a hook has no secret,
because the key is simply absent from config. It cannot prove a secret is
correct: the value comes back masked, so a wrong secret and a right one are
indistinguishable until deliveries start failing.
"""
import argparse
import logging
import os
import sys

import requests

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

API = "https://api.github.com"
UA = "github-hook-secret-audit/1.0"

# What GitHub returns in place of a secret that is set. Its presence is the only
# positive signal available; its value carries no information at all.
MASK = "********"


def secret_state(hook):
    """Is a secret configured on this hook? Pure.

    GitHub masks a configured secret and omits the key when there is none, so
    absence is a real finding rather than an inference. Anything else about the
    secret, including whether it is the right one, is not knowable from here.
    """
    config = hook.get("config")
    if not isinstance(config, dict):
        return "unknown"
    if "secret" not in config:
        return "absent"
    value = config.get("secret")
    if value is None or str(value).strip() == "":
        return "absent"
    return "set"


def unauthorized(deliveries):
    """Count deliveries the receiver refused with 401 or 403. Pure.

    Returns (rejected, total). These are the responses your own server gave, so
    on a hook that has a secret they are the only visible trace of a mismatch
    between the value GitHub signs with and the value the receiver checks.
    """
    rejected = total = 0
    for d in deliveries or []:
        total += 1
        try:
            code = int(d.get("status_code"))
        except (TypeError, ValueError):
            continue
        if code in (401, 403):
            rejected += 1
    return rejected, total


def verdict(hook, rejected=0, delivered=0):
    """Classify one hook. Pure, so the asymmetry is visible and testable.

    Returns (state, detail). "unsigned" is a fact about the configuration.
    "signed" is the absence of evidence and says so.
    """
    state = secret_state(hook)
    url = (hook.get("config") or {}).get("url") or "the configured URL"

    if state == "unknown":
        return ("unknown", "no config on this hook, which should not happen; "
                           "re-read it with GET /repos/{owner}/{repo}/hooks/{id}")

    if state == "absent":
        return ("unsigned",
                "config has no secret key, so GitHub sends no X-Hub-Signature-256 "
                "header with these payloads. A receiver that verifies only when "
                "the header is present verifies nothing, and anyone who learns %s "
                "can post to it." % url)

    if rejected and delivered and rejected * 2 >= delivered:
        return ("rejected",
                "a secret is set and %d of %d recent deliveries came back 401 or "
                "403 from your server. That is what a mismatched secret looks "
                "like from here; the value itself is masked and cannot be "
                "compared." % (rejected, delivered))

    detail = ("a secret is set, so payloads are signed. The value is masked as "
              "%s, so this says nothing about whether it matches the one your "
              "receiver holds." % MASK)
    if rejected:
        detail += (" %d of %d recent deliveries were refused with 401 or 403, "
                   "which is worth reading before you trust it."
                   % (rejected, delivered))
    return ("signed", detail)


def next_link(response):
    """The rel=next URL from the Link header, or None."""
    for part in (response.headers.get("Link") or "").split(","):
        chunk = part.strip()
        if chunk.startswith("<") and chunk.endswith('rel="next"'):
            return chunk[1:chunk.index(">")]
    return None


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code == 401:
        raise SystemExit("401 from GitHub: GITHUB_TOKEN is missing, expired or "
                         "malformed")
    if r.status_code in (403, 404):
        raise SystemExit("%d from %s: listing hooks needs admin:repo_hook for a "
                         "repository or admin:org_hook for an organization"
                         % (r.status_code, url))
    r.raise_for_status()
    return r


def page(session, url, limit=500, **params):
    out = []
    while url and len(out) < limit:
        r = get(session, url, **params)
        out.extend(r.json())
        url, params = next_link(r), {}
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--repo", action="append", default=[],
                    help="owner/name; repeat for several repositories")
    ap.add_argument("--org", action="append", default=[],
                    help="organization login; repeat for several orgs")
    ap.add_argument("--max-deliveries", type=int, default=50,
                    help="deliveries to read per hook when looking for 401s "
                         "(0 to skip that read entirely)")
    args = ap.parse_args()

    if not (args.repo or args.org):
        log.error("pass at least one --repo owner/name or --org login")
        return 2

    token = os.environ.get("GITHUB_TOKEN")
    if not token:
        log.error("set GITHUB_TOKEN (a read-only token is enough)")
        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,
    })

    scopes = []
    for repo in args.repo:
        owner, _, name = repo.partition("/")
        if not (owner and name):
            log.error("--repo takes owner/name, for example acme/api")
            return 2
        scopes.append(("repo " + repo, "%s/repos/%s/%s/hooks" % (API, owner, name)))
    for org in args.org:
        scopes.append(("org " + org, "%s/orgs/%s/hooks" % (API, org)))

    unsigned = refusing = total = 0
    for label, base in scopes:
        for hook in page(session, base, per_page=100):
            total += 1
            rejected = delivered = 0
            if args.max_deliveries:
                rejected, delivered = unauthorized(
                    page(session, "%s/%s/deliveries" % (base, hook.get("id")),
                         limit=args.max_deliveries, per_page=100))
            state, detail = verdict(hook, rejected, delivered)
            url = (hook.get("config") or {}).get("url", "?")
            line = "%-8s %s %s  %s" % (state, label, url, detail)
            if state == "signed":
                log.info(line)
                continue
            log.warning(line)
            if state == "unsigned":
                unsigned += 1
                log.warning("  repair: set a high-entropy secret on this hook, "
                            "then make the receiver reject any request without "
                            "X-Hub-Signature-256 rather than skipping the check")
            elif state == "rejected":
                refusing += 1
                log.warning("  repair: compare the secret in your receiver's "
                            "environment with the one on the hook, then replay "
                            "with POST %s/%s/deliveries/{delivery_id}/attempts",
                            base, hook.get("id"))

    log.info("%d hook(s), %d unsigned, %d rejecting deliveries",
             total, unsigned, refusing)
    return 1 if (unsigned or refusing) else 0


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

/**
 * Find GitHub webhooks with no secret, and hooks whose secret is being rejected.
 *
 * Read only. The script can prove a hook has no secret, because the key is
 * absent from config. It cannot prove a secret is correct: the value comes back
 * masked, so a wrong secret and a right one are indistinguishable until
 * deliveries start failing.
 */
const API = 'https://api.github.com';
const UA = 'github-hook-secret-audit/1.0';

// What GitHub returns in place of a secret that is set. Its presence is the only
// positive signal available; its value carries no information at all.
const MASK = '********';

/**
 * Is a secret configured on this hook? Pure. GitHub masks a configured secret
 * and omits the key when there is none, so absence is a real finding.
 */
export function secretState(hook) {
  const config = hook.config;
  if (config === null || typeof config !== 'object') return 'unknown';
  if (!Object.prototype.hasOwnProperty.call(config, 'secret')) return 'absent';
  const value = config.secret;
  if (value === null || value === undefined || String(value).trim() === '') {
    return 'absent';
  }
  return 'set';
}

/**
 * Count deliveries the receiver refused with 401 or 403. Pure. On a hook that
 * has a secret these are the only visible trace of a mismatch.
 */
export function unauthorized(deliveries) {
  let rejected = 0;
  let total = 0;
  for (const d of deliveries ?? []) {
    total += 1;
    const code = Number.parseInt(d.status_code, 10);
    if (code === 401 || code === 403) rejected += 1;
  }
  return { rejected, total };
}

/**
 * Classify one hook. Pure. "unsigned" is a fact about the configuration;
 * "signed" is the absence of evidence and says so.
 */
export function verdict(hook, rejected = 0, delivered = 0) {
  const state = secretState(hook);
  const url = hook.config?.url ?? 'the configured URL';

  if (state === 'unknown') {
    return ['unknown', 'no config on this hook, which should not happen; ' +
      're-read it with GET /repos/{owner}/{repo}/hooks/{id}'];
  }

  if (state === 'absent') {
    return ['unsigned',
      'config has no secret key, so GitHub sends no X-Hub-Signature-256 header ' +
      'with these payloads. A receiver that verifies only when the header is ' +
      `present verifies nothing, and anyone who learns ${url} can post to it.`];
  }

  if (rejected && delivered && rejected * 2 >= delivered) {
    return ['rejected',
      `a secret is set and ${rejected} of ${delivered} recent deliveries came ` +
      'back 401 or 403 from your server. That is what a mismatched secret looks ' +
      'like from here; the value itself is masked and cannot be compared.'];
  }

  let detail = 'a secret is set, so payloads are signed. The value is masked as ' +
    `${MASK}, so this says nothing about whether it matches the one your ` +
    'receiver holds.';
  if (rejected) {
    detail += ` ${rejected} of ${delivered} recent deliveries were refused with ` +
      '401 or 403, which is worth reading before you trust it.';
  }
  return ['signed', detail];
}

function nextLink(res) {
  for (const part of (res.headers.get('link') ?? '').split(',')) {
    const chunk = part.trim();
    if (chunk.startsWith('<') && chunk.endsWith('rel="next"')) {
      return chunk.slice(1, chunk.indexOf('>'));
    }
  }
  return null;
}

async function get(token, url) {
  const res = await fetch(url, {
    headers: {
      Authorization: `Bearer ${token}`,
      Accept: 'application/vnd.github+json',
      'X-GitHub-Api-Version': '2022-11-28',
      'User-Agent': UA,
    },
  });
  if (res.status === 401) {
    throw new Error('401 from GitHub: GITHUB_TOKEN is missing, expired or malformed');
  }
  if (res.status === 403 || res.status === 404) {
    throw new Error(`${res.status} from ${url}: listing hooks needs ` +
      'admin:repo_hook for a repository or admin:org_hook for an organization');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url}`);
  return res;
}

async function page(token, url, limit = 500) {
  const out = [];
  let next = url;
  while (next && out.length < limit) {
    const res = await get(token, next);
    out.push(...(await res.json()));
    next = nextLink(res);
  }
  return out.slice(0, limit);
}

async function main() {
  const token = process.env.GITHUB_TOKEN;
  if (!token) {
    console.error('set GITHUB_TOKEN (a read-only token is enough)');
    process.exitCode = 2;
    return;
  }

  const scopes = [];
  for (const arg of process.argv.slice(2)) {
    if (arg.includes('/')) scopes.push([`repo ${arg}`, `${API}/repos/${arg}/hooks`]);
    else scopes.push([`org ${arg}`, `${API}/orgs/${arg}/hooks`]);
  }
  if (scopes.length === 0) {
    console.error('usage: node github-hook-secret-audit.mjs owner/name [org ...]');
    process.exitCode = 2;
    return;
  }

  let unsigned = 0;
  let refusing = 0;
  let total = 0;
  for (const [label, base] of scopes) {
    for (const hook of await page(token, `${base}?per_page=100`)) {
      total += 1;
      const deliveries = await page(token,
        `${base}/${hook.id}/deliveries?per_page=100`, 50);
      const { rejected, total: delivered } = unauthorized(deliveries);
      const [state, detail] = verdict(hook, rejected, delivered);
      const url = hook.config?.url ?? '?';
      const line = `${state.padEnd(8)} ${label} ${url}  ${detail}`;
      if (state === 'signed') { console.log(line); continue; }
      console.warn(line);
      if (state === 'unsigned') {
        unsigned += 1;
        console.warn('  repair: set a high-entropy secret on this hook, then ' +
          'make the receiver reject any request without X-Hub-Signature-256 ' +
          'rather than skipping the check');
      } else if (state === 'rejected') {
        refusing += 1;
        console.warn("  repair: compare the secret in your receiver's " +
          'environment with the one on the hook, then replay with POST ' +
          `${base}/${hook.id}/deliveries/{delivery_id}/attempts`);
      }
    }
  }

  console.log(`${total} hook(s), ${unsigned} unsigned, ${refusing} rejecting deliveries`);
  process.exitCode = (unsigned || refusing) ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not run main(), fail on the missing token, and fail the test file with it.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from github_hook_secret_audit import secret_state, unauthorized, verdict

SIGNED = {"id": 1, "config": {"url": "https://hooks.example.com/gh",
                              "secret": "********", "content_type": "json"}}
UNSIGNED = {"id": 2, "config": {"url": "https://hooks.example.com/gh",
                                "content_type": "json"}}


def test_a_missing_key_is_the_finding():
    # GitHub omits the key entirely rather than returning an empty string.
    assert secret_state(UNSIGNED) == "absent"


def test_a_masked_value_means_a_secret_exists():
    assert secret_state(SIGNED) == "set"


def test_an_empty_secret_counts_as_absent():
    assert secret_state({"config": {"secret": "  "}}) == "absent"


def test_a_hook_without_config_is_not_silently_signed():
    assert secret_state({"id": 3}) == "unknown"
    assert verdict({"id": 3})[0] == "unknown"


def test_the_unsigned_detail_names_the_missing_header():
    state, detail = verdict(UNSIGNED)
    assert state == "unsigned"
    assert "X-Hub-Signature-256" in detail
    assert "hooks.example.com" in detail


def test_signed_admits_it_cannot_check_the_value():
    state, detail = verdict(SIGNED)
    assert state == "signed"
    assert "masked" in detail
    assert "whether it matches" in detail


def test_a_run_of_refusals_on_a_signed_hook_is_its_own_state():
    state, detail = verdict(SIGNED, rejected=18, delivered=20)
    assert state == "rejected"
    assert "mismatched secret" in detail


def test_one_refusal_in_fifty_is_not_a_mismatch():
    state, detail = verdict(SIGNED, rejected=1, delivered=50)
    assert state == "signed"
    assert "1 of 50" in detail


def test_unauthorized_counts_only_auth_failures():
    rejected, total = unauthorized([{"status_code": 401}, {"status_code": 403},
                                    {"status_code": 500}, {"status_code": 200},
                                    {"status_code": None}])
    assert (rejected, total) == (2, 5)

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  secretState, unauthorized, verdict,
} from './github-hook-secret-audit.mjs';

const SIGNED = {
  id: 1,
  config: { url: 'https://hooks.example.com/gh', secret: '********', content_type: 'json' },
};
const UNSIGNED = {
  id: 2,
  config: { url: 'https://hooks.example.com/gh', content_type: 'json' },
};

test('a missing key is the finding', () => {
  assert.equal(secretState(UNSIGNED), 'absent');
});

test('a masked value means a secret exists', () => {
  assert.equal(secretState(SIGNED), 'set');
});

test('an empty secret counts as absent', () => {
  assert.equal(secretState({ config: { secret: '  ' } }), 'absent');
});

test('a hook without config is not silently signed', () => {
  assert.equal(secretState({ id: 3 }), 'unknown');
  assert.equal(verdict({ id: 3 })[0], 'unknown');
});

test('the unsigned detail names the missing header', () => {
  const [state, detail] = verdict(UNSIGNED);
  assert.equal(state, 'unsigned');
  assert.match(detail, /X-Hub-Signature-256/);
  assert.match(detail, /hooks\.example\.com/);
});

test('signed admits it cannot check the value', () => {
  const [state, detail] = verdict(SIGNED);
  assert.equal(state, 'signed');
  assert.match(detail, /masked/);
  assert.match(detail, /whether it matches/);
});

test('a run of refusals on a signed hook is its own state', () => {
  const [state, detail] = verdict(SIGNED, 18, 20);
  assert.equal(state, 'rejected');
  assert.match(detail, /mismatched secret/);
});

test('one refusal in fifty is not a mismatch', () => {
  const [state, detail] = verdict(SIGNED, 1, 50);
  assert.equal(state, 'signed');
  assert.match(detail, /1 of 50/);
});

test('unauthorized counts only auth failures', () => {
  const { rejected, total } = unauthorized([{ status_code: 401 },
    { status_code: 403 }, { status_code: 500 }, { status_code: 200 },
    { status_code: null }]);
  assert.equal(rejected, 2);
  assert.equal(total, 5);
});