the receiver still checks the legacy SHA-1 signature

"""Report which webhook signature header your receiver actually verifies.

Read only in both senses. Every API call is a GET, and the local source scan
opens files for reading and prints line numbers rather than lines.

GitHub sends two signature headers on every delivery from a hook that has a
secret set:

    X-Hub-Signature        HMAC-SHA1, kept for backwards compatibility
    X-Hub-Signature-256    HMAC-SHA256, the one to verify

Which of them your receiver checks is a decision in your own code. No API read
can see it, so this script establishes from the API that both headers were
sent, then searches your receiver's source for the header names and reports
what it found. That is a proxy and it is described as one: a header name built
at runtime, read from configuration or hidden inside a framework helper is
invisible to a text search.

The one subtlety worth the code: "x-hub-signature" is a prefix of
"x-hub-signature-256", so a plain substring search reports every correct
receiver as a legacy one. The modern name is removed from each line before the
legacy name is looked for.

The secret is never printed. config.secret comes back masked and this script
reports its presence only.

Environment:

    GITHUB_TOKEN    a read-only token that can see the repository's hooks
"""
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_hook_signature_headers")

API = "https://api.github.com"
UA = "github-hook-signature-headers/1.0"

MODERN = "x-hub-signature-256"
LEGACY = "x-hub-signature"

# Source files worth opening. Everything else in a repository is noise, and a
# scan that opens every file is a scan that reads secrets out of .env by
# accident.
SUFFIXES = (".py", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".rb", ".go", ".php",
            ".java", ".kt", ".cs", ".rs", ".ex", ".exs")

# States that mean the reader has something to change.
FINDINGS = ("sha1-only", "both-accepted", "no-verification-found")


def normalized(text):
    """Lower-cased, with underscores folded to hyphens. Pure.

    The same header reaches code as X-Hub-Signature-256, x-hub-signature-256
    and HTTP_X_HUB_SIGNATURE_256 depending on the runtime, and all three should
    count as the same reference.
    """
    return str(text or "").lower().replace("_", "-")


def secret_state(hook):
    """Whether the hook has a secret set: set, absent or unknown. Pure.

    The value is masked by GitHub and is never returned by this function. A set
    secret is a key present in config; an unset one is a key that is not there
    at all, which is a different note.
    """
    if not isinstance(hook, dict):
        return "unknown"
    config = hook.get("config")
    if not isinstance(config, dict):
        return "unknown"
    return "set" if "secret" in config else "absent"


def redacted_config(config):
    """A copy of a hook config safe to print. Pure.

    The secret arrives masked, so printing it would leak nothing today. It is
    replaced anyway, because the guarantee this section makes is about what the
    script can emit rather than about what GitHub happens to send.
    """
    if not isinstance(config, dict):
        return {}
    out = dict(config)
    if "secret" in out:
        out["secret"] = "<set>"
    return out


def header_names(headers):
    """Normalised header names from a delivery record. Pure. Values discarded.

    request.headers is an object, but the same data reaches this function as a
    list of name/value pairs or as raw header lines depending on what stored
    it, so all three shapes are accepted and only the names survive.
    """
    names = []
    if isinstance(headers, dict):
        names = list(headers.keys())
    elif isinstance(headers, list):
        for row in headers:
            if isinstance(row, dict) and row.get("name"):
                names.append(row["name"])
            elif isinstance(row, str) and ":" in row:
                names.append(row.split(":", 1)[0])
            elif isinstance(row, str):
                names.append(row)
    return [normalized(n).strip() for n in names]


def signature_headers(headers):
    """Which signature headers GitHub sent on a delivery. Pure.

    Exact name matching, so the prefix problem does not arise here; it arises
    in the source scan below, where the names appear inside other text.
    """
    names = header_names(headers)
    return {"sha256": MODERN in names, "sha1": LEGACY in names}


def scan_line(line):
    """Which signature header names a single line refers to. Pure.

    The modern name is removed first. Looking for the legacy name in the raw
    line instead finds a hit inside every correct reference, which turns a
    passing receiver into a finding on every line where it is right.
    """
    norm = normalized(line)
    kinds = []
    if MODERN in norm:
        kinds.append("sha256")
        norm = norm.replace(MODERN, " ")
    if LEGACY in norm:
        kinds.append("sha1")
    return kinds


