Silence a finding with an inline coretrace ignore comment

docs/usage.md

@@ -139,6 +139,21 @@ sources and in configuration files.
 | `denied-dependency` | A package the policy denies. |
 | `unpinned-dependency` | A requirement without an exact pin when the policy requires pins. |
 
+## Suppressing a finding
+
+A finding whose line carries a `# coretrace: ignore` comment is suppressed. With a list
+of rules, only those are:
+
+```python
+data = yaml.load(text)  # coretrace: ignore[insecure-deserialization]
+```
+
+The comment also works in requirements files, where a line
+`pyyaml==5.3.1  # coretrace: ignore[vulnerable-dependency]` silences that requirement's
+finding. Suppressed findings are kept apart: the text report counts them, the JSON report
+lists them under `suppressed`, the SARIF log marks them as suppressed in source, and
+they never affect the exit status.
+
 ## Reports
 
 `--format` accepts `text`, `json` and `sarif`.

src/coretrace_python/cli.py

@@ -196,6 +196,7 @@ def main(argv: list[str] | None = None) -> int:
                 )
                 findings = analysis.findings
                 coverage = analysis.coverage
+                suppressed = analysis.suppressed
                 if args.sbom is not None:
                     args.sbom.write_text(
                         render_sbom(analysis.dependencies, analysis.advisories, engine.TOOL_NAME, __version__),
@@ -204,8 +205,10 @@ def main(argv: list[str] | None = None) -> int:
             else:
                 file_analysis = engine.analyze_file(SourceManager().load_file(args.path), plugin_roots)
                 findings, coverage = file_analysis.findings, file_analysis.coverage
+                suppressed = file_analysis.suppressed
             root = args.path.resolve() if args.path.is_dir() else args.path.resolve().parent
-            print(render(args.format or "text", engine.report(findings, coverage, root)), end="")
+            rendered = render(args.format or "text", engine.report(findings, coverage, root, suppressed))
+            print(rendered, end="")
             return EXIT_FINDINGS if findings else EXIT_CLEAN
     except _ANALYSIS_ERRORS as error:
         print(f"error: {error}", file=sys.stderr)

src/coretrace_python/engine.py

@@ -8,7 +8,7 @@
 
 import concurrent.futures
 import multiprocessing
-from collections.abc import Iterable, Mapping, Sequence
+from collections.abc import Callable, Iterable, Mapping, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
 from types import MappingProxyType
@@ -58,6 +58,7 @@
     Severity,
 )
 from coretrace_python.findings.refutation import RefutationAnalysis
