a firewall allow-list no longer matches GitHub's hook IPs

"""Compare GitHub's published webhook source ranges against your allow-list.

Read only, and unauthenticated: GET /meta needs no token, which means the person
who owns the firewall can run this without being issued a GitHub credential.

GET /meta returns the current CIDR ranges GitHub uses, split by purpose. The
hooks array is where webhook deliveries come from. A firewall allow-list copied
out of the documentation once goes stale as those ranges change, and the failure
is partial: only deliveries leaving from a new range are blocked, which reads as
flakiness rather than as a configuration problem.

The allow-list is a file you export, because the one thing this API can never
read is your own network. The accuracy of the answer is the accuracy of that
export.

Usage:

    python3 github_meta_hook_ranges.py --allowlist firewall-github.txt
"""
import argparse
import ipaddress
import json
import logging
import sys

import requests

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

META = "https://api.github.com/meta"
UA = "github-meta-hook-ranges/1.0"

# The array that answers this question, and the ones a list is most often built
# from by mistake. api and web are inbound: they are where you connect to.
HOOKS = "hooks"
OTHER_ARRAYS = ("api", "web", "git", "packages", "actions", "dependabot")
# How much better another array has to score before the finding is "you copied
# the wrong list" rather than "your list has drifted".
WRONG_ARRAY_MARGIN = 0.5


def parse_cidr(text):
    """(version, first_address, last_address) for one entry, or None. Pure.

    Host bits are tolerated, because firewall exports are full of 10.0.0.1/8 and
    refusing them would drop real rules on the floor.
    """
    raw = str(text or "").split("#")[0].strip()
    if not raw:
        return None
    try:
        net = ipaddress.ip_network(raw, strict=False)
    except ValueError:
        return None
    return (net.version, int(net.network_address), int(net.broadcast_address))


def read_allowlist(lines):
    """(ranges, unreadable) from an exported rule list. Pure.

    Unreadable lines are returned rather than skipped. A rule this script cannot
    parse is a hole in the audit, and silently ignoring it would report better
    coverage than the caller actually has.
    """
    ranges, unreadable = [], []
    for line in lines or []:
        text = str(line).split("#")[0].strip()
        if not text:
            continue
        parsed = parse_cidr(text)
        if parsed is None:
            unreadable.append(text)
        else:
            ranges.append(parsed)
    return ranges, unreadable


def size_of(rng):
    """How many addresses a parsed range holds. Pure."""
    return rng[2] - rng[1] + 1


def overlap(a, b):
    """The addresses two ranges share, or None. Pure."""
    if a[0] != b[0]:
        return None
    start, end = max(a[1], b[1]), min(a[2], b[2])
    return (start, end) if start <= end else None


def merge(intervals):
    """Merge overlapping and adjacent intervals. Pure.

    Without this, two allow-list entries that overlap would have their shared
    addresses counted twice and a range could report as more than fully covered.
    """
    out = []
    for start, end in sorted(intervals):
        if out and start <= out[-1][1] + 1:
            out[-1][1] = max(out[-1][1], end)
        else:
            out.append([start, end])
    return [(start, end) for start, end in out]


def covered_addresses(published, allowed):
    """How many addresses of one published range the allow-list permits. Pure."""
    pieces = [piece for piece in (overlap(published, a) for a in allowed) if piece]
    return sum(end - start + 1 for start, end in merge(pieces))


def coverage(published, allowed):
    """(state, fraction) for one published range. Pure.

    Measured over addresses rather than over the text of the CIDRs, so an
    equivalent rewrite of a range is full coverage and a subset is partial with
    the fraction it really permits.
    """
    total = size_of(published)
    covered = covered_addresses(published, allowed)
    if covered <= 0:
        return ("none", 0.0)
    if covered >= total:
        return ("full", 1.0)
    return ("partial", covered / total)


def allows_everything(allowed):
    """Whether a default route makes the allow-list decorative. Pure."""
    for version, start, end in allowed:
        bits = 32 if version == 4 else 128
        if start == 0 and end == (1 << bits) - 1:
            return True
    return False


