🛡️ Sentinel: Fix symlink hijacking in telemetry stream log handlers

.jules/sentinel.md

@@ -32,3 +32,8 @@
 **Vulnerability:** In `archwiz/activity_listener.py`, auto-executed code block scripts written into `SANDBOX` ran `script.chmod(0o755)` without checking whether `script` was a symlink, allowing local symlink hijacking.
 **Learning:** Creating temporary execution scripts in shared or local user directories without verifying `is_symlink()` allows local users to pre-create symlinks pointing to sensitive system files, causing `chmod` or `write_text` to modify permissions on unexpected target files.
 **Prevention:** Always check `script.is_symlink()` before writing or executing temporary scripts, and wrap top-level polling loops in `if __name__ == '__main__':` to allow safe test module imports.
+
+## 2026-09-20 - Symlink Hijacking Prevention in Telemetry Log Stream Handlers
+**Vulnerability:** In `src/telemetry.py` and `termux-multi-agent/src/telemetry.py`, `TermuxTelemetryLogger.notify` opened and appended log entries to `TELEMETRY_LOG` ("agent_telemetry_stream.json") and applied `os.chmod(..., 0o600)` without validating whether `TELEMETRY_LOG` was a symlink, allowing local symlink hijacking. Similarly, `read_latest_telemetry` in `termux-multi-agent/dashboard.py` read telemetry from unvalidated symlink targets.
+**Learning:** Shared telemetry stream log files created in default working directories are vulnerable to symlink pre-creation by unprivileged local processes, which could redirect log appends and permission modifications to target files.
+**Prevention:** Check `os.path.islink(path)` before reading, writing, or adjusting permissions on telemetry stream log files.

src/telemetry.py

@@ -1,5 +1,6 @@
 import json
 import time
+import os
 
 TELEMETRY_LOG = "agent_telemetry_stream.json"
 
@@ -14,5 +15,12 @@ def notify(level, agent_id, message, target_file=None, attempt=None):
         print(f"{timestamp} {color_tag} [{agent_id}]{context_str}: {message}")
         log_entry = {"timestamp": timestamp, "level": level, "agent": agent_id,
                      "target": target_file, "attempt": attempt, "message": message}
+        if os.path.islink(TELEMETRY_LOG):
+            return
         with open(TELEMETRY_LOG, "a") as f:
-            f.write(json.dumps(log_entry) + "\n")
+            f.write(json.dumps(log_entry) + "\n")
+        if not os.path.islink(TELEMETRY_LOG):
+            try:
+                os.chmod(TELEMETRY_LOG, 0o600)
+            except Exception:
+                pass

termux-multi-agent/dashboard.py

@@ -35,8 +35,8 @@ def read_latest_telemetry():
     to perform incremental I/O, yielding massive performance gains on large log streams.
     """
     global _last_file_pos, _active_jobs_cache, _sorted_telemetry_cache, _last_file_ino, _last_file_mtime
-    if not os.path.exists(TELEMETRY_LOG):
-        # Reset cache if file is missing
+    if not os.path.exists(TELEMETRY_LOG) or os.path.islink(TELEMETRY_LOG):
+        # Reset cache if file is missing or is a symlink
         _active_jobs_cache = {}
         _sorted_telemetry_cache = None
         _last_file_pos = 0

termux-multi-agent/src/telemetry.py

@@ -15,9 +15,12 @@ def notify(level, agent_id, message, target_file=None, attempt=None):
         print(f"{timestamp} {color_tag} [{agent_id}]{context_str}: {message}")
         log_entry = {"timestamp": timestamp, "level": level, "agent": agent_id,
                      "target": target_file, "attempt": attempt, "message": message}
+        if os.path.islink(TELEMETRY_LOG):
+            return
         with open(TELEMETRY_LOG, "a") as f:
             f.write(json.dumps(log_entry) + "\n")
-        try:
-            os.chmod(TELEMETRY_LOG, 0o600)
-        except Exception:
-            pass
+        if not os.path.islink(TELEMETRY_LOG):
+            try:
+                os.chmod(TELEMETRY_LOG, 0o600)
+            except Exception:
+                pass

tests/test_telemetry_optimized.py

@@ -143,3 +143,38 @@ def test_dashboard_status_tag_rendering(tmp_path, monkeypatch):
 
     panel = dashboard.make_dashboard()
     assert panel is not None
+
+
+def test_telemetry_symlink_hijacking_prevention(tmp_path, monkeypatch):
+    from src.telemetry import TermuxTelemetryLogger
+
+    # Create target secret file
+    secret_target = tmp_path / "secret.txt"
+    secret_target.write_text("sensitivedata")
+    if os.name != "nt":
+        secret_target.chmod(0o644)
+
+    # Create symlink pointing to target secret file
+    symlink_file = tmp_path / "symlink_agent_telemetry.json"
+    symlink_file.symlink_to(secret_target)
+
+    # Mock TELEMETRY_LOG in both telemetry logger and dashboard
+    monkeypatch.setattr("src.telemetry.TELEMETRY_LOG", str(symlink_file))
+    monkeypatch.setattr(dashboard, "TELEMETRY_LOG", str(symlink_file))
+
+    monkeypatch.setattr(dashboard, "_last_file_pos", 0)
+    monkeypatch.setattr(dashboard, "_active_jobs_cache", {})
+    monkeypatch.setattr(dashboard, "_last_file_ino", None)
+    monkeypatch.setattr(dashboard, "_last_file_mtime", 0)
+
+    # 1. Attempt notify write on symlink
+    TermuxTelemetryLogger.notify("INFO", "AgentX", "Malicious message", target_file="foo.py", attempt=1)
+
+    # Verify target file content and permissions were unchanged
+    assert secret_target.read_text() == "sensitivedata"
+    if os.name != "nt":
+        assert (secret_target.stat().st_mode & 0o777) == 0o644
+
+    # 2. Attempt dashboard read on symlink
+    jobs = dashboard.read_latest_telemetry()
+    assert jobs == []