diff --git a/docs/plugins.md b/docs/plugins.md index c843e77..ebb442b 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -320,8 +320,12 @@ any analysis result: `SSAAnalysis` (`coretrace_python.ir.ssa`) for the code, ## Findings A `Finding` (`coretrace_python.findings`) has a `rule_id`, a `message`, a `Severity`, a -`Confidence`, a `SourceSpan` (file, start and end line and column), the enclosing -`function` and string `metadata`. Rule ids are lower-case words joined by dashes. +`Confidence`, a location, the enclosing `function` and string `metadata`. Rule ids are +lower-case words joined by dashes. The location (`span`) is a `SourceSpan` (file, start +and end line and column) when the line is established, or a `FileLocation` (file and the +JSON pointer of a value) for a value of a structured file whose line cannot be: a plugin +that reads a finding's line checks which it has, and never invents a line for a +`FileLocation` (`coretrace_python.source.positions` places JSON and TOML values). ## Testing a plugin diff --git a/docs/usage.md b/docs/usage.md index 9b96275..4ca5464 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -215,7 +215,8 @@ 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. +they never affect the exit status. A comment only suppresses a finding located at a +verified line; a finding with a file location (see Reports) is never suppressed this way. ## Adopting the analyzer on an existing code base @@ -232,7 +233,16 @@ above it does not make it new; a change to the line itself does. Commit the file shrink it as findings are fixed: an entry without a matching finding is simply unused. Baselined findings are counted in the text report, listed under `baselined` in the JSON report and marked `baselineState: unchanged` in the SARIF log, where new results are -marked `new`. +marked `new`. A finding with a file location is recognised by the JSON pointer of its +value and a digest of the value (never the value itself) instead of the text of a line, +so a new value at the same place is a new finding. + +Baselines are written in schema 2. A schema 1 file, written by an earlier version, is +still read and its entries match exactly as they did; but it recorded a finding in a +JSON, TOML or lock file with the text of its key's first line, which was its own only +when the key was not repeated. A finding such an entry cannot identify with certainty is +reported as new, and the check then warns on standard error: review the new findings, +delete the baseline and run the check again to record it in schema 2. ## Reports @@ -267,6 +277,17 @@ the directory of the checked file; a path outside it is printed as it is. } ``` +Lines are counted as Python counts them, by `\n`, `\r\n` or a lone `\r` (never U+2028 +or a form feed), in every report, suppression and baseline. A value of a JSON or TOML +file, a package of a lock file, or a requirement of `pyproject.toml`, is located at its own line, +which a separate component establishes from the value's structural path and checks +against the value. When it cannot (a dotted TOML key, an inline table, a multi-line +value), the finding gets a file location instead of a guessed line: `line`, `column`, +`end_line` and `end_column` are `null`, and `pointer` gives the JSON pointer of the value +in the file (`"/tool/app/api_token"`). The SARIF log gives such a result no `region`, +with the pointer under the location's `properties`, and the text report shows the file +alone (`settings.toml: high hardcoded-credential: …`). + `root` is the directory the paths are relative to. A file's `status` is `analysed`, `ambiguous` (analysed, but another file has its module name, see above), `syntax-error` or `unreadable`; only `analysed` files count in `files_analysed`. diff --git a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py index 347943c..f5751d5 100644 --- a/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py +++ b/src/coretrace_python/bundled/dependency/vulnerable_dependency/vulnerable_dependency.py @@ -3,10 +3,11 @@ from __future__ import annotations from collections.abc import Sequence +from pathlib import PurePath from typing import ClassVar from coretrace_python.analysis import AnyAnalysis -from coretrace_python.dependency import DependencyAnalysis +from coretrace_python.dependency import Advisory, DependencyAnalysis, Requirement from coretrace_python.dependency.correlation import ( affected_entries, affected_symbols, @@ -44,11 +45,7 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: findings.append( Finding( rule_id="vulnerable-dependency", - message=( - f"{advisory.id}: {advisory.package} {advisory.vulnerable} is " - f"{'required' if pinned else 'allowed'} by {requirement.name}" - f"{requirement.specifier or ''}: {advisory.summary}" - ), + message=_message(ctx, advisory, requirement), severity=advisory.severity, confidence=Confidence.HIGH if pinned else Confidence.MEDIUM, span=requirement.span, @@ -58,6 +55,26 @@ def analyze_project(self, ctx: ProjectContext) -> Sequence[Finding]: return findings +def _message(ctx: ProjectContext, advisory: Advisory, requirement: Requirement) -> str: + """The evidence: for a locked package, the version the lock file pins and the + packages that require it; for a declared requirement, the specifier that allows or + requires a vulnerable version.""" + + if requirement.locked is None: + verb = "required" if requirement.pinned is not None else "allowed" + return ( + f"{advisory.id}: {advisory.package} {advisory.vulnerable} is {verb} by " + f"{requirement.name}{requirement.specifier or ''}: {advisory.summary}" + ) + lock = PurePath(str(requirement.span.source_id)).name + dependents = sorted(ctx.dependencies.required_by(requirement.name) or ()) + required = f" (required by {', '.join(dependents)})" if dependents else "" + return ( + f"{advisory.id}: {advisory.package} {requirement.locked}, pinned in {lock}, " + f"is in {advisory.vulnerable}{required}: {advisory.summary}" + ) + + def _ruled_out(ctx: ProjectContext) -> dict[str, list[str]]: """By advisory, the calls to its entry points whose arguments contradict one of its conditions: the evidence that the requirement, imported, is not reached there.""" diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index e34ae7d..90ee0af 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -41,9 +41,9 @@ UnknownTarget, ) from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.source import SourceId, SourceSpan +from coretrace_python.source import FileLocation, Location, SourceId, SourceSpan -CACHE_FORMAT = 13 +CACHE_FORMAT = 14 @dataclass(frozen=True) @@ -185,6 +185,24 @@ def _encode_span(span: SourceSpan) -> list[Any]: return [str(span.source_id), span.start_line, span.start_column, span.end_line, span.end_column] +def _encode_location(location: Location) -> Any: + if isinstance(location, FileLocation): + return { + "file": str(location.source_id), + "pointer": location.pointer, + "digest": location.digest, + } + return _encode_span(location) + + +def _decode_location(data: Any) -> Location: + if isinstance(data, Mapping): + return FileLocation( + SourceId(_string(data["file"])), _string(data["pointer"]), _string(data["digest"]) + ) + return _decode_span(data) + + def _decode_span(data: Any) -> SourceSpan: file, line, column, end_line, end_column = data return SourceSpan( @@ -217,7 +235,7 @@ def _encode_finding(finding: Finding) -> dict[str, Any]: "message": finding.message, "severity": finding.severity.value, "confidence": finding.confidence.value, - "span": _encode_span(finding.span), + "span": _encode_location(finding.span), "function": finding.function, "metadata": dict(finding.metadata), } @@ -230,7 +248,7 @@ def _decode_finding(data: Mapping[str, Any]) -> Finding: _string(data["message"]), Severity(data["severity"]), Confidence(data["confidence"]), - _decode_span(data["span"]), + _decode_location(data["span"]), None if function is None else _string(function), {_string(k): _string(v) for k, v in data["metadata"].items()}, ) diff --git a/src/coretrace_python/cli.py b/src/coretrace_python/cli.py index c57d350..2a0bccb 100644 --- a/src/coretrace_python/cli.py +++ b/src/coretrace_python/cli.py @@ -274,7 +274,11 @@ def main(argv: list[str] | None = None) -> int: baselined = None if args.baseline is not None: if args.baseline.is_file(): - findings, baselined = Baseline.load(args.baseline).partition(findings, root) + baseline = Baseline.load(args.baseline) + findings, baselined = baseline.partition(findings, root) + notice = baseline.transition_notice(findings) + if notice is not None: + print(f"coretrace: {args.baseline}: {notice}", file=sys.stderr) else: Baseline.of(findings, root).save(args.baseline) findings, baselined = (), findings diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 4d42e62..e9fffa9 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -13,7 +13,7 @@ import re import tomllib from collections.abc import Iterable, Iterator, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import PurePath, PurePosixPath from types import MappingProxyType from typing import Any, ClassVar @@ -21,7 +21,8 @@ from coretrace_python.analysis import Analysis, AnalysisContext from coretrace_python.findings import Severity from coretrace_python.semantic.symbols import SymbolId -from coretrace_python.source import SourceFile, SourceId, SourceSpan +from coretrace_python.source import Location, SourceFile, SourceId, SourceSpan, lines_of +from coretrace_python.source.positions import Pointer, TomlPositions, file_location _REQUIREMENT = re.compile(r"^\s*([A-Za-z0-9][A-Za-z0-9._-]*)\s*(\[[^\]]*\])?\s*(.*)$") _CLAUSE = re.compile(r"^(===|==|!=|<=|>=|~=|<|>|\^)\s*([0-9][0-9A-Za-z.*+!-]*)$") @@ -115,11 +116,16 @@ def _lower_bounds(specifier: str) -> list[Version]: @dataclass(frozen=True) class Requirement: + """``span`` is where the requirement is written: a verified line, or, for a locked + package whose entry cannot be placed, a file location. ``locked`` is the version a + lock file pins, as it writes it; None for a declared requirement.""" + name: str specifier: str - span: SourceSpan + span: Location pinned: Version | None = None optional: bool = False + locked: str | None = None @classmethod def parse(cls, text: str, source_id: SourceId, line: int, optional: bool = False) -> Requirement | None: @@ -313,12 +319,14 @@ def merge(self, other: DependencyGraph) -> DependencyGraph: if current is None: merged[name] = requirement continue + declared = current if current.specifier else requirement merged[name] = Requirement( name, current.specifier or requirement.specifier, - current.span if current.specifier else requirement.span, + declared.span, requirement.pinned or current.pinned, current.optional and requirement.optional, + declared.locked, ) dependents = dict(self._dependents) for name, found in other._dependents.items(): @@ -341,7 +349,7 @@ def parse_dependencies(source: SourceFile) -> DependencyGraph: def _parse_requirements_txt(source: SourceFile) -> DependencyGraph: found: dict[str, Requirement] = {} - for number, line in enumerate(source.text.splitlines(), start=1): + for number, line in enumerate(lines_of(source.text), start=1): stripped = line.strip() if not stripped or stripped.startswith(("#", "-")): continue @@ -362,25 +370,32 @@ def _parse_toml(source: SourceFile, extract: Any, dependents: Any = None) -> Dep return DependencyGraph(found, dependents=None if dependents is None else dependents(data)) -def _line_of(source: SourceFile, key: str, default: int = 1) -> int: - for number, line in enumerate(source.text.splitlines(), start=1): - if key in line: - return number - return default +def _pyproject_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requirement]: + """The requirements ``pyproject.toml`` declares (PEP 621 and Poetry), each located at + its own array element or key, which the position component checks, or given a file + location when it cannot be placed.""" + positions = TomlPositions(source.text) -def _pyproject_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requirement]: - found: list[Requirement] = [] + def declared(text: str, path: Pointer, value: object, optional: bool = False) -> Requirement | None: + requirement = Requirement.parse(text, source.source_id, 1, optional) + if requirement is None: + return None + place = positions.locate(path, value) + location: Location = ( + SourceSpan(source.source_id, *place) + if place is not None + else file_location(source.source_id, path, text) + ) + return replace(requirement, span=location) + + found: list[Requirement | None] = [] project = data.get("project", {}) - for text in project.get("dependencies", []) or []: - requirement = Requirement.parse(text, source.source_id, _line_of(source, text)) - if requirement is not None: - found.append(requirement) - for group in (project.get("optional-dependencies", {}) or {}).values(): - for text in group or []: - requirement = Requirement.parse(text, source.source_id, _line_of(source, text), True) - if requirement is not None: - found.append(requirement) + for index, text in enumerate(project.get("dependencies", []) or []): + found.append(declared(text, ("project", "dependencies", index), text)) + for group, texts in (project.get("optional-dependencies", {}) or {}).items(): + for index, text in enumerate(texts or []): + found.append(declared(text, ("project", "optional-dependencies", group, index), text, True)) poetry = data.get("tool", {}).get("poetry", {}) for section, optional in (("dependencies", False), ("dev-dependencies", True)): for name, spec in (poetry.get(section, {}) or {}).items(): @@ -389,23 +404,29 @@ def _pyproject_requirements(data: Mapping[str, Any], source: SourceFile) -> list specifier = spec.get("version", "") if isinstance(spec, dict) else str(spec) if specifier == "*": specifier = "" - requirement = Requirement.parse( - f"{name}{specifier}", source.source_id, _line_of(source, name), optional - ) - if requirement is not None: - found.append(requirement) - return found + found.append(declared(f"{name}{specifier}", ("tool", "poetry", section, name), spec, optional)) + return [requirement for requirement in found if requirement is not None] def _lock_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requirement]: + """The packages a lock file pins, each located at the ``name`` of its own + ``[[package]]`` entry, which the position component checks.""" + found: list[Requirement] = [] - for package in data.get("package", []) or []: + positions = TomlPositions(source.text) + for index, package in enumerate(data.get("package", []) or []): name, version = package.get("name"), package.get("version") if not isinstance(name, str) or not isinstance(version, str): continue - line = _line_of(source, f'name = "{name}"') + path = ("package", index, "name") + place = positions.locate(path, name) + location: Location = ( + SourceSpan(source.source_id, *place) + if place is not None + else file_location(source.source_id, path, name) + ) found.append( - Requirement(normalize(name), "", SourceSpan(source.source_id, line, 1), Version.parse(version)) + Requirement(normalize(name), "", location, Version.parse(version), locked=version) ) return found diff --git a/src/coretrace_python/dependency/vex.py b/src/coretrace_python/dependency/vex.py index 614864a..39ac007 100644 --- a/src/coretrace_python/dependency/vex.py +++ b/src/coretrace_python/dependency/vex.py @@ -26,6 +26,7 @@ from coretrace_python.dependency.sbom import purl from coretrace_python.findings import Component, Finding from coretrace_python.findings.coverage import Coverage +from coretrace_python.source import SourceSpan from coretrace_python.taint import FILTER_FUNCTIONS CONTEXT = "https://openvex.dev/ns/v0.2.0" @@ -179,10 +180,12 @@ def _reached(findings: Sequence[Finding], root: Path) -> str: level = finding.metadata.get("level") if level in REACHED: path = _relative(str(finding.span.source_id), root) - place = (path, finding.span.start_line, str(finding.metadata.get("symbol"))) + line = finding.span.start_line if isinstance(finding.span, SourceSpan) else 0 + place = (path, line, str(finding.metadata.get("symbol"))) levels[place] = max(level, levels.get(place, level), key=REACHED.index) return "; ".join( - f"{path}:{line} {symbol} ({level})" for (path, line, symbol), level in sorted(levels.items()) + f"{path}{f':{line}' if line else ''} {symbol} ({level})" + for (path, line, symbol), level in sorted(levels.items()) ) diff --git a/src/coretrace_python/findings/baseline.py b/src/coretrace_python/findings/baseline.py index 0c22f0d..11dae3a 100644 --- a/src/coretrace_python/findings/baseline.py +++ b/src/coretrace_python/findings/baseline.py @@ -2,8 +2,17 @@ A finding is recognised by its file relative to the root, its rule, its function and the text of its line, never by its line number: code inserted above it does not make it -new, a change to the line itself does. Entries are counted, so two identical findings on -identical lines need two entries. +new, a change to the line itself does. A finding with a file location, whose line is not +established, is recognised by the JSON pointer of its value and a digest of the value +instead, never the value itself: a new value there is a new finding. Entries are +counted, so two identical findings on identical lines need two entries. + +Baselines are written in schema 2, whose entries carry the pointer. Schema 1 files are +still read, and their entries match exactly as they did. There, a finding in a JSON, TOML +or lock file was recorded with the text of its key's first line, its own only when the +key was not repeated: such an entry still identifies that finding, and no other. A +finding it cannot identify with certainty is new, and ``transition_notice`` asks for the +baseline to be recorded again. """ from __future__ import annotations @@ -15,11 +24,16 @@ from pathlib import Path from coretrace_python.findings.model import Finding -from coretrace_python.source import decode_text +from coretrace_python.source import SourceSpan, decode_text, lines_of -BASELINE_SCHEMA = 1 +BASELINE_SCHEMA = 2 +_LEGACY_SCHEMA = 1 +# Files whose findings schema 1 placed at their key's first line. +_STRUCTURED = (".json", ".toml", ".lock") -Fingerprint = tuple[str, str, str, str] +# File, rule, function, text of the line (for a file location, the digest of its +# value), pointer (empty for a line). +Fingerprint = tuple[str, str, str, str, str] class BaselineError(Exception): @@ -32,12 +46,18 @@ def fingerprint(finding: Finding, root: Path) -> Fingerprint: relative = path.relative_to(root).as_posix() except ValueError: relative = path.as_posix() - return (relative, finding.rule_id, finding.function or "", _line_text(path, finding.span.start_line)) + head = (relative, finding.rule_id, finding.function or "") + if isinstance(finding.span, SourceSpan): + return (*head, _line_text(path, finding.span.start_line), "") + return (*head, finding.span.digest, finding.span.pointer) def _line_text(path: Path, line: int) -> str: + """The text of ``line``, lines counted as every line number of the engine counts + them (``lines_of``).""" + try: - lines = decode_text(path.read_bytes()).splitlines() + lines = lines_of(decode_text(path.read_bytes())) except (OSError, UnicodeDecodeError): return str(line) if not 1 <= line <= len(lines): @@ -48,6 +68,8 @@ def _line_text(path: Path, line: int) -> str: @dataclass(frozen=True) class Baseline: entries: Counter[Fingerprint] = field(default_factory=Counter) + # Read from a schema 1 file, whose entries on structured files may no longer match. + legacy: bool = False @classmethod def of(cls, findings: Iterable[Finding], root: Path) -> Baseline: @@ -57,11 +79,18 @@ def of(cls, findings: Iterable[Finding], root: Path) -> Baseline: def load(cls, path: Path) -> Baseline: try: document = json.loads(path.read_text(encoding="utf-8")) - if document.get("schema") != BASELINE_SCHEMA: - raise BaselineError(f"{path}: unsupported baseline schema {document.get('schema')!r}") + schema = document.get("schema") + if schema not in (_LEGACY_SCHEMA, BASELINE_SCHEMA): + raise BaselineError(f"{path}: unsupported baseline schema {schema!r}") entries = Counter( { - (str(e["path"]), str(e["rule"]), str(e["function"]), str(e["line"])): int(e["count"]) + ( + str(e["path"]), + str(e["rule"]), + str(e["function"]), + str(e.get("line", "")), + str(e.get("pointer", "")), + ): int(e["count"]) for e in document["findings"] } ) @@ -69,14 +98,14 @@ def load(cls, path: Path) -> Baseline: raise except (OSError, ValueError, KeyError, TypeError, AttributeError) as error: raise BaselineError(f"{path}: {error}") from error - return cls(entries) + return cls(entries, legacy=schema == _LEGACY_SCHEMA) def save(self, path: Path) -> None: document = { "schema": BASELINE_SCHEMA, "findings": [ - {"path": p, "rule": r, "function": f, "line": line, "count": count} - for (p, r, f, line), count in sorted(self.entries.items()) + {"path": p, "rule": r, "function": f, "line": line, "pointer": pointer, "count": count} + for (p, r, f, line, pointer), count in sorted(self.entries.items()) ], } path.write_text(json.dumps(document, indent=2) + "\n", encoding="utf-8") @@ -97,3 +126,17 @@ def partition( else: new.append(finding) return tuple(new), tuple(baselined) + + def transition_notice(self, new: Iterable[Finding]) -> str | None: + """What to tell when a schema 1 baseline leaves findings of JSON, TOML or lock + files new: their old entries may name another line, so the baseline is to be + recorded again; None otherwise.""" + + if not self.legacy or not any(str(finding.span.source_id).endswith(_STRUCTURED) for finding in new): + return None + return ( + "schema 1 baseline: findings in JSON, TOML and lock files are now located at their own " + "line or by their pointer, and an old entry that cannot identify one with certainty no " + "longer accounts for it; review the new findings, then delete the baseline and run the " + "check again to record it in schema 2" + ) diff --git a/src/coretrace_python/findings/model.py b/src/coretrace_python/findings/model.py index 2b2ca19..5bb5b4e 100644 --- a/src/coretrace_python/findings/model.py +++ b/src/coretrace_python/findings/model.py @@ -12,7 +12,7 @@ from enum import Enum from types import MappingProxyType -from coretrace_python.source import SourceSpan +from coretrace_python.source import Location FINDING_SCHEMA_VERSION = 1 @@ -50,7 +50,8 @@ class Finding: message: str severity: Severity confidence: Confidence - span: SourceSpan + # A verified range of lines, or a file location when the line is not established. + span: Location function: str | None = None metadata: Mapping[str, str] = field(default_factory=_no_metadata) diff --git a/src/coretrace_python/findings/suppressions.py b/src/coretrace_python/findings/suppressions.py index c9bd213..e5be1b3 100644 --- a/src/coretrace_python/findings/suppressions.py +++ b/src/coretrace_python/findings/suppressions.py @@ -14,7 +14,7 @@ from collections.abc import Callable, Iterable, Mapping from coretrace_python.findings.model import Finding -from coretrace_python.source import SourceId +from coretrace_python.source import SourceId, SourceSpan, lines_of MARKER = re.compile(r"#\s*coretrace:\s*ignore(?:\[([^\]]*)\])?") @@ -38,10 +38,14 @@ def suppressions_in(text: str) -> Suppressions: def _comments(text: str) -> Iterable[tuple[int, str]]: + """Every comment of ``text`` with its line, lines counted as every line number of the + engine counts them (``lines_of``).""" + + lines = lines_of(text) try: - tokens = list(tokenize.generate_tokens(io.StringIO(text).readline)) + tokens = list(tokenize.generate_tokens(io.StringIO("".join(f"{line}\n" for line in lines)).readline)) except (tokenize.TokenError, SyntaxError): - for number, line in enumerate(text.splitlines(), 1): + for number, line in enumerate(lines, 1): if "#" in line: yield number, line[line.index("#") :] return @@ -59,6 +63,10 @@ def partition( kept: list[Finding] = [] suppressed: list[Finding] = [] for finding in findings: + if not isinstance(finding.span, SourceSpan): + # A suppression names a line; a file location has no verified one. + kept.append(finding) + continue source_id = finding.span.source_id if source_id not in cache: text = text_of(source_id) diff --git a/src/coretrace_python/plugins/secrets.py b/src/coretrace_python/plugins/secrets.py index 2608b68..21eb72c 100644 --- a/src/coretrace_python/plugins/secrets.py +++ b/src/coretrace_python/plugins/secrets.py @@ -28,9 +28,12 @@ from coretrace_python.hir.visitors import Node, children from coretrace_python.interprocedural import discover_files from coretrace_python.plugins.api import Plugin, PluginContext -from coretrace_python.source import SourceId, SourceSpan, decode_text +from coretrace_python.source import Location, SourceId, SourceSpan, decode_text, lines_of +from coretrace_python.source.positions import JsonPositions, Pointer, TomlPositions, file_location -Literal = tuple[str, str | None, SourceSpan, str | None] +# A value, the name it is bound to, where it is, and its enclosing function. A value of +# a JSON or TOML file whose line cannot be established is at a file location. +Literal = tuple[str, str | None, Location, str | None] _HEX = re.compile(r"^[0-9a-fA-F]+$") _TOKEN = re.compile(r"^[A-Za-z0-9+/=_-]+$") @@ -293,9 +296,9 @@ def config_literals(root: Path) -> Iterator[Literal]: source = SourceId(str(path)) if path.suffix == ".json": maps = _NPM_NAME_MAPS if path.name in _NPM_LOCK_FILES else frozenset() - yield from _structured(source, text, _load_json(text), maps) + yield from _structured(source, _load_json(text), JsonPositions(text), maps) elif path.suffix == ".toml": - yield from _structured(source, text, _load_toml(text)) + yield from _structured(source, _load_toml(text), TomlPositions(text)) else: yield from _pairs(source, text) @@ -315,34 +318,36 @@ def _load_toml(text: str) -> object: def _structured( - source: SourceId, text: str, data: object, name_maps: frozenset[str] = frozenset() + source: SourceId, + data: object, + positions: JsonPositions | TomlPositions, + name_maps: frozenset[str] = frozenset(), ) -> Iterator[Literal]: - """Every string value of ``data`` with its key, except that a value directly under - one of ``name_maps`` is bound to no name: its key names a package or a command.""" + """Every string value of ``data`` with its key, where ``positions`` places it from + its structural path, except that a value directly under one of ``name_maps`` is + bound to no name: its key names a package or a command.""" - lines = text.splitlines() - - def line_of(key: str) -> int: - for number, line in enumerate(lines, start=1): - if re.match(rf'^\s*"?{re.escape(key)}"?\s*[:=]', line) or f'"{key}"' in line: - return number - return 1 - - def walk(node: object, key: str | None, named: bool = True) -> Iterator[Literal]: + def walk(node: object, path: Pointer, key: str | None, named: bool = True) -> Iterator[Literal]: if isinstance(node, dict): for name, value in node.items(): - yield from walk(value, str(name), key not in name_maps) + yield from walk(value, (*path, str(name)), str(name), key not in name_maps) elif isinstance(node, list): - for item in node: - yield from walk(item, key, named) + for index, item in enumerate(node): + yield from walk(item, (*path, index), key, named) elif isinstance(node, str) and key is not None: - yield node, key if named else None, SourceSpan(source, line_of(key), 1), None + place = positions.locate(path, node) + location: Location = ( + SourceSpan(source, *place) + if place is not None + else file_location(source, path, node) + ) + yield node, key if named else None, location, None - yield from walk(data, None) + yield from walk(data, (), None) def _pairs(source: SourceId, text: str) -> Iterator[Literal]: - for number, line in enumerate(text.splitlines(), start=1): + for number, line in enumerate(lines_of(text), start=1): stripped = line.strip() if not stripped or stripped[0] in "#;[": continue @@ -385,7 +390,7 @@ def analyze(self, ctx: PluginContext) -> Sequence[Finding]: findings.append(finding) return findings - def judge(self, value: str, name: str | None, span: SourceSpan, function: str | None) -> Finding | None: + def judge(self, value: str, name: str | None, span: Location, function: str | None) -> Finding | None: where = f"in {name}" if name is not None else "in a string literal" for pattern in self.patterns: if pattern.matches(value): diff --git a/src/coretrace_python/reporters/json_format.py b/src/coretrace_python/reporters/json_format.py index 1042add..864b20a 100644 --- a/src/coretrace_python/reporters/json_format.py +++ b/src/coretrace_python/reporters/json_format.py @@ -6,23 +6,30 @@ from coretrace_python.findings import FINDING_SCHEMA_VERSION, Component, Finding from coretrace_python.reporters.report import Report +from coretrace_python.source import SourceSpan def finding_record(finding: Finding, report: Report | None = None) -> dict[str, object]: span = finding.span path = str(span.source_id) + location: dict[str, object] = {"path": report.locate(path) if report is not None else path} + if isinstance(span, SourceSpan): + location.update( + line=span.start_line, + column=span.start_column, + end_line=span.end_line, + end_column=span.end_column, + ) + else: + location.update( + line=None, column=None, end_line=None, end_column=None, pointer=span.pointer + ) return { "rule_id": finding.rule_id, "message": finding.message, "severity": finding.severity.value, "confidence": finding.confidence.value, - "location": { - "path": report.locate(path) if report is not None else path, - "line": span.start_line, - "column": span.start_column, - "end_line": span.end_line, - "end_column": span.end_column, - }, + "location": location, "function": finding.function, "metadata": dict(finding.metadata), } diff --git a/src/coretrace_python/reporters/report.py b/src/coretrace_python/reporters/report.py index 95bb5af..01d37a0 100644 --- a/src/coretrace_python/reporters/report.py +++ b/src/coretrace_python/reporters/report.py @@ -6,11 +6,16 @@ from pathlib import Path, PurePosixPath from coretrace_python.findings import Component, Coverage, Finding +from coretrace_python.source import FileLocation -def _order(finding: Finding) -> tuple[str, int, int, str]: +def _order(finding: Finding) -> tuple[str, int, int, str, str]: + """By file, file locations first, then by line and column.""" + span = finding.span - return (str(span.source_id), span.start_line, span.start_column, finding.rule_id) + if isinstance(span, FileLocation): + return (str(span.source_id), 0, 0, span.pointer, finding.rule_id) + return (str(span.source_id), span.start_line, span.start_column, "", finding.rule_id) @dataclass(frozen=True) diff --git a/src/coretrace_python/reporters/sarif.py b/src/coretrace_python/reporters/sarif.py index 9139a5f..9e32abf 100644 --- a/src/coretrace_python/reporters/sarif.py +++ b/src/coretrace_python/reporters/sarif.py @@ -6,6 +6,7 @@ from coretrace_python.findings import Component, Finding, Severity from coretrace_python.reporters.report import Report +from coretrace_python.source import SourceSpan SARIF_VERSION = "2.1.0" SARIF_SCHEMA = "https://json.schemastore.org/sarif-2.1.0.json" @@ -34,23 +35,23 @@ def _result( baseline_state: str | None = None, ) -> 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 + physical: dict[str, object] = {"artifactLocation": _artifact(report, str(span.source_id))} + location: dict[str, object] = {"physicalLocation": physical} + if isinstance(span, SourceSpan): + 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 + physical["region"] = region + else: + # No region: the line is not established; the pointer says where in the file. + location["properties"] = {"pointer": span.pointer} result: dict[str, object] = { "ruleId": finding.rule_id, "ruleIndex": rule_index, "level": _LEVELS[finding.severity], "message": {"text": finding.message}, - "locations": [ - { - "physicalLocation": { - "artifactLocation": _artifact(report, str(span.source_id)), - "region": region, - } - } - ], + "locations": [location], } if suppressed: result["suppressions"] = [{"kind": "inSource"}] diff --git a/src/coretrace_python/reporters/text.py b/src/coretrace_python/reporters/text.py index 5c392c8..35bf45b 100644 --- a/src/coretrace_python/reporters/text.py +++ b/src/coretrace_python/reporters/text.py @@ -3,16 +3,17 @@ from __future__ import annotations from coretrace_python.reporters.report import Report +from coretrace_python.source import SourceSpan def render_text(report: Report) -> str: lines = [] for finding in report.findings: span = finding.span - line = ( - f"{report.locate(str(span.source_id))}:{span.start_line}:{span.start_column}: " - f"{finding.severity.value} {finding.rule_id}: {finding.message}" - ) + where = report.locate(str(span.source_id)) + if isinstance(span, SourceSpan): + where += f":{span.start_line}:{span.start_column}" + line = f"{where}: {finding.severity.value} {finding.rule_id}: {finding.message}" if finding.function is not None: line += f" [{finding.function}]" lines.append(line) diff --git a/src/coretrace_python/source/__init__.py b/src/coretrace_python/source/__init__.py index 28e8acd..c2a75a2 100644 --- a/src/coretrace_python/source/__init__.py +++ b/src/coretrace_python/source/__init__.py @@ -1,13 +1,17 @@ """Source storage and location primitives.""" -from coretrace_python.source.manager import SourceManager, decode_text -from coretrace_python.source.model import SourceFile, SourceId, SourceSpan +from coretrace_python.source.manager import LINE_BREAK, SourceManager, decode_text, lines_of +from coretrace_python.source.model import FileLocation, Location, SourceFile, SourceId, SourceSpan __all__ = [ + "LINE_BREAK", + "FileLocation", + "Location", "SourceFile", "SourceId", "SourceManager", "SourceSpan", "decode_text", + "lines_of", ] diff --git a/src/coretrace_python/source/manager.py b/src/coretrace_python/source/manager.py index 1441bd0..4eba81f 100644 --- a/src/coretrace_python/source/manager.py +++ b/src/coretrace_python/source/manager.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from pathlib import Path from coretrace_python.source.model import SourceFile, SourceId @@ -30,6 +31,21 @@ def decode_text(data: bytes) -> str: return data.decode("utf-8") +# A line break as Python reads one. +LINE_BREAK = re.compile(r"\r\n|\r|\n") + + +def lines_of(text: str) -> list[str]: + """The lines of ``text`` as Python counts them: broken by ``\\n``, ``\\r\\n`` or a lone + ``\\r``, never by U+2028, U+0085, a form feed or the other separators + ``str.splitlines`` also breaks on. Every line number of the engine counts by it.""" + + lines = LINE_BREAK.split(text) + if lines[-1] == "": + lines.pop() + return lines + + class SourceManager: """Own source text and return one canonical object for each source ID.""" diff --git a/src/coretrace_python/source/model.py b/src/coretrace_python/source/model.py index 8836fd0..da39927 100644 --- a/src/coretrace_python/source/model.py +++ b/src/coretrace_python/source/model.py @@ -45,6 +45,26 @@ def display(self) -> str: return f"{self.source_id}:{self.start_line}:{self.start_column}" +@dataclass(frozen=True) +class FileLocation: + """A place in a file whose line could not be established: the file, and the JSON + pointer of the value in it (``/a/api_token``). It is never given a guessed line, + which a report or an inline suppression would then take for a verified one.""" + + source_id: SourceId + pointer: str + # A digest of the value (``sha256:…``), never the value: what the baseline compares + # where it would compare the text of a line. ``positions.file_location`` builds both. + digest: str + + def display(self) -> str: + return f"{self.source_id}{self.pointer and '#' + self.pointer}" + + +# Where a finding is: a verified range of lines, or a file location. +Location = SourceSpan | FileLocation + + @dataclass(frozen=True) class SourceFile: """Decoded source text, its identity and its dotted module name.""" diff --git a/src/coretrace_python/source/positions.py b/src/coretrace_python/source/positions.py new file mode 100644 index 0000000..ef620fe --- /dev/null +++ b/src/coretrace_python/source/positions.py @@ -0,0 +1,415 @@ +"""Where a value of a JSON or TOML document is written, from its structural path. + +Rules judge the values ``json`` and ``tomllib`` decode; this component, apart from them, +finds the line and column of a value from its path (the keys and indexes leading to it) +and checks that the text there holds that very value. JSON is read with the position of +every value. In TOML, the key and its value are checked in the section the path names: +the root, a ``[table]``, or the element of an array of tables that its ``[[header]]`` +opens, with its ``[sub.tables]``. What the component cannot place (a dotted key, an +inline table, a multi-line value, a document it cannot read) has no position: the caller +gives the file, never a guessed line. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import tomllib +from bisect import bisect_right + +from coretrace_python.source.manager import LINE_BREAK, lines_of +from coretrace_python.source.model import FileLocation, SourceId + +Pointer = tuple[str | int, ...] + +_SPACE = re.compile(r"[ \t\n\r]*") +# A JSON string as RFC 8259 writes it; ``json`` decodes it. +_STRING = re.compile(r'"(?:[^"\\\x00-\x1f]|\\(?:["\\/bfnrt]|u[0-9a-fA-F]{4}))*"') +_NUMBER = re.compile(r"-?(?:0|[1-9][0-9]*)(?:\.[0-9]+)?(?:[eE][-+]?[0-9]+)?") +_LITERALS = {"true": True, "false": False, "null": None} +_HEADER = re.compile(r"\s*(\[\[?)\s*(.+?)\s*(\]\]?)\s*(?:#.*)?$") +# A bare scalar ends at a separator or the end of its line; a datetime may hold a space. +_SCALAR = re.compile(r"[^,\]}#\n]+") +_KEY_LINE = re.compile(r"""\s*("(?:[^"\\]|\\.)*"|'[^']*'|[A-Za-z0-9_-]+)\s*=\s*(.*)$""") + + +def file_location(source_id: SourceId, path: Pointer, value: str) -> FileLocation: + """The location of a value whose line could not be established: its file, its JSON + pointer, and a digest of the value, so that a baseline tells a changed value apart + without recording it.""" + + digest = hashlib.sha256(value.encode("utf-8", "surrogatepass")).hexdigest()[:16] + return FileLocation(source_id, pointer_text(path), f"sha256:{digest}") + + +def pointer_text(path: Pointer) -> str: + """``path`` as a JSON pointer (RFC 6901): ``/entries/0/name``.""" + + return "".join("/" + str(part).replace("~", "~0").replace("/", "~1") for part in path) + + +class _Lines: + def __init__(self, text: str) -> None: + self._starts = [0, *(match.end() for match in LINE_BREAK.finditer(text))] + + def position(self, offset: int) -> tuple[int, int]: + """The one-based line and column of ``offset``.""" + + line = bisect_right(self._starts, offset) + return line, offset - self._starts[line - 1] + 1 + + def offset(self, line: int, column: int) -> int: + """The offset of the one-based ``line`` and ``column``.""" + + return self._starts[line - 1] + column - 1 + + +def _same(found: object, value: object) -> bool: + return type(found) is type(value) and found == value + + +class JsonPositions: + """The position of every value of a JSON document, by structural path. A key + repeated in one object keeps its last value, as ``json`` decodes it.""" + + def __init__(self, text: str) -> None: + self._text = text + self._values: dict[Pointer, tuple[int, object]] = {} + try: + end = self._value(_SPACE.match(text, 0).end(), ()) # type: ignore[union-attr] + if _SPACE.match(text, end).end() != len(text): # type: ignore[union-attr] + raise ValueError("trailing text") + except (ValueError, IndexError, RecursionError): + self._values = {} + self._lines = _Lines(text) + + def locate(self, path: Pointer, value: object) -> tuple[int, int] | None: + """The line and column ``value`` starts at under ``path``, or None when the text + there does not hold it.""" + + found = self._values.get(path) + if found is None or not _same(found[1], value): + return None + return self._lines.position(found[0]) + + def _value(self, start: int, path: Pointer) -> int: + text = self._text + first = text[start] + if first == '"': + decoded, end = self._string(start) + self._values[path] = (start, decoded) + return end + if first == "{": + return self._object(start, path) + if first == "[": + return self._array(start, path) + for word, literal in _LITERALS.items(): + if text.startswith(word, start): + self._values[path] = (start, literal) + return start + len(word) + number = _NUMBER.match(text, start) + if number is None: + raise ValueError(f"no JSON value at {start}") + written = number.group(0) + self._values[path] = ( + start, + float(written) if any(c in written for c in ".eE") else int(written), + ) + return number.end() + + def _object(self, start: int, path: Pointer) -> int: + text = self._text + position = self._space(start + 1) + if text[position] == "}": + return position + 1 + while True: + if text[position] != '"': + raise ValueError(f"no key at {position}") + key, position = self._string(position) + position = self._space(position) + if text[position] != ":": + raise ValueError(f"no colon at {position}") + position = self._space(self._value(self._space(position + 1), (*path, key))) + if text[position] == "}": + return position + 1 + if text[position] != ",": + raise ValueError(f"no comma at {position}") + position = self._space(position + 1) + + def _array(self, start: int, path: Pointer) -> int: + text = self._text + position = self._space(start + 1) + if text[position] == "]": + return position + 1 + index = 0 + while True: + position = self._space(self._value(position, (*path, index))) + if text[position] == "]": + return position + 1 + if text[position] != ",": + raise ValueError(f"no comma at {position}") + position = self._space(position + 1) + index += 1 + + def _string(self, start: int) -> tuple[str, int]: + match = _STRING.match(self._text, start) + if match is None: + raise ValueError(f"no JSON string at {start}") + decoded: str = json.loads(match.group(0)) + return decoded, match.end() + + def _space(self, position: int) -> int: + return _SPACE.match(self._text, position).end() # type: ignore[union-attr] + + +class TomlPositions: + """The key lines of each section of a TOML document, by the structural path of its + table: the root, ``[a.b]``, or the element ``[[a]]`` opens, numbered in order. Lines + are counted as the rest of the engine counts them (``lines_of``); a line that starts + inside a multi-line string or an open array or inline table is a continuation, never + a header or a key.""" + + def __init__(self, text: str) -> None: + # Section, key -> the line number, column and written text of each value. + self._keys: dict[Pointer, dict[str, list[tuple[int, int, str]]]] = {(): {}} + arrays: dict[Pointer, int] = {} + current: Pointer = () + self._lines = lines_of(text) + # The text with one line feed per line break, where arrays are read, once each. + self._text = "\n".join(self._lines) + self._offsets = _Lines(self._text) + self._arrays: dict[int, list[tuple[int, int]]] = {} + state = _LexState() + for number, line in enumerate(self._lines, start=1): + continued = state.inside() + state.read(line) + if continued: + continue + header = _HEADER.match(line) + keys = _header_keys(header.group(2)) if header is not None else None + if ( + header is not None + and keys is not None + and (header.group(1) == "[[") == (header.group(3) == "]]") + ): + table = _resolve(keys[:-1], arrays) + (keys[-1],) + if header.group(1) == "[[": + arrays[table] = arrays.get(table, -1) + 1 + table = (*table, arrays[table]) + else: + table = _resolve(keys, arrays) + current = table + self._keys.setdefault(current, {}) + continue + match = _KEY_LINE.match(line) + if match is None: + continue + try: + key = _key(match.group(1)) + except tomllib.TOMLDecodeError: + continue + self._keys[current].setdefault(key, []).append( + (number, match.start(2) + 1, match.group(2)) + ) + + def locate(self, path: Pointer, value: object) -> tuple[int, int] | None: + """The line and column ``value`` starts at under ``path``, or None when the + section its path names has no line setting that key to that value.""" + + if not path: + return None + if isinstance(path[-1], int): + return self._element(path, value) + for number, column, written in self._keys.get(path[:-1], {}).get(path[-1], ()): + if _same(_value(written), value): + return number, column + return None + + def _element(self, path: Pointer, value: object) -> tuple[int, int] | None: + """An element of an array a key holds, the array written over one or several + lines (``dependencies = [\\n "a",\\n "b",\\n]``). An element whose own text + spans lines has no position: the line it starts on does not hold it.""" + + if len(path) < 2 or isinstance(path[-2], int): + return None + index = path[-1] + assert isinstance(index, int) + for number, column, _ in self._keys.get(path[:-2], {}).get(path[-2], ()): + start = self._offsets.offset(number, column) + if start not in self._arrays: + self._arrays[start] = _array_elements(self._text, start) + elements = self._arrays[start] + if index >= len(elements): + continue + first, end = elements[index] + written = self._text[first:end] + if "\n" not in written and _same(_value(written), value): + return self._offsets.position(first) + return None + + +class _LexState: + """Where a TOML line leaves the reader: inside a multi-line string (its delimiter), + or inside arrays and inline tables opened by a value (their depth). Strings and + comments are skipped, so a delimiter or a bracket written in them counts for + nothing.""" + + def __init__(self) -> None: + self.string: str | None = None + self.depth = 0 + + def inside(self) -> bool: + return self.string is not None or self.depth > 0 + + def read(self, line: str) -> None: + position, valued = 0, self.depth > 0 + while position < len(line): + if self.string is not None: + if self.string == '"""' and line[position] == "\\": + position += 2 + elif line.startswith(self.string, position): + # Up to two more quotes before the delimiter belong to the string. + run = len(line[position:]) - len(line[position:].lstrip(self.string[0])) + self.string, position = None, position + min(run, 5) + else: + position += 1 + continue + character = line[position] + if character == "#": + return + if line.startswith(('"""', "'''"), position): + self.string, position = line[position : position + 3], position + 3 + elif character == '"': + position += 1 + while position < len(line) and line[position] != '"': + position += 2 if line[position] == "\\" else 1 + position += 1 + elif character == "'": + closing = line.find("'", position + 1) + position = len(line) if closing < 0 else closing + 1 + else: + if character == "=" and self.depth == 0: + valued = True + elif character in "[{" and valued: + self.depth += 1 + elif character in "]}" and self.depth > 0: + self.depth -= 1 + position += 1 + + +def _value(written: str) -> object: + """The value TOML text denotes, or None when it is not one value.""" + + try: + return tomllib.loads(f"v = {written}")["v"] + except tomllib.TOMLDecodeError: + return None + + +def _array_elements(text: str, start: int) -> list[tuple[int, int]]: + """The start and end offsets of the values of the array that opens at ``start``, + skipping comments; empty when no array opens there or it does not close.""" + + if not text.startswith("[", start): + return [] + elements: list[tuple[int, int]] = [] + position = start + 1 + while position < len(text): + character = text[position] + if character.isspace() or character == ",": + position += 1 + elif character == "#": + newline = text.find("\n", position) + position = len(text) if newline < 0 else newline + elif character == "]": + return elements + else: + end = _value_end(text, position) + elements.append((position, end)) + position = end + return [] + + +def _value_end(text: str, start: int) -> int: + """The offset just past the TOML value that starts at ``start``: a string, an array + or inline table with what it holds, or a bare scalar.""" + + if text[start] in "\"'": + return _string_end(text, start) + if text[start] not in "[{": + scalar = _SCALAR.match(text, start) + return start + len(scalar.group(0).rstrip()) if scalar is not None else start + 1 + depth, position = 0, start + while position < len(text): + character = text[position] + if character in "\"'": + position = _string_end(text, position) + continue + if character == "#": + newline = text.find("\n", position) + position = len(text) if newline < 0 else newline + continue + if character in "[{": + depth += 1 + elif character in "]}": + depth -= 1 + if depth == 0: + return position + 1 + position += 1 + return len(text) + + +def _string_end(text: str, start: int) -> int: + """The offset just past the TOML string that starts at ``start``.""" + + quote = text[start] + if text.startswith(quote * 3, start): + position = start + 3 + while position < len(text): + if quote == '"' and text[position] == "\\": + position += 2 + elif text.startswith(quote * 3, position): + run = 3 + while run < 5 and text.startswith(quote, position + run): + run += 1 + return position + run + else: + position += 1 + return len(text) + position = start + 1 + while position < len(text) and text[position] not in (quote, "\n"): + position += 2 if quote == '"' and text[position] == "\\" else 1 + return position + 1 + + +def _resolve(keys: tuple[str, ...], arrays: dict[Pointer, int]) -> Pointer: + """The structural path of a table named by ``keys``: a key that names an array of + tables stands for its last element.""" + + path: Pointer = () + for key in keys: + path = (*path, key) + if path in arrays: + path = (*path, arrays[path]) + return path + + +def _header_keys(written: str) -> tuple[str, ...] | None: + """The keys of a table header, read as ``tomllib`` reads them; None when the text + is not a header (an array literal on its own line).""" + + try: + table: object = tomllib.loads(f"[{written}]") + except tomllib.TOMLDecodeError: + return None + keys: list[str] = [] + while isinstance(table, dict) and len(table) == 1: + key = next(iter(table)) + keys.append(key) + table = table[key] + return tuple(keys) if keys else None + + +def _key(written: str) -> str: + return next(iter(tomllib.loads(f"{written} = 0"))) diff --git a/tests/test_baseline.py b/tests/test_baseline.py index e57b887..90d37d9 100644 --- a/tests/test_baseline.py +++ b/tests/test_baseline.py @@ -56,13 +56,13 @@ def test_fingerprint_ignores_the_line_number_but_not_the_line_text(tmp_path: Pat source.write_text(EVAL.replace("eval(code)", "eval(code.strip())"), encoding="utf-8") changed = fingerprint(finding(str(source), 4), tmp_path) - assert before == after == ("app.py", "dangerous-eval", "run", "eval(code)") + assert before == after == ("app.py", "dangerous-eval", "run", "eval(code)", "") assert changed != before assert fingerprint(finding(str(source), 4, "weak-crypto"), tmp_path) != before def test_fingerprint_falls_back_to_the_line_number_without_the_file(tmp_path: Path) -> None: - assert fingerprint(finding(str(tmp_path / "gone.py"), 7), tmp_path) == ("gone.py", "dangerous-eval", "run", "7") + assert fingerprint(finding(str(tmp_path / "gone.py"), 7), tmp_path) == ("gone.py", "dangerous-eval", "run", "7", "") def test_baseline_round_trips_and_partitions_with_counts(tmp_path: Path) -> None: @@ -75,7 +75,7 @@ def test_baseline_round_trips_and_partitions_with_counts(tmp_path: Path) -> None loaded = Baseline.load(path) new, baselined = loaded.partition((first, second), tmp_path) - assert json.loads(path.read_text(encoding="utf-8"))["schema"] == 1 + assert json.loads(path.read_text(encoding="utf-8"))["schema"] == 2 assert baselined == (first,) and new == (second,) diff --git a/tests/test_baseline_schema_1.py b/tests/test_baseline_schema_1.py new file mode 100644 index 0000000..b11ce2f --- /dev/null +++ b/tests/test_baseline_schema_1.py @@ -0,0 +1,100 @@ +"""Acceptance tests for the transition from schema 1 baselines (issues #207, #210). + +Schema 1 files are still read, and their entries match exactly as they did: by file, +rule, function and the text of the line. Before #210, a finding in a JSON, TOML or lock +file was recorded with the text of its key's first line, which was its own line only +when the key was not repeated. Such an entry still identifies that finding with +certainty; any other finding, a new secret included, is new: an old entry never +accounts for a finding it cannot identify. When a schema 1 baseline leaves findings of +such files new, the check says so and asks for the baseline to be recorded again. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from coretrace_python.cli import main + +TOKEN = "Zx81kQpLw0RtY7vBn3MsD9cF2hJ6gK4a" +OTHER = "Pq72mWnE5rTy8uIo1aSdF3gH6jK9lZxC" +NOTICE = "schema 1 baseline" + + +def legacy(path: Path, *entries: tuple[str, str, str, str, int]) -> Path: + findings = [ + {"path": file, "rule": rule, "function": function, "line": line, "count": count} + for file, rule, function, line, count in entries + ] + path.write_text(json.dumps({"schema": 1, "findings": findings}), encoding="utf-8") + return path + + +@pytest.fixture +def root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + root = tmp_path / "proj" + root.mkdir() + (root / "app.py").write_text("def run(code):\n eval(code)\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + return root + + +def test_a_new_secret_is_never_accounted_for_by_an_old_entry( + root: Path, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + (root / "settings.json").write_text( + f'{{\n "db": {{"password": "{OTHER}"}}\n}}\n', encoding="utf-8" + ) + baseline = legacy( + tmp_path / "baseline.json", + ("settings.json", "hardcoded-credential", "", f'"db": {{"api_token": "{TOKEN}"}}', 1), + ("app.py", "dangerous-eval", "run", "eval(code)", 1), + ) + + assert main(["--check", "proj", "--baseline", str(baseline), "--format", "json"]) == 1 + captured = capsys.readouterr() + report = json.loads(captured.out) + + assert [f["metadata"]["name"] for f in report["findings"]] == ["password"] + assert [f["rule_id"] for f in report["baselined"]] == ["dangerous-eval"] + assert NOTICE in captured.err + + +def test_a_repeated_key_is_recognised_only_where_its_old_entry_was_certain( + root: Path, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + text = ( + f'{{\n "first": {{"api_token": "{TOKEN}"}},\n "second": {{"api_token": "{OTHER}"}}\n}}\n' + ) + (root / "settings.json").write_text(text, encoding="utf-8") + baseline = legacy( + tmp_path / "baseline.json", + ("settings.json", "hardcoded-credential", "", f'"first": {{"api_token": "{TOKEN}"}},', 2), + ("app.py", "dangerous-eval", "run", "eval(code)", 1), + ) + + assert main(["--check", "proj", "--baseline", str(baseline), "--format", "json"]) == 1 + captured = capsys.readouterr() + report = json.loads(captured.out) + + assert [f["location"]["line"] for f in report["findings"]] == [3] + assert [ + f["location"]["line"] for f in report["baselined"] if f["rule_id"] != "dangerous-eval" + ] == [2] + assert NOTICE in captured.err + + +def test_a_schema_1_baseline_of_python_findings_matches_as_before( + root: Path, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + baseline = legacy( + tmp_path / "baseline.json", ("app.py", "dangerous-eval", "run", "eval(code)", 1) + ) + + assert main(["--check", "proj", "--baseline", str(baseline)]) == 0 + captured = capsys.readouterr() + + assert captured.out.startswith("no findings, 1 baselined\n") + assert NOTICE not in captured.err diff --git a/tests/test_file_locations.py b/tests/test_file_locations.py new file mode 100644 index 0000000..05e1106 --- /dev/null +++ b/tests/test_file_locations.py @@ -0,0 +1,213 @@ +"""Acceptance tests for issue #210: findings in structured files, located or not. + +A value found in a JSON or TOML file is reported at its own line, from its structural +path. A value whose line cannot be established gets a file location: the file and the +JSON pointer of the value, never a guessed line. Reports show it without a line (JSON +``null``, SARIF without ``region``, text ``path: …``); an inline suppression, which +names a line, does not apply to it; the baseline recognises it by its pointer, in a +schema 2 file (schema 1 files: ``test_baseline_schema_1.py``). +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.findings import Finding +from coretrace_python.source import FileLocation, SourceSpan +from coretrace_python.source.positions import file_location + +PLUGINS = Path(__file__).resolve().parent.parent / "src" / "coretrace_python" / "bundled" +TOKEN = "Zx81kQpLw0RtY7vBn3MsD9cF2hJ6gK4a" +OTHER = "Pq72mWnE5rTy8uIo1aSdF3gH6jK9lZxC" + + +def project(root: Path, files: dict[str, str]) -> Path: + for name, text in files.items(): + (root / name).write_text(text, encoding="utf-8") + (root / "app.py").write_text("x = 1\n", encoding="utf-8") + return root + + +def credentials(root: Path) -> list[Finding]: + return [ + f + for f in engine.analyze_project(root, [PLUGINS]).findings + if f.rule_id == "hardcoded-credential" + ] + + +def test_values_of_one_key_are_located_at_their_own_lines(tmp_path: Path) -> None: + text = ( + f'{{\n "first": {{"api_token": "{TOKEN}"}},\n "second": {{"api_token": "{OTHER}"}}\n}}\n' + ) + findings = credentials(project(tmp_path, {"settings.json": text})) + + assert sorted( + (f.span.start_line, f.span.start_column) for f in findings if isinstance(f.span, SourceSpan) + ) == [ + (2, 26), + (3, 27), + ] + + +def test_a_value_that_cannot_be_located_gets_a_file_location(tmp_path: Path) -> None: + (finding,) = credentials(project(tmp_path, {"settings.toml": f'a.api_token = "{TOKEN}"\n'})) + + assert finding.span == file_location(finding.span.source_id, ("a", "api_token"), TOKEN) + assert Path(str(finding.span.source_id)).name == "settings.toml" + + +@pytest.fixture +def unlocated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """A project with one credential that cannot be located (a dotted TOML key, with an + inline suppression on its line) and one that can, suppressed on its line.""" + + root = tmp_path / "proj" + root.mkdir() + project( + root, + { + "settings.toml": ( + f'a.api_token = "{TOKEN}" # coretrace: ignore\n' + f'[db]\npassword = "{OTHER}" # coretrace: ignore\n' + ) + }, + ) + monkeypatch.chdir(tmp_path) + return root + + +def test_reports_show_a_file_location_without_a_line( + unlocated: Path, capsys: pytest.CaptureFixture[str] +) -> None: + main(["--check", "proj", "--format", "json"]) + (record,) = [ + f + for f in json.loads(capsys.readouterr().out)["findings"] + if f["rule_id"] == "hardcoded-credential" + ] + assert record["location"] == { + "path": "settings.toml", + "line": None, + "column": None, + "end_line": None, + "end_column": None, + "pointer": "/a/api_token", + } + + main(["--check", "proj", "--format", "sarif"]) + results = json.loads(capsys.readouterr().out)["runs"][0]["results"] + (result,) = [ + r for r in results if r["ruleId"] == "hardcoded-credential" and "suppressions" not in r + ] + (location,) = result["locations"] + assert "region" not in location["physicalLocation"] + assert location["physicalLocation"]["artifactLocation"]["uri"] == "settings.toml" + assert location["properties"] == {"pointer": "/a/api_token"} + + main(["--check", "proj"]) + lines = capsys.readouterr().out.splitlines() + assert ( + "settings.toml: high hardcoded-credential: Hardcoded credential in api_token: Zx81… (32 characters)" + in lines + ) + + +def test_an_inline_suppression_applies_to_verified_lines_only(unlocated: Path) -> None: + analysis = engine.analyze_project(unlocated, [PLUGINS]) + + assert [type(f.span) for f in analysis.findings if f.rule_id == "hardcoded-credential"] == [ + FileLocation + ] + assert [f.metadata["name"] for f in analysis.suppressed] == ["password"] + + +def test_the_baseline_recognises_a_file_location_by_its_pointer( + unlocated: Path, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + baseline = tmp_path / "baseline.json" + assert main(["--check", "proj", "--baseline", str(baseline)]) == 0 + document = json.loads(baseline.read_text(encoding="utf-8")) + capsys.readouterr() + + assert document["schema"] == 2 + assert { + "path": "settings.toml", + "rule": "hardcoded-credential", + "function": "", + "pointer": "/a/api_token", + }.items() <= next( + e for e in document["findings"] if e["rule"] == "hardcoded-credential" + ).items() + assert main(["--check", "proj", "--baseline", str(baseline)]) == 0 + assert capsys.readouterr().out.startswith("no findings, 1 suppressed, 1 baselined\n") + + +def test_a_changed_value_at_a_file_location_is_a_new_finding( + unlocated: Path, tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + """The pointer says where the value is, not what it is: the baseline also keeps a + digest of the value, never the value itself, so a new secret there is new.""" + + baseline = tmp_path / "baseline.json" + main(["--check", "proj", "--baseline", str(baseline)]) + assert TOKEN not in baseline.read_text(encoding="utf-8") + settings = unlocated / "settings.toml" + settings.write_text( + settings.read_text(encoding="utf-8").replace(TOKEN, OTHER[::-1]), encoding="utf-8" + ) + capsys.readouterr() + + assert main(["--check", "proj", "--baseline", str(baseline)]) == 1 + assert capsys.readouterr().out.startswith("settings.toml: high hardcoded-credential") + + +def test_the_baseline_reads_the_line_a_finding_is_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Lines are counted by line feeds everywhere: a U+2028 in a string above a secret + does not make the baseline record another line's text.""" + + root = tmp_path / "proj" + root.mkdir() + settings = ( + project(root, {"settings.toml": f'note = "a\u2028b"\napi_token = "{TOKEN}"\n'}) + / "settings.toml" + ) + monkeypatch.chdir(tmp_path) + baseline = tmp_path / "baseline.json" + main(["--check", "proj", "--baseline", str(baseline)]) + settings.write_text( + settings.read_text(encoding="utf-8").replace(TOKEN, OTHER), encoding="utf-8" + ) + capsys.readouterr() + + assert main(["--check", "proj", "--baseline", str(baseline)]) == 1 + assert capsys.readouterr().out.startswith("settings.toml:2:13: high hardcoded-credential") + + +def test_a_changed_secret_written_over_several_lines_in_an_array_is_new( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """A multi-line element is not placed at the line it starts on, which does not hold + it: it gets a file location, whose digest tells a changed value apart.""" + + key = "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEA1b2c3d4e5f6\n-----END RSA PRIVATE KEY-----\n" + root = tmp_path / "proj" + root.mkdir() + config = project(root, {"config.toml": f'[deploy]\nkeys = ["""\n{key}"""]\n'}) / "config.toml" + monkeypatch.chdir(tmp_path) + baseline = tmp_path / "baseline.json" + main(["--check", "proj", "--baseline", str(baseline)]) + config.write_text( + config.read_text(encoding="utf-8").replace("1b2c3d4e5f6", "9z8y7x6w5v4"), encoding="utf-8" + ) + capsys.readouterr() + + assert main(["--check", "proj", "--baseline", str(baseline)]) == 1 + assert capsys.readouterr().out.startswith("config.toml: high hardcoded-secret") diff --git a/tests/test_line_numbering.py b/tests/test_line_numbering.py new file mode 100644 index 0000000..ebf928c --- /dev/null +++ b/tests/test_line_numbering.py @@ -0,0 +1,136 @@ +"""Acceptance tests for one rule of line numbering across the engine (#207, #210). + +Python reads a line break as ``\\n``, ``\\r\\n`` or a lone ``\\r``, never as U+2028, U+0085, +a form feed or the other separators ``str.splitlines`` also breaks on. Every producer of +a line number (the parser, the position component, the readers of ``.env``, YAML, INI and +requirements files) and every consumer (inline suppressions, the baseline) counts lines +by that rule, ``coretrace_python.source.lines_of``; otherwise a suppression or a baseline +entry refers to another line, and a new secret can be absorbed. A requirement declared +in ``pyproject.toml`` is located at its own element, never at the first line mentioning +it nor at a default line. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.cli import main +from coretrace_python.dependency import Advisory, dump_advisories +from coretrace_python.findings import Finding, Severity +from coretrace_python.source import SourceSpan, lines_of +from coretrace_python.source.positions import file_location + +PLUGINS = Path(__file__).resolve().parent.parent / "src" / "coretrace_python" / "bundled" +TOKEN = "ghp_" + "a1B2" * 9 +ROTATED = "ghp_" + "Z9y8" * 9 + + +def test_lines_are_counted_as_python_counts_them() -> None: + assert lines_of("a
b\nc\r\nd\re\x0cf\x85g\n") == ["a
b", "c", "d", "e\x0cf\x85g"] + assert lines_of("") == [] + assert lines_of("last") == ["last"] + + +@pytest.fixture +def checked(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + root = tmp_path / "proj" + root.mkdir() + monkeypatch.chdir(tmp_path) + return root + + +@pytest.mark.parametrize( + ("name", "text"), + [ + (".env", f"A=1
B=2\nGITHUB_TOKEN={TOKEN}\nC=3\n"), + ("config.yaml", f"a: 'x
y'\ntoken: {TOKEN}\nc: 3\n"), + ("app.ini", f"[main]\x0c\ntoken = {TOKEN}\nc = 3\n"), + ("app.py", f"# a\rb\nTOKEN = '{TOKEN}'\nC = 3\n"), + ], +) +def test_a_rotated_secret_below_a_separator_is_a_new_finding( + checked: Path, tmp_path: Path, name: str, text: str, capsys: pytest.CaptureFixture[str] +) -> None: + (checked / name).write_text(text, encoding="utf-8", newline="") + baseline = tmp_path / "baseline.json" + main(["--check", "proj", "--baseline", str(baseline)]) + (checked / name).write_text(text.replace(TOKEN, ROTATED), encoding="utf-8", newline="") + capsys.readouterr() + + assert main(["--check", "proj", "--baseline", str(baseline)]) == 1 + assert "hardcoded-secret" in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("name", "text", "line"), + [ + (".env", f"# note
x\nGITHUB_TOKEN={TOKEN}\nC=3 # coretrace: ignore\n", 2), + ("app.py", f"# a\rb\nTOKEN = '{TOKEN}'\nC = 3 # coretrace: ignore\n", 3), + ], +) +def test_a_suppression_on_the_next_line_does_not_apply( + checked: Path, name: str, text: str, line: int +) -> None: + (checked / name).write_text(text, encoding="utf-8", newline="") + if name != "app.py": + (checked / "app.py").write_text("x = 1\n", encoding="utf-8") + + (finding,) = [ + f + for f in engine.analyze_project(checked, [PLUGINS]).findings + if f.rule_id == "hardcoded-secret" + ] + + assert isinstance(finding.span, SourceSpan) and finding.span.start_line == line + + +ADVISORY = Advisory("CVE-2099-0208", "requests", "<2.1", "requests is vulnerable", Severity.MEDIUM) + + +def declared(root: Path, name: str, text: str) -> Finding: + (root / name).write_text(text, encoding="utf-8") + (root / "advisories.json").write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + (root / "app.py").write_text("import requests\n", encoding="utf-8") + (finding,) = [ + f + for f in engine.analyze_project(root, [PLUGINS]).findings + if f.rule_id == "vulnerable-dependency" and f.metadata["advisory"] == ADVISORY.id + ] + return finding + + +def test_a_requirements_file_counts_lines_as_python(tmp_path: Path) -> None: + finding = declared(tmp_path, "requirements.txt", "# deps\x0c\nflask==1.0\nrequests==2.0.0\n") + + assert isinstance(finding.span, SourceSpan) and finding.span.start_line == 3 + + +def test_a_pyproject_requirement_is_located_at_its_own_element(tmp_path: Path) -> None: + text = ( + '# pin requests==2.0.0 later\n[project]\nname = "app"\n' + 'dependencies = [\n "requests-toolbelt",\n "requests==2.0.0", # pinned\n "django>=1.0",\n]\n' + ) + finding = declared(tmp_path, "pyproject.toml", text) + + assert isinstance(finding.span, SourceSpan) + assert (finding.span.start_line, finding.span.start_column) == (6, 5) + + +def test_a_poetry_requirement_is_located_at_its_key(tmp_path: Path) -> None: + text = '[tool.poetry.dependencies]\npython = "^3.11"\nrequests = { version = "<2.1" }\n' + finding = declared(tmp_path, "pyproject.toml", text) + + assert isinstance(finding.span, SourceSpan) + assert (finding.span.start_line, finding.span.start_column) == (3, 12) + + +def test_a_pyproject_requirement_that_cannot_be_placed_has_a_file_location(tmp_path: Path) -> None: + text = 'project.name = "app"\nproject.dependencies = ["requests==2.0.0"]\n' + finding = declared(tmp_path, "pyproject.toml", text) + + assert finding.span == file_location( + finding.span.source_id, ("project", "dependencies", 0), "requests==2.0.0" + ) diff --git a/tests/test_lock_positions.py b/tests/test_lock_positions.py new file mode 100644 index 0000000..4868474 --- /dev/null +++ b/tests/test_lock_positions.py @@ -0,0 +1,108 @@ +"""Acceptance tests for issue #207: a vulnerable package pinned in a lock file. + +The finding is located at the package's own entry, the ``name = "..."`` line of its +``[[package]]`` table, which the position component checks; not at the first line that +mentions the name, which may be another package's dependency list. When the entry cannot +be placed, the finding has a file location, never a guessed line. The message states the +evidence: the pinned version, the lock file, and the packages that require it. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from coretrace_python import engine +from coretrace_python.dependency import Advisory, dump_advisories +from coretrace_python.findings import Finding, Severity +from coretrace_python.source import SourceSpan +from coretrace_python.source.positions import file_location + +PLUGINS = Path(__file__).resolve().parent.parent / "src" / "coretrace_python" / "bundled" +ADVISORY = Advisory( + "CVE-2099-0207", "werkzeug", "<3.1.6", "safe_join accepts a device name", Severity.MEDIUM +) + +UV_LOCK = """version = 1 + +[[package]] +name = "flask" +version = "3.1.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "werkzeug" }, +] + +[[package]] +name = "werkzeug" +version = "3.1.5" +source = { registry = "https://pypi.org/simple" } +""" +POETRY_LOCK = """[[package]] +name = "flask" +version = "3.1.2" + +[package.dependencies] +werkzeug = ">=3.1" + +[[package]] +name = "werkzeug" +version = "3.1.5" +""" +INLINE_LOCK = 'version = 1\npackage = [{ name = "werkzeug", version = "3.1.5" }]\n' + + +def vulnerable(tmp_path: Path, lock: str, text: str) -> Finding: + (tmp_path / lock).write_text(text, encoding="utf-8") + (tmp_path / "advisories.json").write_text(dump_advisories((ADVISORY,)), encoding="utf-8") + (tmp_path / "app.py").write_text("import werkzeug\n", encoding="utf-8") + (finding,) = [ + f + for f in engine.analyze_project(tmp_path, [PLUGINS]).findings + if f.rule_id == "vulnerable-dependency" and f.metadata["advisory"] == ADVISORY.id + ] + return finding + + +@pytest.mark.parametrize( + ("lock", "text", "line"), [("uv.lock", UV_LOCK, 12), ("poetry.lock", POETRY_LOCK, 9)] +) +def test_a_locked_package_is_located_at_its_own_entry( + tmp_path: Path, lock: str, text: str, line: int +) -> None: + finding = vulnerable(tmp_path, lock, text) + + assert isinstance(finding.span, SourceSpan) + assert (finding.span.start_line, finding.span.start_column) == (line, 8) + + +@pytest.mark.parametrize(("lock", "text"), [("uv.lock", UV_LOCK), ("poetry.lock", POETRY_LOCK)]) +def test_the_message_states_the_pinned_version_the_lock_file_and_its_dependents( + tmp_path: Path, lock: str, text: str +) -> None: + finding = vulnerable(tmp_path, lock, text) + + assert finding.message == ( + f"CVE-2099-0207: werkzeug 3.1.5, pinned in {lock}, is in <3.1.6 (required by flask): " + "safe_join accepts a device name" + ) + + +def test_a_locked_package_that_cannot_be_placed_has_a_file_location(tmp_path: Path) -> None: + finding = vulnerable(tmp_path, "uv.lock", INLINE_LOCK) + + assert finding.span == file_location(finding.span.source_id, ("package", 0, "name"), "werkzeug") + assert finding.message.startswith( + "CVE-2099-0207: werkzeug 3.1.5, pinned in uv.lock, is in <3.1.6: " + ) + + +def test_a_declared_requirement_keeps_its_wording(tmp_path: Path) -> None: + finding = vulnerable(tmp_path, "requirements.txt", "werkzeug>=3.0\n") + + assert ( + finding.message + == "CVE-2099-0207: werkzeug <3.1.6 is allowed by werkzeug>=3.0: safe_join accepts a device name" + ) + assert isinstance(finding.span, SourceSpan) and finding.span.start_line == 1 diff --git a/tests/test_structured_positions.py b/tests/test_structured_positions.py new file mode 100644 index 0000000..b851ae2 --- /dev/null +++ b/tests/test_structured_positions.py @@ -0,0 +1,196 @@ +"""Acceptance tests for issues #207 and #210: where a value of a JSON or TOML file is written. + +The detectors judge the values ``json`` and ``tomllib`` decode; where each value is +written is found by a separate component, from its structural path (the keys and indexes +leading to it) and checked against the value itself. JSON is read with the position of +every value, so repeated keys, nested arrays and escaped strings are told apart. In TOML, +the key and its value are checked in the section the path names, an array of tables +included: the element is the one its ``[[header]]`` opens, and its key must hold the +value. A value the component cannot place is not given a guessed line: it has no +position (None). +""" + +from __future__ import annotations + +import pytest + +from coretrace_python.source.positions import JsonPositions, TomlPositions, pointer_text + +HAR = """{ + "entries": [ + {"request": {"headers": [["Host", "example.org"], ["X-Amz-Cf-Id", "Kk1li6yk9u"]]}}, + {"request": {"headers": [["Host", "example.org"], ["X-Amz-Cf-Id", "PaWnQ2x7ab"]]}} + ] +} +""" + + +def test_a_json_value_is_found_by_its_path_among_repeated_keys() -> None: + text = '{\n "first": {"api_token": "Zx81"},\n "second": {"api_token": "Pq72"}\n}\n' + positions = JsonPositions(text) + + assert positions.locate(("first", "api_token"), "Zx81") == (2, 26) + assert positions.locate(("second", "api_token"), "Pq72") == (3, 27) + + +def test_json_values_in_nested_arrays_are_found_at_their_own_line() -> None: + positions = JsonPositions(HAR) + + assert positions.locate(("entries", 0, "request", "headers", 1, 1), "Kk1li6yk9u") == (3, 71) + assert positions.locate(("entries", 1, "request", "headers", 1, 1), "PaWnQ2x7ab") == (4, 71) + + +def test_escaped_json_strings_are_read_as_json_reads_them() -> None: + text = '{"k\\"q": "\\u00e9t\\u00e9", "plain": "a\\nb"}' + positions = JsonPositions(text) + + assert positions.locate(('k"q',), "été") == (1, 10) + assert positions.locate(("plain",), "a\nb") == (1, 36) + + +def test_a_duplicate_json_key_is_the_last_one_as_json_decodes_it() -> None: + positions = JsonPositions('{\n"k": "one",\n"k": "two"\n}') + + assert positions.locate(("k",), "two") == (3, 6) + assert positions.locate(("k",), "one") is None + + +def test_a_json_value_that_does_not_match_has_no_position() -> None: + positions = JsonPositions('{"a": {"b": "x"}}') + + assert positions.locate(("a", "b"), "y") is None + assert positions.locate(("a", "c"), "x") is None + assert JsonPositions("{not json").locate(("a",), "x") is None + + +LOCK = """version = 1 + +[[package]] +name = "flask" +version = "3.1.2" +dependencies = [ + { name = "werkzeug" }, +] + +[[package]] +name = "werkzeug" +version = "3.1.5" + +[package.metadata] +token = "abc123" +""" + + +def test_a_toml_value_is_found_in_the_section_its_path_names() -> None: + positions = TomlPositions( + 'title = "app"\n\n[db]\npassword = "s3cr3t" # set by ops\nport = 5432\n' + ) + + assert positions.locate(("title",), "app") == (1, 9) + assert positions.locate(("db", "password"), "s3cr3t") == (4, 12) + assert positions.locate(("db", "port"), 5432) == (5, 8) + + +def test_an_element_of_an_array_of_tables_is_checked_by_its_key_and_value() -> None: + positions = TomlPositions(LOCK) + + assert positions.locate(("package", 0, "name"), "flask") == (4, 8) + assert positions.locate(("package", 1, "name"), "werkzeug") == (11, 8) + assert positions.locate(("package", 1, "name"), "flask") is None + assert positions.locate(("package", 2, "name"), "werkzeug") is None + + +def test_a_subtable_of_an_array_element_belongs_to_that_element() -> None: + positions = TomlPositions(LOCK) + + assert positions.locate(("package", 1, "metadata", "token"), "abc123") == (15, 9) + assert positions.locate(("package", 0, "metadata", "token"), "abc123") is None + + +def test_quoted_toml_keys_and_headers_are_read() -> None: + positions = TomlPositions('[servers."alpha.example"]\n"api-key" = "k-1234"\n') + + assert positions.locate(("servers", "alpha.example", "api-key"), "k-1234") == (2, 13) + + +def test_toml_values_the_component_cannot_place_have_no_position() -> None: + positions = TomlPositions('a.token = "x1"\ndep = { token = "x2" }\nnote = """\nx3\n"""\n') + + assert positions.locate(("a", "token"), "x1") is None + assert positions.locate(("dep", "token"), "x2") is None + assert positions.locate(("note",), "x3\n") is None + assert TomlPositions("[unclosed").locate(("a",), "x") is None + + +def test_a_header_or_key_written_inside_a_toml_string_is_not_one() -> None: + commented = 'note = "a" # use """ for docs\nexample = """\n[t]\nk = "S"\n"""\n[t]\nk = "S"\n' + escaped = 'doc = """\nescaped \\"""\n[t]\nk = "S"\n"""\n[t]\nk = "S"\n' + one_line = 's = \'"""\'\n[t]\nk = "S"\n' + + assert TomlPositions(commented).locate(("t", "k"), "S") == (7, 5) + assert TomlPositions(escaped).locate(("t", "k"), "S") == (7, 5) + assert TomlPositions(one_line).locate(("t", "k"), "S") == (3, 5) + + +@pytest.mark.parametrize("quote", ['"', "'"]) +def test_quotes_before_a_closing_delimiter_belong_to_the_string(quote: str) -> None: + triple = quote * 3 + text = ( + f"a = [{triple}x{triple}{quote}, {triple}y{triple}]\n" + f'c = {triple}\n]\n[t]\nk = "v"\n{triple}\n[t]\nk = "v"\n' + ) + + assert TomlPositions(text).locate(("t", "k"), "v") == (8, 5) + + +def test_an_array_element_alone_on_its_line_is_not_a_toml_header() -> None: + positions = TomlPositions('arr = [\n ["b"]\n]\nk = "v"\n') + + assert positions.locate(("k",), "v") == (4, 5) + + +def test_toml_lines_are_counted_by_line_feeds_only() -> None: + positions = TomlPositions('# caf\u2028e\napi_token = "S"\r\nport = 1\r\n') + + assert positions.locate(("api_token",), "S") == (2, 13) + assert positions.locate(("port",), 1) == (3, 8) + + +def test_an_element_of_a_toml_array_is_found_and_checked() -> None: + text = ( + '[project]\ndependencies = [\n "a", # first, ["x"]\n { n = [1, 2] },\n' + ' \'lit, "b"\', "c"\n]\noptional = ["d", "e"]\nname = "app"\n' + ) + positions = TomlPositions(text) + + assert positions.locate(("project", "dependencies", 0), "a") == (3, 5) + assert positions.locate(("project", "dependencies", 1), {"n": [1, 2]}) == (4, 5) + assert positions.locate(("project", "dependencies", 2), 'lit, "b"') == (5, 5) + assert positions.locate(("project", "dependencies", 3), "c") == (5, 17) + assert positions.locate(("project", "optional", 1), "e") == (7, 18) + assert positions.locate(("project", "dependencies", 0), "b") is None + assert positions.locate(("project", "dependencies", 4), "c") is None + assert positions.locate(("project", "name", 0), "app") is None + + +def test_an_array_element_written_over_several_lines_has_no_position() -> None: + positions = TomlPositions('keys = ["""\nfirst\n""", "b"]\nother = [[1,\n 2]]\n') + + assert positions.locate(("keys", 0), "first\n") is None + assert positions.locate(("keys", 1), "b") == (3, 6) + assert positions.locate(("other", 0), [1, 2]) is None + + +def test_a_datetime_written_with_a_space_is_one_array_element() -> None: + positions = TomlPositions('a = [1979-05-27 07:32:00, "s", "s" ]\n') + + assert positions.locate(("a", 1), "s") == (1, 27) + assert positions.locate(("a", 2), "s") == (1, 32) + + +def test_a_structural_path_is_written_as_a_json_pointer() -> None: + assert pointer_text(("packages", "node_modules/a", "dependencies", "js-tokens")) == ( + "/packages/node_modules~1a/dependencies/js-tokens" + ) + assert pointer_text(("entries", 0, "headers", 1, 1)) == "/entries/0/headers/1/1" + assert pointer_text(("a~b",)) == "/a~0b"