From 5000adf8e4e9e9b9804e6c5fa125f5df59b7a22d Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 12:21:24 +0200 Subject: [PATCH 1/8] feat(source): locate JSON and TOML values by their structural path A component apart from the rules finds the line and column of a value from its path and checks that the text there holds that very value. 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 element of an array of tables included. What it cannot place has no position, never a guessed line. Refs #207, #210 --- src/coretrace_python/source/positions.py | 231 +++++++++++++++++++++++ tests/test_structured_positions.py | 128 +++++++++++++ 2 files changed, 359 insertions(+) create mode 100644 src/coretrace_python/source/positions.py create mode 100644 tests/test_structured_positions.py diff --git a/src/coretrace_python/source/positions.py b/src/coretrace_python/source/positions.py new file mode 100644 index 0000000..ee3bf05 --- /dev/null +++ b/src/coretrace_python/source/positions.py @@ -0,0 +1,231 @@ +"""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 json +import re +import tomllib +from bisect import bisect_right + +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*(?:#.*)?$") +_KEY_LINE = re.compile(r"""\s*("(?:[^"\\]|\\.)*"|'[^']*'|[A-Za-z0-9_-]+)\s*=\s*(.*)$""") + + +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 re.finditer("\n", 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 _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.""" + + def __init__(self, text: str) -> None: + self._lines = text.splitlines() + self._sections: dict[Pointer, list[int]] = {(): []} + arrays: dict[Pointer, int] = {} + current: Pointer = () + quoted: str | None = None + for number, line in enumerate(self._lines): + if quoted is not None: + if line.count(quoted) % 2: + quoted = None + 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._sections.setdefault(current, []) + continue + self._sections[current].append(number) + for delimiter in ('"""', "'''"): + if line.count(delimiter) % 2: + quoted = delimiter + + 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 or isinstance(path[-1], int): + return None + for number in self._sections.get(path[:-1], ()): + line = self._lines[number] + match = _KEY_LINE.match(line) + if match is None or _key(match.group(1)) != path[-1]: + continue + try: + found = tomllib.loads(f"v = {match.group(2)}")["v"] + except tomllib.TOMLDecodeError: + continue + if _same(found, value): + return number + 1, match.start(2) + 1 + return None + + +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_structured_positions.py b/tests/test_structured_positions.py new file mode 100644 index 0000000..4d96c1a --- /dev/null +++ b/tests/test_structured_positions.py @@ -0,0 +1,128 @@ +"""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 + +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_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" From 1f74d1c9c257e55b889bf1a82a4dd0b566aec508 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 12:21:24 +0200 Subject: [PATCH 2/8] fix(secrets): locate configuration values at their own line, or by pointer Values of JSON and TOML configuration files were all located at their key's first occurrence, so findings of one key shared a line that was often not theirs. The configuration scan now places each value with the position component. A value whose line cannot be established gets a FileLocation, the file and the JSON pointer of the value, never a guessed line: reports show it without a line (JSON null and pointer, SARIF without region, text path only), an inline suppression does not apply to it, and the baseline recognises it by its pointer. Baselines are written in schema 2, whose entries carry the pointer; schema 1 files are still read and match exactly as they did. An old entry that cannot identify a finding with certainty no longer accounts for it, and the check then asks on standard error for the baseline to be recorded again. Closes #210 --- docs/plugins.md | 8 +- docs/usage.md | 22 ++- src/coretrace_python/cache.py | 20 ++- src/coretrace_python/cli.py | 6 +- src/coretrace_python/dependency/vex.py | 7 +- src/coretrace_python/findings/baseline.py | 63 ++++++-- src/coretrace_python/findings/model.py | 5 +- src/coretrace_python/findings/suppressions.py | 6 +- src/coretrace_python/plugins/secrets.py | 47 +++--- src/coretrace_python/reporters/json_format.py | 14 +- src/coretrace_python/reporters/report.py | 9 +- src/coretrace_python/reporters/sarif.py | 25 +-- src/coretrace_python/reporters/text.py | 9 +- src/coretrace_python/source/__init__.py | 4 +- src/coretrace_python/source/model.py | 17 ++ tests/test_baseline.py | 6 +- tests/test_baseline_schema_1.py | 100 ++++++++++++ tests/test_file_locations.py | 147 ++++++++++++++++++ 18 files changed, 438 insertions(+), 77 deletions(-) create mode 100644 tests/test_baseline_schema_1.py create mode 100644 tests/test_file_locations.py 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..292a4fb 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,15 @@ 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 instead of the text of a line. + +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 +276,15 @@ the directory of the checked file; a path outside it is printed as it is. } ``` +A value of a JSON or TOML file, or a package of a lock file, 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/cache.py b/src/coretrace_python/cache.py index e34ae7d..85ee5be 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,18 @@ 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} + return _encode_span(location) + + +def _decode_location(data: Any) -> Location: + if isinstance(data, Mapping): + return FileLocation(SourceId(_string(data["file"])), _string(data["pointer"])) + return _decode_span(data) + + def _decode_span(data: Any) -> SourceSpan: file, line, column, end_line, end_column = data return SourceSpan( @@ -217,7 +229,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 +242,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/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..50c0daf 100644 --- a/src/coretrace_python/findings/baseline.py +++ b/src/coretrace_python/findings/baseline.py @@ -2,8 +2,16 @@ 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 instead. 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 +23,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 -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 (empty for a file location), pointer (empty +# for a line). +Fingerprint = tuple[str, str, str, str, str] class BaselineError(Exception): @@ -32,7 +45,10 @@ 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.pointer) def _line_text(path: Path, line: int) -> str: @@ -48,6 +64,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 +75,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 +94,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 +122,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..c6039ae 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 MARKER = re.compile(r"#\s*coretrace:\s*ignore(?:\[([^\]]*)\])?") @@ -59,6 +59,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..915cae9 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 FileLocation, Location, SourceId, SourceSpan, decode_text +from coretrace_python.source.positions import JsonPositions, Pointer, TomlPositions, pointer_text -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,30 +318,30 @@ 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 FileLocation(source, pointer_text(path)) + ) + 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]: @@ -385,7 +388,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..85b17f8 100644 --- a/src/coretrace_python/reporters/json_format.py +++ b/src/coretrace_python/reporters/json_format.py @@ -6,23 +6,23 @@ 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..448fcac 100644 --- a/src/coretrace_python/source/__init__.py +++ b/src/coretrace_python/source/__init__.py @@ -1,9 +1,11 @@ """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.model import FileLocation, Location, SourceFile, SourceId, SourceSpan __all__ = [ + "FileLocation", + "Location", "SourceFile", "SourceId", "SourceManager", diff --git a/src/coretrace_python/source/model.py b/src/coretrace_python/source/model.py index 8836fd0..ba9894d 100644 --- a/src/coretrace_python/source/model.py +++ b/src/coretrace_python/source/model.py @@ -45,6 +45,23 @@ 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 = "" + + 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/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..0535180 --- /dev/null +++ b/tests/test_file_locations.py @@ -0,0 +1,147 @@ +"""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 + +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 == FileLocation(finding.span.source_id, "/a/api_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") From c3477c9c3c0bb0fb327c5f66cf8ef19a9ba6502f Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 12:21:24 +0200 Subject: [PATCH 3/8] fix(vulnerable-dependency): locate a locked package at its own entry A package pinned in uv.lock or poetry.lock was located at the first line mentioning its name, often another package's dependency list, and the message named it as its own requirer. It is now located at the name of its own [[package]] entry, which the position component checks, or given a file location when the entry cannot be placed; the message states the pinned version, the lock file and the packages that require it. Closes #207 --- .../vulnerable_dependency.py | 29 ++++- src/coretrace_python/dependency/graph.py | 30 ++++- tests/test_lock_positions.py | 107 ++++++++++++++++++ 3 files changed, 154 insertions(+), 12 deletions(-) create mode 100644 tests/test_lock_positions.py 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/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 4d42e62..73c5d96 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -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 FileLocation, Location, SourceFile, SourceId, SourceSpan +from coretrace_python.source.positions import TomlPositions, pointer_text _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(): @@ -398,14 +406,24 @@ def _pyproject_requirements(data: Mapping[str, Any], source: SourceFile) -> list 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 FileLocation(source.source_id, pointer_text(path)) + ) 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/tests/test_lock_positions.py b/tests/test_lock_positions.py new file mode 100644 index 0000000..7d30683 --- /dev/null +++ b/tests/test_lock_positions.py @@ -0,0 +1,107 @@ +"""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 FileLocation, SourceSpan + +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 == FileLocation(finding.span.source_id, "/package/0/name") + 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 From c3c8b380a7024c32aa21d37c8cec1a740e898bce Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 12:36:31 +0200 Subject: [PATCH 4/8] fix(source): tell file locations apart by value, and read TOML strings Review of the structured positions: a file location was fingerprinted by its pointer alone, so a new secret at the same place was baselined; it now carries a digest of its value, which the baseline compares, never the value itself. Multi-line TOML strings were tracked by counting triple quotes, so a delimiter in a comment, a one-line string or an escape put a value on a line inside a string; lines are now read by a small lexer that skips strings and comments and follows arrays and inline tables opened by a value, so an array element alone on its line is no header either. TOML lines are counted by line feeds, as the rest of the engine counts them, and each section's keys are indexed once. Refs #207, #210 --- docs/usage.md | 3 +- src/coretrace_python/cache.py | 10 +- src/coretrace_python/dependency/graph.py | 6 +- src/coretrace_python/findings/baseline.py | 9 +- src/coretrace_python/plugins/secrets.py | 8 +- src/coretrace_python/reporters/json_format.py | 11 +- src/coretrace_python/source/model.py | 3 + src/coretrace_python/source/positions.py | 106 ++++++++++++++---- tests/test_file_locations.py | 21 +++- tests/test_lock_positions.py | 2 +- tests/test_structured_positions.py | 23 ++++ 11 files changed, 165 insertions(+), 37 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index 292a4fb..bb08f50 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -234,7 +234,8 @@ shrink it as findings are fixed: an entry without a matching finding is simply u 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`. A finding with a file location is recognised by the JSON pointer of its -value instead of the text of a line. +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 diff --git a/src/coretrace_python/cache.py b/src/coretrace_python/cache.py index 85ee5be..90ee0af 100644 --- a/src/coretrace_python/cache.py +++ b/src/coretrace_python/cache.py @@ -187,13 +187,19 @@ def _encode_span(span: SourceSpan) -> list[Any]: def _encode_location(location: Location) -> Any: if isinstance(location, FileLocation): - return {"file": str(location.source_id), "pointer": location.pointer} + 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"])) + return FileLocation( + SourceId(_string(data["file"])), _string(data["pointer"]), _string(data["digest"]) + ) return _decode_span(data) diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 73c5d96..0b0d7f1 100644 --- a/src/coretrace_python/dependency/graph.py +++ b/src/coretrace_python/dependency/graph.py @@ -21,8 +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 FileLocation, Location, SourceFile, SourceId, SourceSpan -from coretrace_python.source.positions import TomlPositions, pointer_text +from coretrace_python.source import Location, SourceFile, SourceId, SourceSpan +from coretrace_python.source.positions import 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.*+!-]*)$") @@ -420,7 +420,7 @@ def _lock_requirements(data: Mapping[str, Any], source: SourceFile) -> list[Requ location: Location = ( SourceSpan(source.source_id, *place) if place is not None - else FileLocation(source.source_id, pointer_text(path)) + else file_location(source.source_id, path, name) ) found.append( Requirement(normalize(name), "", location, Version.parse(version), locked=version) diff --git a/src/coretrace_python/findings/baseline.py b/src/coretrace_python/findings/baseline.py index 50c0daf..7575dbe 100644 --- a/src/coretrace_python/findings/baseline.py +++ b/src/coretrace_python/findings/baseline.py @@ -3,7 +3,8 @@ 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. A finding with a file location, whose line is not -established, is recognised by the JSON pointer of its value instead. Entries are +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 @@ -30,8 +31,8 @@ # Files whose findings schema 1 placed at their key's first line. _STRUCTURED = (".json", ".toml", ".lock") -# File, rule, function, text of the line (empty for a file location), pointer (empty -# for a line). +# 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] @@ -48,7 +49,7 @@ def fingerprint(finding: Finding, root: Path) -> Fingerprint: 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.pointer) + return (*head, finding.span.digest, finding.span.pointer) def _line_text(path: Path, line: int) -> str: diff --git a/src/coretrace_python/plugins/secrets.py b/src/coretrace_python/plugins/secrets.py index 915cae9..e6cb04c 100644 --- a/src/coretrace_python/plugins/secrets.py +++ b/src/coretrace_python/plugins/secrets.py @@ -28,8 +28,8 @@ 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 FileLocation, Location, SourceId, SourceSpan, decode_text -from coretrace_python.source.positions import JsonPositions, Pointer, TomlPositions, pointer_text +from coretrace_python.source import Location, SourceId, SourceSpan, decode_text +from coretrace_python.source.positions import JsonPositions, Pointer, TomlPositions, file_location # 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. @@ -337,7 +337,9 @@ def walk(node: object, path: Pointer, key: str | None, named: bool = True) -> It elif isinstance(node, str) and key is not None: place = positions.locate(path, node) location: Location = ( - SourceSpan(source, *place) if place is not None else FileLocation(source, pointer_text(path)) + SourceSpan(source, *place) + if place is not None + else file_location(source, path, node) ) yield node, key if named else None, location, None diff --git a/src/coretrace_python/reporters/json_format.py b/src/coretrace_python/reporters/json_format.py index 85b17f8..864b20a 100644 --- a/src/coretrace_python/reporters/json_format.py +++ b/src/coretrace_python/reporters/json_format.py @@ -14,9 +14,16 @@ def finding_record(finding: Finding, report: Report | None = None) -> dict[str, 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) + 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) + location.update( + line=None, column=None, end_line=None, end_column=None, pointer=span.pointer + ) return { "rule_id": finding.rule_id, "message": finding.message, diff --git a/src/coretrace_python/source/model.py b/src/coretrace_python/source/model.py index ba9894d..d9bb281 100644 --- a/src/coretrace_python/source/model.py +++ b/src/coretrace_python/source/model.py @@ -53,6 +53,9 @@ class FileLocation: 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. + digest: str = "" def display(self) -> str: return f"{self.source_id}{self.pointer and '#' + self.pointer}" diff --git a/src/coretrace_python/source/positions.py b/src/coretrace_python/source/positions.py index ee3bf05..8952d9b 100644 --- a/src/coretrace_python/source/positions.py +++ b/src/coretrace_python/source/positions.py @@ -12,11 +12,14 @@ from __future__ import annotations +import hashlib import json import re import tomllib from bisect import bisect_right +from coretrace_python.source.model import FileLocation, SourceId + Pointer = tuple[str | int, ...] _SPACE = re.compile(r"[ \t\n\r]*") @@ -28,6 +31,16 @@ _KEY_LINE = re.compile(r"""\s*("(?:[^"\\]|\\.)*"|'[^']*'|[A-Za-z0-9_-]+)\s*=\s*(.*)$""") +def file_location(source_id: SourceId, path: Pointer, value: object) -> 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.""" + + written = value if isinstance(value, str) else json.dumps(value) + digest = hashlib.sha256(written.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``.""" @@ -145,18 +158,22 @@ def _space(self, position: int) -> int: 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.""" + table: the root, ``[a.b]``, or the element ``[[a]]`` opens, numbered in order. Lines + are counted by line feeds, as the rest of the engine counts them; 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: - self._lines = text.splitlines() - self._sections: dict[Pointer, list[int]] = {(): []} + # 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 = () - quoted: str | None = None - for number, line in enumerate(self._lines): - if quoted is not None: - if line.count(quoted) % 2: - quoted = None + state = _LexState() + for number, line in enumerate(text.split("\n"), start=1): + line = line.removesuffix("\r") + 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 @@ -172,12 +189,18 @@ def __init__(self, text: str) -> None: else: table = _resolve(keys, arrays) current = table - self._sections.setdefault(current, []) + 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._sections[current].append(number) - for delimiter in ('"""', "'''"): - if line.count(delimiter) % 2: - quoted = delimiter + 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 @@ -185,20 +208,63 @@ def locate(self, path: Pointer, value: object) -> tuple[int, int] | None: if not path or isinstance(path[-1], int): return None - for number in self._sections.get(path[:-1], ()): - line = self._lines[number] - match = _KEY_LINE.match(line) - if match is None or _key(match.group(1)) != path[-1]: - continue + for number, column, written in self._keys.get(path[:-1], {}).get(path[-1], ()): try: - found = tomllib.loads(f"v = {match.group(2)}")["v"] + found = tomllib.loads(f"v = {written}")["v"] except tomllib.TOMLDecodeError: continue if _same(found, value): - return number + 1, match.start(2) + 1 + return number, column 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): + self.string, position = None, position + 3 + 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 _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.""" diff --git a/tests/test_file_locations.py b/tests/test_file_locations.py index 0535180..7b8b08f 100644 --- a/tests/test_file_locations.py +++ b/tests/test_file_locations.py @@ -57,7 +57,7 @@ def test_values_of_one_key_are_located_at_their_own_lines(tmp_path: Path) -> Non 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 == FileLocation(finding.span.source_id, "/a/api_token") + assert isinstance(finding.span, FileLocation) and finding.span.pointer == "/a/api_token" assert Path(str(finding.span.source_id)).name == "settings.toml" @@ -145,3 +145,22 @@ def test_the_baseline_recognises_a_file_location_by_its_pointer( ).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") diff --git a/tests/test_lock_positions.py b/tests/test_lock_positions.py index 7d30683..b48fd8a 100644 --- a/tests/test_lock_positions.py +++ b/tests/test_lock_positions.py @@ -91,7 +91,7 @@ def test_the_message_states_the_pinned_version_the_lock_file_and_its_dependents( 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 == FileLocation(finding.span.source_id, "/package/0/name") + assert isinstance(finding.span, FileLocation) and finding.span.pointer == "/package/0/name" assert finding.message.startswith( "CVE-2099-0207: werkzeug 3.1.5, pinned in uv.lock, is in <3.1.6: " ) diff --git a/tests/test_structured_positions.py b/tests/test_structured_positions.py index 4d96c1a..7c34754 100644 --- a/tests/test_structured_positions.py +++ b/tests/test_structured_positions.py @@ -120,6 +120,29 @@ def test_toml_values_the_component_cannot_place_have_no_position() -> 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) + + +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_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" From 1e794c8494dfbb978745823fed692744a6a0cec4 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 12:51:30 +0200 Subject: [PATCH 5/8] fix(baseline): count lines by line feeds, and close TOML strings on their quotes Review of the structured positions: the baseline read a finding's line with splitlines, which also breaks on U+2028 and U+0085, while the parser and the position component count line feeds, so a separator above a secret recorded another line's text and could absorb a new secret. It now counts line feeds. A multi-line TOML string closed by four or five quotes left the lexer inverted, and a key written inside a later string could be located; the quotes before a closing delimiter now belong to the string. A FileLocation now requires its pointer and digest, which file_location builds from a string value. Refs #207, #210 --- src/coretrace_python/findings/baseline.py | 5 ++++- src/coretrace_python/source/model.py | 6 ++--- src/coretrace_python/source/positions.py | 9 ++++---- tests/test_file_locations.py | 27 ++++++++++++++++++++++- tests/test_lock_positions.py | 5 +++-- tests/test_structured_positions.py | 13 +++++++++++ 6 files changed, 54 insertions(+), 11 deletions(-) diff --git a/src/coretrace_python/findings/baseline.py b/src/coretrace_python/findings/baseline.py index 7575dbe..3b85cad 100644 --- a/src/coretrace_python/findings/baseline.py +++ b/src/coretrace_python/findings/baseline.py @@ -53,8 +53,11 @@ def fingerprint(finding: Finding, root: Path) -> Fingerprint: def _line_text(path: Path, line: int) -> str: + """The text of ``line``, lines counted by line feeds as the parser and the position + component count them (``splitlines`` also breaks on U+2028 and other separators).""" + try: - lines = decode_text(path.read_bytes()).splitlines() + lines = [text.removesuffix("\r") for text in decode_text(path.read_bytes()).split("\n")] except (OSError, UnicodeDecodeError): return str(line) if not 1 <= line <= len(lines): diff --git a/src/coretrace_python/source/model.py b/src/coretrace_python/source/model.py index d9bb281..da39927 100644 --- a/src/coretrace_python/source/model.py +++ b/src/coretrace_python/source/model.py @@ -52,10 +52,10 @@ class FileLocation: which a report or an inline suppression would then take for a verified one.""" source_id: SourceId - pointer: str = "" + pointer: str # A digest of the value (``sha256:…``), never the value: what the baseline compares - # where it would compare the text of a line. - digest: str = "" + # 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}" diff --git a/src/coretrace_python/source/positions.py b/src/coretrace_python/source/positions.py index 8952d9b..d9eaeb6 100644 --- a/src/coretrace_python/source/positions.py +++ b/src/coretrace_python/source/positions.py @@ -31,13 +31,12 @@ _KEY_LINE = re.compile(r"""\s*("(?:[^"\\]|\\.)*"|'[^']*'|[A-Za-z0-9_-]+)\s*=\s*(.*)$""") -def file_location(source_id: SourceId, path: Pointer, value: object) -> FileLocation: +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.""" - written = value if isinstance(value, str) else json.dumps(value) - digest = hashlib.sha256(written.encode("utf-8", "surrogatepass")).hexdigest()[:16] + digest = hashlib.sha256(value.encode("utf-8", "surrogatepass")).hexdigest()[:16] return FileLocation(source_id, pointer_text(path), f"sha256:{digest}") @@ -238,7 +237,9 @@ def read(self, line: str) -> None: if self.string == '"""' and line[position] == "\\": position += 2 elif line.startswith(self.string, position): - self.string, position = None, position + 3 + # 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 diff --git a/tests/test_file_locations.py b/tests/test_file_locations.py index 7b8b08f..e79b6e8 100644 --- a/tests/test_file_locations.py +++ b/tests/test_file_locations.py @@ -19,6 +19,7 @@ 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" @@ -57,7 +58,7 @@ def test_values_of_one_key_are_located_at_their_own_lines(tmp_path: Path) -> Non 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 isinstance(finding.span, FileLocation) and finding.span.pointer == "/a/api_token" + assert finding.span == file_location(finding.span.source_id, ("a", "api_token"), TOKEN) assert Path(str(finding.span.source_id)).name == "settings.toml" @@ -164,3 +165,27 @@ def test_a_changed_value_at_a_file_location_is_a_new_finding( 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") diff --git a/tests/test_lock_positions.py b/tests/test_lock_positions.py index b48fd8a..4868474 100644 --- a/tests/test_lock_positions.py +++ b/tests/test_lock_positions.py @@ -16,7 +16,8 @@ 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 FileLocation, SourceSpan +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( @@ -91,7 +92,7 @@ def test_the_message_states_the_pinned_version_the_lock_file_and_its_dependents( 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 isinstance(finding.span, FileLocation) and finding.span.pointer == "/package/0/name" + 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: " ) diff --git a/tests/test_structured_positions.py b/tests/test_structured_positions.py index 7c34754..6299d88 100644 --- a/tests/test_structured_positions.py +++ b/tests/test_structured_positions.py @@ -12,6 +12,8 @@ from __future__ import annotations +import pytest + from coretrace_python.source.positions import JsonPositions, TomlPositions, pointer_text HAR = """{ @@ -130,6 +132,17 @@ def test_a_header_or_key_written_inside_a_toml_string_is_not_one() -> None: 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') From 408e8a24cc7c32bc8b8d1cb64048e4d7cd461566 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 13:06:56 +0200 Subject: [PATCH 6/8] fix(source): count every line number by one rule, and place pyproject requirements Review of the baseline line fix: line numbers had no single rule. The baseline now counted line feeds while the .env, YAML, INI and requirements readers used splitlines and the parser also breaks on a lone carriage return, so a fingerprint or an inline suppression could refer to another line and a rotated secret could be absorbed. lines_of, in the source layer, counts lines as Python does (\n, \r\n, a lone \r; never U+2028 or a form feed), and every producer and consumer of a line number now uses it: the baseline, inline suppressions, the configuration and requirements readers, and the JSON and TOML positions. pyproject.toml requirements were placed at the first line containing their text, or line 1 by default; the position component now finds the element of an array a key holds, and a requirement it cannot place gets a file location. Refs #207, #210 --- docs/usage.md | 4 +- src/coretrace_python/dependency/graph.py | 55 +++---- src/coretrace_python/findings/baseline.py | 8 +- src/coretrace_python/findings/suppressions.py | 10 +- src/coretrace_python/plugins/secrets.py | 4 +- src/coretrace_python/source/__init__.py | 4 +- src/coretrace_python/source/manager.py | 16 +++ src/coretrace_python/source/positions.py | 109 ++++++++++++-- tests/test_line_numbering.py | 136 ++++++++++++++++++ tests/test_structured_positions.py | 17 +++ 10 files changed, 316 insertions(+), 47 deletions(-) create mode 100644 tests/test_line_numbering.py diff --git a/docs/usage.md b/docs/usage.md index bb08f50..4ca5464 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -277,7 +277,9 @@ the directory of the checked file; a path outside it is printed as it is. } ``` -A value of a JSON or TOML file, or a package of a lock file, is located at its own line, +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`, diff --git a/src/coretrace_python/dependency/graph.py b/src/coretrace_python/dependency/graph.py index 0b0d7f1..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,8 +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 Location, SourceFile, SourceId, SourceSpan -from coretrace_python.source.positions import TomlPositions, file_location +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.*+!-]*)$") @@ -349,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 @@ -370,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(): @@ -397,12 +404,8 @@ 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]: diff --git a/src/coretrace_python/findings/baseline.py b/src/coretrace_python/findings/baseline.py index 3b85cad..11dae3a 100644 --- a/src/coretrace_python/findings/baseline.py +++ b/src/coretrace_python/findings/baseline.py @@ -24,7 +24,7 @@ from pathlib import Path from coretrace_python.findings.model import Finding -from coretrace_python.source import SourceSpan, decode_text +from coretrace_python.source import SourceSpan, decode_text, lines_of BASELINE_SCHEMA = 2 _LEGACY_SCHEMA = 1 @@ -53,11 +53,11 @@ def fingerprint(finding: Finding, root: Path) -> Fingerprint: def _line_text(path: Path, line: int) -> str: - """The text of ``line``, lines counted by line feeds as the parser and the position - component count them (``splitlines`` also breaks on U+2028 and other separators).""" + """The text of ``line``, lines counted as every line number of the engine counts + them (``lines_of``).""" try: - lines = [text.removesuffix("\r") for text in decode_text(path.read_bytes()).split("\n")] + lines = lines_of(decode_text(path.read_bytes())) except (OSError, UnicodeDecodeError): return str(line) if not 1 <= line <= len(lines): diff --git a/src/coretrace_python/findings/suppressions.py b/src/coretrace_python/findings/suppressions.py index c6039ae..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, SourceSpan +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 diff --git a/src/coretrace_python/plugins/secrets.py b/src/coretrace_python/plugins/secrets.py index e6cb04c..21eb72c 100644 --- a/src/coretrace_python/plugins/secrets.py +++ b/src/coretrace_python/plugins/secrets.py @@ -28,7 +28,7 @@ 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 Location, 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 # A value, the name it is bound to, where it is, and its enclosing function. A value of @@ -347,7 +347,7 @@ def walk(node: object, path: Pointer, key: str | None, named: bool = True) -> It 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 diff --git a/src/coretrace_python/source/__init__.py b/src/coretrace_python/source/__init__.py index 448fcac..c2a75a2 100644 --- a/src/coretrace_python/source/__init__.py +++ b/src/coretrace_python/source/__init__.py @@ -1,9 +1,10 @@ """Source storage and location primitives.""" -from coretrace_python.source.manager import SourceManager, decode_text +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", @@ -11,5 +12,6 @@ "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/positions.py b/src/coretrace_python/source/positions.py index d9eaeb6..629c0c5 100644 --- a/src/coretrace_python/source/positions.py +++ b/src/coretrace_python/source/positions.py @@ -18,6 +18,7 @@ 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, ...] @@ -48,7 +49,7 @@ def pointer_text(path: Pointer) -> str: class _Lines: def __init__(self, text: str) -> None: - self._starts = [0, *(match.end() for match in re.finditer("\n", text))] + 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``.""" @@ -158,7 +159,7 @@ def _space(self, position: int) -> int: 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 by line feeds, as the rest of the engine counts them; a line that starts + 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.""" @@ -167,9 +168,9 @@ def __init__(self, text: str) -> None: self._keys: dict[Pointer, dict[str, list[tuple[int, int, str]]]] = {(): {}} arrays: dict[Pointer, int] = {} current: Pointer = () + self._lines = lines_of(text) state = _LexState() - for number, line in enumerate(text.split("\n"), start=1): - line = line.removesuffix("\r") + for number, line in enumerate(self._lines, start=1): continued = state.inside() state.read(line) if continued: @@ -205,17 +206,36 @@ 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 or isinstance(path[-1], int): + 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], ()): - try: - found = tomllib.loads(f"v = {written}")["v"] - except tomllib.TOMLDecodeError: - continue - if _same(found, value): + 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, written over one or several lines: + ``dependencies = [\\n "a",\\n "b",\\n]``.""" + + 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], ()): + text = "\n".join(self._lines[number - 1 :])[column - 1 :] + elements = _array_elements(text) + if index >= len(elements): + continue + start, end = elements[index] + if not _same(_value(text[start:end]), value): + continue + before = text[:start] + line = number + before.count("\n") + return line, (start - before.rfind("\n") if "\n" in before else column + start) + return None + class _LexState: """Where a TOML line leaves the reader: inside a multi-line string (its delimiter), @@ -266,6 +286,75 @@ def read(self, line: str) -> None: 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) -> list[tuple[int, int]]: + """The start and end offsets of the elements of the array ``text`` opens, skipping + strings and comments; empty when ``text`` does not open an array.""" + + if not text.startswith("["): + return [] + elements: list[tuple[int, int]] = [] + depth, start, position = 0, -1, 0 + while position < len(text): + character = text[position] + if character in "\"'": + if depth == 1 and start < 0: + start = position + 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 "[{": + if depth == 1 and start < 0: + start = position + depth += 1 + elif character in "]}": + depth -= 1 + if depth == 0: + if start >= 0: + elements.append((start, position)) + return elements + elif character == "," and depth == 1: + if start >= 0: + elements.append((start, position)) + start = -1 + elif depth == 1 and start < 0 and not character.isspace(): + start = position + position += 1 + return [] + + +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 = len(text[position:]) - len(text[position:].lstrip(quote)) + return position + min(run, 5) + 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.""" 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_structured_positions.py b/tests/test_structured_positions.py index 6299d88..da3d982 100644 --- a/tests/test_structured_positions.py +++ b/tests/test_structured_positions.py @@ -156,6 +156,23 @@ def test_toml_lines_are_counted_by_line_feeds_only() -> None: 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_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" From f6edf1fc42eac51a9564d3de3585ec56baf3520b Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 13:22:14 +0200 Subject: [PATCH 7/8] fix(source): leave multi-line array elements unplaced, and read each array once Review of the array elements: an element written over several lines (a key in a multi-line string) was placed at the line it starts on, which does not hold it, so a changed value kept that line's text and was baselined. An element is now delimited by the end of its own value, and one that spans lines has no position: it gets a file location, whose digest of the value the baseline compares. Each array's elements were found again for every value, scanning the rest of the file each time; they are now read once per array over a single copy of the text (4,000 elements in 0.04 s instead of 40 s). Refs #207, #210 --- src/coretrace_python/source/positions.py | 83 +++++++++++++++--------- tests/test_file_locations.py | 22 +++++++ tests/test_structured_positions.py | 8 +++ 3 files changed, 84 insertions(+), 29 deletions(-) diff --git a/src/coretrace_python/source/positions.py b/src/coretrace_python/source/positions.py index 629c0c5..181747c 100644 --- a/src/coretrace_python/source/positions.py +++ b/src/coretrace_python/source/positions.py @@ -29,6 +29,7 @@ _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*(?:#.*)?$") +_SCALAR = re.compile(r"[^,\]}#\s]+") _KEY_LINE = re.compile(r"""\s*("(?:[^"\\]|\\.)*"|'[^']*'|[A-Za-z0-9_-]+)\s*=\s*(.*)$""") @@ -57,6 +58,11 @@ def position(self, offset: int) -> tuple[int, int]: 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 @@ -169,6 +175,10 @@ def __init__(self, text: str) -> None: 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() @@ -216,24 +226,25 @@ def locate(self, path: Pointer, value: object) -> tuple[int, int] | None: return None def _element(self, path: Pointer, value: object) -> tuple[int, int] | None: - """An element of an array a key holds, written over one or several lines: - ``dependencies = [\\n "a",\\n "b",\\n]``.""" + """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], ()): - text = "\n".join(self._lines[number - 1 :])[column - 1 :] - elements = _array_elements(text) + 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 - start, end = elements[index] - if not _same(_value(text[start:end]), value): - continue - before = text[:start] - line = number + before.count("\n") - return line, (start - before.rfind("\n") if "\n" in before else column + start) + 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 @@ -295,19 +306,43 @@ def _value(written: str) -> object: return None -def _array_elements(text: str) -> list[tuple[int, int]]: - """The start and end offsets of the elements of the array ``text`` opens, skipping - strings and comments; empty when ``text`` does not open an array.""" +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("["): + if not text.startswith("[", start): return [] elements: list[tuple[int, int]] = [] - depth, start, position = 0, -1, 0 + 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 scalar.end() if scalar is not None else start + 1 + depth, position = 0, start while position < len(text): character = text[position] if character in "\"'": - if depth == 1 and start < 0: - start = position position = _string_end(text, position) continue if character == "#": @@ -315,23 +350,13 @@ def _array_elements(text: str) -> list[tuple[int, int]]: position = len(text) if newline < 0 else newline continue if character in "[{": - if depth == 1 and start < 0: - start = position depth += 1 elif character in "]}": depth -= 1 if depth == 0: - if start >= 0: - elements.append((start, position)) - return elements - elif character == "," and depth == 1: - if start >= 0: - elements.append((start, position)) - start = -1 - elif depth == 1 and start < 0 and not character.isspace(): - start = position + return position + 1 position += 1 - return [] + return len(text) def _string_end(text: str, start: int) -> int: diff --git a/tests/test_file_locations.py b/tests/test_file_locations.py index e79b6e8..05e1106 100644 --- a/tests/test_file_locations.py +++ b/tests/test_file_locations.py @@ -189,3 +189,25 @@ def test_the_baseline_reads_the_line_a_finding_is_at( 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_structured_positions.py b/tests/test_structured_positions.py index da3d982..2ecefa8 100644 --- a/tests/test_structured_positions.py +++ b/tests/test_structured_positions.py @@ -173,6 +173,14 @@ def test_an_element_of_a_toml_array_is_found_and_checked() -> 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_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" From 8ce2575a421426a9303ebe840b0e28ceea94b017 Mon Sep 17 00:00:00 2001 From: Hugo Date: Wed, 7 Oct 2026 13:34:59 +0200 Subject: [PATCH 8/8] fix(source): read a TOML datetime with a space as one array element Review of the array elements: a bare scalar ended at its first space, so a datetime written with a space (1979-05-27 07:32:00) counted as two elements and shifted every later index. A bare scalar now ends at a separator or the end of its line, its trailing spaces trimmed. Closing a multi-line string counts its quotes in place instead of copying the rest of the text. Refs #207, #210 --- src/coretrace_python/source/positions.py | 11 +++++++---- tests/test_structured_positions.py | 7 +++++++ 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/src/coretrace_python/source/positions.py b/src/coretrace_python/source/positions.py index 181747c..ef620fe 100644 --- a/src/coretrace_python/source/positions.py +++ b/src/coretrace_python/source/positions.py @@ -29,7 +29,8 @@ _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*(?:#.*)?$") -_SCALAR = re.compile(r"[^,\]}#\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*(.*)$""") @@ -338,7 +339,7 @@ def _value_end(text: str, start: int) -> int: return _string_end(text, start) if text[start] not in "[{": scalar = _SCALAR.match(text, start) - return scalar.end() if scalar is not None else start + 1 + 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] @@ -369,8 +370,10 @@ def _string_end(text: str, start: int) -> int: if quote == '"' and text[position] == "\\": position += 2 elif text.startswith(quote * 3, position): - run = len(text[position:]) - len(text[position:].lstrip(quote)) - return position + min(run, 5) + run = 3 + while run < 5 and text.startswith(quote, position + run): + run += 1 + return position + run else: position += 1 return len(text) diff --git a/tests/test_structured_positions.py b/tests/test_structured_positions.py index 2ecefa8..b851ae2 100644 --- a/tests/test_structured_positions.py +++ b/tests/test_structured_positions.py @@ -181,6 +181,13 @@ def test_an_array_element_written_over_several_lines_has_no_position() -> None: 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"