def scan_source(text, path="<source>"):
    """Every signature header reference in a file, as (path, line, kind). Pure.

    Line numbers only. The contents of a line that handles signatures are the
    contents most likely to sit next to a secret, and a diagnostic that pastes
    them into a terminal or a CI log has made the problem worse.
    """
    hits = []
    for number, line in enumerate(str(text or "").splitlines(), start=1):
        for kind in scan_line(line):
            hits.append((path, number, kind))
    return hits


def receiver_state(hits):
    """What the scan says the receiver names. Pure.

    none means the header names do not appear in the source that was scanned,
    which is not the same as "does not verify" and is not reported as if it
    were.
    """
    kinds = {kind for _, _, kind in hits or []}
    if not kinds:
        return "none"
    if kinds == {"sha256"}:
        return "sha256-only"
    if kinds == {"sha1"}:
        return "sha1-only"
    return "both"


def format_hit(hit):
    """One line of the scan report. Pure. Never includes source text."""
    path, number, kind = hit
    name = "X-Hub-Signature-256" if kind == "sha256" else "X-Hub-Signature"
    label = "modern" if kind == "sha256" else "legacy"
    return "%s:%d %s %s" % (path, number, label, name)


def verdict(secret, sig=None, receiver=None):
    """Combine the API evidence and the source scan into a finding. Pure.

    secret is the output of secret_state, sig the output of signature_headers
    or None when no delivery was read, receiver the output of receiver_state or
    None when no source was scanned.
    """
    if secret == "absent":
        return ("no-secret",
                "this hook has no secret, so GitHub sends neither signature "
                "header and there is nothing for the receiver to verify. That "
                "is a different and larger problem than which digest you use.")
    if sig is not None and not sig.get("sha256") and not sig.get("sha1"):
        return ("headers-missing",
                "the delivery that was read carries no signature header at "
                "all. Either it predates the secret being set, or the record "
                "is not a delivery from this hook.")
    if receiver is None:
        return ("not-scanned",
                "GitHub sent the SHA-256 header. Which header the receiver "
                "verifies is not visible from the API, so point the scan at "
                "the receiver's source to get an answer rather than a "
                "recommendation.")
    if receiver == "none":
        return ("no-verification-found",
                "neither signature header name appears in the source that was "
                "scanned. Either the receiver does not verify, or it builds "
                "the header name at runtime, or the verification lives "
                "somewhere the scan was not pointed at.")
    if receiver == "sha1-only":
        return ("sha1-only",
                "the receiver names only the legacy SHA-1 header. GitHub sent "
                "the SHA-256 header on the same request and it is being "
                "ignored.")
    if receiver == "both":
        return ("both-accepted",
                "the receiver names both headers. A receiver that accepts "
                "either is exactly as strong as the weaker one, so this is a "
                "migration state rather than a finished one.")
    return ("sha256-only",
            "the receiver names only X-Hub-Signature-256, which is the header "
            "to verify.")


def repair(state):
    """The change to make, in the reader's own code. Pure."""
    if state == "no-secret":
        return ("set a secret on the hook first. Until there is one, GitHub "
                "sends no signature and no digest choice matters.")
    if state == "headers-missing":
        return ("read a delivery from after the secret was set, then re-run.")
    if state == "not-scanned":
        return ("re-run with --receiver pointed at the source tree that "
                "handles the webhook.")
    if state == "no-verification-found":
        return ("confirm by hand that the receiver verifies at all. If it does "
                "not, verify X-Hub-Signature-256 over the raw request bytes "
                "with a constant-time comparison and reject a request whose "
                "header is missing.")
    if state in ("sha1-only", "both-accepted"):
        return ("verify X-Hub-Signature-256 over the raw request bytes with a "
                "constant-time comparison, then delete the SHA-1 branch rather "
                "than keeping it as a fallback.")
    return "nothing. This receiver verifies the header GitHub wants it to."


