Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 118 additions & 18 deletions src/skillspector/references.py
Original file line number Diff line number Diff line change
Expand Up @@ -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})(?:`|'|\")"
)
Expand Down Expand Up @@ -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)

Copy link
Copy Markdown
Collaborator

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.

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

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Consume the full link title before finding another opening

consumed_until stops at destination_end, even though _MARKDOWN_INLINE_END may have validated a quoted title and the link's closing parenthesis. Consequently [guide](docs/guide.md "see [sample](missing.md)") yields missing.md as a second required reference even though that text is only the outer link title. Retain the full end position from the successful inline-end match for overlap suppression (while keeping the candidate's destination span unchanged), and add a title regression.

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,
*,
Expand All @@ -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:
Expand All @@ -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,
),
)
Expand All @@ -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
Expand Down Expand Up @@ -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]
Expand Down
Loading
Loading