def audit(published_cidrs, allowed):
    """[(cidr, state, fraction)] for every published range. Pure."""
    rows = []
    for cidr in published_cidrs or []:
        parsed = parse_cidr(cidr)
        if parsed is None:
            rows.append((str(cidr), "unreadable", 0.0))
            continue
        state, fraction = coverage(parsed, allowed)
        rows.append((str(cidr), state, fraction))
    return rows


def uncovered(rows):
    """The published ranges that are not fully covered. Pure."""
    return [cidr for cidr, state, _ in rows if state != "full"]


def array_score(meta, allowed, key):
    """Mean coverage of one /meta array by the allow-list, 0 to 1. Pure."""
    values = (meta or {}).get(key)
    if not isinstance(values, list) or not values:
        return 0.0
    rows = audit(values, allowed)
    return sum(fraction for _, _, fraction in rows) / len(rows)


def best_other_array(meta, allowed):
    """(key, score) for the non-hooks array the allow-list matches best. Pure."""
    best, score = None, 0.0
    for key in OTHER_ARRAYS:
        value = array_score(meta, allowed, key)
        if value > score:
            best, score = key, value
    return (best, score)


def verdict(meta, allowed, unreadable=0):
    """Turn the comparison into a finding. Pure."""
    published = (meta or {}).get(HOOKS)
    if not isinstance(published, list) or not published:
        return ("no-hooks-array",
                "GET /meta did not return a hooks array. Nothing can be "
                "compared until it does.")
    if not allowed:
        return ("no-allowlist",
                "the allow-list is empty, so either nothing is permitted or the "
                "export is wrong. Check the export before reading anything else "
                "here.")
    if allows_everything(allowed):
        return ("allow-all",
                "the allow-list contains a default route, so every published "
                "range is covered and the control is not filtering anything. "
                "This audit will pass forever and mean nothing.")
    rows = audit(published, allowed)
    missing = uncovered(rows)
    hooks_score = array_score(meta, allowed, HOOKS)
    other, other_score = best_other_array(meta, allowed)
    if missing and other and other_score > hooks_score + WRONG_ARRAY_MARGIN:
        return ("wrong-array",
                "the allow-list covers the %s ranges %d%% and the hooks ranges "
                "%d%%. This list was built from the wrong section of GET /meta: "
                "%s is inbound traffic, and webhooks arrive from hooks."
                % (other, round(other_score * 100), round(hooks_score * 100), other))
    if missing:
        return ("drifted",
                "%d of %d published hook ranges are not fully covered by the "
                "allow-list. Partial coverage fails intermittently, which is "
                "why this reads as flakiness rather than as a blocked range."
                % (len(missing), len(rows)))
    if unreadable:
        return ("current-with-gaps",
                "every published hook range is covered, but %d allow-list "
                "entries could not be parsed and were left out of the audit."
                % unreadable)
    return ("current",
            "every published hook range is fully covered by the allow-list.")


def repair(state):
    """The sentence a reader has to act on. Pure."""
    if state in ("drifted", "current-with-gaps"):
        return ("generate the allow-list from GET /meta on a schedule rather "
                "than maintaining it by hand, and alert when the published set "
                "changes so the next change is a pull request instead of an "
                "incident. The current set is printed below in full.")
    if state == "wrong-array":
        return ("rebuild the allow-list from the hooks array. The array in use "
                "is where GitHub serves traffic you connect to, not where "
                "webhook deliveries come from.")
    if state == "allow-all":
        return ("remove the default route or accept that this control does "
                "nothing. Either way, verify X-Hub-Signature-256 on every "
                "request: the signature is what authenticates an event, and an "
                "IP list never was.")
    if state == "no-allowlist":
        return ("export the rules from the device or the infrastructure code "
                "that defines them, one CIDR per line, and run this again.")
    if state == "current":
        return ("nothing today. Put this on a schedule so the answer stays "
                "true, and keep signature verification as the real control.")
    return "nothing."