def scan_paths(paths, suffixes=SUFFIXES, max_bytes=2_000_000):
    """Walk files and directories, scanning source for header references.

    Read only: files are opened for reading and only line numbers leave this
    function.
    """
    hits = []
    scanned = []
    for root in paths or []:
        for path in _walk(root, suffixes):
            try:
                if os.path.getsize(path) > max_bytes:
                    continue
                with open(path, "r", encoding="utf-8", errors="replace") as fh:
                    text = fh.read()
            except OSError as exc:
                log.warning("could not read %s: %s", path, exc)
                continue
            scanned.append(path)
            hits.extend(scan_source(text, path))
    return hits, scanned


def _walk(root, suffixes):
    """Every candidate source file under a path."""
    if os.path.isfile(root):
        return [root]
    out = []
    for base, dirs, files in os.walk(root):
        dirs[:] = [d for d in dirs if d not in
                   (".git", "node_modules", "venv", ".venv", "__pycache__", "dist", "build")]
        for name in files:
            if name.endswith(suffixes):
                out.append(os.path.join(base, name))
    return out


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 latest_delivery_headers(session, repo, hook_id):
    """The request headers of the most recent delivery, or None. Read only."""
    status, rows = get(session, "/repos/%s/hooks/%s/deliveries?per_page=10"
                       % (repo, hook_id))
    if status != 200 or not isinstance(rows, list) or not rows:
        return None
    for row in rows:
        if not isinstance(row, dict) or not row.get("id"):
            continue
        status, body = get(session, "/repos/%s/hooks/%s/deliveries/%s"
                           % (repo, hook_id, row["id"]))
        if status == 200 and isinstance(body, dict):
            request = body.get("request")
            if isinstance(request, dict):
                return request.get("headers")
    return None


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--repo", required=True, help="owner/name")
    ap.add_argument("--hook-id", help="one hook; omit to check every hook")
    ap.add_argument("--receiver", action="append", default=[],
                    help="file or directory to scan for header names; repeatable")
    args = ap.parse_args()

    token = os.environ.get("GITHUB_TOKEN")
    if not token:
        log.error("set GITHUB_TOKEN to a read-only token that can see the "
                  "repository's hooks")
        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,
    })

    status, hooks = get(session, "/repos/%s/hooks?per_page=100" % args.repo)
    if status != 200 or not isinstance(hooks, list):
        log.error("GET /repos/%s/hooks returned %d", args.repo, status)
        return 2
    if args.hook_id:
        hooks = [h for h in hooks if str(h.get("id")) == str(args.hook_id)]

    hits, scanned = scan_paths(args.receiver)
    state_of_receiver = receiver_state(hits) if args.receiver else None
    if args.receiver:
        log.info("source scan: %d reference(s) across %d file(s)",
                 len(hits), len(set(p for p, _, _ in hits)))
        for hit in hits:
            log.info("  %s", format_hit(hit))

    findings = []
    for hook in hooks:
        secret = secret_state(hook)
        headers = None
        if secret == "set":
            headers = latest_delivery_headers(session, args.repo, hook.get("id"))
        sig = signature_headers(headers) if headers is not None else None
        state, detail = verdict(secret, sig, state_of_receiver)
        log.info("hook %s: secret is %s, %s", hook.get("id"), secret,
                 "GitHub sent both signature headers"
                 if sig and sig["sha1"] and sig["sha256"]
                 else "no delivery headers were read")
        log.info("%s: %s", state, detail)
        log.info("repair: %s", repair(state))
        findings.append({"hook_id": hook.get("id"), "secret": secret,
                         "signature_headers": sig, "state": state,
                         "detail": detail,
                         "config": redacted_config(hook.get("config"))})

    print(json.dumps({"repo": args.repo, "files_scanned": len(scanned),
                      "references": [format_hit(h) for h in hits],
                      "findings": findings}, indent=2, default=str))
    return 1 if any(f["state"] in FINDINGS for f in findings) else 0


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

/**
 * Report which webhook signature header your receiver actually verifies.
 *
 * Read only in both senses: every API call is a GET, and the local source scan
 * opens files for reading and prints line numbers rather than lines.
 *
 * GitHub sends X-Hub-Signature (HMAC-SHA1, legacy) and X-Hub-Signature-256
 * (HMAC-SHA256) on every delivery from a hook that has a secret. Which one the
 * receiver checks lives in your source and no API read can see it, so the API
 * half of this script only establishes that both were sent.
 *
 * The secret is never printed; its presence is reported and nothing else.
 *
 * Environment:
 *   GITHUB_TOKEN   a read-only token that can see the repository's hooks
 *
 * Usage:
 *   node github-hook-signature-headers.mjs acme-corp/api ../receiver/src
 */
