From 77d4c5b94e22a79d3b745ebb66745ad1432b097e Mon Sep 17 00:00:00 2001 From: stacknil <219124580+stacknil@users.noreply.github.com> Date: Mon, 19 Oct 2026 14:27:29 +0800 Subject: [PATCH] fix(process): reject partially malformed ss users fields Signed-off-by: stacknil <219124580+stacknil@users.noreply.github.com> --- projects/linux-process-observe/README.md | 16 +++ .../linux_process_observe/parsers/ss_text.py | 6 +- .../linux-process-observe/tests/test_cli.py | 104 ++++++++++++++++++ .../tests/test_parsers.py | 68 ++++++++++++ 4 files changed, 192 insertions(+), 2 deletions(-) diff --git a/projects/linux-process-observe/README.md b/projects/linux-process-observe/README.md index c30f5b1..a40985c 100644 --- a/projects/linux-process-observe/README.md +++ b/projects/linux-process-observe/README.md @@ -23,6 +23,22 @@ proc/ `cmdline` may contain real NUL separators or the text escape `\0`. `exe` contains the resolved `/proc//exe` target as UTF-8 text; a real symlink is also accepted. +### Saved ss users field + +After the six socket columns, process context may be absent or a complete users field: + +```text +tail := empty | "users:(" reference ("," reference)* ")" +reference := "(" '"' nonempty-name '"' ",pid=" digits ",fd=" digits ")" +``` + +For example, `users:(("worker-a",pid=120,fd=3),("worker-b",pid=220,fd=4))` is accepted. Names may contain spaces, commas and parentheses inside quotes, but not double quotes. PID and FD text use decimal digits; key order and delimiters must match the grammar, without extra internal spacing. PID conversion is unchanged, and FD is not stored. References are sorted and deduplicated by `(pid, name)`, including when only FD differs. + +An absent field retains an unlinked socket. A present empty list, malformed member, truncated delimiter, missing wrapper or unconsumed suffix rejects the saved input rather than admitting a matching subset. The CLI returns 1 with contextual `input=ss` stderr and no new snapshot output. Error line numbers count nonblank lines, including an optional header. + +This is a bounded saved-input profile. Thread attributes, alternate key order, quote escapes, fd-less references and extra tail metadata are unsupported; acceptance of every native ss variant is not claimed. + + ## Outputs | Artifact | Contents | diff --git a/projects/linux-process-observe/src/linux_process_observe/parsers/ss_text.py b/projects/linux-process-observe/src/linux_process_observe/parsers/ss_text.py index 300b55f..48b5a43 100644 --- a/projects/linux-process-observe/src/linux_process_observe/parsers/ss_text.py +++ b/projects/linux-process-observe/src/linux_process_observe/parsers/ss_text.py @@ -6,6 +6,8 @@ PROCESS_RE = re.compile(r'"(?P[^"]+)",pid=(?P\d+)') +_PROCESS_REF_PATTERN = r'\("[^"]+",pid=\d+,fd=\d+\)' +_USERS_FIELD_RE = re.compile(rf'users:\({_PROCESS_REF_PATTERN}(?:,{_PROCESS_REF_PATTERN})*\)') def parse_ss_text(text: str) -> list[SocketObservation]: @@ -79,8 +81,8 @@ def _parse_endpoint(token: str, line_number: int) -> tuple[str, str | int]: def _parse_process_refs(token: str, line_number: int) -> list[ProcessRef]: - matches = [ProcessRef(name=match.group("name"), pid=int(match.group("pid"))) for match in PROCESS_RE.finditer(token)] - if "users:" in token and not matches: + if token and _USERS_FIELD_RE.fullmatch(token) is None: raise ValueError(f"malformed ss process field on line {line_number}") + matches = [ProcessRef(name=match.group("name"), pid=int(match.group("pid"))) for match in PROCESS_RE.finditer(token)] unique = {(item.pid, item.name): item for item in matches} return [unique[key] for key in sorted(unique)] diff --git a/projects/linux-process-observe/tests/test_cli.py b/projects/linux-process-observe/tests/test_cli.py index 4bb2b89..e3a5f16 100644 --- a/projects/linux-process-observe/tests/test_cli.py +++ b/projects/linux-process-observe/tests/test_cli.py @@ -1,6 +1,8 @@ from pathlib import Path import json +import pytest + from linux_process_observe.cli import main @@ -112,3 +114,105 @@ def _snapshot(name: str, observed_at: str, output_dir: Path) -> int: str(output_dir), ] ) + + +@pytest.mark.parametrize("malformed_first", [False, True]) +@pytest.mark.parametrize("existing_output", [False, True]) +def test_cli_rejects_partial_ss_users_without_touching_outputs( + tmp_path, capsys, malformed_first, existing_output +): + good = '("sshd",pid=120,fd=3)' + bad = '("app-worker",pid=bad,fd=7)' + members = f"{bad},{good}" if malformed_first else f"{good},{bad}" + ss = tmp_path / "ss.txt" + ss.write_text( + "udp UNCONN 0 0 0.0.0.0:68 0.0.0.0:*\n\n" + f"tcp LISTEN 0 64 127.0.0.1:9000 0.0.0.0:* users:({members})\n", + encoding="utf-8", + ) + original_input = ss.read_bytes() + output = tmp_path / "output" + preserved = {} + if existing_output: + output.mkdir() + for name in ("process_snapshot.json", "process_socket_links.json", + "process_diff.json", "report.md", "events.jsonl"): + path = output / name + path.write_text(f"preserve {name}\n", encoding="utf-8") + preserved[name] = path.read_bytes() + + result = main([ + "snapshot", "--proc-root", str(FIXTURES / "baseline" / "proc"), + "--ss", str(ss), "--host-id", "lab-host", + "--observed-at", "2026-07-05T00:05:00Z", "--output-dir", str(output), + ]) + captured = capsys.readouterr() + + assert result == 1 + assert captured.out == "" + assert "error command=snapshot input=ss" in captured.err + assert f"path={ss}" in captured.err + assert "type=ValueError message=malformed ss process field on line 2" in captured.err + assert "Traceback" not in captured.err + assert ss.read_bytes() == original_input + if existing_output: + assert {path.name: path.read_bytes() for path in output.iterdir()} == preserved + else: + assert not output.exists() + + +def test_complete_ss_users_keeps_valid_diff_and_blocks_malformed_after(tmp_path, capsys): + prefix = "tcp LISTEN 0 64 127.0.0.1:9000 0.0.0.0:*" + first = '("sshd",pid=120,fd=3)' + second = '("app-worker",pid=220,fd=7)' + ss = tmp_path / "ss.txt" + pairs = [] + for label, time, members in [ + ("before", "2026-07-05T00:00:00Z", f"{first},{second}"), + ("after", "2026-07-05T00:05:00Z", f"{second},{first}"), + ]: + ss.write_text(f"{prefix} users:({members})\n", encoding="utf-8") + output = tmp_path / label + assert main([ + "snapshot", "--proc-root", str(FIXTURES / "baseline" / "proc"), + "--ss", str(ss), "--host-id", "lab-host", "--observed-at", time, + "--output-dir", str(output), + ]) == 0 + links = json.loads((output / "process_socket_links.json").read_text(encoding="utf-8")) + assert [row["pid"] for row in links["records"]] == [120, 220] + assert all(row["linked"] is True for row in links["records"]) + pairs.append(output) + + diff_args = [] + for label, output in zip(("before", "after"), pairs): + diff_args.extend([ + f"--{label}-processes", str(output / "process_snapshot.json"), + f"--{label}-links", str(output / "process_socket_links.json"), + ]) + diff_dir = tmp_path / "diff" + assert main(["diff", *diff_args, "--output-dir", str(diff_dir)]) == 0 + assert json.loads((diff_dir / "process_diff.json").read_text(encoding="utf-8"))["records"] == [] + events = diff_dir / "events.jsonl" + assert main(["adapt", "--input", str(diff_dir / "process_diff.json"), + "--output", str(events)]) == 0 + assert events.read_bytes() == b"" + assert "Process/socket links: +0 / -0" in (diff_dir / "report.md").read_text(encoding="utf-8") + preserved = {path: path.read_bytes() for directory in [*pairs, diff_dir] for path in directory.iterdir()} + capsys.readouterr() + + ss.write_text(f'{prefix} users:({first},("app-worker",pid=bad,fd=7))\n', encoding="utf-8") + original_input = ss.read_bytes() + failed_output = tmp_path / "malformed-after" + assert main([ + "snapshot", "--proc-root", str(FIXTURES / "baseline" / "proc"), + "--ss", str(ss), "--host-id", "lab-host", + "--observed-at", "2026-07-05T00:05:00Z", "--output-dir", str(failed_output), + ]) == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "error command=snapshot input=ss" in captured.err + assert "malformed ss process field on line 1" in captured.err + assert "Traceback" not in captured.err + assert not failed_output.exists() + assert all(path.read_bytes() == content for path, content in preserved.items()) + assert ss.read_bytes() == original_input diff --git a/projects/linux-process-observe/tests/test_parsers.py b/projects/linux-process-observe/tests/test_parsers.py index 18c11a7..404e040 100644 --- a/projects/linux-process-observe/tests/test_parsers.py +++ b/projects/linux-process-observe/tests/test_parsers.py @@ -45,3 +45,71 @@ def test_malformed_procfs_and_ss_inputs_fail_closed() -> None: with pytest.raises(ValueError, match="unsupported ss protocol on line 1: u_str"): parse_ss_text("u_str ESTAB 0 0 /run/example.sock * users:((\"example\",pid=10,fd=3))") + + +SS_SOCKET = "tcp LISTEN 0 64 127.0.0.1:9000 0.0.0.0:*" +SS_WORKER_A = '("worker-a",pid=120,fd=3)' +SS_WORKER_B = '("worker-b",pid=220,fd=4)' +SS_BAD_WORKER = '("worker-b",pid=bad,fd=4)' + + +@pytest.mark.parametrize( + ("tail", "expected"), + [ + pytest.param(f"users:({SS_WORKER_A})", [("worker-a", 120)], id="single"), + pytest.param(f"users:({SS_WORKER_A},{SS_WORKER_B})", + [("worker-a", 120), ("worker-b", 220)], id="multiple"), + pytest.param(f"users:({SS_WORKER_B},{SS_WORKER_A})", + [("worker-a", 120), ("worker-b", 220)], id="reversed"), + pytest.param(f"users:({SS_WORKER_A},{SS_WORKER_A})", + [("worker-a", 120)], id="duplicate"), + pytest.param(f'users:({SS_WORKER_A},("worker-a",pid=120,fd=9))', + [("worker-a", 120)], id="different-fd"), + pytest.param("", [], id="missing"), + pytest.param('users:(("worker, (a)",pid=120,fd=3))', + [("worker, (a)", 120)], id="quoted-delimiters"), + pytest.param('users:(("worker-a",pid=000120,fd=00))', + [("worker-a", 120)], id="leading-zeroes"), + pytest.param('users:(("worker-a",pid=\u0661\u0662\u0660,fd=3))', + [("worker-a", 120)], id="decimal-digits"), + pytest.param('users:(("worker-a",pid=0,fd=0))', + [("worker-a", 0)], id="zero"), + ], +) +def test_complete_ss_users_preserves_reference_projection(tail, expected): + line = SS_SOCKET + (f" {tail}" if tail else "") + refs = parse_ss_text(line)[0].process_refs + + assert [(item.name, item.pid) for item in refs] == expected + + +@pytest.mark.parametrize( + "tail", + [ + pytest.param(f"users:({SS_WORKER_A},{SS_BAD_WORKER})", id="valid-then-malformed"), + pytest.param(f"users:({SS_BAD_WORKER},{SS_WORKER_A})", id="malformed-then-valid"), + pytest.param(f"users:({SS_WORKER_A}", id="truncated-wrapper"), + pytest.param('users:(("worker-a",pid=120', id="truncated-reference"), + pytest.param(f"users:({SS_WORKER_A},garbage)", id="garbage-inside"), + pytest.param(f"users:({SS_WORKER_A}) trailing", id="garbage-after"), + pytest.param('users:(("worker-a",pid=120junk,fd=3))', id="pid-suffix"), + pytest.param('users:(("worker-a",pid=120,fd=bad))', id="nonnumeric-fd"), + pytest.param(SS_WORKER_A, id="missing-wrapper"), + pytest.param(f"users:({SS_BAD_WORKER})", id="malformed-only"), + pytest.param("users:()", id="empty-users"), + ], +) +def test_incomplete_ss_users_rejects_entire_field(tail): + with pytest.raises(ValueError, match=r"^malformed ss process field on line 1$"): + parse_ss_text(f"{SS_SOCKET} {tail}") + + +@pytest.mark.parametrize("malformed", [False, True]) +def test_ss_users_long_repeated_list_is_consumed_completely(malformed): + tail = "users:(" + ",".join([SS_WORKER_A] * 2000) + ")" + if malformed: + with pytest.raises(ValueError, match=r"^malformed ss process field on line 1$"): + parse_ss_text(f"{SS_SOCKET} {tail} trailing") + else: + refs = parse_ss_text(f"{SS_SOCKET} {tail}")[0].process_refs + assert [(item.name, item.pid) for item in refs] == [("worker-a", 120)]