def fetch_meta(session):
    """The published ranges. Unauthenticated on purpose."""
    r = session.get(META, timeout=30)
    if r.status_code != 200:
        log.error("GET /meta returned %d", r.status_code)
        return None
    try:
        return r.json()
    except ValueError:
        log.error("GET /meta returned a body that is not JSON")
        return None


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--allowlist", required=True,
                    help="file of CIDRs your firewall permits, one per line")
    args = ap.parse_args()

    with open(args.allowlist, "r", encoding="utf-8") as fh:
        allowed, unreadable = read_allowlist(fh.readlines())
    for line in unreadable:
        log.warning("allow-list entry not understood, left out of the audit: %s", line)

    session = requests.Session()
    session.headers.update({"Accept": "application/vnd.github+json",
                            "User-Agent": UA})
    meta = fetch_meta(session)
    if meta is None:
        return 2

    published = meta.get(HOOKS) or []
    log.info("GET /meta: %d hooks range(s) published, allow-list holds %d entry/entries",
             len(published), len(allowed))
    rows = audit(published, allowed)
    for cidr, state, fraction in rows:
        log.info("%-22s %-9s %3d%% covered", cidr, state, round(fraction * 100))

    state, detail = verdict(meta, allowed, len(unreadable))
    log.info("%s: %s", state, detail)
    log.info("repair: %s", repair(state))
    if state in ("drifted", "wrong-array", "current-with-gaps"):
        log.info("the published hooks ranges, in full:")
        for cidr in published:
            log.info("  %s", cidr)

    print(json.dumps({
        "published_hooks_ranges": published,
        "allowlist_entries": len(allowed),
        "allowlist_unreadable": unreadable,
        "coverage": [{"cidr": c, "state": s, "fraction": round(f, 4)}
                     for c, s, f in rows],
        "not_fully_covered": uncovered(rows),
        "hooks_score": round(array_score(meta, allowed, HOOKS), 4),
        "best_other_array": best_other_array(meta, allowed)[0],
        "state": state,
        "detail": detail,
        "repair": repair(state),
    }, indent=2, default=str))
    return 1 if state in ("drifted", "wrong-array", "allow-all", "no-allowlist") else 0


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

/**
 * Compare GitHub's published webhook source ranges against your allow-list.
 *
 * Read only, and unauthenticated: GET /meta needs no token, so the person who
 * owns the firewall can run this without being issued a GitHub credential.
 *
 * The address arithmetic is written out here because there is no address type
 * in the standard library and GitHub publishes IPv6 ranges as well as IPv4.
 *
 * Usage:
 *   node github-meta-hook-ranges.mjs ./firewall-github.txt
 */
import { readFile } from 'node:fs/promises';

const META = 'https://api.github.com/meta';
const UA = 'github-meta-hook-ranges/1.0';

export const HOOKS = 'hooks';
export const OTHER_ARRAYS = ['api', 'web', 'git', 'packages', 'actions', 'dependabot'];
export const WRONG_ARRAY_MARGIN = 0.5;

function isDigits(text, maxLength) {
  if (typeof text !== 'string' || text.length === 0 || text.length > maxLength) return false;
  for (const ch of text) if (ch < '0' || ch > '9') return false;
  return true;
}

function ipv4ToBig(addr) {
  const parts = String(addr).split('.');
  if (parts.length !== 4) return null;
  let n = 0n;
  for (const part of parts) {
    if (!isDigits(part, 3)) return null;
    const value = Number(part);
    if (value > 255) return null;
    n = (n << 8n) | BigInt(value);
  }
  return n;
}

function ipv6ToBig(addr) {
  let text = String(addr).toLowerCase();
  if (text.includes('.')) {
    // An embedded IPv4 tail, as in ::ffff:192.0.2.1.
    const cut = text.lastIndexOf(':');
    const tail = ipv4ToBig(text.slice(cut + 1));
    if (tail === null) return null;
    text = `${text.slice(0, cut + 1)}${(tail >> 16n).toString(16)}:${(tail & 0xffffn).toString(16)}`;
  }
  const halves = text.split('::');
  if (halves.length > 2) return null;
  const head = halves[0] ? halves[0].split(':') : [];
  const tail = halves.length === 2 && halves[1] ? halves[1].split(':') : [];
  if (halves.length === 1 && head.length !== 8) return null;
  if (head.length + tail.length > 8) return null;
  const filler = new Array(8 - head.length - tail.length).fill('0');
  const groups = halves.length === 1 ? head : [...head, ...filler, ...tail];
  let n = 0n;
  for (const group of groups) {
    if (group.length === 0 || group.length > 4) return null;
    for (const ch of group) if (!'0123456789abcdef'.includes(ch)) return null;
    n = (n << 16n) | BigInt(parseInt(group, 16));
  }
  return n;
}