import { readFile, readdir, stat } from 'node:fs/promises';
import path from 'node:path';

const API = 'https://api.github.com';
const UA = 'github-hook-signature-headers/1.0';

export const MODERN = 'x-hub-signature-256';
export const LEGACY = 'x-hub-signature';

const SUFFIXES = ['.py', '.js', '.mjs', '.cjs', '.ts', '.tsx', '.rb', '.go',
  '.php', '.java', '.kt', '.cs', '.rs', '.ex', '.exs'];
const SKIP_DIRS = ['.git', 'node_modules', 'venv', '.venv', '__pycache__',
  'dist', 'build'];

/** States that mean the reader has something to change. */
export const FINDINGS = ['sha1-only', 'both-accepted', 'no-verification-found'];

/** Lower-cased, with underscores folded to hyphens. Pure. */
export function normalized(text) {
  return String(text ?? '').toLowerCase().replaceAll('_', '-');
}

/** Whether the hook has a secret set: set, absent or unknown. Pure. */
export function secretState(hook) {
  if (!hook || typeof hook !== 'object') return 'unknown';
  const config = hook.config;
  if (!config || typeof config !== 'object') return 'unknown';
  return 'secret' in config ? 'set' : 'absent';
}

/** A copy of a hook config safe to print. Pure. */
export function redactedConfig(config) {
  if (!config || typeof config !== 'object') return {};
  const out = { ...config };
  if ('secret' in out) out.secret = '<set>';
  return out;
}

/** Normalised header names from a delivery record. Pure. Values discarded. */
export function headerNames(headers) {
  let names = [];
  if (Array.isArray(headers)) {
    for (const row of headers) {
      if (row && typeof row === 'object' && row.name) names.push(row.name);
      else if (typeof row === 'string') names.push(row.includes(':') ? row.split(':')[0] : row);
    }
  } else if (headers && typeof headers === 'object') {
    names = Object.keys(headers);
  }
  return names.map((n) => normalized(n).trim());
}

/** Which signature headers GitHub sent on a delivery. Pure. */
export function signatureHeaders(headers) {
  const names = headerNames(headers);
  return { sha256: names.includes(MODERN), sha1: names.includes(LEGACY) };
}

/**
 * Which signature header names a single line refers to. Pure.
 * The modern name is removed first, because the legacy name is a prefix of it.
 */
export function scanLine(line) {
  let norm = normalized(line);
  const kinds = [];
  if (norm.includes(MODERN)) {
    kinds.push('sha256');
    norm = norm.replaceAll(MODERN, ' ');
  }
  if (norm.includes(LEGACY)) kinds.push('sha1');
  return kinds;
}

/** Every signature header reference in a file, as [path, line, kind]. Pure. */
export function scanSource(text, filePath = '<source>') {
  const hits = [];
  const lines = String(text ?? '').split('\n');
  for (let i = 0; i < lines.length; i += 1) {
    for (const kind of scanLine(lines[i])) hits.push([filePath, i + 1, kind]);
  }
  return hits;
}

/** What the scan says the receiver names. Pure. */
export function receiverState(hits) {
  const kinds = new Set((hits ?? []).map(([, , kind]) => kind));
  if (kinds.size === 0) return 'none';
  if (kinds.size === 1 && kinds.has('sha256')) return 'sha256-only';
  if (kinds.size === 1 && kinds.has('sha1')) return 'sha1-only';
  return 'both';
}

/** One line of the scan report. Pure. Never includes source text. */
export function formatHit(hit) {
  const [filePath, number, kind] = hit;
  const name = kind === 'sha256' ? 'X-Hub-Signature-256' : 'X-Hub-Signature';
  const label = kind === 'sha256' ? 'modern' : 'legacy';
  return `${filePath}:${number} ${label} ${name}`;
}

