fix(process): validate saved process/socket attribution consistency

notes/process-evidence-schema.md

@@ -76,6 +76,27 @@ Saved `ss -H -a -n -t -u -p` output can connect a TCP or UDP socket to a PID whe
 
 Missing process context is not converted into false certainty. The socket remains in the artifact with `linked=false`.
 
+`process_name` on a socket link is the name reported by `ss`, including when it
+differs from the PID-associated procfs process's `name`. This preserves both
+observations. A name disagreement can follow legitimate changes and does not prove
+PID reuse or misconduct; agreement cannot rule out reuse between saved captures.
+
+`linked=true` describes an unambiguous PID association in the supplied records.
+It does not independently verify that `ss` observed the procfs execution instance.
+The link's `process_id`, executable, UID and GID are procfs-derived context. Duplicate
+PID candidates fail instead of selecting the first or last record. During `diff`,
+process IDs must agree with their declared host/PID/start fields, and linked records
+must reference the paired process record with matching PID, executable, UID and GID.
+An unlinked record must not claim that procfs identity or context. Contradictory raw
+artifacts remain available; a failed comparison produces no downstream diff.
+
+The telemetry adapter checks change-identity self-consistency. Its diff input does
+not contain the complete paired procfs artifacts, so it cannot repeat cohort
+validation or prove instance ownership. Names are contextual and excluded from the
+existing link identity: a name-only difference does not create a link-change event.
+Rejecting name mismatches as an identity test would both lose legitimate observations
+and still miss PID/name reuse. No confidence or verification field is introduced.
+
 `wildcard` means the socket was bound to an any-address value such as `0.0.0.0` or `::`. It does not prove external reachability because firewall, namespace, routing, and host policy are outside this snapshot.
 
 ## Diff interpretation

projects/linux-process-observe/README.md

@@ -95,7 +95,10 @@ The JSONL can be supplied as `input_path` to telemetry-lab's existing `run windo
 - `parent_process_id` links to the same snapshot identity when the parent PID is present in the saved procfs export; otherwise it is `null`.
 - UID/GID fields come from the saved procfs `status` record. Parent PID and start time come from `stat`.
 - The executable path is context, not proof of the executable file's contents or integrity.
-- Socket links use PID context reported by `ss`. A socket is retained with `linked=false` when no process context is available.
+- Socket links use PID context reported by `ss`. `linked=true` means an unambiguous PID association within the supplied records, not independent verification of the socket's execution instance. A socket is retained with `linked=false` when no matching procfs context is available.
+- A link's `process_name` preserves the name reported by `ss`; the selected process's name remains in `process_snapshot.json`. A disagreement is retained as context, without proving PID reuse or malicious behavior. Names can legitimately change, and matching names cannot exclude PID reuse between source captures.
+- Link `process_id`, executable, UID and GID come from the PID-associated procfs record. Duplicate PID candidates are rejected rather than reconciled by input order. `diff` rejects process IDs inconsistent with their host/PID/start fields, dangling linked identities, and links whose procfs-derived fields disagree with the paired process artifact, even when timestamps match.
+- The standalone telemetry adapter validates identity self-consistency; it does not have the paired procfs artifact and cannot independently revalidate a socket's process attribution. Link-name-only changes do not create new events: the link diff remains added/removed by the existing identity and endpoint key.
 - `exposure=loopback`, `wildcard`, or `specific` describes the local bind address; it is not a firewall or reachability verdict.
 
 ## Validation Status

projects/linux-process-observe/src/linux_process_observe/adapters.py

@@ -4,7 +4,7 @@
 from typing import Any
 import json
 
