You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@magpie.apache.org by po...@apache.org on 2026/10/17 00:21:57 UTC
(magpie) 25/27: fix(github-body-field): verify the rewritten body and keep CRLF spacing
This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch perf/security-1-import
in repository https://gitbox.apache.org/repos/asf/magpie.git
commit 8bce3aefb92b0fa579cd57da0a1ca0c026961ec2
Author: Jarek Potiuk <po...@apache.org>
AuthorDate: Sat Oct 17 01:30:49 2026 +0200
fix(github-body-field): verify the rewritten body and keep CRLF spacing
replace_field checked the rewrite against the lenient parser only, so
replacing A in "### A ... ### B" with an indented fence closed by a
column-0 run passed: CommonMark closes the fence there, while a
column-0 reader opens one and loses B. The rewritten body now goes
through the full strict-vs-lenient and nested-heading check, with every
field of the old body as a known name, and any disagreement refuses the
rewrite before vetted-ops body-field-set can PATCH.
Spacer detection and replacement assumed LF. On a CRLF body, get
returned the blank spacer as part of the value, and set dropped it and
wrote mixed line endings. Blank spacers are now recognised under any
line ending, and the replacement value and spacers are written in the
body's own convention (CRLF, lone CR or LF), so setting an unchanged
value is byte-identical and the CLI writes nothing.
The vetted-ops copy is re-synced.
Part of #1672
Generated-by: Claude Opus 5
---
.../vetted-ops/src/vetted_ops/body_field_format.py | 53 +++++++++++++++++-----
.../src/github_body_field/parser.py | 53 +++++++++++++++++-----
tools/github-body-field/tests/test_normalise.py | 46 +++++++++++++++++++
.../vetted-ops/src/vetted_ops/body_field_format.py | 53 +++++++++++++++++-----
tools/vetted-ops/tests/test_procedures.py | 7 +++
5 files changed, 179 insertions(+), 33 deletions(-)
diff --git a/plugins/magpie-vetted-ops/tools/vetted-ops/src/vetted_ops/body_field_format.py b/plugins/magpie-vetted-ops/tools/vetted-ops/src/vetted_ops/body_field_format.py
index 4e627c48..37244773 100644
--- a/plugins/magpie-vetted-ops/tools/vetted-ops/src/vetted_ops/body_field_format.py
+++ b/plugins/magpie-vetted-ops/tools/vetted-ops/src/vetted_ops/body_field_format.py
@@ -503,6 +503,24 @@ def normalise(
return NormaliseResult(text=text, changes=changes, needs_decision=needs_decision)
+_EOLS = ("\r\n", "\r", "\n")
+_ANY_EOL = re.compile(r"\r\n|\r|\n")
+
+
+def _is_blank_eol(line: str) -> bool:
+ """Whether ``line`` is an empty line: nothing but its ending (LF, CRLF or CR)."""
+ return line in _EOLS
+
+
+def _eol_of(body: str, line: str) -> str:
+ """The line ending ``line`` uses, else the body's first one, else LF."""
+ ending = line[len(_content(line)) :]
+ if ending:
+ return ending
+ first = _ANY_EOL.search(body)
+ return first.group(0) if first else "\n"
+
+
def _strip_spacer_blanks(body_lines: list[str]) -> tuple[bool, list[str], bool]:
"""Strip up to one leading and one trailing blank-line spacer.
@@ -510,10 +528,10 @@ def _strip_spacer_blanks(body_lines: list[str]) -> tuple[bool, list[str], bool]:
caller can later re-emit with the same convention.
"""
lines = list(body_lines)
- had_leading = bool(lines) and lines[0] == "\n"
+ had_leading = bool(lines) and _is_blank_eol(lines[0])
if had_leading:
lines = lines[1:]
- had_trailing = bool(lines) and lines[-1] == "\n" and len(lines) > 1
+ had_trailing = bool(lines) and _is_blank_eol(lines[-1]) and len(lines) > 1
# The `len(lines) > 1` guard is so a field whose *only* line
# is a blank one (i.e. an empty value with one spacer) doesn't
# get the same line both stripped and counted as "trailing".
@@ -606,19 +624,22 @@ def replace_field(body: str, field: str, new_value: str) -> str:
# - the empty-value case (caller passed "") produces an
# entirely empty content payload so the heading sits
# directly against the spacer / next heading.
- normalised = new_value
- while normalised.endswith("\n\n"):
- normalised = normalised[:-1]
- if normalised and not normalised.endswith("\n"):
- normalised += "\n"
+ # Written in the body's own line-ending convention (the heading line's),
+ # so a CRLF body stays CRLF and an unchanged value is byte-identical.
+ eol = _eol_of(body, target.heading_line)
+ normalised = _ANY_EOL.sub(eol, new_value)
+ while normalised.endswith(eol * 2):
+ normalised = normalised[: -len(eol)]
+ if normalised and not normalised.endswith(eol):
+ normalised += eol
rebuilt_body_lines: list[str] = []
if normalised:
if had_leading:
- rebuilt_body_lines.append("\n")
- rebuilt_body_lines.extend(normalised.splitlines(keepends=True))
+ rebuilt_body_lines.append(eol)
+ rebuilt_body_lines.extend(split_lines(normalised))
if had_trailing and not is_last_section:
- rebuilt_body_lines.append("\n")
+ rebuilt_body_lines.append(eol)
else:
# Empty value: collapse to a single blank-line spacer so the
# next heading isn't glued to the current one. Don't stack
@@ -627,7 +648,7 @@ def replace_field(body: str, field: str, new_value: str) -> str:
# UI). For the last section, emit nothing — the bare heading
# line is sufficient.
if not is_last_section and (had_leading or had_trailing):
- rebuilt_body_lines.append("\n")
+ rebuilt_body_lines.append(eol)
out: list[str] = list(preamble)
for index, s in enumerate(sections):
@@ -668,6 +689,16 @@ def _refuse_unsafe_rewrite(
)
_, before = _parse(old_body)
_, after = _parse(new_body)
+ # The rewritten body must read the same to every reader: run the full
+ # strict-vs-lenient and nested-heading check over it, with every field
+ # of the old body as a known name.
+ known = [s.name for s in before if s.name]
+ disputes = [d for d in normalise(new_body, known).needs_decision if d["kind"] == "container-heading"]
+ if disputes:
+ raise AmbiguousFieldError(
+ f"field {field_name!r}: after the rewrite, markdown readers would disagree on the body's fields "
+ f"({', '.join(sorted({str(d['field']) for d in disputes}))}); refusing to rewrite it"
+ )
if [s.name for s in before] != [s.name for s in after] or any(
(b.heading_line, b.body_lines) != (a.heading_line, a.body_lines)
for b, a in zip(before, after, strict=True)
diff --git a/tools/github-body-field/src/github_body_field/parser.py b/tools/github-body-field/src/github_body_field/parser.py
index 4e627c48..37244773 100644
--- a/tools/github-body-field/src/github_body_field/parser.py
+++ b/tools/github-body-field/src/github_body_field/parser.py
@@ -503,6 +503,24 @@ def normalise(
return NormaliseResult(text=text, changes=changes, needs_decision=needs_decision)
+_EOLS = ("\r\n", "\r", "\n")
+_ANY_EOL = re.compile(r"\r\n|\r|\n")
+
+
+def _is_blank_eol(line: str) -> bool:
+ """Whether ``line`` is an empty line: nothing but its ending (LF, CRLF or CR)."""
+ return line in _EOLS
+
+
+def _eol_of(body: str, line: str) -> str:
+ """The line ending ``line`` uses, else the body's first one, else LF."""
+ ending = line[len(_content(line)) :]
+ if ending:
+ return ending
+ first = _ANY_EOL.search(body)
+ return first.group(0) if first else "\n"
+
+
def _strip_spacer_blanks(body_lines: list[str]) -> tuple[bool, list[str], bool]:
"""Strip up to one leading and one trailing blank-line spacer.
@@ -510,10 +528,10 @@ def _strip_spacer_blanks(body_lines: list[str]) -> tuple[bool, list[str], bool]:
caller can later re-emit with the same convention.
"""
lines = list(body_lines)
- had_leading = bool(lines) and lines[0] == "\n"
+ had_leading = bool(lines) and _is_blank_eol(lines[0])
if had_leading:
lines = lines[1:]
- had_trailing = bool(lines) and lines[-1] == "\n" and len(lines) > 1
+ had_trailing = bool(lines) and _is_blank_eol(lines[-1]) and len(lines) > 1
# The `len(lines) > 1` guard is so a field whose *only* line
# is a blank one (i.e. an empty value with one spacer) doesn't
# get the same line both stripped and counted as "trailing".
@@ -606,19 +624,22 @@ def replace_field(body: str, field: str, new_value: str) -> str:
# - the empty-value case (caller passed "") produces an
# entirely empty content payload so the heading sits
# directly against the spacer / next heading.
- normalised = new_value
- while normalised.endswith("\n\n"):
- normalised = normalised[:-1]
- if normalised and not normalised.endswith("\n"):
- normalised += "\n"
+ # Written in the body's own line-ending convention (the heading line's),
+ # so a CRLF body stays CRLF and an unchanged value is byte-identical.
+ eol = _eol_of(body, target.heading_line)
+ normalised = _ANY_EOL.sub(eol, new_value)
+ while normalised.endswith(eol * 2):
+ normalised = normalised[: -len(eol)]
+ if normalised and not normalised.endswith(eol):
+ normalised += eol
rebuilt_body_lines: list[str] = []
if normalised:
if had_leading:
- rebuilt_body_lines.append("\n")
- rebuilt_body_lines.extend(normalised.splitlines(keepends=True))
+ rebuilt_body_lines.append(eol)
+ rebuilt_body_lines.extend(split_lines(normalised))
if had_trailing and not is_last_section:
- rebuilt_body_lines.append("\n")
+ rebuilt_body_lines.append(eol)
else:
# Empty value: collapse to a single blank-line spacer so the
# next heading isn't glued to the current one. Don't stack
@@ -627,7 +648,7 @@ def replace_field(body: str, field: str, new_value: str) -> str:
# UI). For the last section, emit nothing — the bare heading
# line is sufficient.
if not is_last_section and (had_leading or had_trailing):
- rebuilt_body_lines.append("\n")
+ rebuilt_body_lines.append(eol)
out: list[str] = list(preamble)
for index, s in enumerate(sections):
@@ -668,6 +689,16 @@ def _refuse_unsafe_rewrite(
)
_, before = _parse(old_body)
_, after = _parse(new_body)
+ # The rewritten body must read the same to every reader: run the full
+ # strict-vs-lenient and nested-heading check over it, with every field
+ # of the old body as a known name.
+ known = [s.name for s in before if s.name]
+ disputes = [d for d in normalise(new_body, known).needs_decision if d["kind"] == "container-heading"]
+ if disputes:
+ raise AmbiguousFieldError(
+ f"field {field_name!r}: after the rewrite, markdown readers would disagree on the body's fields "
+ f"({', '.join(sorted({str(d['field']) for d in disputes}))}); refusing to rewrite it"
+ )
if [s.name for s in before] != [s.name for s in after] or any(
(b.heading_line, b.body_lines) != (a.heading_line, a.body_lines)
for b, a in zip(before, after, strict=True)
diff --git a/tools/github-body-field/tests/test_normalise.py b/tools/github-body-field/tests/test_normalise.py
index ee5579f2..1c1ab733 100644
--- a/tools/github-body-field/tests/test_normalise.py
+++ b/tools/github-body-field/tests/test_normalise.py
@@ -502,3 +502,49 @@ def test_any_number_of_markers_still_needs_a_decision(count: int) -> None:
def test_ten_digit_ordered_markers_are_not_markers() -> None:
assert normalise("### A\n\n1234567890. ### Severity\n", KNOWN).needs_decision == []
+
+
+def test_replace_refuses_a_value_that_splits_the_readers() -> None:
+ # Lenient pairing closes the indented fence at the column-0 run; a column-0
+ # reader opens a fence there instead and loses B.
+ body = "### A\n\nx\n\n### B\n\ny\n"
+ with pytest.raises(AmbiguousFieldError, match="disagree"):
+ replace_field(body, "A", " ```\ncode\n```")
+
+
+def test_crlf_bodies_keep_their_spacers_and_line_endings() -> None:
+ body = "### A\r\n\r\nx\r\n\r\n### B\r\n\r\ny\r\n"
+ assert extract_field(body, "A") == "x\r\n"
+ assert replace_field(body, "A", extract_field(body, "A")) == body
+ assert replace_field(body, "A", "x") == body
+ new = replace_field(body, "A", "one\ntwo\n")
+ assert new == "### A\r\n\r\none\r\ntwo\r\n\r\n### B\r\n\r\ny\r\n"
+ lone_cr = "### A\r\rx\r\r### B\r\ry\r"
+ assert replace_field(lone_cr, "A", "x") == lone_cr
+
+
+def test_cli_get_and_set_on_a_remote_crlf_body(monkeypatch, capsys) -> None:
+ import subprocess
+
+ body = "### A\r\n\r\nx\r\n\r\n### B\r\n\r\ny\r\n"
+ writes: list[str] = []
+
+ class Result:
+ def __init__(self, stdout: str = "") -> None:
+ self.returncode = 0
+ self.stdout = stdout
+ self.stderr = ""
+
+ def fake_run(cmd, input=None, **kwargs):
+ if "edit" in cmd:
+ writes.append(input)
+ return Result()
+ return Result(json.dumps({"body": body}))
+
+ monkeypatch.setattr(subprocess, "run", fake_run)
+ assert cli.main(["get", "1", "--field", "A"]) == 0
+ assert capsys.readouterr().out == "x\r\n"
+ assert cli.main(["set", "1", "--field", "A", "--value", "x"]) == 0
+ assert writes == [] # unchanged: byte-identical, nothing written
+ assert cli.main(["set", "1", "--field", "A", "--value", "z"]) == 0
+ assert writes == [body.replace("\r\nx\r\n", "\r\nz\r\n")]
diff --git a/tools/vetted-ops/src/vetted_ops/body_field_format.py b/tools/vetted-ops/src/vetted_ops/body_field_format.py
index 4e627c48..37244773 100644
--- a/tools/vetted-ops/src/vetted_ops/body_field_format.py
+++ b/tools/vetted-ops/src/vetted_ops/body_field_format.py
@@ -503,6 +503,24 @@ def normalise(
return NormaliseResult(text=text, changes=changes, needs_decision=needs_decision)
+_EOLS = ("\r\n", "\r", "\n")
+_ANY_EOL = re.compile(r"\r\n|\r|\n")
+
+
+def _is_blank_eol(line: str) -> bool:
+ """Whether ``line`` is an empty line: nothing but its ending (LF, CRLF or CR)."""
+ return line in _EOLS
+
+
+def _eol_of(body: str, line: str) -> str:
+ """The line ending ``line`` uses, else the body's first one, else LF."""
+ ending = line[len(_content(line)) :]
+ if ending:
+ return ending
+ first = _ANY_EOL.search(body)
+ return first.group(0) if first else "\n"
+
+
def _strip_spacer_blanks(body_lines: list[str]) -> tuple[bool, list[str], bool]:
"""Strip up to one leading and one trailing blank-line spacer.
@@ -510,10 +528,10 @@ def _strip_spacer_blanks(body_lines: list[str]) -> tuple[bool, list[str], bool]:
caller can later re-emit with the same convention.
"""
lines = list(body_lines)
- had_leading = bool(lines) and lines[0] == "\n"
+ had_leading = bool(lines) and _is_blank_eol(lines[0])
if had_leading:
lines = lines[1:]
- had_trailing = bool(lines) and lines[-1] == "\n" and len(lines) > 1
+ had_trailing = bool(lines) and _is_blank_eol(lines[-1]) and len(lines) > 1
# The `len(lines) > 1` guard is so a field whose *only* line
# is a blank one (i.e. an empty value with one spacer) doesn't
# get the same line both stripped and counted as "trailing".
@@ -606,19 +624,22 @@ def replace_field(body: str, field: str, new_value: str) -> str:
# - the empty-value case (caller passed "") produces an
# entirely empty content payload so the heading sits
# directly against the spacer / next heading.
- normalised = new_value
- while normalised.endswith("\n\n"):
- normalised = normalised[:-1]
- if normalised and not normalised.endswith("\n"):
- normalised += "\n"
+ # Written in the body's own line-ending convention (the heading line's),
+ # so a CRLF body stays CRLF and an unchanged value is byte-identical.
+ eol = _eol_of(body, target.heading_line)
+ normalised = _ANY_EOL.sub(eol, new_value)
+ while normalised.endswith(eol * 2):
+ normalised = normalised[: -len(eol)]
+ if normalised and not normalised.endswith(eol):
+ normalised += eol
rebuilt_body_lines: list[str] = []
if normalised:
if had_leading:
- rebuilt_body_lines.append("\n")
- rebuilt_body_lines.extend(normalised.splitlines(keepends=True))
+ rebuilt_body_lines.append(eol)
+ rebuilt_body_lines.extend(split_lines(normalised))
if had_trailing and not is_last_section:
- rebuilt_body_lines.append("\n")
+ rebuilt_body_lines.append(eol)
else:
# Empty value: collapse to a single blank-line spacer so the
# next heading isn't glued to the current one. Don't stack
@@ -627,7 +648,7 @@ def replace_field(body: str, field: str, new_value: str) -> str:
# UI). For the last section, emit nothing — the bare heading
# line is sufficient.
if not is_last_section and (had_leading or had_trailing):
- rebuilt_body_lines.append("\n")
+ rebuilt_body_lines.append(eol)
out: list[str] = list(preamble)
for index, s in enumerate(sections):
@@ -668,6 +689,16 @@ def _refuse_unsafe_rewrite(
)
_, before = _parse(old_body)
_, after = _parse(new_body)
+ # The rewritten body must read the same to every reader: run the full
+ # strict-vs-lenient and nested-heading check over it, with every field
+ # of the old body as a known name.
+ known = [s.name for s in before if s.name]
+ disputes = [d for d in normalise(new_body, known).needs_decision if d["kind"] == "container-heading"]
+ if disputes:
+ raise AmbiguousFieldError(
+ f"field {field_name!r}: after the rewrite, markdown readers would disagree on the body's fields "
+ f"({', '.join(sorted({str(d['field']) for d in disputes}))}); refusing to rewrite it"
+ )
if [s.name for s in before] != [s.name for s in after] or any(
(b.heading_line, b.body_lines) != (a.heading_line, a.body_lines)
for b, a in zip(before, after, strict=True)
diff --git a/tools/vetted-ops/tests/test_procedures.py b/tools/vetted-ops/tests/test_procedures.py
index 16eee51f..905579df 100644
--- a/tools/vetted-ops/tests/test_procedures.py
+++ b/tools/vetted-ops/tests/test_procedures.py
@@ -480,6 +480,13 @@ def test_body_field_set_refuses_to_replace_a_span_hiding_another_field() -> None
assert fake.writes == []
+def test_body_field_set_refuses_a_value_that_splits_the_readers() -> None:
+ fake = FakeGh(body="### A\n\nx\n\n### B\n\ny\n")
+ with pytest.raises(procedures.ProcedureRefused, match="disagree"):
+ procedures.body_field_set(runner(fake), number="7", field="A", text=" ```\ncode\n```")
+ assert fake.writes == []
+
+
def test_body_field_get_prints_only_the_value(capsys: pytest.CaptureFixture[str]) -> None:
fake = FakeGh(body=ISSUE_BODY)
procedures.body_field_get(runner(fake, writes=False), number="7", field="Severity")