/** Combine the API evidence and the source scan into a finding. Pure. */
export function verdict(secret, sig = null, receiver = null) {
  if (secret === 'absent') {
    return ['no-secret',
      'this hook has no secret, so GitHub sends neither signature header and ' +
      'there is nothing for the receiver to verify. That is a different and ' +
      'larger problem than which digest you use.'];
  }
  if (sig !== null && !sig.sha256 && !sig.sha1) {
    return ['headers-missing',
      'the delivery that was read carries no signature header at all. Either ' +
      'it predates the secret being set, or the record is not a delivery from ' +
      'this hook.'];
  }
  if (receiver === null) {
    return ['not-scanned',
      'GitHub sent the SHA-256 header. Which header the receiver verifies is ' +
      'not visible from the API, so point the scan at the receiver\'s source ' +
      'to get an answer rather than a recommendation.'];
  }
  if (receiver === 'none') {
    return ['no-verification-found',
      'neither signature header name appears in the source that was scanned. ' +
      'Either the receiver does not verify, or it builds the header name at ' +
      'runtime, or the verification lives somewhere the scan was not pointed at.'];
  }
  if (receiver === 'sha1-only') {
    return ['sha1-only',
      'the receiver names only the legacy SHA-1 header. GitHub sent the ' +
      'SHA-256 header on the same request and it is being ignored.'];
  }
  if (receiver === 'both') {
    return ['both-accepted',
      'the receiver names both headers. A receiver that accepts either is ' +
      'exactly as strong as the weaker one, so this is a migration state ' +
      'rather than a finished one.'];
  }
  return ['sha256-only',
    'the receiver names only X-Hub-Signature-256, which is the header to verify.'];
}

/** The change to make, in the reader's own code. Pure. */
export function repair(state) {
  if (state === 'no-secret') {
    return 'set a secret on the hook first. Until there is one, GitHub sends ' +
      'no signature and no digest choice matters.';
  }
  if (state === 'headers-missing') {
    return 'read a delivery from after the secret was set, then re-run.';
  }
  if (state === 'not-scanned') {
    return 're-run with a path to the source tree that handles the webhook.';
  }
  if (state === 'no-verification-found') {
    return 'confirm by hand that the receiver verifies at all. If it does ' +
      'not, verify X-Hub-Signature-256 over the raw request bytes with a ' +
      'constant-time comparison and reject a request whose header is missing.';
  }
  if (state === 'sha1-only' || state === 'both-accepted') {
    return 'verify X-Hub-Signature-256 over the raw request bytes with a ' +
      'constant-time comparison, then delete the SHA-1 branch rather than ' +
      'keeping it as a fallback.';
  }
  return 'nothing. This receiver verifies the header GitHub wants it to.';
}

async function walk(root) {
  const info = await stat(root).catch(() => null);
  if (!info) return [];
  if (info.isFile()) return [root];
  const out = [];
  const entries = await readdir(root, { withFileTypes: true }).catch(() => []);
  for (const entry of entries) {
    if (entry.isDirectory()) {
      if (SKIP_DIRS.includes(entry.name)) continue;
      out.push(...await walk(path.join(root, entry.name)));
    } else if (SUFFIXES.some((s) => entry.name.endsWith(s))) {
      out.push(path.join(root, entry.name));
    }
  }
  return out;
}

async function scanPaths(paths) {
  const hits = [];
  const scanned = [];
  for (const root of paths ?? []) {
    for (const file of await walk(root)) {
      const text = await readFile(file, 'utf8').catch(() => null);
      if (text === null) continue;
      scanned.push(file);
      hits.push(...scanSource(text, file));
    }
  }
  return { hits, scanned };
}

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

async function latestDeliveryHeaders(token, repo, hookId) {
  const list = await get(token, `/repos/${repo}/hooks/${hookId}/deliveries?per_page=10`);
  if (list.status !== 200 || !Array.isArray(list.body)) return null;
  for (const row of list.body) {
    if (!row || typeof row !== 'object' || !row.id) continue;
    const one = await get(token, `/repos/${repo}/hooks/${hookId}/deliveries/${row.id}`);
    if (one.status === 200 && one.body && typeof one.body === 'object') {
      const request = one.body.request;
      if (request && typeof request === 'object') return request.headers;
    }
  }
  return null;
}

