check_quotes.py: add --coverage tracking; add devtools/bolt-coverage.py.
What changed, and why it matters
This commit adds two developer-only tools that track whether the project's source-code comments correctly quote the BOLT (Bitcoin Lightning specification) documents. It does not change any network, wallet, or consensus code, and it introduces no security-relevant behavior in the running node.
No security action required; this is a normal development-tooling enhancement.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The patch extends devtools/check_quotes.py with an optional –coverage=FILE flag that appends a coverage record for each successful BOLT quote match, and adds devtools/bolt-coverage.py to report gaps in BOLT text not covered by source comments. The changes are confined to build/dev tooling, use atomic os.write() for parallel-safe append, and only read/write local files specified by the caller. No runtime code, cryptography, or protocol handling is modified.
Changed components
devtools/check_quotes.pydevtools/bolt-coverage.pyInspect captured patch +349 / −25
diff --git a/devtools/bolt-coverage.py b/devtools/bolt-coverage.py
new file mode 100755
index 00000000..841169f9
--- /dev/null
+++ b/devtools/bolt-coverage.py
@@ -0,0 +1,278 @@
+#!/usr/bin/env python3
+"""Report BOLT text (by default, Requirements sections) not quoted by any source comment.
+
+Coverage data is produced by check_quotes.py --coverage=FILE; this tool reads
+that file and highlights text that no comment quotes, showing adjacent quoted
+text for context.
+
+Output is in compiler-error format so Emacs/Vim can navigate directly:
+ src/foo.c:13:...text covered just before the gap...
+ .tmp.lightningrfc/02-peer-protocol.md:5:uncovered text here
+ src/bar.c:99:text covered just after the gap...
+
+Usage example:
+ make check-requirements-coverage
+"""
+
+import glob
+import re
+import sys
+from argparse import ArgumentParser
+from collections import defaultdict
+from typing import Dict, List, Tuple
+
+whitespace_re = re.compile(r"\s+")
+
+
+def collapse_whitespace(string: str) -> str:
+ return whitespace_re.sub(" ", string)
+
+
+def collapse_with_linemap(raw_lines: List[Tuple[int, str]]) -> Tuple[str, List[int]]:
+ """Collapse whitespace across a list of (lineno, text) pairs.
+
+ Returns (collapsed_text, linemap) where linemap[i] is the original line
+ number for character i in collapsed_text. Matches the behaviour of
+ collapse_whitespace() so positions are compatible with coverage records.
+ """
+ result: List[str] = []
+ linemap: List[int] = []
+ in_ws = False
+ ws_lineno = 1
+
+ for lineno, line in raw_lines:
+ for ch in line:
+ if whitespace_re.match(ch):
+ if not in_ws:
+ in_ws = True
+ ws_lineno = lineno
+ else:
+ if in_ws:
+ result.append(" ")
+ linemap.append(ws_lineno)
+ in_ws = False
+ result.append(ch)
+ linemap.append(lineno)
+
+ if in_ws:
+ result.append(" ")
+ linemap.append(ws_lineno)
+
+ return "".join(result), linemap
+
+
+def load_bolt(boltdir: str, num: int) -> Tuple[str, List[Tuple[str, List[int]]]]:
+ """Load a BOLT file, split into sections.
+
+ Returns (boltpath, sections) where sections is a list of
+ (collapsed_text, linemap) pairs; linemap[i] is the original line number
+ for collapsed_text[i].
+ """
+ boltfile = glob.glob("{}/{}-*md".format(boltdir, str(num).zfill(2)))
+ if not boltfile:
+ print("Cannot find bolt {} in {}".format(num, boltdir), file=sys.stderr)
+ sys.exit(1)
+ if len(boltfile) > 1:
+ print("More than one bolt {} in {}? {}".format(num, boltdir, boltfile),
+ file=sys.stderr)
+ sys.exit(1)
+
+ boltpath = boltfile[0]
+ with open(boltpath) as f:
+ raw = list(enumerate(f.readlines(), 1)) # [(lineno, line), ...]
+
+ # Split into sections on lines that start with '#'.
+ raw_sections: List[List[Tuple[int, str]]] = []
+ cur: List[Tuple[int, str]] = []
+ for lineno, line in raw:
+ if line.startswith("#"):
+ raw_sections.append(cur)
+ cur = []
+ cur.append((lineno, line))
+ raw_sections.append(cur)
+
+ sections = [collapse_with_linemap(s) for s in raw_sections]
+ return boltpath, sections
+
+
+def is_requirements_section(text: str) -> bool:
+ """True if the section's header names a Requirements section."""
+ return bool(re.match(r"#+\s+requirements\b", text.lstrip(), re.IGNORECASE))
+
+
+# Coverage record: (section_idx, start, end, src_file, src_line)
+CovRecord = Tuple[int, int, int, str, int]
+
+
+def load_coverage(coverage_file: str) -> Dict[int, List[CovRecord]]:
+ """Return {bolt: [CovRecord, ...]} from the coverage file."""
+ coverage: Dict[int, List[CovRecord]] = defaultdict(list)
+ try:
+ with open(coverage_file) as f:
+ for lineno, line in enumerate(f, 1):
+ line = line.strip()
+ if not line:
+ continue
+ parts = line.split()
+ if len(parts) != 6:
+ print("{}:{}: bad coverage record (expected 6 fields): {!r}".format(
+ coverage_file, lineno, line), file=sys.stderr)
+ continue
+ bolt = int(parts[0])
+ si, start, end, src_line = int(parts[1]), int(parts[2]), int(parts[3]), int(parts[5])
+ src_file = parts[4]
+ coverage[bolt].append((si, start, end, src_file, src_line))
+ except FileNotFoundError:
+ print("Coverage file not found: {}".format(coverage_file), file=sys.stderr)
+ sys.exit(1)
+ return coverage
+
+
+def merge_intervals(intervals: List[Tuple[int, int]]) -> List[Tuple[int, int]]:
+ """Merge overlapping/adjacent [start, end) intervals."""
+ merged: List[List[int]] = []
+ for start, end in sorted(intervals):
+ if merged and start <= merged[-1][1]:
+ merged[-1][1] = max(merged[-1][1], end)
+ else:
+ merged.append([start, end])
+ return [(s, e) for s, e in merged]
+
+
+def section_content_start(text: str) -> int:
+ """Offset past the leading '### SectionTitle ' header."""
+ m = re.match(r"(#+\s+\S+\s+)", text.lstrip())
+ return m.end() if m else 0
+
+
+def adjacent_before(records: List[CovRecord], gap_start: int) -> List[CovRecord]:
+ """All records whose covered range ends closest to (but not after) gap_start."""
+ candidates = [r for r in records if r[2] <= gap_start]
+ if not candidates:
+ return []
+ max_end = max(r[2] for r in candidates)
+ return [r for r in candidates if r[2] == max_end]
+
+
+def adjacent_after(records: List[CovRecord], gap_end: int) -> List[CovRecord]:
+ """All records whose covered range starts closest to (but not before) gap_end."""
+ candidates = [r for r in records if r[1] >= gap_end]
+ if not candidates:
+ return []
+ min_start = min(r[1] for r in candidates)
+ return [r for r in candidates if r[1] == min_start]
+
+
+def snippet(text: str, start: int, end: int, tail: bool = False, maxlen: int = 60) -> str:
+ """Return a short excerpt of text[start:end], from the tail or head."""
+ s = text[start:end].strip()
+ if tail:
+ return ("..." + s[-maxlen:]) if len(s) > maxlen else ("..." + s)
+ return (s[:maxlen] + "...") if len(s) > maxlen else (s + "...")
+
+
+def show_gaps(boltpath: str, bolt_num: int, si: int,
+ text: str, linemap: List[int],
+ section_records: List[CovRecord]) -> bool:
+ """Print uncovered gaps in text with adjacent-mention context.
+
+ Returns True if any uncovered text was found.
+ """
+ content_start = section_content_start(text)
+
+ # Build merged intervals to find gaps, but keep raw records for adjacency.
+ merged = merge_intervals([(r[1], r[2]) for r in section_records])
+
+ # Walk the merged intervals and find gaps.
+ any_gap = False
+ pos = content_start
+ for m_start, m_end in merged:
+ if m_start > pos:
+ gap_text = text[pos:m_start].strip()
+ if not gap_text:
+ pos = max(pos, m_end)
+ continue
+
+ any_gap = True
+ bolt_lineno = linemap[pos] if pos < len(linemap) else 1
+
+ before = adjacent_before(section_records, pos)
+ after = adjacent_after(section_records, m_start)
+
+ for r in before:
+ print("{}:{}:{}".format(r[3], r[4], snippet(text, r[1], r[2], tail=True)))
+ print("{}:{}:{}".format(boltpath, bolt_lineno, gap_text[:120]))
+ for r in after:
+ print("{}:{}:{}".format(r[3], r[4], snippet(text, r[1], r[2], tail=False)))
+
+ pos = max(pos, m_end)
+
+ # Tail gap after last interval.
+ if pos < len(text):
+ gap_text = text[pos:].strip()
+ if gap_text:
+ any_gap = True
+ bolt_lineno = linemap[pos] if pos < len(linemap) else 1
+ before = adjacent_before(section_records, pos)
+
+ for r in before:
+ print("{}:{}:{}".format(r[3], r[4], snippet(text, r[1], r[2], tail=True)))
+ print("{}:{}:{}".format(boltpath, bolt_lineno, gap_text[:120]))
+
+ return any_gap
+
+
+def main() -> None:
+ parser = ArgumentParser(
+ description="Show BOLT text not covered by any source code comment"
+ )
+ parser.add_argument(
+ "--boltdir", default="../lightning-rfc",
+ help="Directory containing BOLT spec files (default: ../lightning-rfc)"
+ )
+ parser.add_argument(
+ "--coverage", required=True, metavar="FILE",
+ help="Coverage file produced by check_quotes.py --coverage=FILE"
+ )
+ parser.add_argument(
+ "--bolt", type=int, action="append", dest="bolts", metavar="N",
+ help="Restrict to BOLT #N (may be repeated; default: all bolts in coverage file)"
+ )
+ parser.add_argument(
+ "--all-sections", action="store_true",
+ help="Check all sections, not just Requirements sections"
+ )
+ args = parser.parse_args()
+
+ coverage = load_coverage(args.coverage)
+ bolts_to_check = sorted(args.bolts if args.bolts else coverage.keys())
+
+ any_uncovered = False
+ for bolt_num in bolts_to_check:
+ try:
+ boltpath, sections = load_bolt(args.boltdir, bolt_num)
+ except SystemExit:
+ continue
+
+ # Group records by section index.
+ by_section: Dict[int, List[CovRecord]] = defaultdict(list)
+ for rec in coverage.get(bolt_num, []):
+ si = rec[0]
+ if 0 <= si < len(sections):
+ by_section[si].append(rec)
+
+ for si, (text, linemap) in enumerate(sections):
+ if not text.strip():
+ continue
+ if not args.all_sections and not is_requirements_section(text):
+ continue
+
+ records = by_section.get(si, [])
+ if show_gaps(boltpath, bolt_num, si, text, linemap, records):
+ any_uncovered = True
+
+ sys.exit(1 if any_uncovered else 0)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/devtools/check_quotes.py b/devtools/check_quotes.py
index c644a214..9e847b9c 100755
--- a/devtools/check_quotes.py
+++ b/devtools/check_quotes.py
@@ -1,6 +1,7 @@
#! /usr/bin/python3
import fileinput
import glob
+import os
import re
import sys
from argparse import ArgumentParser, REMAINDER, Namespace
@@ -157,17 +158,22 @@ def load_bolt(boltdir: str, num: int) -> List[str]:
def find_quote(
text: str, boltsections: List[str]
-) -> Tuple[Optional[str], Optional[int]]:
- # '...' means "match anything, but prefer within a single section".
- # When a part is not found in the current section, we try subsequent
- # sections so that quotes can explicitly span a section header.
+) -> Tuple[int, int, int]:
+ """Search for text (with '...' wildcards) across boltsections.
+
+ Returns (section_idx, start, end) of the match, or (-1, 0, 0) on failure.
+ When a '...' part is not found in the current section we try subsequent
+ sections, so quotes can explicitly span a section header using '*...'.
+ For a cross-section match the start is credited as 0 in the final section.
+ """
textparts = text.split("...")
for start_si, start_b in enumerate(boltsections):
cur_si = start_si
cur_section = start_b
off = 0
+ match_start = -1
success = True
- for part in textparts:
+ for i, part in enumerate(textparts):
new_off = cur_section.find(part, off)
if new_off == -1:
# Try subsequent sections; strip leading whitespace since we're
@@ -179,6 +185,8 @@ def find_quote(
if new_off != -1:
cur_si = next_si
cur_section = boltsections[next_si]
+ # Cross-section: credit coverage from start of this section.
+ match_start = 0
off = new_off + len(search_part)
found = True
break
@@ -186,18 +194,22 @@ def find_quote(
success = False
break
else:
+ if i == 0 and match_start < 0:
+ match_start = new_off
off = new_off + len(part)
if success:
- return cur_section, off
- return None, None
+ return cur_si, (match_start if match_start >= 0 else 0), off
+ return -1, 0, 0
def find_quote_immediate(
text: str, section: str, start: int
-) -> Tuple[Optional[str], Optional[int]]:
+) -> Tuple[int, int]:
"""Find text in section starting immediately at position start.
- Allows a single space separator (whitespace is already collapsed to single spaces).
- The text may still contain '...' wildcards for its own internal matching.
+
+ Allows a single space separator (whitespace is already collapsed to single
+ spaces). The text may still contain '...' wildcards for its own internal
+ matching. Returns (match_start, end) or (-1, -1) on failure.
"""
textparts = text.split("...")
off = start
@@ -206,16 +218,33 @@ def find_quote_immediate(
# continuation line joining), so strip both together.
if off < len(section) and section[off] == " ":
off += 1
+ match_start = off
first_part = textparts[0].lstrip(" ")
if not section[off:].startswith(first_part):
- return None, None
+ return -1, -1
off += len(first_part)
for part in textparts[1:]:
new_off = section.find(part, off)
if new_off == -1:
- return None, None
+ return -1, -1
off = new_off + len(part)
- return section, off
+ return match_start, off
+
+
+def write_coverage(filename: str, bolt: int, section_idx: int, start: int, end: int,
+ src_file: str, src_line: int) -> None:
+ """Atomically append one coverage record to filename.
+
+ Each record is a single line '{bolt} {si} {start} {end} {src_file} {src_line}\\n',
+ written via a single os.write() call so parallel invocations don't
+ interleave partial lines.
+ """
+ record = "{} {} {} {} {} {}\n".format(bolt, section_idx, start, end, src_file, src_line).encode()
+ fd = os.open(filename, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o666)
+ try:
+ os.write(fd, record)
+ finally:
+ os.close(fd)
def main(args: Namespace) -> None:
@@ -224,12 +253,14 @@ def main(args: Namespace) -> None:
for bolt in boltquotes:
boltsections = load_bolt(args.boltdir, bolt)
last_section: Optional[str] = None
+ last_section_idx: int = -1
last_end: int = 0
last_filename: Optional[str] = None
for quote in boltquotes[bolt]:
# Reset per-file tracking when the file changes.
if quote.filename != last_filename:
last_section = None
+ last_section_idx = -1
last_end = 0
last_filename = quote.filename
@@ -246,11 +277,12 @@ def main(args: Namespace) -> None:
if not args.keep_going:
sys.exit(1)
failed = True
- sect, end = None, None
+ sect, istart, end = None, -1, 0
else:
text_after = quote.text[3:]
- sect, end = find_quote_immediate(text_after, last_section, last_end)
- if sect is None:
+ istart, end = find_quote_immediate(text_after, last_section, last_end)
+ if istart < 0:
+ sect = None
print(
"{}:{}:cannot find match (must immediately follow previous quote)".format(
quote.filename, quote.line
@@ -270,8 +302,14 @@ def main(args: Namespace) -> None:
if not args.keep_going:
sys.exit(1)
failed = True
+ else:
+ sect = last_section
+ if args.coverage:
+ write_coverage(args.coverage, bolt, last_section_idx, istart, end,
+ quote.filename, quote.line)
else:
- sect, end = find_quote(quote.text, boltsections)
+ si, start, end = find_quote(quote.text, boltsections)
+ sect = boltsections[si] if si >= 0 else None
if sect is None:
print(
"{}:{}:cannot find match".format(quote.filename, quote.line),
@@ -279,14 +317,15 @@ def main(args: Namespace) -> None:
)
# Reduce the text until we find a match.
for n in range(len(quote.text), -1, -1):
- sect, end = find_quote(quote.text[:n], boltsections)
- if sect:
+ si2, _, end2 = find_quote(quote.text[:n], boltsections)
+ if si2 >= 0:
+ s2 = boltsections[si2]
print(
" common prefix: {}...".format(quote.text[:n]),
file=sys.stderr,
)
print(
- " expected ...{:.45}".format(sect[end:]), file=sys.stderr
+ " expected ...{:.45}".format(s2[end2:]), file=sys.stderr
)
print(
" but have ...{:.45}".format(quote.text[n:]),
@@ -296,12 +335,17 @@ def main(args: Namespace) -> None:
if not args.keep_going:
sys.exit(1)
failed = True
- elif args.verbose:
- print(
- "{}:{}:Matched {} in {}".format(
- quote.filename, quote.line, quote.text, sect
+ else:
+ if args.coverage:
+ write_coverage(args.coverage, bolt, si, start, end,
+ quote.filename, quote.line)
+ if args.verbose:
+ print(
+ "{}:{}:Matched {} in {}".format(
+ quote.filename, quote.line, quote.text, sect
+ )
)
- )
+ last_section_idx = si
if sect is not None:
last_section = sect
@@ -318,6 +362,8 @@ if __name__ == "__main__":
parser.add_argument("-v", "--verbose", action="store_true")
parser.add_argument("-k", "--keep-going", action="store_true",
help="Report all errors instead of stopping at first")
+ parser.add_argument("--coverage", metavar="FILE",
+ help="Append coverage records to FILE (bolt section_idx start end)")
# e.g. for C code these are '/* ', '*' and '*/'
parser.add_argument(
"--comment-start", help='marker for start of "BOLT #N" quote', default="# "
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.