fix: remediate 65 security and correctness audit findings

Implements the remediation plan for the validated audit findings
(1 Critical, 15 High, 29 Medium, 20 Low) across five phases, each fix
behind a TDD test and gated by `make check` plus a security review.

Phase 0 — shell permission classifier: close the read-only / auto-mode
  bypass cluster (interior &/|& separators, casefolded base commands,
  wrapper value-options, find/xargs/awk payloads, glued output
  redirection, unsafe `git -c`, uv sub-namespaces and `uv run`
  option-prefix bypass).
Phase 1 — confinement, egress, telemetry: symlink-resolve file
  read/write/edit and grep before workspace/sensitive checks; fail-closed
  SSRF with a connection-pinned resolver; bounded shell wait; Sentry
  path/home redaction; invisible-char and case-folded sensitive-file
  handling; sensitive-import gate; untrusted-output wrapping; subagent-id
  path validation.
Phase 2 — tool dispatch, lifecycle, context integrity: MCP-vs-builtin
  tool collisions; run_soul task-leak cleanup; restore-id/path traversal
  guards; compaction rollback; mid-tool-cancel turn balance; cyclic-extend
  detection.
Phase 3 — wire server, auth, web-server: wire read-loop hardening; OAuth
  refresh / device-id / 403 handling; provider base_url validation; replay
  watermark; session-leak cleanup; ZIP-import validation.
Phase 4 — UI/usage/CLI: ANSI sanitization at render boundaries (incl.
  generic tool arg-key names); usage-meter consumed-vs-remaining;
  reset-window loop guard; RunAgents approval fingerprint over child
  prompts; owner-only MCP config; live Typer help; /restore traversal and
  error handling; bounded approval-request store.

Review-found gaps were fixed with regression tests: uv-run option bypass,
grep symlink escape, SSH-key import gate, ANSI arg-key injection, and the
MCP-config / share-dir permission race. Also hardens auth JSON parsing.

Author: elkaix

packages/pythinker-host/src/pythinker_host/__init__.py

@@ -156,6 +156,10 @@ async def chdir(self, path: StrOrHostPath) -> None:
         """Change the current working directory."""
         ...
 
+    async def realpath(self, path: StrOrHostPath) -> HostPath:
+        """Resolve symlinks and return the real absolute path."""
+        ...
+
     async def stat(self, path: StrOrHostPath, *, follow_symlinks: bool = True) -> StatResult:
         """Get the stat result for a path."""
         ...
@@ -282,6 +286,10 @@ async def chdir(path: StrOrHostPath) -> None:
     await get_current_host().chdir(path)
 
 
+async def realpath(path: StrOrHostPath) -> HostPath:
+    return await get_current_host().realpath(path)
+
+
 async def stat(path: StrOrHostPath, *, follow_symlinks: bool = True) -> StatResult:
     return await get_current_host().stat(path, follow_symlinks=follow_symlinks)
 

packages/pythinker-host/src/pythinker_host/local.py

@@ -99,6 +99,12 @@ async def chdir(self, path: StrOrHostPath) -> None:
         local_path = path.unsafe_to_local_path() if isinstance(path, HostPath) else Path(path)
         os.chdir(local_path)
 
+    async def realpath(self, path: StrOrHostPath) -> HostPath:
+        """Resolve symlinks and return the real path (follows symlinks)."""
+        local = path.unsafe_to_local_path() if isinstance(path, HostPath) else Path(path)
+        resolved = await asyncio.to_thread(os.path.realpath, str(local))
+        return HostPath.unsafe_from_local_path(Path(resolved))
+
     async def stat(self, path: StrOrHostPath, *, follow_symlinks: bool = True) -> StatResult:
         local_path = path.unsafe_to_local_path() if isinstance(path, HostPath) else Path(path)
         st = await aiofiles.os.stat(local_path, follow_symlinks=follow_symlinks)
@@ -143,7 +149,7 @@ async def readtext(
         errors: Literal["strict", "ignore", "replace"] = "strict",
     ) -> str:
         local_path = path.unsafe_to_local_path() if isinstance(path, HostPath) else Path(path)