/** {version, start, end} for one entry, or null. Host bits tolerated. Pure. */
export function parseCidr(text) {
  const raw = String(text ?? '').split('#')[0].trim();
  if (!raw) return null;
  const slash = raw.indexOf('/');
  const addr = slash === -1 ? raw : raw.slice(0, slash);
  const prefixText = slash === -1 ? null : raw.slice(slash + 1);
  const version = addr.includes(':') ? 6 : 4;
  const bits = version === 4 ? 32 : 128;
  const base = version === 4 ? ipv4ToBig(addr) : ipv6ToBig(addr);
  if (base === null) return null;
  let prefix = bits;
  if (prefixText !== null) {
    if (!isDigits(prefixText, 3)) return null;
    prefix = Number(prefixText);
    if (prefix > bits) return null;
  }
  const hostBits = BigInt(bits - prefix);
  const start = (base >> hostBits) << hostBits;
  return { version, start, end: start + (1n << hostBits) - 1n };
}

/** [ranges, unreadable] from an exported rule list. Pure. */
export function readAllowlist(lines) {
  const ranges = [];
  const unreadable = [];
  for (const line of lines || []) {
    const text = String(line).split('#')[0].trim();
    if (!text) continue;
    const parsed = parseCidr(text);
    if (parsed === null) unreadable.push(text);
    else ranges.push(parsed);
  }
  return [ranges, unreadable];
}

/** How many addresses a parsed range holds. Pure. */
export function sizeOf(range) {
  return range.end - range.start + 1n;
}

/** The addresses two ranges share, or null. Pure. */
export function overlap(a, b) {
  if (a.version !== b.version) return null;
  const start = a.start > b.start ? a.start : b.start;
  const end = a.end < b.end ? a.end : b.end;
  return start <= end ? { version: a.version, start, end } : null;
}

/** Merge overlapping and adjacent intervals so nothing is counted twice. Pure. */
export function merge(intervals) {
  const sorted = [...intervals].sort((a, b) => {
    if (a.start < b.start) return -1;
    if (a.start > b.start) return 1;
    return 0;
  });
  const out = [];
  for (const piece of sorted) {
    const last = out[out.length - 1];
    if (last && piece.start <= last.end + 1n) {
      last.end = piece.end > last.end ? piece.end : last.end;
    } else {
      out.push({ version: piece.version, start: piece.start, end: piece.end });
    }
  }
  return out;
}

/** How many addresses of one published range the allow-list permits. Pure. */
export function coveredAddresses(published, allowed) {
  const pieces = (allowed || []).map((a) => overlap(published, a)).filter(Boolean);
  return merge(pieces).reduce((total, piece) => total + (piece.end - piece.start + 1n), 0n);
}

/** [state, fraction] for one published range, measured over addresses. Pure. */
export function coverage(published, allowed) {
  const total = sizeOf(published);
  const covered = coveredAddresses(published, allowed);
  if (covered <= 0n) return ['none', 0];
  if (covered >= total) return ['full', 1];
  return ['partial', Number((covered * 10000n) / total) / 10000];
}

/** Whether a default route makes the allow-list decorative. Pure. */
export function allowsEverything(allowed) {
  for (const range of allowed || []) {
    const bits = range.version === 4 ? 32n : 128n;
    if (range.start === 0n && range.end === (1n << bits) - 1n) return true;
  }
  return false;
}

/** [[cidr, state, fraction]] for every published range. Pure. */
export function audit(publishedCidrs, allowed) {
  const rows = [];
  for (const cidr of publishedCidrs || []) {
    const parsed = parseCidr(cidr);
    if (parsed === null) {
      rows.push([String(cidr), 'unreadable', 0]);
      continue;
    }
    const [state, fraction] = coverage(parsed, allowed);
    rows.push([String(cidr), state, fraction]);
  }
  return rows;
}

/** The published ranges that are not fully covered. Pure. */
export function uncovered(rows) {
  return (rows || []).filter(([, state]) => state !== 'full').map(([cidr]) => cidr);
}

/** Mean coverage of one /meta array by the allow-list, 0 to 1. Pure. */
export function arrayScore(meta, allowed, key) {
  const values = (meta || {})[key];
  if (!Array.isArray(values) || values.length === 0) return 0;
  const rows = audit(values, allowed);
  return rows.reduce((sum, [, , fraction]) => sum + fraction, 0) / rows.length;
}

