diff --git a/src/openjd/sessions/_embedded_files.py b/src/openjd/sessions/_embedded_files.py index 8f8d26b8..6ca5c2fd 100644 --- a/src/openjd/sessions/_embedded_files.py +++ b/src/openjd/sessions/_embedded_files.py @@ -47,10 +47,14 @@ def _convert_line_endings(data: str, end_of_line: Optional[str]) -> str: Args: data: The string data to convert - end_of_line: One of None, "AUTO", "LF", or "CRLF" + end_of_line: One of None, "AUTO", "LF", or "CRLF". Any other value is + rejected rather than silently leaving the data unconverted. Returns: The data with converted line endings + + Raises: + ValueError: if ``end_of_line`` is not a recognized value. """ if end_of_line is None or end_of_line == "AUTO": # AUTO: use OS native line endings @@ -63,7 +67,10 @@ def _convert_line_endings(data: str, end_of_line: Optional[str]) -> str: return data.replace("\r\n", "\n") elif end_of_line == "CRLF": return _LF_NOT_CRLF.sub("\r\n", data) - return data + raise ValueError( + f"Unrecognized end_of_line value {end_of_line!r}. Expected one of " + "None, 'AUTO', 'LF', or 'CRLF'." + ) def chown_group(path: Path, group: str) -> None: @@ -142,7 +149,47 @@ def write_file_for_user( user: Optional[SessionUser], additional_permissions: int = 0, end_of_line: Optional[str] = None, + *, + verbatim: bool = False, ) -> None: + """Write ``data`` to ``filename`` with owner-only read/write permissions, + optionally extended to a session user's group. + + Arguments: + filename: Absolute path of the file to create or overwrite. + data: The text to write. Encoded as UTF-8 by this function. + user: If given, the session user whose group gains read/write access. + additional_permissions: Extra mode bits to set, masked to the owner + triad, and extended to the group triad only when ``user`` is given + on POSIX. + end_of_line: One of ``None``, ``"AUTO"``, ``"LF"``, or ``"CRLF"``. + ``None`` and ``"AUTO"`` select the host's native line ending. Any + other value is rejected. + verbatim: Write ``data`` exactly as given. The content is not escaped, + not re-interpreted, and its line endings are not converted: + ``_convert_line_endings`` is not called at all, and the bytes on + disk are ``data.encode("utf-8")`` unmodified. This is for a caller + writing opaque content it does not own and must not alter, such as + a script fetched from a third party. This is a writer-level option + only and deliberately not an ``openjd-model`` ``EndOfLine`` enum + value -- that enum is a specification surface describing what a job + template may request, and verbatim is not a line-ending mode a + template can ask for. It is a statement by the calling code that + the bytes it holds are already final. + + Raises: + ValueError: if ``verbatim`` is True and ``end_of_line`` is not None + (the two requests contradict each other), or if ``end_of_line`` + is not a recognized value. + OSError: if the file could not be created or written, or if the final + path component is a symbolic link. + """ + if verbatim and end_of_line is not None: + raise ValueError( + f"verbatim=True writes data unmodified, so end_of_line={end_of_line!r} " + "cannot also be applied. Pass one or the other." + ) + # File should only be r/w by the owner, by default # flags: @@ -166,7 +213,8 @@ def write_file_for_user( # while still allowing the file to be rewritten across Tasks. flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC # On Windows, use O_BINARY to prevent automatic line ending conversion - # since we handle line endings explicitly via _convert_line_endings + # since line endings are handled explicitly here via _convert_line_endings, + # when they are converted at all (see 'verbatim'). flags |= getattr(os, "O_BINARY", 0) # O_NOFOLLOW is not defined on Windows; getattr(..., 0) makes this a no-op there. flags |= getattr(os, "O_NOFOLLOW", 0) @@ -174,9 +222,9 @@ def write_file_for_user( # S_IRUSR - Read by owner # S_IWUSR - Write by owner mode = stat.S_IRUSR | stat.S_IWUSR | (additional_permissions & stat.S_IRWXU) - converted_data = _convert_line_endings(data, end_of_line) + data_to_write = data if verbatim else _convert_line_endings(data, end_of_line) with _open_context(filename, flags, mode=mode) as fd: - os.write(fd, converted_data.encode("utf-8")) + os.write(fd, data_to_write.encode("utf-8")) if os.name == "posix": if user is not None: diff --git a/test/openjd/sessions_v0/test_embedded_files.py b/test/openjd/sessions_v0/test_embedded_files.py index 8cbf373d..5e8964a2 100644 --- a/test/openjd/sessions_v0/test_embedded_files.py +++ b/test/openjd/sessions_v0/test_embedded_files.py @@ -6,6 +6,7 @@ import uuid from dataclasses import dataclass from pathlib import Path +from typing import Optional from unittest.mock import MagicMock, patch from openjd.sessions._os_checker import is_posix, is_windows import pytest @@ -26,6 +27,7 @@ from openjd.sessions._embedded_files import ( EmbeddedFiles, EmbeddedFilesScope, + _convert_line_endings, _validate_embedded_filename, chown_group, write_file_for_user, @@ -599,6 +601,219 @@ def test_auto_uses_crlf_on_windows(self, tmp_path: Path) -> None: result_bytes = file.read() assert result_bytes == b"line1\r\nline2" + class TestVerbatimWrite: + """Tests for write_file_for_user()'s verbatim option.""" + + def test_verbatim_preserves_mixed_line_endings(self, tmp_path: Path) -> None: + # verbatim=True must write the input bytes exactly, on POSIX and + # Windows alike. A mixed LF/CRLF body is rewritten by every + # end_of_line mode, so it fails if verbatim stops bypassing the + # conversion. + + # GIVEN + filename = tmp_path / uuid.uuid4().hex + data = "line1\nline2\r\nline3\n" + + # WHEN + write_file_for_user(filename, data, user=None, verbatim=True) + + # THEN + with open(filename, "rb") as file: + result_bytes = file.read() + assert result_bytes == b"line1\nline2\r\nline3\n" + + @pytest.mark.skipif(not is_posix(), reason="posix file mode bits") + def test_verbatim_matches_non_verbatim_permissions(self, tmp_path: Path) -> None: + # The verbatim branch changes only which string is encoded, so the + # resulting file mode must be identical to the default path's. + + # GIVEN + verbatim_file = tmp_path / uuid.uuid4().hex + converted_file = tmp_path / uuid.uuid4().hex + + # WHEN + write_file_for_user( + verbatim_file, + "data", + user=None, + additional_permissions=stat.S_IXUSR, + verbatim=True, + ) + write_file_for_user( + converted_file, "data", user=None, additional_permissions=stat.S_IXUSR + ) + + # THEN + verbatim_mode = os.stat(verbatim_file).st_mode & 0o777 + converted_mode = os.stat(converted_file).st_mode & 0o777 + assert verbatim_mode == converted_mode + # Asserted absolutely as well, so the equality cannot hold by both + # paths regressing together. + assert verbatim_mode == (stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR) + + @pytest.mark.skipif(not is_posix(), reason="O_NOFOLLOW is posix-specific") + def test_verbatim_still_refuses_to_follow_symlink(self, tmp_path: Path) -> None: + # The verbatim branch must go through the same O_NOFOLLOW-guarded + # writer, not a convenience writer that follows symlinks. + + # GIVEN + target = tmp_path / "target.txt" + target.write_text("original-contents") + link = tmp_path / "link.txt" + link.symlink_to(target) + + # WHEN / THEN + with pytest.raises(OSError): + write_file_for_user(link, "overwritten", user=None, verbatim=True) + # AND the symlink target is untouched + assert target.read_text() == "original-contents" + + @pytest.mark.parametrize( + "end_of_line", + [ + pytest.param("AUTO", id="AUTO"), + pytest.param("LF", id="LF"), + pytest.param("CRLF", id="CRLF"), + ], + ) + def test_verbatim_with_explicit_end_of_line_raises( + self, tmp_path: Path, end_of_line: str + ) -> None: + # verbatim=True and an explicit end_of_line are contradictory + # requests. Neither silently wins. + + # GIVEN + filename = tmp_path / uuid.uuid4().hex + + # WHEN + with pytest.raises(ValueError) as excinfo: + write_file_for_user( + filename, "data", user=None, end_of_line=end_of_line, verbatim=True + ) + + # THEN + assert "verbatim" in str(excinfo.value) + assert "end_of_line" in str(excinfo.value) + # AND the guard ran before the O_TRUNC open, so the destination was + # not created or truncated on the way to raising. + assert not filename.exists() + + def test_verbatim_with_explicit_none_end_of_line_is_allowed(self, tmp_path: Path) -> None: + # Negative control for the guard: it fires on a non-None + # end_of_line, not on the keyword being present at all. + + # GIVEN + filename = tmp_path / uuid.uuid4().hex + + # WHEN + write_file_for_user(filename, "a\nb\r\nc\n", user=None, end_of_line=None, verbatim=True) + + # THEN + with open(filename, "rb") as file: + result_bytes = file.read() + assert result_bytes == b"a\nb\r\nc\n" + + @pytest.mark.parametrize( + "end_of_line,expected_bytes", + [ + pytest.param("LF", b"a\nb\nc\n", id="LF"), + pytest.param("CRLF", b"a\r\nb\r\nc\r\n", id="CRLF"), + ], + ) + def test_non_verbatim_still_converts( + self, tmp_path: Path, end_of_line: str, expected_bytes: bytes + ) -> None: + # Negative control: conversion still happens when verbatim is not + # requested. AUTO is covered by TestEndOfLine, whose expectation is + # platform-dependent. + + # GIVEN + filename = tmp_path / uuid.uuid4().hex + + # WHEN + write_file_for_user(filename, "a\nb\r\nc\n", user=None, end_of_line=end_of_line) + + # THEN + with open(filename, "rb") as file: + result_bytes = file.read() + assert result_bytes == expected_bytes + + class TestConvertLineEndings: + """Tests for _convert_line_endings()'s handling of unrecognized values.""" + + @pytest.mark.parametrize( + "end_of_line", + [ + pytest.param("crlf", id="case-wrong-crlf"), + pytest.param("Lf", id="case-wrong-Lf"), + pytest.param("", id="empty-string"), + pytest.param("typo", id="typo"), + pytest.param("VERBATIM", id="not-a-mode"), + ], + ) + def test_rejects_unrecognized_value(self, end_of_line: str) -> None: + # An unrecognized end_of_line must be rejected, not silently + # ignored. Previously each of these returned the data unconverted + # with no error, so a case-wrong value looked like it worked. + + # WHEN + with pytest.raises(ValueError) as excinfo: + _convert_line_endings("a\nb\r\nc\n", end_of_line) + + # THEN + assert repr(end_of_line) in str(excinfo.value) + + @pytest.mark.parametrize( + "end_of_line", + [ + pytest.param(None, id="None"), + pytest.param("AUTO", id="AUTO"), + ], + ) + def test_accepts_native_values(self, end_of_line: Optional[str]) -> None: + # Negative control: None and AUTO still select the host's native + # line ending and do not raise. + + # GIVEN + expected = "a\r\nb\r\nc\r\n" if is_windows() else "a\nb\nc\n" + + # WHEN + result = _convert_line_endings("a\nb\r\nc\n", end_of_line) + + # THEN + assert result == expected + + @pytest.mark.parametrize( + "end_of_line,expected", + [ + pytest.param("LF", "a\nb\nc\n", id="LF"), + pytest.param("CRLF", "a\r\nb\r\nc\r\n", id="CRLF"), + ], + ) + def test_accepts_explicit_values(self, end_of_line: str, expected: str) -> None: + # Negative control: the explicit modes still convert and do not + # raise, so the strict check is not over-broad. + + # WHEN + result = _convert_line_endings("a\nb\r\nc\n", end_of_line) + + # THEN + assert result == expected + + def test_write_file_for_user_rejects_unrecognized_value(self, tmp_path: Path) -> None: + # End to end through the writer: the conversion happens before the + # O_TRUNC open, so a rejected value leaves no file behind. + + # GIVEN + filename = tmp_path / uuid.uuid4().hex + + # WHEN + with pytest.raises(ValueError): + write_file_for_user(filename, "a\nb\r\nc\n", user=None, end_of_line="crlf") + + # THEN + assert not filename.exists() + class TestMaterialize: """Tests for EmbeddedFiles.materialize()"""