diff --git a/session_peer.py b/session_peer.py index d045f89..718022f 100644 --- a/session_peer.py +++ b/session_peer.py @@ -6056,25 +6056,56 @@ def skill_update_cache_path() -> Path: return update_cache_path().with_name("skill-update.json") +SKILL_FRONTMATTER_MAX_BYTES = 16384 + + def installed_skill_metadata(path: Path) -> dict: """Read only the small, published frontmatter fields; never execute a skill.""" try: - if path.stat().st_size > 16384: - return {"version": None, "runtimeMinVersion": None} - content = path.read_text(encoding="utf-8") + # Follow manager-owned symlinks, but never block on a POSIX FIFO or + # interpret nonregular data. Read a finite prefix, not the skill body. + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_BINARY", 0) + | getattr(os, "O_NONBLOCK", 0)) + try: + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + return {"version": None, "runtimeMinVersion": None} + chunks, remaining = [], SKILL_FRONTMATTER_MAX_BYTES + 1 + while remaining: + chunk = os.read(descriptor, remaining) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + content = b"".join(chunks) + finally: + os.close(descriptor) except (OSError, UnicodeError): return {"version": None, "runtimeMinVersion": None} - match = re.match(r"\A---\r?\n(.*?)\r?\n---(?:\r?\n|\Z)", content, re.DOTALL) - if match is None: + # The extra byte distinguishes a closing delimiter at actual EOF from a + # prefix cut just before more text. Delimiters and line endings count. + match = re.match(rb"\A---(?:\r\n|\r|\n)(.*?)(?:\r\n|\r|\n)---(?:(?:\r\n|\r|\n)|\Z)", + content, re.DOTALL) + if match is None or match.end() > SKILL_FRONTMATTER_MAX_BYTES: + return {"version": None, "runtimeMinVersion": None} + try: + # Preserve read_text's universal-newline behavior, but decode only the + # complete header: a cut/invalid UTF-8 body cannot erase valid metadata. + header = match.group(1).decode("utf-8").replace("\r\n", "\n").replace("\r", "\n") + except UnicodeError: return {"version": None, "runtimeMinVersion": None} - block = re.search(r"(?m)^metadata:\s*\n((?:[ \t]+[^\n]*\n?)*)", match.group(1)) + block = re.search(r"(?m)^metadata:\s*\n((?:[ \t]+[^\n]*\n?)*)", header) if block is None: return {"version": None, "runtimeMinVersion": None} fields = dict(re.findall(r"(?m)^ ([a-z-]+):\s*[\"']?([0-9]+\.[0-9]+\.[0-9]+)[\"']?\s*$", block.group(1))) - version = fields.get("version") - minimum = fields.get("runtime-min-version") - return {"version": version if stable_version(version) else None, - "runtimeMinVersion": minimum if stable_version(minimum) else None} + def known_version(value): + try: + return value if stable_version(value) else None + except ValueError: + # Python's integer digit limit must not turn malformed metadata + # into an exception, or erase a separately valid published field. + return None + return {"version": known_version(fields.get("version")), + "runtimeMinVersion": known_version(fields.get("runtime-min-version"))} def installed_skills() -> list[dict]: diff --git a/session_peer_core/updates.py b/session_peer_core/updates.py index 8060499..e31756e 100644 --- a/session_peer_core/updates.py +++ b/session_peer_core/updates.py @@ -104,25 +104,56 @@ def skill_update_cache_path() -> Path: return update_cache_path().with_name("skill-update.json") +SKILL_FRONTMATTER_MAX_BYTES = 16384 + + def installed_skill_metadata(path: Path) -> dict: """Read only the small, published frontmatter fields; never execute a skill.""" try: - if path.stat().st_size > 16384: - return {"version": None, "runtimeMinVersion": None} - content = path.read_text(encoding="utf-8") + # Follow manager-owned symlinks, but never block on a POSIX FIFO or + # interpret nonregular data. Read a finite prefix, not the skill body. + descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_BINARY", 0) + | getattr(os, "O_NONBLOCK", 0)) + try: + if not stat.S_ISREG(os.fstat(descriptor).st_mode): + return {"version": None, "runtimeMinVersion": None} + chunks, remaining = [], SKILL_FRONTMATTER_MAX_BYTES + 1 + while remaining: + chunk = os.read(descriptor, remaining) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + content = b"".join(chunks) + finally: + os.close(descriptor) except (OSError, UnicodeError): return {"version": None, "runtimeMinVersion": None} - match = re.match(r"\A---\r?\n(.*?)\r?\n---(?:\r?\n|\Z)", content, re.DOTALL) - if match is None: + # The extra byte distinguishes a closing delimiter at actual EOF from a + # prefix cut just before more text. Delimiters and line endings count. + match = re.match(rb"\A---(?:\r\n|\r|\n)(.*?)(?:\r\n|\r|\n)---(?:(?:\r\n|\r|\n)|\Z)", + content, re.DOTALL) + if match is None or match.end() > SKILL_FRONTMATTER_MAX_BYTES: + return {"version": None, "runtimeMinVersion": None} + try: + # Preserve read_text's universal-newline behavior, but decode only the + # complete header: a cut/invalid UTF-8 body cannot erase valid metadata. + header = match.group(1).decode("utf-8").replace("\r\n", "\n").replace("\r", "\n") + except UnicodeError: return {"version": None, "runtimeMinVersion": None} - block = re.search(r"(?m)^metadata:\s*\n((?:[ \t]+[^\n]*\n?)*)", match.group(1)) + block = re.search(r"(?m)^metadata:\s*\n((?:[ \t]+[^\n]*\n?)*)", header) if block is None: return {"version": None, "runtimeMinVersion": None} fields = dict(re.findall(r"(?m)^ ([a-z-]+):\s*[\"']?([0-9]+\.[0-9]+\.[0-9]+)[\"']?\s*$", block.group(1))) - version = fields.get("version") - minimum = fields.get("runtime-min-version") - return {"version": version if stable_version(version) else None, - "runtimeMinVersion": minimum if stable_version(minimum) else None} + def known_version(value): + try: + return value if stable_version(value) else None + except ValueError: + # Python's integer digit limit must not turn malformed metadata + # into an exception, or erase a separately valid published field. + return None + return {"version": known_version(fields.get("version")), + "runtimeMinVersion": known_version(fields.get("runtime-min-version"))} def installed_skills() -> list[dict]: diff --git a/tests/core/test_skill_frontmatter.py b/tests/core/test_skill_frontmatter.py new file mode 100644 index 0000000..585c94f --- /dev/null +++ b/tests/core/test_skill_frontmatter.py @@ -0,0 +1,177 @@ +"""Bounded metadata only; no skill instruction execution or manager changes.""" +import errno +import os +from pathlib import Path +import tempfile +import unittest +from unittest import mock + +import session_peer as peer + + +# Exact frontmatter from the released v0.3.2 and v0.3.3 skills. Bodies below +# are synthetic padding at the observed document sizes, not copied instructions. +RELEASED_HEADER = '''--- +name: session-peer +description: Send user-requested messages to Claude Code, Codex, or a live registered Antigravity TUI on this machine, an SSH host, or a paired device using session-peer. Use for cross-session handoffs and notifications when the available native tools do not cover the requested target. +allowed-tools: Bash, Read +metadata: + version: "{version}" + runtime-min-version: "0.9.1" + runtime-full-version: "1.0.1" +--- +''' +LIMIT = 16384 +UNKNOWN = {"version": None, "runtimeMinVersion": None} + + +class SkillFrontmatter(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory(prefix="skill-metadata-") + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.path = self.root / "SKILL.md" + + def write(self, data): + self.path.write_bytes(data) + return self.path + + def metadata(self, data): + return peer.installed_skill_metadata(self.write(data)) + + def test_released_headers_survive_their_real_document_sizes(self): + for version, size in (("0.3.2", 16459), ("0.3.3", 18824)): + with self.subTest(version=version): + header = RELEASED_HEADER.format(version=version).encode("utf-8") + content = header + b"\n# Synthetic body\n" + b"x" * (size - len(header) - 18) + self.assertEqual(len(content), size) + self.assertEqual(self.metadata(content), {"version": version, "runtimeMinVersion": "0.9.1"}) + + def test_large_invalid_or_split_utf8_body_is_not_decoded_as_metadata(self): + header = RELEASED_HEADER.format(version="0.3.3").encode() + body = b"x" * (LIMIT - len(header)) + "ν•œ".encode() + b"\xff" * 100000 + self.assertEqual(self.metadata(header + body), {"version": "0.3.3", "runtimeMinVersion": "0.9.1"}) + + def test_valid_utf8_header_is_decoded_with_a_byte_not_character_limit(self): + prefix = '---\nname: ν•œκΈ€\nmetadata:\n version: "0.3.3"\n# ' + self.assertEqual(self.metadata((prefix + '\n---\n').encode("utf-8")), + {"version": "0.3.3", "runtimeMinVersion": None}) + text = prefix + "ν•œ" * (LIMIT // 3) + "\n---\n" + self.assertLess(len(text), LIMIT) + self.assertGreater(len(text.encode("utf-8")), LIMIT) + self.assertEqual(self.metadata(text.encode("utf-8")), UNKNOWN) + + def test_exact_budget_includes_both_delimiters_and_line_endings(self): + for eol in (b"\n", b"\r\n", b"\r"): + for final in (True, False): + with self.subTest(eol=eol, final_newline=final): + prefix = eol.join([b"---", b"metadata:", b' version: "0.3.3"', b"# "]) + suffix = eol + b"---" + (eol if final else b"") + header = prefix + b"x" * (LIMIT - len(prefix) - len(suffix)) + suffix + self.assertEqual(len(header), LIMIT) + self.assertEqual(self.metadata(header), {"version": "0.3.3", "runtimeMinVersion": None}) + self.assertEqual(self.metadata(prefix + b"x" + header[len(prefix):]), UNKNOWN) + + def test_closing_delimiter_at_read_limit_is_not_assumed_to_be_eof(self): + prefix = b'---\nmetadata:\n version: "0.3.3"\n# ' + header = prefix + b"x" * (LIMIT - len(prefix) - 4) + b"\n---" + self.assertEqual(len(header), LIMIT) + self.assertEqual(self.metadata(header + b"not-a-delimiter\n"), UNKNOWN) + self.assertEqual(self.metadata(header + b"\n"), UNKNOWN) + + def test_missing_malformed_truncated_and_bad_utf8_header_stay_unknown(self): + self.assertEqual(peer.installed_skill_metadata(self.path), UNKNOWN) + for data in (b"", b"not frontmatter", b"---\nmetadata:\n version: 0.3.3\n", + b"\xef\xbb\xbf---\nmetadata:\n version: 0.3.3\n---\n", + b"---\nname: fixture\n---\n", b"---\n\xff\nmetadata:\n version: 0.3.3\n---\n", + b"---\nmetadata: !!python/object/apply:fixture\n---\n"): + with self.subTest(case=len(data)): + self.assertEqual(self.metadata(data), UNKNOWN) + + def test_existing_field_selection_and_partial_unknown_semantics_are_preserved(self): + data = b'---\nmetadata:\n version: "bad"\n runtime-min-version: "0.9.1"\n---\n' + self.assertEqual(self.metadata(data), {"version": None, "runtimeMinVersion": "0.9.1"}) + duplicate = b'---\nmetadata:\n version: "0.3.2"\n version: "0.3.3"\n---\n' + self.assertEqual(self.metadata(duplicate), {"version": "0.3.3", "runtimeMinVersion": None}) + + def test_oversized_numeric_fields_are_unknown_independently(self): + for value in ("9" * 5000 + ".0.0", str(peer.sys.maxsize + 1) + ".0.0"): + for field in ("version", "runtime-min-version"): + with self.subTest(field=field, digits=len(value)): + version = value if field == "version" else "0.3.3" + minimum = value if field == "runtime-min-version" else "0.9.1" + content = ('---\nmetadata:\n version: "' + version + '"\n runtime-min-version: "' + + minimum + '"\n---\n').encode() + self.assertLess(len(content), LIMIT) + expected = {"version": None if field == "version" else "0.3.3", + "runtimeMinVersion": None if field == "runtime-min-version" else "0.9.1"} + self.assertEqual(self.metadata(content), expected) + + def test_reads_at_most_budget_plus_one_without_read_text(self): + path = self.write(RELEASED_HEADER.format(version="0.3.3").encode() + b"x" * 100000) + original = os.read + reads = [] + def bounded_read(fd, count): + self.assertGreater(count, 0) + reads.append(count) + return original(fd, count) + with mock.patch.object(peer.os, "read", side_effect=bounded_read), \ + mock.patch.object(Path, "read_text", side_effect=AssertionError("No whole text read")): + result = peer.installed_skill_metadata(path) + self.assertEqual(result["version"], "0.3.3") + self.assertLessEqual(sum(reads), LIMIT + 1) + + def test_short_reads_are_accumulated_with_a_total_byte_bound(self): + path = self.write(RELEASED_HEADER.format(version="0.3.3").encode() + b"x" * 100000) + original = os.read + returned = [] + def short(fd, count): + data = original(fd, min(count, 137)) + returned.append(len(data)) + return data + with mock.patch.object(peer.os, "read", side_effect=short): + result = peer.installed_skill_metadata(path) + self.assertEqual(result["version"], "0.3.3") + self.assertLessEqual(sum(returned), LIMIT + 1) + + @unittest.skipIf(os.name == "nt", "symlink privilege is not assumed") + def test_manager_symlink_is_followed_without_mutation(self): + target = self.write(RELEASED_HEADER.format(version="0.3.3").encode() + b"x" * 18824) + alias = self.root / "manager-link.md" + alias.symlink_to(target) + original = target.read_bytes() + self.assertEqual(peer.installed_skill_metadata(alias)["version"], "0.3.3") + self.assertTrue(alias.is_symlink()) + self.assertEqual(target.read_bytes(), original) + + def test_unreadable_and_directory_paths_are_unknown(self): + self.assertEqual(peer.installed_skill_metadata(self.root), UNKNOWN) + with mock.patch.object(peer.os, "open", side_effect=PermissionError("fixture")): + self.assertEqual(peer.installed_skill_metadata(self.path), UNKNOWN) + + @unittest.skipUnless(hasattr(os, "mkfifo"), "POSIX nonblocking FIFO") + def test_fifo_is_unknown_without_blocking_on_open(self): + os.mkfifo(self.path) + original_open = os.open + def nonblocking(path, flags): + self.assertTrue(flags & os.O_NONBLOCK, "FIFO open must be nonblocking") + return original_open(path, flags) + with mock.patch.object(peer.os, "open", side_effect=nonblocking), \ + mock.patch.object(Path, "open", side_effect=AssertionError("No blocking buffered FIFO open")): + self.assertEqual(peer.installed_skill_metadata(self.path), UNKNOWN) + + def test_descriptor_is_closed_after_read_failure_or_nonregular_rejection(self): + self.write(RELEASED_HEADER.format(version="0.3.3").encode()) + for failure in ("read", "stat", "nonregular"): + with self.subTest(failure=failure), mock.patch.object(peer.os, "close", wraps=os.close) as closed: + if failure == "nonregular": + patcher = mock.patch.object(peer.stat, "S_ISREG", return_value=False) + else: + patcher = mock.patch.object(peer.os, "fstat" if failure == "stat" else "read", + side_effect=OSError("fixture")) + with patcher: + self.assertEqual(peer.installed_skill_metadata(self.path), UNKNOWN) + closed.assert_called_once() + with self.assertRaises(OSError) as caught: + os.fstat(closed.call_args.args[0]) + self.assertEqual(caught.exception.errno, errno.EBADF)