-from .models import EVIDENCE_SCHEMA, EvidenceEnvelope
+from .models import EVIDENCE_SCHEMA, EvidenceEnvelope, socket_link_identity, validate_process_record_identity
 
 
 _EVENT_TYPES = {
@@ -90,10 +90,12 @@ def _map_record(
     if event_prefix == "process":
         for process_record in (before, after):
             if process_record is not None:
-                _validate_process_identity(process_record, identity, index)
+                _validate_process_identity(process_record, identity, index, diff.host_id)
         source = _required_string(selected, "process_id", index)
         target = _required_string(selected, "executable", index)
     else:
+        if socket_link_identity(selected) != identity:
+            raise ValueError(f"record {index} socket identity must match observation")
         source = _socket_source(diff.host_id, selected, index)
         target = _socket_target(selected, index)
 
@@ -166,11 +168,12 @@ def _change_records(
 
 
 def _validate_process_identity(
-    record: dict[str, Any], identity: str, index: int
+    record: dict[str, Any], identity: str, index: int, host_id: str
 ) -> None:
     process_id = _required_string(record, "process_id", index)
     if process_id != identity:
         raise ValueError(f"record {index} process_id must match identity")
+    validate_process_record_identity(record, host_id)
 
 
 def _socket_source(host_id: str, record: dict[str, Any], index: int) -> str:

projects/linux-process-observe/src/linux_process_observe/diff.py

@@ -3,7 +3,10 @@
 from datetime import datetime
 from typing import Any
 
-from .models import EVIDENCE_SCHEMA, EvidenceEnvelope, parse_observed_at
+from .models import (
+    EVIDENCE_SCHEMA, EvidenceEnvelope, parse_observed_at,
+    socket_link_identity, validate_process_record_identity,
+)
 
 
 def build_diff_envelope(
@@ -15,10 +18,10 @@ def build_diff_envelope(
 ) -> EvidenceEnvelope:
     _validate_artifacts(before_processes, after_processes, before_links, after_links)
 
-    before_process_map = _records_by_key(before_processes, record_type="process", key="process_id")
-    after_process_map = _records_by_key(after_processes, record_type="process", key="process_id")
-    before_link_map = _links_by_identity(before_links)
-    after_link_map = _links_by_identity(after_links)
+    before_process_map = _processes_by_identity(before_processes)
+    after_process_map = _processes_by_identity(after_processes)
+    before_link_map = _links_by_identity(before_links, before_process_map)
+    after_link_map = _links_by_identity(after_links, after_process_map)
 
     changes: list[dict[str, Any]] = []
     for identity in sorted(after_process_map.keys() - before_process_map.keys()):
@@ -88,37 +91,55 @@ def _records(envelope: EvidenceEnvelope, record_type: str) -> list[dict[str, Any
     return records
 
 
-def _records_by_key(envelope: EvidenceEnvelope, *, record_type: str, key: str) -> dict[str, dict[str, Any]]:
+def _processes_by_identity(envelope: EvidenceEnvelope) -> dict[str, dict[str, Any]]:
     records: dict[str, dict[str, Any]] = {}
-    for record in _records(envelope, record_type):
-        value = record.get(key)
+    pids: set[int] = set()
+    for record in _records(envelope, "process"):
+        value = record.get("process_id")
         if not isinstance(value, str) or not value:
-            raise ValueError(f"{record_type} record missing {key}")
+            raise ValueError("process record missing process_id")
         if value in records:
-            raise ValueError(f"duplicate {record_type} {key}: {value}")
+            raise ValueError(f"duplicate process process_id: {value}")
+        validate_process_record_identity(record, envelope.host_id)
+        if record["pid"] in pids:
+            raise ValueError(f"duplicate process pid: {record['pid']}")
+        pids.add(record["pid"])
         records[value] = record
     return records
 
 
-def _link_identity(record: dict[str, Any]) -> str:
-    process = record.get("process_id") or f"pid={record.get('pid', 'unknown')}"
-    return (
-        f"{process}|{record.get('protocol')}|{record.get('state')}|"
-        f"{record.get('local_address')}:{record.get('local_port')}|"
-        f"{record.get('remote_address')}:{record.get('remote_port')}"
-    )
-
-
-def _links_by_identity(envelope: EvidenceEnvelope) -> dict[str, dict[str, Any]]:
+def _links_by_identity(
+    envelope: EvidenceEnvelope, processes: dict[str, dict[str, Any]]
+) -> dict[str, dict[str, Any]]:
     links: dict[str, dict[str, Any]] = {}
     for record in _records(envelope, "process_socket_link"):
-        identity = _link_identity(record)
+        _validate_link_context(record, processes)
+        identity = socket_link_identity(record)
         if identity in links:
             raise ValueError(f"duplicate process_socket_link identity: {identity}")
         links[identity] = record
     return links
 
 
+def _validate_link_context(record: dict[str, Any], processes: dict[str, dict[str, Any]]) -> None:
+    if record.get("linked") is False:
+        if any(record.get(field) is not None for field in ("process_id", "executable", "uid", "gid")):
+            raise ValueError("unlinked socket must not claim procfs identity or context")
+        return
+    if record.get("linked") is not True:
+        raise ValueError("socket linked must be boolean")
+    process_id = record.get("process_id")
+    process = processes.get(process_id) if isinstance(process_id, str) else None
+    if process is None:
+        raise ValueError("linked socket process_id must reference the paired process artifact")
+    for field in ("pid", "executable", "uid", "gid"):
+        if (
+            field not in process or field not in record
+            or type(record[field]) is not type(process[field]) or record[field] != process[field]
+        ):
+            raise ValueError(f"linked socket {field} must match procfs process")
+
+
 def _field_changes(before: dict[str, Any], after: dict[str, Any]) -> dict[str, dict[str, Any]]:
     immutable = {"record_type", "process_id", "pid", "start_time_ticks"}
     changes: dict[str, dict[str, Any]] = {}

projects/linux-process-observe/src/linux_process_observe/models.py

@@ -11,6 +11,25 @@
 _FRACTIONAL_COMPONENT = re.compile(r"[.,](\d+)")
 
 
+def validate_process_record_identity(record: Mapping[str, Any], host_id: str) -> None:
+    """Check declared fields, without asserting independent instance verification."""
+    pid = record.get("pid")
+    ticks = record.get("start_time_ticks")
+    if type(pid) is not int or type(ticks) is not int:
+        raise ValueError("process identity requires integer pid and start_time_ticks")
+    if record.get("process_id") != f"{host_id}:{pid}:{ticks}":
+        raise ValueError("process_id must match host_id, pid and start_time_ticks")
+
+
+def socket_link_identity(record: Mapping[str, Any]) -> str:
+    process = record.get("process_id") or f"pid={record.get('pid', 'unknown')}"
+    return (
+        f"{process}|{record.get('protocol')}|{record.get('state')}|"
+        f"{record.get('local_address')}:{record.get('local_port')}|"
+        f"{record.get('remote_address')}:{record.get('remote_port')}"
+    )
+
+
 def parse_observed_at(value: str) -> datetime:
     if not isinstance(value, str) or not value.strip():
         raise ValueError("observed_at must be a non-empty string")

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

@@ -24,7 +24,11 @@ def parse_procfs_root(root: str | Path, *, host_id: str) -> list[ProcessRecord]:
             key=lambda item: int(item.name),
         )
     ]
-    processes_by_pid = {item.pid: item for item in records}
+    processes_by_pid: dict[int, ProcessRecord] = {}
+    for item in records:
+        if item.pid in processes_by_pid:
+            raise ValueError(f"duplicate procfs pid: {item.pid}")
+        processes_by_pid[item.pid] = item
     return [
         replace(
             item,

projects/linux-process-observe/src/linux_process_observe/snapshot.py

@@ -73,7 +73,11 @@ def build_snapshot_artifacts(
 def link_processes_to_sockets(
     processes: list[ProcessRecord], observations: list[SocketObservation]
 ) -> list[ProcessSocketLink]:
-    processes_by_pid = {item.pid: item for item in processes}
+    processes_by_pid: dict[int, ProcessRecord] = {}
+    for item in processes:
+        if item.pid in processes_by_pid:
+            raise ValueError(f"duplicate procfs pid: {item.pid}")
+        processes_by_pid[item.pid] = item
     links: list[ProcessSocketLink] = []
 
     for socket in observations:
@@ -86,7 +90,7 @@ def link_processes_to_sockets(
                     linked=process is not None,
                     process_id=process.process_id if process else None,
                     pid=process.pid if process else (ref.pid if ref else None),
-                    process_name=process.name if process else (ref.name if ref else None),
+                    process_name=ref.name if ref else None,
                     executable=process.executable if process else None,
                     uid=process.uid if process else None,
                     gid=process.gid if process else None,