+from coretrace_python.findings.suppressions import partition
 from coretrace_python.frontend import HIRBuildError, ParseError, build_hir
 from coretrace_python.hir import HIR_SCHEMA_VERSION, nodes
 from coretrace_python.interprocedural import (
@@ -96,7 +97,7 @@
 from coretrace_python.semantic.imports import ImportAnalysis, ImportResolutionError, ImportTable
 from coretrace_python.semantic.scopes import ScopeAnalysis, ScopeError
 from coretrace_python.semantic.symbols import SymbolAnalysis, SymbolId
-from coretrace_python.source import SourceFile, SourceManager, SourceSpan, decode_text
+from coretrace_python.source import SourceFile, SourceId, SourceManager, SourceSpan, decode_text
 from coretrace_python.taint import (
     EntryPoint,
     ModelTable,
@@ -189,12 +190,32 @@ class ProjectAnalysis:
     reused: tuple[str, ...] = ()
     advisories: tuple[Advisory, ...] = ()
     coverage: Coverage = field(default_factory=Coverage)
+    # Findings silenced by an inline ``# coretrace: ignore`` comment.
+    suppressed: tuple[Finding, ...] = ()
 
 
 @dataclass(frozen=True)
 class FileAnalysis:
     findings: tuple[Finding, ...]
     coverage: Coverage
+    suppressed: tuple[Finding, ...] = ()
+
+
+def _text_of(sources: SourceManager) -> Callable[[SourceId], str | None]:
+    """The text of a source by identifier: loaded sources first, then the file on disk."""
+
+    def text_of(source_id: SourceId) -> str | None:
+        try:
+            return sources.get(source_id).text
+        except KeyError:
+            pass
+        path = Path(str(source_id))
+        try:
+            return decode_text(path.read_bytes()) if path.is_file() else None
+        except (OSError, UnicodeDecodeError):
+            return None
+
+    return text_of
 
 
 def build_manager(
@@ -257,7 +278,8 @@ def analyze_file(source: SourceFile, plugin_roots: Sequence[Path]) -> FileAnalys
     findings, supported = _check_module(manager, tuple(loaded.plugin for loaded in registry))
     functions = len(analyzable_functions(manager.module))
     coverage = Coverage((FileCoverage(str(source.source_id), "analysed", functions, len(supported)),))
-    return FileAnalysis(findings, coverage)
+    kept, suppressed = partition(findings, lambda sid: source.text if sid == source.source_id else None)
+    return FileAnalysis(kept, coverage, suppressed)
 
 
 def resolve_dependencies(root: Path, sources: SourceManager) -> DependencyGraph:
@@ -438,15 +460,17 @@ def analyze_project(
     for plugin in all_plugins:
         if isinstance(plugin, ProjectPlugin):
             findings.extend(plugin.analyze_project(context))
+    kept, suppressed = partition(apply_policy(policy, findings), _text_of(sources))
     return ProjectAnalysis(
         graph,
         index,
-        apply_policy(policy, findings),
+        kept,
         dependencies,
         MappingProxyType(keys),
         reused,
         advisories,
         Coverage(tuple(sorted(coverage, key=lambda c: c.path))),
+        suppressed,
     )
 
 
@@ -685,8 +709,11 @@ def _register_all(module: nodes.Module) -> AnalysisManager:
 
 
 def report(
-    findings: Sequence[Finding], coverage: Coverage | None = None, root: Path | None = None
+    findings: Sequence[Finding],
+    coverage: Coverage | None = None,
+    root: Path | None = None,
+    suppressed: Sequence[Finding] = (),
 ) -> Report:
     """The report of a check; paths under ``root`` are rendered relative to it."""
 
-    return Report(tuple(findings), TOOL_NAME, __version__, coverage, root)
+    return Report(tuple(findings), TOOL_NAME, __version__, coverage, root, tuple(suppressed))

src/coretrace_python/findings/suppressions.py

@@ -0,0 +1,71 @@
+"""Inline suppressions: ``# coretrace: ignore`` on the line of a finding.
+
+A bare ``ignore`` silences every finding reported on that line; ``ignore[rule, rule]``
+silences the listed rules only. Python sources are tokenized so a string that merely
+contains the marker is not a suppression; other text files (requirements files,
+configuration) are read line by line with the same comment syntax.
+"""
+
+from __future__ import annotations
+
+import io
+import re
+import tokenize
+from collections.abc import Callable, Iterable, Mapping
+
+from coretrace_python.findings.model import Finding
+from coretrace_python.source import SourceId
+
+MARKER = re.compile(r"#\s*coretrace:\s*ignore(?:\[([^\]]*)\])?")
+
+Suppressions = Mapping[int, frozenset[str] | None]
+"""Line number to the suppressed rules, ``None`` meaning every rule."""
+
+
+def suppressions_in(text: str) -> Suppressions:
+    found: dict[int, frozenset[str] | None] = {}
+    for line, comment in _comments(text):
+        match = MARKER.search(comment)
+        if match is None:
+            continue
+        rules = match.group(1)
+        if rules is None:
+            found[line] = None
+        elif found.get(line, ()) is not None:
+            listed = frozenset(r.strip() for r in rules.split(",") if r.strip())
+            found[line] = frozenset(found.get(line) or ()) | listed
+    return found
+
+
+def _comments(text: str) -> Iterable[tuple[int, str]]:
+    try:
+        tokens = list(tokenize.generate_tokens(io.StringIO(text).readline))
+    except (tokenize.TokenError, SyntaxError):
+        for number, line in enumerate(text.splitlines(), 1):
+            if "#" in line:
+                yield number, line[line.index("#") :]
+        return
+    for token in tokens:
+        if token.type == tokenize.COMMENT:
+            yield token.start[0], token.string
+
+
+def partition(
+    findings: Iterable[Finding], text_of: Callable[[SourceId], str | None]
+) -> tuple[tuple[Finding, ...], tuple[Finding, ...]]:
+    """Split findings into the kept and the suppressed, reading each file's text once."""
+
+    cache: dict[SourceId, Suppressions] = {}
+    kept: list[Finding] = []
+    suppressed: list[Finding] = []
+    for finding in findings:
+        source_id = finding.span.source_id
+        if source_id not in cache:
+            text = text_of(source_id)
+            cache[source_id] = suppressions_in(text) if text is not None else {}
+        rules = cache[source_id].get(finding.span.start_line, frozenset())
+        if rules is None or finding.rule_id in rules:
+            suppressed.append(finding)
+        else:
+            kept.append(finding)
+    return tuple(kept), tuple(suppressed)

src/coretrace_python/reporters/json_format.py

@@ -36,6 +36,8 @@ def render_json(report: Report) -> str:
     if report.root is not None:
         document["root"] = str(report.root)
     document["findings"] = [finding_record(finding, report) for finding in report.findings]
+    if report.suppressed:
+        document["suppressed"] = [finding_record(finding, report) for finding in report.suppressed]
     if report.coverage is not None:
         coverage = report.coverage
         document["coverage"] = {

src/coretrace_python/reporters/report.py

@@ -21,9 +21,12 @@ class Report:
     coverage: Coverage | None = None
     # The directory the report is about: paths under it are rendered relative to it.
     root: Path | None = None
+    # Findings silenced by an inline suppression; reported apart, never counted.
+    suppressed: tuple[Finding, ...] = ()
 
     def __post_init__(self) -> None:
         object.__setattr__(self, "findings", tuple(sorted(self.findings, key=_order)))
+        object.__setattr__(self, "suppressed", tuple(sorted(self.suppressed, key=_order)))
 
     def locate(self, path: str) -> str:
         """``path`` relative to the root, POSIX style, when it lies under the root."""

src/coretrace_python/reporters/sarif.py

@@ -26,13 +26,15 @@ def _artifact(report: Report, path: str) -> dict[str, str]:
     return {"uri": path}
 
 
-def _result(report: Report, finding: Finding, rule_index: int) -> dict[str, object]:
+def _result(
+    report: Report, finding: Finding, rule_index: int, suppressed: bool = False
+) -> dict[str, object]:
     span = finding.span
     region: dict[str, int] = {"startLine": span.start_line, "startColumn": span.start_column}
     if span.end_line is not None and span.end_column is not None:
         region["endLine"] = span.end_line
         region["endColumn"] = span.end_column
-    return {
+    result: dict[str, object] = {
         "ruleId": finding.rule_id,
         "ruleIndex": rule_index,
         "level": _LEVELS[finding.severity],
@@ -46,11 +48,14 @@ def _result(report: Report, finding: Finding, rule_index: int) -> dict[str, obje
             }
         ],
     }
+    if suppressed:
+        result["suppressions"] = [{"kind": "inSource"}]
+    return result
 
 
 def render_sarif(report: Report) -> str:
     rule_ids: list[str] = []
-    for finding in report.findings:
+    for finding in (*report.findings, *report.suppressed):
         if finding.rule_id not in rule_ids:
             rule_ids.append(finding.rule_id)
     run: dict[str, object] = {}
@@ -69,8 +74,8 @@ def render_sarif(report: Report) -> str:
                     }
                 },
             "results": [
-                _result(report, finding, rule_ids.index(finding.rule_id))
-                for finding in report.findings
+                *(_result(report, f, rule_ids.index(f.rule_id)) for f in report.findings),
+                *(_result(report, f, rule_ids.index(f.rule_id), True) for f in report.suppressed),
             ],
         }
     )

src/coretrace_python/reporters/text.py

@@ -18,6 +18,8 @@ def render_text(report: Report) -> str:
         lines.append(line)
     count = len(report.findings)
     summary = "no findings" if count == 0 else f"{count} finding{'s' if count > 1 else ''}"
+    if report.suppressed:
+        summary += f", {len(report.suppressed)} suppressed"
     lines.append(summary)
     if report.coverage is not None:
         lines.append(report.coverage.summary())