/** [key, score] for the non-hooks array the allow-list matches best. Pure. */
export function bestOtherArray(meta, allowed) {
  let best = null;
  let score = 0;
  for (const key of OTHER_ARRAYS) {
    const value = arrayScore(meta, allowed, key);
    if (value > score) { best = key; score = value; }
  }
  return [best, score];
}

/** Turn the comparison into a finding. Pure. */
export function verdict(meta, allowed, unreadable = 0) {
  const published = (meta || {})[HOOKS];
  if (!Array.isArray(published) || published.length === 0) {
    return ['no-hooks-array',
      'GET /meta did not return a hooks array. Nothing can be compared until it does.'];
  }
  if (!allowed || allowed.length === 0) {
    return ['no-allowlist',
      'the allow-list is empty, so either nothing is permitted or the export is '
      + 'wrong. Check the export before reading anything else here.'];
  }
  if (allowsEverything(allowed)) {
    return ['allow-all',
      'the allow-list contains a default route, so every published range is '
      + 'covered and the control is not filtering anything. This audit will pass '
      + 'forever and mean nothing.'];
  }
  const rows = audit(published, allowed);
  const missing = uncovered(rows);
  const hooksScore = arrayScore(meta, allowed, HOOKS);
  const [other, otherScore] = bestOtherArray(meta, allowed);
  if (missing.length && other && otherScore > hooksScore + WRONG_ARRAY_MARGIN) {
    return ['wrong-array',
      `the allow-list covers the ${other} ranges ${Math.round(otherScore * 100)}% `
      + `and the hooks ranges ${Math.round(hooksScore * 100)}%. This list was built `
      + `from the wrong section of GET /meta: ${other} is inbound traffic, and `
      + 'webhooks arrive from hooks.'];
  }
  if (missing.length) {
    return ['drifted',
      `${missing.length} of ${rows.length} published hook ranges are not fully `
      + 'covered by the allow-list. Partial coverage fails intermittently, which '
      + 'is why this reads as flakiness rather than as a blocked range.'];
  }
  if (unreadable) {
    return ['current-with-gaps',
      `every published hook range is covered, but ${unreadable} allow-list entries `
      + 'could not be parsed and were left out of the audit.'];
  }
  return ['current', 'every published hook range is fully covered by the allow-list.'];
}

/** The sentence a reader has to act on. Pure. */
export function repair(state) {
  if (['drifted', 'current-with-gaps'].includes(state)) {
    return 'generate the allow-list from GET /meta on a schedule rather than '
      + 'maintaining it by hand, and alert when the published set changes so the '
      + 'next change is a pull request instead of an incident. The current set is '
      + 'printed below in full.';
  }
  if (state === 'wrong-array') {
    return 'rebuild the allow-list from the hooks array. The array in use is '
      + 'where GitHub serves traffic you connect to, not where webhook '
      + 'deliveries come from.';
  }
  if (state === 'allow-all') {
    return 'remove the default route or accept that this control does nothing. '
      + 'Either way, verify X-Hub-Signature-256 on every request: the signature '
      + 'is what authenticates an event, and an IP list never was.';
  }
  if (state === 'no-allowlist') {
    return 'export the rules from the device or the infrastructure code that '
      + 'defines them, one CIDR per line, and run this again.';
  }
  if (state === 'current') {
    return 'nothing today. Put this on a schedule so the answer stays true, and '
      + 'keep signature verification as the real control.';
  }
  return 'nothing.';
}

