diff --git a/Makefile b/Makefile index d99a327f..f2502b15 100644 --- a/Makefile +++ b/Makefile @@ -57,6 +57,7 @@ help: @printf "\n" @printf " $(BOLD)make init$(RESET) Create/update repo-local virtualenv\n" @printf " $(BOLD)make phase-1$(RESET) Run reconnaissance\n" + @printf " $(BOLD)ID_SCHEME=CC-0142$(RESET) First finding id for this audit\n" @printf " $(BOLD)make phase-2$(RESET) Run hypothesis generation\n" @printf " $(BOLD)make phase-3$(RESET) Run counter-analysis\n" @printf " $(BOLD)make phase-4 FINDING=CC-0001$(RESET) Validate one finding\n" @@ -188,6 +189,7 @@ env-check: # --------------------------------------------------------------------------- phase-1: env-check + @$(PYTHON) tools/record-id-scheme.py $(if $(ID_SCHEME),--scheme "$(ID_SCHEME)") $(if $(ID_START),--start $(ID_START)) $(if $(ID_LEDGER),--ledger "$(ID_LEDGER)") @$(PYTHON) tools/gate-check.py 1 @$(PYTHON) tools/run-agent.py --phase 1 --label "Phase 1: Reconnaissance" --agent recon diff --git a/docs/workflow.md b/docs/workflow.md index 53b1b74d..f67356aa 100644 --- a/docs/workflow.md +++ b/docs/workflow.md @@ -37,6 +37,15 @@ Run: make phase-1 +To line finding ids up with an internal tracker, pass the first id you want +when reconnaissance starts. Later findings continue from that number instead +of minting another `CC-0001`. The request is saved in +`itemdb/notes/id-scheme.yml` and reused by `make findings-create` and +`make next-id`. + + make phase-1 ID_SCHEME=CC-0142 + make phase-1 ID_SCHEME=INT-0007 ID_LEDGER=$HOME/.codecome/finding-ids.json + `make phase-1` is the recommended automatic path (it runs Phase 1a, CodeQL enrichment, Phase 1b, and Phase 1c). For manual invocation, use the individual subphase prompts: opencode run --agent recon "$(cat prompts/phase-1a-profile.md)" diff --git a/tests/test_id_scheme.py b/tests/test_id_scheme.py new file mode 100644 index 00000000..2e70adce --- /dev/null +++ b/tests/test_id_scheme.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import sys +from argparse import Namespace +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "tools")) + +from findings.constants import FindingsContext # noqa: E402 +from findings.create import create_finding # noqa: E402 +from findings.ids import next_finding_id # noqa: E402 +from findings.scheme import ( # noqa: E402 + IdSchemeError, + allocate_next_id, + load_id_scheme, + parse_id_scheme, + save_id_scheme, +) + + +def _workspace(tmp_path: Path) -> FindingsContext: + notes = tmp_path / "itemdb" / "notes" + findings = tmp_path / "itemdb" / "findings" + evidence = tmp_path / "itemdb" / "evidence" + template_dir = tmp_path / "templates" + notes.mkdir(parents=True) + (findings / "PENDING").mkdir(parents=True) + evidence.mkdir(parents=True) + template_dir.mkdir(parents=True) + (template_dir / "finding.md").write_text( + "---\n" + 'id: "CC-0000"\n' + 'title: "Short vulnerability title"\n' + 'status: "PENDING"\n' + 'severity: "MEDIUM"\n' + 'confidence: "LOW"\n' + 'category: "Unclassified"\n' + 'language: "unknown"\n' + 'target_area: "unknown"\n' + "validation:\n" + ' evidence_dir: "itemdb/evidence/CC-0000"\n' + "exploitation:\n" + ' artifacts_dir: "itemdb/evidence/CC-0000/exploits"\n' + 'created_at: "YYYY-MM-DD"\n' + 'updated_at: "YYYY-MM-DD"\n' + "---\n\n" + "Briefly describe the suspected vulnerability.\n", + encoding="utf-8", + ) + return FindingsContext( + root=tmp_path, + findings_root=findings, + evidence_root=evidence, + notes_root=notes, + template_path=template_dir / "finding.md", + evidence_template_path=template_dir / "evidence-readme.md", + ) + + +def test_parse_specimen_sets_width_and_start(): + scheme = parse_id_scheme("CC-0142") + assert scheme.prefix == "CC" + assert scheme.width == 4 + assert scheme.start == 142 + assert scheme.format(142) == "CC-0142" + + +def test_parse_rejects_free_form_text(): + with pytest.raises(IdSchemeError): + parse_id_scheme("start at one") + + +def test_default_numbering_still_starts_at_cc_0001(tmp_path): + ctx = _workspace(tmp_path) + assert next_finding_id(findings_root=ctx.findings_root, notes_root=ctx.notes_root) == "CC-0001" + + +def test_phase1_scheme_is_reused_by_the_next_finding(tmp_path): + ctx = _workspace(tmp_path) + save_id_scheme(parse_id_scheme("CC-0142"), ctx.notes_root) + assert next_finding_id(findings_root=ctx.findings_root, notes_root=ctx.notes_root) == "CC-0142" + + pending = ctx.findings_root / "PENDING" + (pending / "CC-0142-first.md").write_text("id: CC-0142\n", encoding="utf-8") + assert next_finding_id(findings_root=ctx.findings_root, notes_root=ctx.notes_root) == "CC-0143" + + +def test_requested_start_does_not_reuse_a_taken_number(tmp_path): + ctx = _workspace(tmp_path) + pending = ctx.findings_root / "PENDING" + (pending / "CC-0001-old.md").write_text("id: CC-0001\n", encoding="utf-8") + (pending / "CC-0002-old.md").write_text("id: CC-0002\n", encoding="utf-8") + save_id_scheme(parse_id_scheme("CC-0001"), ctx.notes_root) + assert next_finding_id(findings_root=ctx.findings_root, notes_root=ctx.notes_root) == "CC-0003" + + +def test_other_prefix_does_not_collide_with_cc(tmp_path): + ctx = _workspace(tmp_path) + pending = ctx.findings_root / "PENDING" + (pending / "CC-0001-old.md").write_text("id: CC-0001\n", encoding="utf-8") + save_id_scheme(parse_id_scheme("INT-0007"), ctx.notes_root) + assert next_finding_id(findings_root=ctx.findings_root, notes_root=ctx.notes_root) == "INT-0007" + + +def test_ledger_continues_across_audits(tmp_path): + ledger = tmp_path / "finding-ids.json" + first = parse_id_scheme("CC-0001") + first = type(first)(first.prefix, first.width, first.start, ledger) + assert allocate_next_id(first, []) == "CC-0001" + second_notes = tmp_path / "other" / "notes" + second_notes.mkdir(parents=True) + save_id_scheme(first, second_notes) + loaded = load_id_scheme(second_notes) + assert loaded is not None + assert allocate_next_id(loaded, []) == "CC-0002" + + +def test_create_finding_uses_saved_scheme(tmp_path): + ctx = _workspace(tmp_path) + save_id_scheme(parse_id_scheme("CC-0142"), ctx.notes_root) + out = create_finding( + Namespace( + title="Missing auth check", + id=None, + slug=None, + severity="HIGH", + confidence="MEDIUM", + category="Auth", + language="python", + target_area="api", + force=False, + ), + ctx=ctx, + ) + assert out.name.startswith("CC-0142-") + assert 'id: "CC-0142"' in out.read_text(encoding="utf-8") diff --git a/tools/codecome.py b/tools/codecome.py index 15a426b0..a9784956 100755 --- a/tools/codecome.py +++ b/tools/codecome.py @@ -690,9 +690,9 @@ def command_status(_: argparse.Namespace) -> int: def command_next_id(_: argparse.Namespace) -> int: - ids = collect_finding_ids(iter_finding_files()) - next_id = max(ids, default=0) + 1 - print(f"CC-{next_id:04d}") + from findings.ids import next_finding_id + + print(next_finding_id()) return 0 diff --git a/tools/findings/create.py b/tools/findings/create.py index 0d3b08b4..cd7f18f9 100644 --- a/tools/findings/create.py +++ b/tools/findings/create.py @@ -20,6 +20,7 @@ ) from findings.constants import FindingsContext from findings.ids import next_finding_id, slugify +from findings.scheme import load_id_scheme from findings.frontmatter import replace_scalar_frontmatter, replace_nested_value @@ -39,10 +40,20 @@ def create_finding( if not ctx.template_path.exists(): raise FileNotFoundError(f"Template not found: {ctx.template_path}") - finding_id = args.id or next_finding_id(findings_root=ctx.findings_root) - - if args.id and not FINDING_ID_FORMAT_RE.fullmatch(args.id): - raise ValueError(f"Invalid finding id format: {args.id!r} (expected CC-NNNN)") + scheme = load_id_scheme(ctx.notes_root) + if args.id: + if scheme is not None and not scheme.matches(args.id): + raise ValueError( + f"Invalid finding id format: {args.id!r} " + f"(expected {scheme.prefix}- and {scheme.width} digits)" + ) + if scheme is None and not FINDING_ID_FORMAT_RE.fullmatch(args.id): + raise ValueError(f"Invalid finding id format: {args.id!r} (expected CC-NNNN)") + + finding_id = args.id or next_finding_id( + findings_root=ctx.findings_root, + notes_root=ctx.notes_root, + ) slug = args.slug or slugify(args.title) today = date.today().isoformat() diff --git a/tools/findings/ids.py b/tools/findings/ids.py index c2f32ea8..38057463 100644 --- a/tools/findings/ids.py +++ b/tools/findings/ids.py @@ -7,7 +7,8 @@ from pathlib import Path from typing import Iterable, List, Optional -from findings.constants import FINDINGS_ROOT, FINDING_ID_RE, FINDING_ID_STRICT_RE, ROOT, STATUSES +from findings.constants import FINDINGS_ROOT, FINDING_ID_RE, FINDING_ID_STRICT_RE, NOTES_ROOT, ROOT, STATUSES +from findings.scheme import IdScheme, allocate_next_id, load_id_scheme def extract_id_from_path(path: Path) -> str: @@ -25,17 +26,27 @@ def slugify(value: str) -> str: def iter_finding_files( *, findings_root: Optional[Path] = None, + notes_root: Optional[Path] = None, ) -> List[Path]: """Recursive CC-*.md glob, sorted. Used by create (next_id), checks.""" findings_root = findings_root if findings_root is not None else FINDINGS_ROOT + notes = notes_root if notes_root is not None else notes_root_for(findings_root) files: List[Path] = [] if not findings_root.exists(): return files - for path in findings_root.rglob("CC-*.md"): - if path.is_file(): - files.append(path) + patterns = ["CC-*.md"] + scheme = _scheme_for(notes) + if scheme is not None and scheme.prefix != "CC": + patterns.append(f"{scheme.prefix}-*.md") + + seen: set[Path] = set() + for pattern in patterns: + for path in findings_root.rglob(pattern): + if path.is_file() and path not in seen: + seen.add(path) + files.append(path) return sorted(files) @@ -56,22 +67,56 @@ def iter_findings( if not status_dir.exists(): continue - yield from sorted(status_dir.glob("CC-*.md")) + patterns = ["CC-*.md"] + scheme = _scheme_for(notes_root_for(findings_root)) + if scheme is not None and scheme.prefix != "CC": + patterns.append(f"{scheme.prefix}-*.md") + for pattern in patterns: + yield from sorted(status_dir.glob(pattern)) + + +def notes_root_for(findings_root: Optional[Path]) -> Path: + if findings_root is None: + return NOTES_ROOT + return findings_root.parent / "notes" + + +def _scheme_for(notes_root: Optional[Path]) -> Optional[IdScheme]: + return load_id_scheme(notes_root) + + +def issued_numbers( + scheme: IdScheme, + *, + findings_root: Optional[Path] = None, + notes_root: Optional[Path] = None, +) -> List[int]: + numbers: List[int] = [] + pattern = scheme.number_re() + notes = notes_root if notes_root is not None else notes_root_for(findings_root) + for path in iter_finding_files(findings_root=findings_root, notes_root=notes): + match = pattern.search(path.name) + if match: + numbers.append(int(match.group(1))) + return numbers def next_finding_id( *, findings_root: Optional[Path] = None, + notes_root: Optional[Path] = None, ) -> str: - ids: List[int] = [] - - for path in iter_finding_files(findings_root=findings_root): - match = FINDING_ID_RE.search(path.name) - if match: - ids.append(int(match.group(1))) + notes = notes_root if notes_root is not None else notes_root_for(findings_root) + scheme = load_id_scheme(notes) + if scheme is None: + ids: List[int] = [] + for path in iter_finding_files(findings_root=findings_root, notes_root=notes): + match = FINDING_ID_RE.search(path.name) + if match: + ids.append(int(match.group(1))) + return f"CC-{max(ids, default=0) + 1:04d}" - next_id = max(ids, default=0) + 1 - return f"CC-{next_id:04d}" + return allocate_next_id(scheme, issued_numbers(scheme, findings_root=findings_root, notes_root=notes)) def find_finding( @@ -87,7 +132,8 @@ def find_finding( if candidate.exists(): return candidate.resolve() - if not FINDING_ID_STRICT_RE.fullmatch(identifier): + scheme = _scheme_for(notes_root_for(findings_root)) + if not FINDING_ID_STRICT_RE.fullmatch(identifier) and not (scheme and scheme.matches(identifier)): raise FileNotFoundError(f"Invalid finding id or path: {identifier}") matches = sorted(findings_root.rglob(f"{identifier}-*.md")) diff --git a/tools/findings/package.py b/tools/findings/package.py index d2d09a5c..b6f82665 100644 --- a/tools/findings/package.py +++ b/tools/findings/package.py @@ -12,10 +12,14 @@ import _colors as C from findings.constants import FINDING_ID_STRICT_RE, ROOT, FindingsContext +from findings.scheme import load_id_scheme -def validate_finding_id(finding_id: str) -> str: +def validate_finding_id(finding_id: str, *, notes_root: Optional[Path] = None) -> str: stripped = finding_id.strip() + scheme = load_id_scheme(notes_root) + if scheme is not None and scheme.matches(stripped): + return stripped if not FINDING_ID_STRICT_RE.fullmatch(stripped): raise ValueError(f"Invalid finding id format: {finding_id!r} (expected CC-NNNN)") return stripped diff --git a/tools/findings/scheme.py b/tools/findings/scheme.py new file mode 100644 index 00000000..0cb03e8b --- /dev/null +++ b/tools/findings/scheme.py @@ -0,0 +1,227 @@ +# Copyright (C) 2025-2026 Pablo Ruiz García +# SPDX-License-Identifier: GPL-3.0-or-later OR AGPL-3.0-or-later +"""Finding-id scheme requested when reconnaissance starts. + +Pass the first id you want (``CC-0142``, ``INT-0007``) to ``make phase-1``. +The request is stored in ``itemdb/notes/id-scheme.yml``. Later findings +continue from that sequence instead of minting another ``CC-0001``. +""" + +from __future__ import annotations + +import fcntl +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + +import yaml + +from findings.constants import NOTES_ROOT + +STAMP_NAME = "id-scheme.yml" + +_SPECIMEN_RE = re.compile(r"^([A-Za-z][A-Za-z0-9]{0,15})-(\d{1,8})$") + + +class IdSchemeError(ValueError): + """The requested scheme cannot be used.""" + + +@dataclass(frozen=True) +class IdScheme: + prefix: str + width: int + start: int + ledger: Optional[Path] = None + + @property + def template(self) -> str: + return f"{self.prefix}-{{n:0{self.width}d}}" + + def format(self, number: int) -> str: + if number < 0 or number >= 10 ** self.width: + raise IdSchemeError( + f"{number} does not fit in {self.prefix}-{'0' * self.width} " + f"({self.width} digits)." + ) + return f"{self.prefix}-{number:0{self.width}d}" + + def matches(self, finding_id: str) -> bool: + return bool(re.fullmatch(rf"{re.escape(self.prefix)}-\d{{{self.width}}}", finding_id)) + + def number_re(self) -> re.Pattern[str]: + return re.compile(rf"\b{re.escape(self.prefix)}-(\d{{{self.width}}})\b") + + +def stamp_path(notes_root: Optional[Path] = None) -> Path: + root = notes_root if notes_root is not None else NOTES_ROOT + return root / STAMP_NAME + + +def parse_id_scheme(spec: str, start_override: Optional[int] = None) -> IdScheme: + """Parse ``CC-0142`` or ``INT-0007``. + + The digits set the width and the requested start. ``--start`` replaces + the number and keeps the width. + """ + text = (spec or "").strip() + match = _SPECIMEN_RE.fullmatch(text) + if not match: + raise IdSchemeError( + "ID_SCHEME must be the first id you want, such as CC-0142 or " + f"INT-0007. Got {spec!r}." + ) + prefix, digits = match.group(1), match.group(2) + width = len(digits) + start = int(digits) + if start_override is not None: + if start_override < 0: + raise IdSchemeError("--start must be zero or greater.") + start = start_override + scheme = IdScheme(prefix=prefix, width=width, start=start) + scheme.format(start) + return scheme + + +def load_id_scheme(notes_root: Optional[Path] = None) -> Optional[IdScheme]: + path = stamp_path(notes_root) + if not path.exists(): + return None + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError) as exc: + raise IdSchemeError(f"could not read {path}: {exc}") from exc + if not isinstance(data, dict) or not isinstance(data.get("specimen"), str): + raise IdSchemeError(f"{path} must contain a specimen id.") + start = data.get("start") + scheme = parse_id_scheme(data["specimen"], start if isinstance(start, int) else None) + ledger = data.get("ledger") + if ledger: + scheme = IdScheme( + prefix=scheme.prefix, + width=scheme.width, + start=scheme.start, + ledger=Path(str(ledger)), + ) + return scheme + + +def save_id_scheme(scheme: IdScheme, notes_root: Optional[Path] = None) -> Path: + path = stamp_path(notes_root) + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "specimen": scheme.format(scheme.start), + "prefix": scheme.prefix, + "width": scheme.width, + "start": scheme.start, + "ledger": str(scheme.ledger) if scheme.ledger else None, + } + path.write_text(yaml.safe_dump(payload, sort_keys=False), encoding="utf-8") + return path + + +def _with_ledger(scheme: IdScheme, ledger: Optional[Path]) -> IdScheme: + if ledger is None: + return scheme + return IdScheme(scheme.prefix, scheme.width, scheme.start, Path(ledger)) + + +def preview_next_id(scheme: IdScheme, issued: list[int]) -> tuple[str, Optional[str]]: + """Next id from numbers already on disk. Does not write a ledger.""" + local_next = max(issued, default=scheme.start - 1) + 1 + if scheme.start > local_next: + local_next = scheme.start + chosen = local_next + note = None + if scheme.ledger is not None and scheme.ledger.exists(): + slot = _slot(_read_ledger(scheme.ledger), scheme) + if slot is not None and int(slot["next"]) > chosen: + chosen = int(slot["next"]) + note = ( + f"requested {scheme.format(scheme.start)} is already used; " + f"continuing at {scheme.format(chosen)}" + ) + if issued and scheme.start > max(issued) + 1 and chosen == scheme.start and note is None: + note = ( + f"requested start is ahead of existing findings; " + f"the next id will be {scheme.format(chosen)}" + ) + return scheme.format(chosen), note + + +def allocate_next_id(scheme: IdScheme, issued: list[int]) -> str: + """Issue one id. A ledger, when configured, never repeats a number.""" + local_next = max(issued, default=scheme.start - 1) + 1 + if scheme.start > local_next: + local_next = scheme.start + if scheme.ledger is None: + return scheme.format(local_next) + + path = scheme.ledger + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, "a+", encoding="utf-8") as handle: + fcntl.flock(handle, fcntl.LOCK_EX) + try: + handle.seek(0) + raw = handle.read() + data = json.loads(raw) if raw.strip() else {"series": {}} + if not isinstance(data, dict) or not isinstance(data.get("series", {}), dict): + raise IdSchemeError(f"id ledger {path} must be a JSON object with a series object.") + data.setdefault("series", {}) + slot = _slot(data, scheme) + if slot is None: + slot = {"width": scheme.width, "next": local_next, "issued": []} + data["series"][scheme.prefix] = slot + nxt = max(int(slot["next"]), local_next) + finding_id = scheme.format(nxt) + issued_list = slot.setdefault("issued", []) + if not isinstance(issued_list, list): + raise IdSchemeError(f"id ledger series {scheme.prefix!r} has a corrupt issued list.") + issued_list.append(finding_id) + slot["next"] = nxt + 1 + slot["width"] = scheme.width + handle.seek(0) + handle.truncate() + json.dump(data, handle, indent=2) + handle.write("\n") + handle.flush() + finally: + fcntl.flock(handle, fcntl.LOCK_UN) + return finding_id + + +def _read_ledger(path: Path) -> dict: + try: + raw = path.read_text(encoding="utf-8") + except OSError as exc: + raise IdSchemeError(f"could not read id ledger {path}: {exc}") from exc + if not raw.strip(): + return {"series": {}} + try: + data = json.loads(raw) + except json.JSONDecodeError as exc: + raise IdSchemeError(f"id ledger {path} is not valid JSON: {exc}") from exc + if not isinstance(data, dict) or not isinstance(data.get("series", {}), dict): + raise IdSchemeError(f"id ledger {path} must be a JSON object with a series object.") + data.setdefault("series", {}) + return data + + +def _slot(data: dict, scheme: IdScheme) -> Optional[dict]: + slot = data.get("series", {}).get(scheme.prefix) + if slot is None: + return None + if not isinstance(slot, dict) or "width" not in slot or "next" not in slot: + raise IdSchemeError(f"id ledger series {scheme.prefix!r} is missing width/next.") + if int(slot["width"]) != scheme.width: + raise IdSchemeError( + f"id ledger already numbers {scheme.prefix}- ids with width {slot['width']}, " + f"not {scheme.width}." + ) + return slot + + +def bind_ledger(scheme: IdScheme, ledger: Optional[Path]) -> IdScheme: + return _with_ledger(scheme, ledger) diff --git a/tools/record-id-scheme.py b/tools/record-id-scheme.py new file mode 100644 index 00000000..34855364 --- /dev/null +++ b/tools/record-id-scheme.py @@ -0,0 +1,66 @@ +#!/usr/bin/env python3 +# Copyright (C) 2025-2026 Pablo Ruiz García +# SPDX-License-Identifier: GPL-3.0-or-later OR AGPL-3.0-or-later + +"""Record or show the finding-id scheme for this audit. + +Called at the start of ``make phase-1``. With no arguments, prints the +saved scheme or the default ``CC-0001`` sequence. +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from findings.ids import issued_numbers, next_finding_id +from findings.scheme import ( + IdSchemeError, + bind_ledger, + load_id_scheme, + parse_id_scheme, + preview_next_id, + save_id_scheme, +) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Record a finding-id scheme before reconnaissance.") + parser.add_argument("--scheme", help="First id you want, for example CC-0142 or INT-0007.") + parser.add_argument("--start", type=int, default=None, help="Override the counter in --scheme.") + parser.add_argument("--ledger", type=Path, default=None, help="JSON ledger shared across audits.") + return parser + + +def main() -> int: + args = build_parser().parse_args() + try: + if args.scheme: + scheme = bind_ledger(parse_id_scheme(args.scheme, args.start), args.ledger) + save_id_scheme(scheme) + issued = issued_numbers(scheme) + nxt, note = preview_next_id(scheme, issued) + print(f"finding ids: {scheme.template}") + print(f"next finding will be at least {nxt}") + if note: + print(note) + return 0 + if args.start is not None or args.ledger is not None: + raise IdSchemeError("ID_START and ID_LEDGER require ID_SCHEME.") + saved = load_id_scheme() + if saved is None: + print("finding ids: CC-0001 (default)") + return 0 + print(f"finding ids: {saved.template} (saved at phase 1)") + print(f"next finding will be at least {next_finding_id()}") + return 0 + except IdSchemeError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main())