-        async with aiofiles.open(local_path, encoding=encoding, errors=errors) as f:
+        async with aiofiles.open(local_path, encoding=encoding, errors=errors, newline="") as f:
             return await f.read()
 
     async def readlines(
@@ -154,7 +160,7 @@ async def readlines(
         errors: Literal["strict", "ignore", "replace"] = "strict",
     ) -> AsyncGenerator[str]:
         local_path = path.unsafe_to_local_path() if isinstance(path, HostPath) else Path(path)
-        async with aiofiles.open(local_path, encoding=encoding, errors=errors) as f:
+        async with aiofiles.open(local_path, encoding=encoding, errors=errors, newline="") as f:
             async for line in f:
                 yield line
 

packages/pythinker-host/src/pythinker_host/path.py

@@ -118,6 +118,10 @@ def expanduser(self) -> HostPath:
             return home
         return home.joinpath(*parts[1:])
 
+    async def realpath(self) -> HostPath:
+        """Resolve symlinks and return the real absolute path."""
+        return await pythinker_host.realpath(self)
+
     async def stat(self, follow_symlinks: bool = True) -> pythinker_host.StatResult:
         """Return an os.stat_result for the path."""
         return await pythinker_host.stat(self, follow_symlinks=follow_symlinks)

packages/pythinker-host/src/pythinker_host/ssh.py

@@ -179,6 +179,11 @@ async def chdir(self, path: StrOrHostPath) -> None:
         await self._sftp.chdir(str(path))
         self._cwd = await self._sftp.realpath(".")
 
+    async def realpath(self, path: StrOrHostPath) -> HostPath:
+        """Resolve symlinks and return the real path via SFTP realpath."""
+        real = await self._sftp.realpath(str(path))
+        return HostPath(real)
+
     async def stat(
         self,
         path: StrOrHostPath,

packages/pythinker-review/tests/unit/test_security_intel.py

@@ -48,7 +48,8 @@ def test_intel_client_disables_implicit_redirects() -> None:
     from pythinker_review.security_intel.client import IntelHttpClient
 
     client = IntelHttpClient()
-    assert any(isinstance(h, _NoRedirectHandler) for h in client._opener.handlers)
+    handlers = client._opener.handlers  # pyright: ignore[reportAttributeAccessIssue]
+    assert any(isinstance(h, _NoRedirectHandler) for h in handlers)
 
 
 def test_intel_cache_roundtrip(tmp_path: Path) -> None:

src/pythinker_code/__main__.py

@@ -8,60 +8,6 @@
 if TYPE_CHECKING:
     from typing import TextIO
 
-ROOT_HELP = """Usage: pythinker [OPTIONS] COMMAND [ARGS]...
-
-  Pythinker, your next CLI agent.
-
-Options:
-  -h, --help                       Show this message and exit.
-  -V, --version                    Show version and exit.
-  --verbose                        Print verbose information.
-  --debug                          Log debug information.
-  -w, --work-dir DIRECTORY         Working directory for the agent.
-  --add-dir DIRECTORY              Add an additional workspace directory.
-  -S, -r, --session, --resume TEXT Resume a session.
-  -C, --continue                   Continue the previous session.
-  --config TEXT                    Config TOML/JSON string to load.
-  --config-file FILE               Config TOML/JSON file to load.
-  -m, --model TEXT                 LLM model to use.
-  --thinking / --no-thinking       Enable or disable thinking mode.
-  -y, --yolo, --yes, --auto-approve
-                                   Dangerously skip permission approvals.
-  --plan                           Start in plan mode.
-  --auto                           Run in auto mode (no user present).
-  -p, -c, --prompt, --command TEXT User prompt to the agent.
-  --print                          Run in print mode.
-  --acp                            Deprecated; use `pythinker acp`.
-  --wire                           Run as Wire server.
-  --quiet                          Print only the final assistant message.
-  --agent [default|okabe]          Builtin agent specification to use.
-  --agent-file FILE                Custom agent specification file.
-  --mcp-config-file FILE           MCP config file to load; repeatable.
-  --mcp-config TEXT                MCP config JSON to load; repeatable.
-  --skills-dir DIRECTORY           Custom skills directory; repeatable.
-  --no-telemetry                   Disable anonymous telemetry & error reporting.
-
-Commands:
-  acp      Run Pythinker CLI ACP server.
-  term     Run Toad TUI backed by Pythinker CLI ACP server.
-  login    Login with a model provider.
-  logout   Logout from a model provider.
-  info     Show version and protocol information.
-  export   Export session data.
-  mcp      Manage MCP server configurations.
-  plugin   Manage plugins.
-  review          Diff-focused code review (delegates to pythinker-review).
-  secscan         Diff-focused security review (delegates to pythinker-review).
-  security-scan   Repo-wide Pythinker Security Scan pipeline (Python-native).
-  debug           Failure/log root-cause analysis (delegates to pythinker-review).
-  update   Check for and install Pythinker CLI updates.
-  vis      Run Pythinker Agent Tracing Visualizer.
-  web      Run Pythinker CLI web interface.
-
-Documentation:        https://pythoughts-labs.github.io/pythinker-code/
-LLM friendly version: https://pythoughts-labs.github.io/pythinker-code/llms.txt
-"""
-
 
 def _prog_name() -> str:
     return Path(sys.argv[0]).name or "pythinker"
@@ -126,10 +72,6 @@ def main(argv: Sequence[str] | None = None) -> int | str | None:
         print(f"pythinker, version {get_version()} — by {ORGANIZATION}")
         return 0
 
-    if len(args) == 1 and args[0] in {"--help", "-h"}:
-        print(ROOT_HELP, end="")
-        return 0
-
     from pythinker_code.telemetry.crash import install_crash_handlers, set_phase
     from pythinker_code.utils.proxy import normalize_proxy_env
 

src/pythinker_code/acp/host.py

@@ -212,6 +212,9 @@ def getcwd(self) -> HostPath:
     async def chdir(self, path: StrOrHostPath) -> None:
         await self._fallback.chdir(path)
 
+    async def realpath(self, path: StrOrHostPath) -> HostPath:
+        return await self._fallback.realpath(path)
+
     async def stat(self, path: StrOrHostPath, *, follow_symlinks: bool = True) -> StatResult:
         return await self._fallback.stat(path, follow_symlinks=follow_symlinks)
 

src/pythinker_code/agentspec.py

@@ -129,7 +129,13 @@ def load_agent_spec(agent_file: Path) -> ResolvedAgentSpec:
     )
 
 
-def _load_agent_spec(agent_file: Path) -> AgentSpec:
+def _load_agent_spec(agent_file: Path, _visited: set[Path] | None = None) -> AgentSpec:
+    resolved = agent_file.resolve()
+    if _visited is None:
+        _visited = set()
+    if resolved in _visited:
+        raise AgentSpecError(f"Cyclic agent extend chain detected at {agent_file}")
+    _visited.add(resolved)
     if not agent_file.exists():
         raise AgentSpecError(f"Agent spec file not found: {agent_file}")
     if not agent_file.is_file():
@@ -160,7 +166,7 @@ def _load_agent_spec(agent_file: Path) -> AgentSpec:
             base_agent_file = DEFAULT_AGENT_FILE
         else:
             base_agent_file = (agent_file.parent / agent_spec.extend).absolute()
-        base_agent_spec = _load_agent_spec(base_agent_file)
+        base_agent_spec = _load_agent_spec(base_agent_file, _visited)
         if not isinstance(agent_spec.name, Inherit):
             base_agent_spec.name = agent_spec.name
         if not isinstance(agent_spec.system_prompt_path, Inherit):

src/pythinker_code/app.py

@@ -338,20 +338,6 @@ async def create(
         _cleanup_stale_foreground_subagents(runtime)
         _phase_timings_ms["init_ms"] = int((time.monotonic() - _phase_t) * 1000)
 
-        # Refresh plugin configs with fresh credentials (e.g. OAuth tokens)
-        try:
-            from pythinker_code.plugin.manager import (
-                collect_host_values,
-                get_plugins_dir,
-                refresh_plugin_configs,
-            )
-
-            host_values = collect_host_values(config, oauth)
-            if host_values.get("api_key"):
-                refresh_plugin_configs(get_plugins_dir(), host_values)
-        except Exception:
-            logger.debug("Failed to refresh plugin configs, skipping")
-
         if agent_file is None:
             agent_file = DEFAULT_AGENT_FILE
         if startup_progress is not None:

src/pythinker_code/approval_runtime/runtime.py

@@ -23,6 +23,9 @@
     from pythinker_code.wire.types import DisplayBlock
 
 
+_MAX_TERMINAL_RECORDS = 256
+
+
 class ApprovalCancelledError(Exception):
     """Raised when a pending approval is cancelled by its source lifecycle."""
 
@@ -143,6 +146,7 @@ def resolve(self, request_id: str, response: ApprovalResponseKind, feedback: str
             waiter.set_result((response, feedback))
         self._publish_event(ApprovalRuntimeEvent(kind="request_resolved", request=request))
         self._publish_wire_response(request_id, response, feedback)
+        self._evict_terminal_overflow()
         return True
 
     def _cancel_request(self, request_id: str, feedback: str = "") -> None:
@@ -161,6 +165,7 @@ def _cancel_request(self, request_id: str, feedback: str = "") -> None:
             waiter.set_exception(ApprovalCancelledError(request_id))
         self._publish_event(ApprovalRuntimeEvent(kind="request_resolved", request=request))
         self._publish_wire_response(request_id, "reject", feedback)
+        self._evict_terminal_overflow()
 
     def cancel_by_source(self, source_kind: ApprovalSourceKind, source_id: str) -> int:
         cancelled = 0
@@ -180,8 +185,18 @@ def cancel_by_source(self, source_kind: ApprovalSourceKind, source_id: str) -> i
             self._publish_event(ApprovalRuntimeEvent(kind="request_resolved", request=request))
             self._publish_wire_response(request_id, "reject")
             cancelled += 1
+        self._evict_terminal_overflow()
         return cancelled
 
+    def _evict_terminal_overflow(self) -> None:
+        terminal_ids = [rid for rid, r in self._requests.items() if r.status != "pending"]
+        overflow = len(terminal_ids) - _MAX_TERMINAL_RECORDS
+        if overflow <= 0:
+            return
+        # dict preserves insertion order; terminal_ids is already oldest-first
+        for rid in terminal_ids[:overflow]:
+            self._requests.pop(rid, None)
+
     def list_pending(self) -> list[ApprovalRequestRecord]:
         pending = [request for request in self._requests.values() if request.status == "pending"]
         pending.sort(key=lambda request: request.created_at)