async function main() {
  const path = process.argv[2] || process.env.GITHUB_ALLOWLIST;
  if (!path) {
    console.error('usage: node github-meta-hook-ranges.mjs ./firewall-github.txt');
    process.exitCode = 2;
    return;
  }
  const text = await readFile(path, 'utf8');
  const [allowed, unreadable] = readAllowlist(text.split(String.fromCharCode(10)));
  for (const line of unreadable) {
    console.error(`allow-list entry not understood, left out of the audit: ${line}`);
  }

  const res = await fetch(META, {
    headers: { Accept: 'application/vnd.github+json', 'User-Agent': UA },
  });
  if (res.status !== 200) {
    console.error(`GET /meta returned ${res.status}`);
    process.exitCode = 2;
    return;
  }
  const meta = await res.json();
  const published = meta[HOOKS] || [];
  console.log(`GET /meta: ${published.length} hooks range(s) published, `
    + `allow-list holds ${allowed.length} entry/entries`);
  const rows = audit(published, allowed);
  for (const [cidr, state, fraction] of rows) {
    console.log(`${cidr.padEnd(22)} ${state.padEnd(9)} ${Math.round(fraction * 100)}% covered`);
  }
  const [state, detail] = verdict(meta, allowed, unreadable.length);
  console.log(`${state}: ${detail}`);
  console.log(`repair: ${repair(state)}`);
  if (['drifted', 'wrong-array', 'current-with-gaps'].includes(state)) {
    console.log('the published hooks ranges, in full:');
    for (const cidr of published) console.log(`  ${cidr}`);
  }
  console.log(JSON.stringify({
    published_hooks_ranges: published,
    allowlist_entries: allowed.length,
    allowlist_unreadable: unreadable,
    not_fully_covered: uncovered(rows),
    hooks_score: Number(arrayScore(meta, allowed, HOOKS).toFixed(4)),
    best_other_array: bestOtherArray(meta, allowed)[0],
    state,
  }, null, 2));
  process.exitCode = ['drifted', 'wrong-array', 'allow-all', 'no-allowlist']
    .includes(state) ? 1 : 0;
}

// Guarded so importing this file from the test runner does not start a run and
// fail 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_meta_hook_ranges import (
    allows_everything, array_score, audit, best_other_array, coverage,
    covered_addresses, merge, overlap, parse_cidr, read_allowlist, repair,
    size_of, uncovered, verdict,
)

META = {
    "hooks": ["192.30.252.0/22", "140.82.112.0/20", "2a0a:a440::/29"],
    "api": ["10.10.0.0/16", "10.20.0.0/16"],
    "web": ["10.30.0.0/16"],
}


def parsed(*entries):
    return [parse_cidr(e) for e in entries]


def test_a_cidr_is_parsed_into_a_range_of_addresses():
    version, start, end = parse_cidr("192.30.252.0/22")
    assert version == 4
    assert end - start + 1 == 1024


def test_host_bits_and_bare_addresses_are_tolerated():
    assert parse_cidr("192.30.252.7/22") == parse_cidr("192.30.252.0/22")
    assert size_of(parse_cidr("140.82.112.5")) == 1
    assert parse_cidr("not-an-address") is None
    assert parse_cidr("   ") is None
    assert parse_cidr("# a comment") is None


def test_ipv6_ranges_are_understood():
    version, start, end = parse_cidr("2a0a:a440::/29")
    assert version == 6
    assert end > start


def test_the_two_families_never_cover_each_other():
    assert overlap(parse_cidr("0.0.0.0/0"), parse_cidr("2a0a:a440::/29")) is None
    assert coverage(parse_cidr("2a0a:a440::/29"), parsed("0.0.0.0/0")) == ("none", 0.0)


def test_a_subset_is_partial_with_the_fraction_it_permits():
    state, fraction = coverage(parse_cidr("192.30.252.0/22"), parsed("192.30.252.0/24"))
    assert state == "partial"
    assert round(fraction, 4) == 0.25


def test_a_superset_is_full_coverage_not_a_mismatch():
    assert coverage(parse_cidr("140.82.112.0/20"), parsed("140.82.0.0/16")) == ("full", 1.0)


def test_overlapping_rules_are_never_counted_twice():
    published = parse_cidr("192.30.252.0/22")
    allowed = parsed("192.30.252.0/24", "192.30.252.0/23")
    assert covered_addresses(published, allowed) == 512
    assert coverage(published, allowed)[0] == "partial"


def test_adjacent_rules_add_up_to_full_coverage():
    published = parse_cidr("192.30.252.0/23")
    allowed = parsed("192.30.252.0/24", "192.30.253.0/24")
    assert coverage(published, allowed) == ("full", 1.0)
    assert len(merge([overlap(published, a) for a in allowed])) == 1


def test_a_default_route_is_recognised_in_both_families():
    assert allows_everything(parsed("0.0.0.0/0"))
    assert allows_everything(parsed("::/0"))
    assert not allows_everything(parsed("10.0.0.0/8"))


