fix(D): address Sourcery review findings on PR #410

Three review fixes:

1. Security: validate lane/split/version against ^[a-z0-9_]+$ before
   building the runner module name. The runner_args list is passed to
   subprocess.run without shell=True (no shell injection possible),
   but defense-in-depth blocks arbitrary token characters from
   reaching Python's -m module loader. Bad input now errors at the
   CLI boundary with a clear message.

2. Bug-risk: _classify_refusal docstring referenced a
   no_admissible_candidate bucket that the implementation never
   emitted. Aligned docstring with actual buckets
   (no_admissible_question / no_admissible_statement). Also made all
   matching consistently case-insensitive (was mixed — some checks
   used raw reason, one used .lower()).

3. Bug-risk: fetch_committed_baseline wrote to
   .git/coverage_baseline_tmp.json. Replaced with tempfile.mkstemp in
   the system temp dir — avoids (a) failures in non-git worktrees
   where .git is a file pointer, (b) concurrent-access collisions
   between simultaneous operators.

Tests (+3 new):
- test_classify_refusal_is_case_insensitive
- test_classify_docstring_matches_implementation_buckets
- test_fetch_committed_baseline_uses_system_temp

All 16 coverage tests green. Verified the validation:
  core teaching coverage --lane 'evil; rm -rf /'
  → ERROR: lane='evil; rm -rf /' must match ^[a-z0-9_]+$
This commit is contained in:
Shay 2026-05-27 21:06:46 -07:00
parent d91ea3d36e
commit 36f3dbfc4e
3 changed files with 101 additions and 10 deletions

View file

@ -2014,6 +2014,23 @@ def cmd_teaching_coverage(args: argparse.Namespace) -> int:
split = args.split or "train_sample"
version = args.version or "v1"
# Validate inputs against a strict whitelist before any path
# construction or subprocess invocation. The runner module name is
# built from these tokens (``f"evals.{lane}.{split}.{version}.runner"``)
# and passed to ``python -m``. Python's module loader would reject
# most malicious payloads, but a strict whitelist is the defense-in-
# depth response to the Sourcery security advisory: reject
# everything except ``[a-z0-9_]+``.
import re as _re
_safe_token_re = _re.compile(r"^[a-z0-9_]+$")
for label, value in (("lane", lane), ("split", split), ("version", version)):
if not _safe_token_re.match(value):
print(
f"ERROR: {label}={value!r} must match ^[a-z0-9_]+$",
file=sys.stderr,
)
return 1
repo_root = Path(__file__).resolve().parent.parent
lane_dir = repo_root / "evals" / lane / split / version
if not lane_dir.is_dir():

View file

@ -11,6 +11,7 @@ from __future__ import annotations
import json
import re
import subprocess
import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Mapping
@ -61,24 +62,30 @@ class CoverageReport:
def _classify_refusal(reason: str) -> str:
"""Map a per-case refusal reason string to a stable category bucket.
Matching is case-insensitive throughout the raw runner output is
consistently lowercase prose today, but normalizing once avoids
drift if upstream casing changes.
Buckets:
- ``recognizer_empty_injection(<ShapeCategory>)`` recognizer
matched but injector returned empty
- ``no_admissible_candidate`` neither parser nor recognizer admitted
- ``no_admissible_question`` statement(s) admitted; question parser refused
matched but the per-category injector returned empty
- ``no_admissible_question`` statement(s) admitted; question
parser refused
- ``no_admissible_statement`` neither parser nor recognizer
admitted any statement
- ``unexpected_question_count`` !=1 question sentence
- ``other`` any unmatched reason text
"""
if not reason:
return "other"
lower = reason.lower()
if "recognizer matched but produced no injection" in reason:
m = _REASON_CATEGORY_RE.search(reason)
if "recognizer matched but produced no injection" in lower:
m = _REASON_CATEGORY_RE.search(lower)
cat = m.group(1) if m else "unknown"
return f"recognizer_empty_injection({cat})"
if "no admissible candidate for question" in reason:
if "no admissible candidate for question" in lower:
return "no_admissible_question"
if "no admissible candidate for statement" in reason:
if "no admissible candidate for statement" in lower:
return "no_admissible_statement"
if "expected exactly one question sentence" in lower:
return "unexpected_question_count"
@ -171,9 +178,21 @@ def fetch_committed_baseline(
return None
if not result.stdout.strip():
return None
tmp = repo_root / ".git" / "coverage_baseline_tmp.json"
tmp.write_text(result.stdout, encoding="utf-8")
return tmp
# Use the system temp dir with a unique filename to avoid:
# (a) failures in non-git checkouts or worktrees where .git is a
# file pointing elsewhere
# (b) concurrent-access collisions if two operators run
# ``core teaching coverage --delta`` simultaneously
fd, tmp_path = tempfile.mkstemp(
prefix="core_coverage_baseline_", suffix=".json"
)
try:
with open(fd, "w", encoding="utf-8") as fh:
fh.write(result.stdout)
except Exception:
Path(tmp_path).unlink(missing_ok=True)
return None
return Path(tmp_path)
__all__ = [

View file

@ -152,3 +152,58 @@ def test_wrong_zero_invariant_visible_via_as_dict(tmp_path: Path):
_write_report(report_path, correct=0, refused=0, wrong=2, per_case=[])
r = build_coverage_report(report_path, lane="t", split="s", version="v1", use_reader=False)
assert r.as_dict()["counts"]["wrong"] == 2
# ---------------------------------------------------------------------------
# Sourcery review follow-ups (PR #410 round 1)
# ---------------------------------------------------------------------------
def test_classify_refusal_is_case_insensitive():
"""Mixed-case reason text classifies into the same bucket."""
mixed = (
"Candidate_Graph: Recognizer Matched But Produced No Injection "
"for statement: 'X.' (category=currency_amount)"
)
assert _classify_refusal(mixed) == "recognizer_empty_injection(currency_amount)"
def test_classify_docstring_matches_implementation_buckets():
"""The docstring's bucket list matches the values _classify_refusal returns."""
expected_buckets = {
"recognizer_empty_injection",
"no_admissible_question",
"no_admissible_statement",
"unexpected_question_count",
"other",
}
doc = _classify_refusal.__doc__ or ""
for bucket in expected_buckets:
assert bucket in doc, f"docstring missing bucket: {bucket}"
def test_fetch_committed_baseline_uses_system_temp(tmp_path, monkeypatch):
"""The baseline temp file lives in tempfile.gettempdir(), not .git/."""
import tempfile as _tempfile
from teaching.coverage import fetch_committed_baseline
class _FakeCompleted:
def __init__(self, stdout: str) -> None:
self.stdout = stdout
import subprocess as _sp
fake_baseline = '{"counts": {"correct": 1, "refused": 1, "wrong": 0}}'
def _fake_run(*_args, **_kwargs):
return _FakeCompleted(fake_baseline)
monkeypatch.setattr(_sp, "run", _fake_run)
out = fetch_committed_baseline("evals/x/report.json", tmp_path)
assert out is not None
sys_temp = Path(_tempfile.gettempdir()).resolve()
assert str(out.resolve()).startswith(str(sys_temp)), (
f"baseline temp must be under {sys_temp}, got {out!r}"
)
out.unlink(missing_ok=True)