fix(encoding): read kernel pseudo-files as plain utf-8
utf-8-sig exists to tolerate BOMs that Windows tooling adds to files users edit. /proc and /sys files are generated by the Linux kernel, never BOM'd and absent on Windows, so -sig there only muddies the read/write policy. Switch every literal /proc/ and /sys/ read to utf-8 and teach the footgun read rule that string literals starting with /proc/ or /sys/ are exempt (user-edited files keep utf-8-sig).
This commit is contained in:
@@ -584,7 +584,7 @@ def _read_pid() -> Optional[int]:
|
||||
def _pid_proc_starttime(pid: int) -> Optional[str]:
|
||||
"""/proc/<pid>/stat starttime (field 22) on Linux, else None — cheap PID-recycling detector."""
|
||||
try:
|
||||
text = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8-sig")
|
||||
text = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return None
|
||||
# comm may contain spaces/parens, so split after the LAST ")"; field 22 -> tail index 19.
|
||||
|
||||
@@ -21,7 +21,7 @@ from pathlib import Path
|
||||
def _own_cgroup_path() -> str | None:
|
||||
"""Return the cgroup v2 path for the calling process, or None."""
|
||||
try:
|
||||
text = Path("/proc/self/cgroup").read_text(encoding="utf-8-sig")
|
||||
text = Path("/proc/self/cgroup").read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return None
|
||||
match = re.search(r"^0::(.+)$", text, re.MULTILINE)
|
||||
@@ -30,7 +30,7 @@ def _own_cgroup_path() -> str | None:
|
||||
|
||||
def _read_cgroup_pids(cgroup_path: str) -> list[int]:
|
||||
try:
|
||||
raw = Path(f"/sys/fs/cgroup{cgroup_path}/cgroup.procs").read_text(encoding="utf-8-sig")
|
||||
raw = Path(f"/sys/fs/cgroup{cgroup_path}/cgroup.procs").read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return []
|
||||
pids: list[int] = []
|
||||
|
||||
@@ -48,11 +48,11 @@ def current_instantiation_epoch() -> str:
|
||||
"""
|
||||
boot_id = pid1_start = ""
|
||||
with contextlib.suppress(OSError):
|
||||
boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="utf-8-sig").strip()
|
||||
boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="utf-8").strip()
|
||||
with contextlib.suppress(OSError, IndexError):
|
||||
# "<pid> (<comm>) <state> ...": comm may contain spaces/parens, so split on the
|
||||
# LAST ')'. starttime is field 22 (1-indexed) = tail index 19.
|
||||
pid1_start = Path("/proc/1/stat").read_text(encoding="utf-8-sig").rsplit(")", 1)[1].split()[19]
|
||||
pid1_start = Path("/proc/1/stat").read_text(encoding="utf-8").rsplit(")", 1)[1].split()[19]
|
||||
return f"{boot_id}:{pid1_start}" if (boot_id or pid1_start) else ""
|
||||
|
||||
|
||||
|
||||
@@ -41,7 +41,7 @@ def _signal_name(sig: Any) -> str:
|
||||
|
||||
def _read_proc_field(pid: int, key: str) -> Optional[str]:
|
||||
"""Read a single field from /proc/<pid>/status. Linux only; None elsewhere."""
|
||||
with contextlib.suppress(OSError), open(f"/proc/{pid}/status", encoding="utf-8-sig") as fh:
|
||||
with contextlib.suppress(OSError), open(f"/proc/{pid}/status", encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
if line.startswith(key + ":"):
|
||||
return line.split(":", 1)[1].strip()
|
||||
@@ -210,7 +210,7 @@ def check_systemd_timing_alignment(
|
||||
return None # Not running under systemd (or at least not directly)
|
||||
# /proc/self/cgroup: "0::/user.slice/.../hermes-gateway.service"
|
||||
unit_name: Optional[str] = None
|
||||
with contextlib.suppress(OSError), open("/proc/self/cgroup", encoding="utf-8-sig") as fh:
|
||||
with contextlib.suppress(OSError), open("/proc/self/cgroup", encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
parts = reversed(line.strip().split("/"))
|
||||
unit_name = next((p for p in parts if p.endswith(".service")), None)
|
||||
|
||||
@@ -896,7 +896,7 @@ def _pid_exists(pid: int) -> bool:
|
||||
def _posix_is_zombie(pid: int) -> bool:
|
||||
"""Zombie via ``/proc/<pid>/stat`` field 3, or ``ps -o state=`` without /proc (macOS/BSD)."""
|
||||
try:
|
||||
stat_fields = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8-sig").split()
|
||||
stat_fields = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8").split()
|
||||
return len(stat_fields) > 2 and stat_fields[2] == "Z"
|
||||
except FileNotFoundError:
|
||||
with contextlib.suppress(Exception):
|
||||
@@ -1515,7 +1515,7 @@ def _scoped_lock_record_is_stale(existing: dict[str, Any], existing_pid: Optiona
|
||||
def _process_is_stopped(pid: int) -> bool:
|
||||
"""True for a stopped / tracing-stop state (T/t) in ``/proc/<pid>/status``."""
|
||||
with contextlib.suppress(OSError):
|
||||
for line in Path(f"/proc/{pid}/status").read_text(encoding="utf-8-sig").splitlines():
|
||||
for line in Path(f"/proc/{pid}/status").read_text(encoding="utf-8").splitlines():
|
||||
if line.startswith("State:"):
|
||||
return line.split()[1] in {"T", "t"}
|
||||
return False
|
||||
|
||||
@@ -328,7 +328,7 @@ def _pid_alive(pid: Optional[int]) -> bool:
|
||||
return False
|
||||
if sys.platform == "linux":
|
||||
try:
|
||||
with open(f"/proc/{int(pid)}/status", "r", encoding="utf-8-sig") as f:
|
||||
with open(f"/proc/{int(pid)}/status", "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
if line.startswith("State:"):
|
||||
# "State:\tZ (zombie)" → dead
|
||||
|
||||
@@ -67,7 +67,7 @@ _LINUX_RAM_KEYS = frozenset({"MemTotal", "MemAvailable", "MemFree"})
|
||||
def _linux_meminfo_text() -> str | None:
|
||||
"""Raw /proc/meminfo, or None when procfs cannot be read."""
|
||||
try:
|
||||
return Path("/proc/meminfo").read_text(encoding="utf-8-sig")
|
||||
return Path("/proc/meminfo").read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
@@ -979,7 +979,7 @@ def _desktop_linux_needs_no_sandbox() -> bool:
|
||||
if hasattr(os, "geteuid") and os.geteuid() == 0:
|
||||
return False
|
||||
try:
|
||||
with open("/proc/sys/kernel/apparmor_restrict_unprivileged_userns", encoding="utf-8-sig") as f:
|
||||
with open("/proc/sys/kernel/apparmor_restrict_unprivileged_userns", encoding="utf-8") as f:
|
||||
return f.read().strip() == "1"
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
@@ -77,7 +77,7 @@ def _read_proc_status() -> str | None:
|
||||
if sys.platform != "linux":
|
||||
return None
|
||||
try:
|
||||
return Path("/proc/self/status").read_text(encoding="utf-8-sig")
|
||||
return Path("/proc/self/status").read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
@@ -74,7 +74,7 @@ def _in_container() -> bool:
|
||||
if os.environ.get("HERMES_DESKTOP_CHILD_PID"):
|
||||
return False # desktop child, not a server container
|
||||
try:
|
||||
cgroup = Path("/proc/1/cgroup").read_text(encoding="utf-8-sig", errors="replace")
|
||||
cgroup = Path("/proc/1/cgroup").read_text(encoding="utf-8", errors="replace")
|
||||
except Exception:
|
||||
return False
|
||||
return any(tok in cgroup for tok in ("docker", "containerd", "kubepods", "libpod"))
|
||||
@@ -89,7 +89,7 @@ def _path_is_mounted(path: Path) -> bool:
|
||||
except Exception:
|
||||
target = path
|
||||
try:
|
||||
mounts = Path("/proc/mounts").read_text(encoding="utf-8-sig", errors="replace").splitlines()
|
||||
mounts = Path("/proc/mounts").read_text(encoding="utf-8", errors="replace").splitlines()
|
||||
except Exception:
|
||||
return True # can't tell — fail safe (no warning)
|
||||
# (mountpoint, fstype) entries at or above target; the longest wins, first one on ties.
|
||||
|
||||
@@ -28,7 +28,7 @@ def _process_start_marker(pid: int) -> str:
|
||||
"""
|
||||
if sys.platform == "linux":
|
||||
try:
|
||||
stat_line = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8-sig")
|
||||
stat_line = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
|
||||
except FileNotFoundError as exc:
|
||||
raise ProcessLookupError(pid) from exc
|
||||
|
||||
|
||||
@@ -464,7 +464,10 @@ FOOTGUNS: list[Footgun] = [
|
||||
"path.read_text(encoding='utf-8-sig') / "
|
||||
"open(path, 'r', encoding='utf-8-sig')"
|
||||
),
|
||||
post_filter=lambda m, line: _is_read_shaped(line),
|
||||
# Literal /proc/ and /sys/ paths are kernel pseudo-files: Linux
|
||||
# generates them, no Windows tool can BOM them, and they do not
|
||||
# exist on Windows at all, so plain utf-8 is the honest encoding.
|
||||
post_filter=lambda m, line: _is_read_shaped(line) and not _KERNEL_PSEUDO_FILE.search(line),
|
||||
),
|
||||
Footgun(
|
||||
name="write with encoding='utf-8-sig' (emits a BOM)",
|
||||
@@ -696,6 +699,10 @@ def _extract_mode(line: str) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
# A string literal (plain or f-string) that starts with /proc/ or /sys/.
|
||||
_KERNEL_PSEUDO_FILE = re.compile(r"""['"]/(?:proc|sys)/""")
|
||||
|
||||
|
||||
def _is_read_shaped(line: str) -> bool:
|
||||
"""Heuristic: does this line READ a file (so utf-8-sig applies)?
|
||||
|
||||
|
||||
@@ -28,6 +28,9 @@ def linter():
|
||||
('data = path.read_text(encoding="utf_8")', [(1, READ)]),
|
||||
("with open(path, mode='r', encoding='utf-8') as f:", [(1, READ)]),
|
||||
('path.read_text(encoding="utf-8-sig")', []),
|
||||
('Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")', []),
|
||||
("with open('/sys/fs/cgroup/cgroup.procs', encoding='utf-8') as f:", []),
|
||||
('Path(proc_root, "stat").read_text(encoding="utf-8") # /proc/ in prose only', [(1, READ)]),
|
||||
('path.write_text(data, encoding="utf-8")', []),
|
||||
("open(path, 'w', encoding='utf-8')", []),
|
||||
("open(path, 'a', encoding='utf-8')", []),
|
||||
|
||||
@@ -89,7 +89,7 @@ def test_timeout_kills_the_package_managers_whole_process_group(monkeypatch):
|
||||
|
||||
def gone() -> bool: # /proc-based: a reparented orphan sits outside our subtree, where os.kill(pid, 0) is guarded
|
||||
try:
|
||||
return "Z" in (Path(f"/proc/{child}/stat").read_text(encoding="utf-8-sig").rsplit(")", 1)[1].split() or ["Z"])[0]
|
||||
return "Z" in (Path(f"/proc/{child}/stat").read_text(encoding="utf-8").rsplit(")", 1)[1].split() or ["Z"])[0]
|
||||
except OSError:
|
||||
return True
|
||||
|
||||
|
||||
@@ -148,7 +148,7 @@ def _launched_by_session(chromium_pid: int) -> Optional[str]:
|
||||
for a human-started (dock) instance. Chromium itself gets a scrubbed environment, so the daemon's
|
||||
``/proc/<ppid>/environ`` is the marker (Linux-only, same user)."""
|
||||
try:
|
||||
with open(f"/proc/{chromium_pid}/status", encoding="utf-8-sig") as fh:
|
||||
with open(f"/proc/{chromium_pid}/status", encoding="utf-8") as fh:
|
||||
ppid = next((int(line.split()[1]) for line in fh if line.startswith("PPid:")), 0)
|
||||
with open(f"/proc/{ppid}/environ", "rb") as fh:
|
||||
raw = fh.read()
|
||||
|
||||
@@ -133,7 +133,7 @@ def _running_in_docker() -> bool:
|
||||
if os.path.exists("/.dockerenv"):
|
||||
return True
|
||||
try:
|
||||
with open("/proc/1/cgroup", "rt", encoding="utf-8-sig") as fp:
|
||||
with open("/proc/1/cgroup", "rt", encoding="utf-8") as fp:
|
||||
return "docker" in fp.read()
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
@@ -41,7 +41,7 @@ def apparmor_restricts_unprivileged_userns() -> bool:
|
||||
"""Ubuntu 23.10+ default: unprivileged user namespaces are denied, so a Chromium whose
|
||||
``chrome_sandbox`` helper is not setuid (Playwright's bundle) dies with 'No usable sandbox'."""
|
||||
try:
|
||||
with open("/proc/sys/kernel/apparmor_restrict_unprivileged_userns", encoding="utf-8-sig") as f:
|
||||
with open("/proc/sys/kernel/apparmor_restrict_unprivileged_userns", encoding="utf-8") as f:
|
||||
return f.read().strip() == "1"
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
@@ -135,7 +135,7 @@ def _worker_memory_max_bytes() -> int:
|
||||
override, _MIN_WORKER_MEMORY_MAX_BYTES // (1024 * 1024))
|
||||
candidates: List[int] = []
|
||||
try:
|
||||
for line in Path("/proc/self/cgroup").read_text(encoding="utf-8-sig").splitlines():
|
||||
for line in Path("/proc/self/cgroup").read_text(encoding="utf-8").splitlines():
|
||||
if line.startswith("0::"):
|
||||
relative = line.partition("::")[2].lstrip("/")
|
||||
raw_limit = (
|
||||
|
||||
Reference in New Issue
Block a user