def test_unreadable_allowlist_lines_are_returned_not_swallowed():
    ranges, unreadable = read_allowlist([
        "192.30.252.0/22", "  # comment", "", "140.82.112.0/20 # inline",
        "hooks.github.com",
    ])
    assert len(ranges) == 2
    assert unreadable == ["hooks.github.com"]


def test_the_audit_names_every_published_range():
    rows = audit(META["hooks"], parsed("140.82.112.0/20"))
    assert [state for _, state, _ in rows] == ["none", "full", "none"]
    assert uncovered(rows) == ["192.30.252.0/22", "2a0a:a440::/29"]


def test_drift_is_the_finding_when_some_ranges_are_short():
    allowed = parsed("192.30.252.0/24", "140.82.112.0/20", "2a0a:a440::/29")
    state, detail = verdict(META, allowed)
    assert state == "drifted"
    assert "1 of 3" in detail
    assert "intermittently" in detail


def test_a_list_built_from_the_wrong_array_is_named_as_such():
    allowed = parsed("10.10.0.0/16", "10.20.0.0/16")
    state, detail = verdict(META, allowed)
    assert state == "wrong-array"
    assert "api" in detail
    assert round(array_score(META, allowed, "api"), 4) == 1.0
    assert best_other_array(META, allowed)[0] == "api"


def test_a_default_route_passes_the_arithmetic_and_still_fails_the_audit():
    state, detail = verdict(META, parsed("0.0.0.0/0", "::/0"))
    assert state == "allow-all"
    assert "not filtering" in detail
    assert "never was" in repair("allow-all")


def test_a_complete_allowlist_is_current():
    allowed = parsed("192.30.252.0/22", "140.82.112.0/20", "2a0a:a440::/29")
    assert verdict(META, allowed)[0] == "current"


def test_unparsed_entries_downgrade_a_clean_result():
    allowed = parsed("192.30.252.0/22", "140.82.112.0/20", "2a0a:a440::/29")
    state, detail = verdict(META, allowed, unreadable=2)
    assert state == "current-with-gaps"
    assert "2 allow-list entries" in detail


def test_an_empty_allowlist_is_reported_rather_than_scored():
    assert verdict(META, [])[0] == "no-allowlist"
    assert verdict({}, parsed("10.0.0.0/8"))[0] == "no-hooks-array"


def test_the_repair_for_drift_is_automation_and_not_a_fresh_paste():
    assert "on a schedule" in repair("drifted")
    assert "hooks array" in repair("wrong-array")

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  allowsEverything, arrayScore, audit, bestOtherArray, coverage,
  coveredAddresses, merge, overlap, parseCidr, readAllowlist, repair, sizeOf,
  uncovered, verdict,
} from './github-meta-hook-ranges.mjs';

const META = {
  hooks: ['192.30.252.0/22', '140.82.112.0/20', '2a0a:a440::/29'],
  api: ['10.10.0.0/16', '10.20.0.0/16'],
  web: ['10.30.0.0/16'],
};

const parsed = (...entries) => entries.map((e) => parseCidr(e));

test('a CIDR is parsed into a range of addresses', () => {
  const range = parseCidr('192.30.252.0/22');
  assert.equal(range.version, 4);
  assert.equal(sizeOf(range), 1024n);
});

test('host bits and bare addresses are tolerated', () => {
  assert.deepEqual(parseCidr('192.30.252.7/22'), parseCidr('192.30.252.0/22'));
  assert.equal(sizeOf(parseCidr('140.82.112.5')), 1n);
  assert.equal(parseCidr('not-an-address'), null);
  assert.equal(parseCidr('   '), null);
  assert.equal(parseCidr('# a comment'), null);
});

test('ipv6 ranges are understood, compressed and expanded alike', () => {
  const range = parseCidr('2a0a:a440::/29');
  assert.equal(range.version, 6);
  assert.ok(range.end > range.start);
  assert.equal(parseCidr('2a0a:a440:0:0:0:0:0:0/29').start, range.start);
  assert.equal(parseCidr('2a0a:a440::zz/29'), null);
});