async function main() {
  const token = process.env.GITHUB_TOKEN;
  if (!token) {
    console.error('set GITHUB_TOKEN to a read-only token that can see the ' +
      "repository's hooks");
    process.exitCode = 2;
    return;
  }
  const [repo, ...receivers] = process.argv.slice(2);
  if (!repo) {
    console.error('usage: node github-hook-signature-headers.mjs owner/name [receiver-path...]');
    process.exitCode = 2;
    return;
  }

  const { status, body: hooks } = await get(token, `/repos/${repo}/hooks?per_page=100`);
  if (status !== 200 || !Array.isArray(hooks)) {
    console.error(`GET /repos/${repo}/hooks returned ${status}`);
    process.exitCode = 2;
    return;
  }

  const { hits, scanned } = await scanPaths(receivers);
  const stateOfReceiver = receivers.length ? receiverState(hits) : null;
  if (receivers.length) {
    const files = new Set(hits.map(([f]) => f));
    console.log(`source scan: ${hits.length} reference(s) across ${files.size} file(s)`);
    for (const hit of hits) console.log(`  ${formatHit(hit)}`);
  }

  const findings = [];
  for (const hook of hooks) {
    const secret = secretState(hook);
    const headers = secret === 'set'
      ? await latestDeliveryHeaders(token, repo, hook.id)
      : null;
    const sig = headers !== null && headers !== undefined ? signatureHeaders(headers) : null;
    const [state, detail] = verdict(secret, sig, stateOfReceiver);
    console.log(`hook ${hook.id}: secret is ${secret}, ` +
      (sig && sig.sha1 && sig.sha256
        ? 'GitHub sent both signature headers'
        : 'no delivery headers were read'));
    console.log(`${state}: ${detail}`);
    console.log(`repair: ${repair(state)}`);
    findings.push({
      hook_id: hook.id,
      secret,
      signature_headers: sig,
      state,
      detail,
      config: redactedConfig(hook.config),
    });
  }

  console.log(JSON.stringify({
    repo,
    files_scanned: scanned.length,
    references: hits.map(formatHit),
    findings,
  }, null, 2));
  process.exitCode = findings.some((f) => FINDINGS.includes(f.state)) ? 1 : 0;
}

// Only run when invoked directly, so importing this module from the test file
// does not execute main() 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_hook_signature_headers import (
    format_hit, header_names, normalized, receiver_state, redacted_config,
    repair, scan_line, scan_source, secret_state, signature_headers, verdict,
)

MODERN_LINE = 'sig = request.headers["X-Hub-Signature-256"]'
LEGACY_LINE = 'sig = request.headers["X-Hub-Signature"]'
WSGI_LINE = 'sig = environ["HTTP_X_HUB_SIGNATURE_256"]'

SECRET_SET = {"id": 1, "config": {"url": "https://example.com/hook",
                                  "secret": "********", "content_type": "json"}}
NO_SECRET = {"id": 2, "config": {"url": "https://example.com/hook",
                                 "content_type": "json"}}


def test_the_modern_header_is_not_read_as_a_legacy_one():
    assert scan_line(MODERN_LINE) == ["sha256"]
    assert scan_line(LEGACY_LINE) == ["sha1"]
    assert scan_line("nothing to see here") == []


def test_a_line_naming_both_headers_reports_both():
    line = 'const h = req.headers["x-hub-signature-256"] ?? req.headers["x-hub-signature"];'
    assert scan_line(line) == ["sha256", "sha1"]


def test_the_runtime_spellings_are_the_same_header():
    assert scan_line(WSGI_LINE) == ["sha256"]
    assert scan_line("X_HUB_SIGNATURE") == ["sha1"]
    assert normalized("X-Hub_Signature-256") == "x-hub-signature-256"


def test_the_scan_reports_line_numbers_and_never_lines():
    text = "\n".join(["import os", LEGACY_LINE, "", MODERN_LINE])
    hits = scan_source(text, "receiver/hooks.py")
    assert hits == [("receiver/hooks.py", 2, "sha1"),
                    ("receiver/hooks.py", 4, "sha256")]
    rendered = [format_hit(h) for h in hits]
    assert rendered[0] == "receiver/hooks.py:2 legacy X-Hub-Signature"
    assert rendered[1] == "receiver/hooks.py:4 modern X-Hub-Signature-256"
    assert not any("request.headers" in line for line in rendered)


