-
Notifications
You must be signed in to change notification settings - Fork 1.5k
fix: preserve complete Markdown reference destinations #553
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
975ca05
f3bc9d4
a429733
88f63eb
bde389f
5a0d419
18c897a
91ffba6
04be7ff
8e0f51f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -23,7 +23,12 @@ | |
| MAX_REFERENCE_RECORDS = 1024 | ||
| MAX_REFERENCE_RUNTIME_SECONDS = 2.0 | ||
| _MAX_EVIDENCE = 160 | ||
| _MARKDOWN_DESTINATION = re.compile(r"\[[^\]\n]{1,200}\]\(([^)\n]{1,512})\)") | ||
| _MAX_MARKDOWN_DESTINATION_CHARS = 512 | ||
| _MARKDOWN_REFERENCE_START = re.compile(r"\[[^\]\n]{1,200}\]\(|^[ \t]{0,3}\[[^\]\n]{1,200}\]:[ \t]*") | ||
| _MARKDOWN_TITLE = r"""(?:"(?:\\.|[^"\\\r\n])*"|'(?:\\.|[^'\\\r\n])*'|\((?:\\.|[^)\\\r\n])*\))""" | ||
| _MARKDOWN_INLINE_END = re.compile(r"[ \t]*(?:" + _MARKDOWN_TITLE + r")?[ \t]*\)") | ||
| _MARKDOWN_DEFINITION_END = re.compile(r"[ \t]*(?:" + _MARKDOWN_TITLE + r")?[ \t]*(?:\r?\n)?\Z") | ||
| _MARKDOWN_STRUCTURAL_ESCAPE = re.compile(r"\\([\\()<>])") | ||
| _QUOTED_OR_CODE_PATH = re.compile( | ||
| r"(?:`|'|\")((?:\./)?(?:[A-Za-z0-9_.-]+/)*[A-Za-z0-9_.-]+\.[A-Za-z0-9]{1,12})(?:`|'|\")" | ||
| ) | ||
|
|
@@ -57,6 +62,90 @@ def _evidence(cleaned_line: str, column: int) -> str: | |
| return cleaned_line[start : start + _MAX_EVIDENCE] | ||
|
|
||
|
|
||
| @dataclass(frozen=True) | ||
| class _ReferenceCandidate: | ||
| raw: str | ||
| start: int | ||
| end: int | ||
|
|
||
|
|
||
| def _markdown_candidates( | ||
| line: str, *, deadline: float, clock: Callable[[], float] | ||
| ) -> Iterator[_ReferenceCandidate]: | ||
| """Read bounded destinations without splitting spaces or balanced parentheses. | ||
|
|
||
| Explicit reference definitions use the same destination grammar as inline | ||
| links. Recognizing them separately keeps slash-separated prose excluded. | ||
| """ | ||
| consumed_until = 0 | ||
| for opening in _MARKDOWN_REFERENCE_START.finditer(line): | ||
| # Malformed openings may never yield a candidate. Bound their work too. | ||
| if clock() >= deadline: | ||
| return | ||
| if opening.start() < consumed_until: | ||
| continue | ||
| start = opening.end() | ||
| while start < len(line) and line[start] in " \t": | ||
| start += 1 | ||
| if start >= len(line): | ||
| continue | ||
| limit = min(len(line), start + _MAX_MARKDOWN_DESTINATION_CHARS + 2) | ||
| end = start | ||
| if line[start] == "<": | ||
| start += 1 | ||
| end = start | ||
| while end < limit and line[end] not in "<>\r\n": | ||
| if line[end] == "\\" and end + 1 < limit: | ||
| end += 1 | ||
| end += 1 | ||
| if end >= limit or line[end] != ">": | ||
| continue | ||
| raw = line[start:end] | ||
| destination_end = end + 1 | ||
| else: | ||
| depth = 0 | ||
| while end < limit and not line[end].isspace(): | ||
| char = line[end] | ||
| if char == "\\" and end + 1 < limit: | ||
| end += 2 | ||
| continue | ||
| if char == "(": | ||
| depth += 1 | ||
| elif char == ")": | ||
| if depth == 0: | ||
| break | ||
| depth -= 1 | ||
| elif char in "<>": | ||
| break | ||
| end += 1 | ||
| if depth or end == start or end - start > _MAX_MARKDOWN_DESTINATION_CHARS: | ||
| continue | ||
| raw = line[start:end] | ||
| destination_end = end | ||
| if len(raw) > _MAX_MARKDOWN_DESTINATION_CHARS: | ||
| continue | ||
| if opening.group().endswith("("): | ||
| if not _MARKDOWN_INLINE_END.match( | ||
| line, destination_end, min(len(line), destination_end + 512) | ||
| ): | ||
| continue | ||
| elif len(line) - destination_end > 512 or not _MARKDOWN_DEFINITION_END.fullmatch( | ||
| line, destination_end | ||
| ): | ||
| # A prose line such as "[status]: All checks passed." is not a | ||
| # reference definition: a destination can only be followed by a title. | ||
| continue | ||
| consumed_until = destination_end | ||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. [P2] Consume the full link title before finding another opening
|
||
| yield _ReferenceCandidate( | ||
| _MARKDOWN_STRUCTURAL_ESCAPE.sub(r"\1", raw), start, destination_end | ||
| ) | ||
|
|
||
|
|
||
| def _pattern_candidates(pattern: re.Pattern[str], line: str) -> Iterator[_ReferenceCandidate]: | ||
| for match in pattern.finditer(line): | ||
| yield _ReferenceCandidate(match.group(1), match.start(1), match.end(1)) | ||
|
|
||
|
|
||
| def _candidate_strings( | ||
| text: str, | ||
| *, | ||
|
|
@@ -65,43 +154,52 @@ def _candidate_strings( | |
| ) -> tuple[list[tuple[str, int, int, str]], tuple[str, ...]]: | ||
| """Extract path-like strings without materializing all matches or lines. | ||
|
|
||
| Each regular expression contributes at most one pending match to a small | ||
| Each candidate iterator contributes at most one pending match to a small | ||
| merge heap. This preserves source ordering while ensuring a dense, | ||
| attacker-controlled line cannot be fully enumerated and sorted before the | ||
| candidate and time ceilings are enforced. | ||
| """ | ||
| candidates: list[tuple[str, int, int, str]] = [] | ||
| seen: set[tuple[int, int, str]] = set() | ||
| patterns = (_MARKDOWN_DESTINATION, _QUOTED_OR_CODE_PATH, _PLAIN_RELATIVE_PATH) | ||
| for line_number, line in enumerate(StringIO(text), 1): | ||
| if clock() >= deadline: | ||
| return candidates, ("runtime",) | ||
| cleaned_line = " ".join(line.strip().split()) | ||
| iterators: list[Iterator[re.Match[str]]] = [pattern.finditer(line) for pattern in patterns] | ||
| pending: list[tuple[int, int, int, re.Match[str]]] = [] | ||
| iterators = [ | ||
| _markdown_candidates(line, deadline=deadline, clock=clock), | ||
| _pattern_candidates(_QUOTED_OR_CODE_PATH, line), | ||
| _pattern_candidates(_PLAIN_RELATIVE_PATH, line), | ||
| ] | ||
| pending: list[tuple[int, int, int, _ReferenceCandidate]] = [] | ||
| markdown_destination_end = 0 | ||
| for pattern_index, iterator in enumerate(iterators): | ||
| match = next(iterator, None) | ||
| if match is not None: | ||
| heapq.heappush( | ||
| pending, | ||
| (match.start(1), match.end(1), pattern_index, match), | ||
| (match.start, pattern_index, match.end, match), | ||
| ) | ||
| if clock() >= deadline: | ||
| return candidates, ("runtime",) | ||
| while pending: | ||
| if clock() >= deadline: | ||
| return candidates, ("runtime",) | ||
| _, _, pattern_index, match = heapq.heappop(pending) | ||
| raw = match.group(1).strip().split(maxsplit=1)[0] | ||
| key = (line_number, match.start(1), raw) | ||
| if key not in seen: | ||
| _, pattern_index, _, match = heapq.heappop(pending) | ||
| raw = match.raw | ||
| key = (line_number, match.start, raw) | ||
| # A plain/quoted substring of an explicit Markdown destination is | ||
| # not a second reference (e.g. <docs/user.md guide.md>). | ||
| inside_destination = pattern_index != 0 and match.start < markdown_destination_end | ||
| if pattern_index == 0: | ||
| markdown_destination_end = max(markdown_destination_end, match.end) | ||
| if not inside_destination and key not in seen: | ||
| seen.add(key) | ||
| candidates.append( | ||
| ( | ||
| raw, | ||
| line_number, | ||
| match.start(1) + 1, | ||
| _evidence(cleaned_line, match.start(1) + 1), | ||
| match.start + 1, | ||
| _evidence(cleaned_line, match.start + 1), | ||
| ) | ||
| ) | ||
| if len(candidates) >= MAX_RAW_REFERENCE_CANDIDATES: | ||
|
|
@@ -111,9 +209,9 @@ def _candidate_strings( | |
| heapq.heappush( | ||
| pending, | ||
| ( | ||
| next_match.start(1), | ||
| next_match.end(1), | ||
| next_match.start, | ||
| pattern_index, | ||
| next_match.end, | ||
| next_match, | ||
| ), | ||
| ) | ||
|
|
@@ -122,12 +220,14 @@ def _candidate_strings( | |
|
|
||
| def _normalize_candidate(raw: str, source_path: str) -> str | None: | ||
| """Return a contained relative POSIX candidate, or None when unsupported.""" | ||
| raw = unquote(raw.strip().strip("<>")) | ||
| raw = raw.strip() | ||
| split = urlsplit(raw) | ||
| if split.scheme or split.netloc or raw.startswith(("/", "\\", "#")): | ||
| return None | ||
| path_part = split.path.replace("\\", "/") | ||
| if not path_part: | ||
| # Split URI syntax before decoding so %23/%3F remain filename characters. | ||
| # Decode exactly once, then apply containment checks to the decoded path. | ||
| path_part = unquote(split.path).replace("\\", "/") | ||
| if not path_part or path_part.startswith("/"): | ||
| return None | ||
| if len(path_part) >= 2 and path_part[1] == ":": | ||
| return None | ||
|
|
@@ -206,7 +306,7 @@ def resolve_bundle_references_with_metadata( | |
| resolved_target = target | ||
| status = "resolved" | ||
| disposition = ArtifactDisposition.ANALYZED | ||
| elif "/" not in raw.replace("\\", "/"): | ||
| elif "/" not in unquote(raw).replace("\\", "/"): | ||
| matches = basename_index.get(PurePosixPath(target).name, []) | ||
| if len(matches) == 1: | ||
| resolved_target = matches[0] | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[P1] Fail closed when an explicit destination exceeds the budget
The 512-character limit silently
continues in both the angle and bare branches. A valid long destination containing encoded spaces (so the plain-path fallback cannot recover it) then produces no reference record and no extraction limitation; a missing required file can still report complete/SAFE. Preserve bounded work, but return an explicit limitation/partial result when a syntactically explicit destination crosses this cap, and cover the resulting CLI/MCP rejection.