test('the two families never cover each other', () => {
  assert.equal(overlap(parseCidr('0.0.0.0/0'), parseCidr('2a0a:a440::/29')), null);
  assert.deepEqual(coverage(parseCidr('2a0a:a440::/29'), parsed('0.0.0.0/0')), ['none', 0]);
});

test('a subset is partial with the fraction it permits', () => {
  const [state, fraction] = coverage(parseCidr('192.30.252.0/22'), parsed('192.30.252.0/24'));
  assert.equal(state, 'partial');
  assert.equal(fraction, 0.25);
});

test('a superset is full coverage, not a mismatch', () => {
  assert.deepEqual(coverage(parseCidr('140.82.112.0/20'), parsed('140.82.0.0/16')), ['full', 1]);
});

test('overlapping rules are never counted twice', () => {
  const published = parseCidr('192.30.252.0/22');
  const allowed = parsed('192.30.252.0/24', '192.30.252.0/23');
  assert.equal(coveredAddresses(published, allowed), 512n);
  assert.equal(coverage(published, allowed)[0], 'partial');
});

test('adjacent rules add up to full coverage', () => {
  const published = parseCidr('192.30.252.0/23');
  const allowed = parsed('192.30.252.0/24', '192.30.253.0/24');
  assert.deepEqual(coverage(published, allowed), ['full', 1]);
  assert.equal(merge(allowed.map((a) => overlap(published, a))).length, 1);
});

test('a default route is recognised in both families', () => {
  assert.ok(allowsEverything(parsed('0.0.0.0/0')));
  assert.ok(allowsEverything(parsed('::/0')));
  assert.ok(!allowsEverything(parsed('10.0.0.0/8')));
});

test('unreadable allow-list lines are returned, not swallowed', () => {
  const [ranges, unreadable] = readAllowlist([
    '192.30.252.0/22', '  # comment', '', '140.82.112.0/20 # inline', 'hooks.github.com',
  ]);
  assert.equal(ranges.length, 2);
  assert.deepEqual(unreadable, ['hooks.github.com']);
});

test('the audit names every published range', () => {
  const rows = audit(META.hooks, parsed('140.82.112.0/20'));
  assert.deepEqual(rows.map(([, state]) => state), ['none', 'full', 'none']);
  assert.deepEqual(uncovered(rows), ['192.30.252.0/22', '2a0a:a440::/29']);
});

test('drift is the finding when some ranges are short', () => {
  const allowed = parsed('192.30.252.0/24', '140.82.112.0/20', '2a0a:a440::/29');
  const [state, detail] = verdict(META, allowed);
  assert.equal(state, 'drifted');
  assert.match(detail, /1 of 3/);
  assert.match(detail, /intermittently/);
});

test('a list built from the wrong array is named as such', () => {
  const allowed = parsed('10.10.0.0/16', '10.20.0.0/16');
  const [state, detail] = verdict(META, allowed);
  assert.equal(state, 'wrong-array');
  assert.match(detail, /api/);
  assert.equal(arrayScore(META, allowed, 'api'), 1);
  assert.equal(bestOtherArray(META, allowed)[0], 'api');
});

test('a default route passes the arithmetic and still fails the audit', () => {
  const [state, detail] = verdict(META, parsed('0.0.0.0/0', '::/0'));
  assert.equal(state, 'allow-all');
  assert.match(detail, /not filtering/);
  assert.match(repair('allow-all'), /never was/);
});

test('a complete allow-list is current', () => {
  const allowed = parsed('192.30.252.0/22', '140.82.112.0/20', '2a0a:a440::/29');
  assert.equal(verdict(META, allowed)[0], 'current');
});

test('unparsed entries downgrade a clean result', () => {
  const allowed = parsed('192.30.252.0/22', '140.82.112.0/20', '2a0a:a440::/29');
  const [state, detail] = verdict(META, allowed, 2);
  assert.equal(state, 'current-with-gaps');
  assert.match(detail, /2 allow-list entries/);
});

test('an empty allow-list is reported rather than scored', () => {
  assert.equal(verdict(META, [])[0], 'no-allowlist');
  assert.equal(verdict({}, parsed('10.0.0.0/8'))[0], 'no-hooks-array');
});

test('the repair for drift is automation and not a fresh paste', () => {
  assert.match(repair('drifted'), /on a schedule/);
  assert.match(repair('wrong-array'), /hooks array/);
});