fix(process): validate complete saved ss users fields

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/<pid>/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 |

projects/linux-process-observe/src/linux_process_observe/parsers/ss_text.py

@@ -6,6 +6,8 @@
 
 
 PROCESS_RE = re.compile(r'"(?P<name>[^"]+)",pid=(?P<pid>\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)]

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

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)]