def test_the_receiver_state_separates_only_legacy_from_both():
    assert receiver_state([("a", 1, "sha1")]) == "sha1-only"
    assert receiver_state([("a", 1, "sha256")]) == "sha256-only"
    assert receiver_state([("a", 1, "sha256"), ("a", 1, "sha1")]) == "both"
    assert receiver_state([]) == "none"


def test_a_masked_secret_is_presence_and_never_a_value():
    assert secret_state(SECRET_SET) == "set"
    assert secret_state(NO_SECRET) == "absent"
    assert secret_state({"id": 3}) == "unknown"
    safe = redacted_config(SECRET_SET["config"])
    assert safe["secret"] == "<set>"
    assert "********" not in str(safe)


def test_header_names_are_matched_exactly_and_values_dropped():
    sent = {"X-Hub-Signature": "sha1=deadbeef",
            "X-Hub-Signature-256": "sha256=deadbeef",
            "Content-Type": "application/json"}
    assert signature_headers(sent) == {"sha256": True, "sha1": True}
    assert signature_headers({"X-Hub-Signature-256": "x"}) == {"sha256": True, "sha1": False}
    assert signature_headers({}) == {"sha256": False, "sha1": False}
    assert "deadbeef" not in str(header_names(sent))


def test_delivery_headers_arrive_in_more_than_one_shape():
    as_list = [{"name": "X-Hub-Signature-256", "value": "sha256=x"},
               {"name": "Content-Type", "value": "application/json"}]
    assert signature_headers(as_list)["sha256"] is True
    assert signature_headers(["X-Hub-Signature: sha1=x"])["sha1"] is True
    assert signature_headers(None) == {"sha256": False, "sha1": False}


def test_a_legacy_receiver_is_the_finding():
    state, detail = verdict("set", {"sha256": True, "sha1": True}, "sha1-only")
    assert state == "sha1-only"
    assert "being ignored" in detail
    assert "constant-time" in repair(state)


def test_accepting_both_headers_is_still_a_finding():
    state, detail = verdict("set", {"sha256": True, "sha1": True}, "both")
    assert state == "both-accepted"
    assert "weaker" in detail


def test_a_hook_with_no_secret_is_sent_to_a_different_note():
    state, detail = verdict("absent", None, "sha1-only")
    assert state == "no-secret"
    assert "different and larger problem" in detail


def test_with_no_source_the_script_declines_to_guess():
    state, detail = verdict("set", {"sha256": True, "sha1": True}, None)
    assert state == "not-scanned"
    assert "not visible from the API" in detail


def test_finding_nothing_is_not_reported_as_finding_a_problem():
    state, detail = verdict("set", {"sha256": True, "sha1": True}, "none")
    assert state == "no-verification-found"
    assert "at runtime" in detail


def test_a_correct_receiver_passes():
    state, _ = verdict("set", {"sha256": True, "sha1": True}, "sha256-only")
    assert state == "sha256-only"
    assert repair(state).startswith("nothing")

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  formatHit, headerNames, normalized, receiverState, redactedConfig,
  repair, scanLine, scanSource, secretState, signatureHeaders, verdict,
} from './github-hook-signature-headers.mjs';

const MODERN_LINE = 'const sig = req.headers["X-Hub-Signature-256"];';
const LEGACY_LINE = 'const sig = req.headers["X-Hub-Signature"];';
const WSGI_LINE = 'sig = environ["HTTP_X_HUB_SIGNATURE_256"]';

const SECRET_SET = {
  id: 1,
  config: { url: 'https://example.com/hook', secret: '********', content_type: 'json' },
};
const NO_SECRET = {
  id: 2,
  config: { url: 'https://example.com/hook', content_type: 'json' },
};

test('the modern header is not read as a legacy one', () => {
  assert.deepEqual(scanLine(MODERN_LINE), ['sha256']);
  assert.deepEqual(scanLine(LEGACY_LINE), ['sha1']);
  assert.deepEqual(scanLine('nothing to see here'), []);
});

test('a line naming both headers reports both', () => {
  const line = 'const h = req.headers["x-hub-signature-256"] ?? req.headers["x-hub-signature"];';
  assert.deepEqual(scanLine(line), ['sha256', 'sha1']);
});

test('the runtime spellings are the same header', () => {
  assert.deepEqual(scanLine(WSGI_LINE), ['sha256']);
  assert.deepEqual(scanLine('X_HUB_SIGNATURE'), ['sha1']);
  assert.equal(normalized('X-Hub_Signature-256'), 'x-hub-signature-256');
});

test('the scan reports line numbers and never lines', () => {
  const text = ['import os', LEGACY_LINE, '', MODERN_LINE].join('\n');
  const hits = scanSource(text, 'receiver/hooks.js');
  assert.deepEqual(hits, [
    ['receiver/hooks.js', 2, 'sha1'],
    ['receiver/hooks.js', 4, 'sha256'],
  ]);
  const rendered = hits.map(formatHit);
  assert.equal(rendered[0], 'receiver/hooks.js:2 legacy X-Hub-Signature');
  assert.equal(rendered[1], 'receiver/hooks.js:4 modern X-Hub-Signature-256');
  assert.ok(!rendered.some((line) => line.includes('req.headers')));
});

test('the receiver state separates only legacy from both', () => {
  assert.equal(receiverState([['a', 1, 'sha1']]), 'sha1-only');
  assert.equal(receiverState([['a', 1, 'sha256']]), 'sha256-only');
  assert.equal(receiverState([['a', 1, 'sha256'], ['a', 1, 'sha1']]), 'both');
  assert.equal(receiverState([]), 'none');
});

test('a masked secret is presence and never a value', () => {
  assert.equal(secretState(SECRET_SET), 'set');
  assert.equal(secretState(NO_SECRET), 'absent');
  assert.equal(secretState({ id: 3 }), 'unknown');
  const safe = redactedConfig(SECRET_SET.config);
  assert.equal(safe.secret, '<set>');
  assert.ok(!JSON.stringify(safe).includes('********'));
});

test('header names are matched exactly and values dropped', () => {
  const sent = {
    'X-Hub-Signature': 'sha1=deadbeef',
    'X-Hub-Signature-256': 'sha256=deadbeef',
    'Content-Type': 'application/json',
  };
  assert.deepEqual(signatureHeaders(sent), { sha256: true, sha1: true });
  assert.deepEqual(signatureHeaders({ 'X-Hub-Signature-256': 'x' }),
    { sha256: true, sha1: false });
  assert.deepEqual(signatureHeaders({}), { sha256: false, sha1: false });
  assert.ok(!JSON.stringify(headerNames(sent)).includes('deadbeef'));
});

test('delivery headers arrive in more than one shape', () => {
  const asList = [
    { name: 'X-Hub-Signature-256', value: 'sha256=x' },
    { name: 'Content-Type', value: 'application/json' },
  ];
  assert.equal(signatureHeaders(asList).sha256, true);
  assert.equal(signatureHeaders(['X-Hub-Signature: sha1=x']).sha1, true);
  assert.deepEqual(signatureHeaders(null), { sha256: false, sha1: false });
});

test('a legacy receiver is the finding', () => {
  const [state, detail] = verdict('set', { sha256: true, sha1: true }, 'sha1-only');
  assert.equal(state, 'sha1-only');
  assert.match(detail, /being ignored/);
  assert.match(repair(state), /constant-time/);
});

test('accepting both headers is still a finding', () => {
  const [state, detail] = verdict('set', { sha256: true, sha1: true }, 'both');
  assert.equal(state, 'both-accepted');
  assert.match(detail, /weaker/);
});

test('a hook with no secret is sent to a different note', () => {
  const [state, detail] = verdict('absent', null, 'sha1-only');
  assert.equal(state, 'no-secret');
  assert.match(detail, /different and larger problem/);
});

test('with no source the script declines to guess', () => {
  const [state, detail] = verdict('set', { sha256: true, sha1: true }, null);
  assert.equal(state, 'not-scanned');
  assert.match(detail, /not visible from the API/);
});

test('finding nothing is not reported as finding a problem', () => {
  const [state, detail] = verdict('set', { sha256: true, sha1: true }, 'none');
  assert.equal(state, 'no-verification-found');
  assert.match(detail, /at runtime/);
});

test('a correct receiver passes', () => {
  const [state] = verdict('set', { sha256: true, sha1: true }, 'sha256-only');
  assert.equal(state, 'sha256-only');
  assert.ok(repair(state).startsWith('nothing'));
});