fix: connect tty context and secure archive extraction

Author: QUSETIONS

minicode/main.py

@@ -376,6 +376,7 @@ def main() -> None:
             resume_session=args.resume,
             list_sessions_only=args.list_sessions,
             memory_manager=memory_mgr,
+            context_manager=context_mgr,
         )
     except KeyboardInterrupt:
         print("\n\nInterrupted by user. Shutting down gracefully...")

minicode/tools/archive_utils.py

@@ -8,6 +8,55 @@
 from pathlib import Path
 
 from minicode.tooling import ToolDefinition, ToolContext, ToolResult
+from minicode.workspace import resolve_tool_path
+
+
+def _resolve_archive_member(destination: Path, member_name: str) -> Path:
+    normalized_name = member_name.replace("\\", "/")
+    member_path = Path(normalized_name)
+    if member_path.is_absolute() or any(part == ".." for part in member_path.parts):
+        raise ValueError(f"Archive member escapes extraction destination: {member_name}")
+
+    destination_root = destination.resolve()
+    target = (destination_root / member_path).resolve()
+    try:
+        target.relative_to(destination_root)
+    except ValueError as error:
+        raise ValueError(f"Archive member escapes extraction destination: {member_name}") from error
+    return target
+
+
+def _safe_extract_zip(source: Path, destination: Path) -> None:
+    destination.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(source, "r") as zf:
+        for info in zf.infolist():
+            target = _resolve_archive_member(destination, info.filename)
+            if info.is_dir():
+                target.mkdir(parents=True, exist_ok=True)
+                continue
+            target.parent.mkdir(parents=True, exist_ok=True)
+            with zf.open(info, "r") as src, open(target, "wb") as dst:
+                shutil.copyfileobj(src, dst)
+
+
+def _safe_extract_tar(source: Path, destination: Path) -> None:
+    destination.mkdir(parents=True, exist_ok=True)
+    with tarfile.open(source, "r:*") as tar:
+        for member in tar.getmembers():
+            if member.issym() or member.islnk():
+                raise ValueError(f"Archive member uses unsupported link: {member.name}")
+            target = _resolve_archive_member(destination, member.name)
+            if member.isdir():
+                target.mkdir(parents=True, exist_ok=True)
+                continue
+            if not member.isfile():
+                continue
+            target.parent.mkdir(parents=True, exist_ok=True)
+            src = tar.extractfile(member)
+            if src is None:
+                continue
+            with src, open(target, "wb") as dst:
+                shutil.copyfileobj(src, dst)
 
 
 # ---------------------------------------------------------------------------
@@ -185,23 +234,20 @@ def _validate_tar_extract(input_data: dict) -> dict:
 
 
 def _run_tar_extract(input_data: dict, context: ToolContext) -> ToolResult:
-    source = Path(context.cwd) / input_data["source"]
+    source = resolve_tool_path(context, input_data["source"], "read")
     dest_dir = input_data.get("destination", "")
 
     if not source.exists():
         return ToolResult(ok=False, output=f"Source not found: {input_data['source']}")
 
     try:
         if dest_dir:
-            destination = Path(context.cwd) / dest_dir
+            destination = resolve_tool_path(context, dest_dir, "write")
         else:
             # Extract to same directory as archive
             destination = source.parent / source.stem
 
-        destination.mkdir(parents=True, exist_ok=True)
-        
-        with tarfile.open(source, "r:*") as tar:
-            tar.extractall(destination)
+        _safe_extract_tar(source, destination)
 
         return ToolResult(ok=True, output=f"Extracted to {destination}")
     except Exception as e:
@@ -291,22 +337,19 @@ def _validate_zip_extract(input_data: dict) -> dict:
 
 
 def _run_zip_extract(input_data: dict, context: ToolContext) -> ToolResult:
-    source = Path(context.cwd) / input_data["source"]
+    source = resolve_tool_path(context, input_data["source"], "read")
     dest_dir = input_data.get("destination", "")
 
     if not source.exists():
         return ToolResult(ok=False, output=f"Source not found: {input_data['source']}")
 
     try:
         if dest_dir:
-            destination = Path(context.cwd) / dest_dir
+            destination = resolve_tool_path(context, dest_dir, "write")
         else:
             destination = source.parent / source.stem
 
-        destination.mkdir(parents=True, exist_ok=True)
-        
-        with zipfile.ZipFile(source, "r") as zf:
-            zf.extractall(destination)
+        _safe_extract_zip(source, destination)
 
         return ToolResult(ok=True, output=f"Extracted to {destination}")
     except Exception as e:
@@ -326,4 +369,4 @@ def _run_zip_extract(input_data: dict, context: ToolContext) -> ToolResult:
     },
     validator=_validate_zip_extract,
     run=_run_zip_extract,
-)
+)

minicode/tty_app.py

@@ -62,6 +62,7 @@ def run_tty_app(
     resume_session: str | None = None,
     list_sessions_only: bool = False,
     memory_manager: Any | None = None,
+    context_manager: Any | None = None,
 ) -> list[ChatMessage]:
     """Event-driven full-screen TTY application, ported from the TypeScript version.
     
@@ -74,7 +75,17 @@ def run_tty_app(
         return messages
 
     session = load_or_create_session(cwd, resume_session)
-    args, state = build_tty_runtime_state(runtime, tools, model, messages, cwd, permissions, session, memory_manager)
+    args, state = build_tty_runtime_state(
+        runtime,
+        tools,
+        model,
+        messages,
+        cwd,
+        permissions,
+        session,
+        memory_manager,
+        context_manager,
+    )
 
     # Throttled renderer: coalesces rapid rerender() calls to reduce flickering
     throttled = _ThrottledRenderer(lambda: _render_screen(args, state), min_interval=0.016)

minicode/tui/input_handler.py

@@ -3,12 +3,14 @@
 import logging
 import os
 import sys
+import threading
 import time
 from typing import Any, Callable
 from minicode.tui.input_parser import KeyEvent, ParsedInputEvent, TextEvent, WheelEvent, parse_input_chunk
 from minicode.tui.state import ScreenState, TtyAppArgs
 from minicode.cli_commands import try_handle_local_command, find_matching_slash_commands
 from minicode.agent_loop import run_agent_turn
+from minicode.context_manager import save_context_state
 from minicode.history import save_history_entries
 from minicode.local_tool_shortcuts import parse_local_tool_shortcut
 from minicode.prompt import build_system_prompt
@@ -579,8 +581,13 @@ def _run_agent_background():
                 on_assistant_message=on_assistant_message,
                 on_progress_message=on_progress_message,
                 on_assistant_stream_chunk=on_assistant_stream_chunk,
+                store=state.app_state,
+                context_manager=args.context_manager,
                 runtime=args.runtime,
             )
+            if args.context_manager is not None:
+                args.context_manager.messages = next_messages
+                save_context_state(args.context_manager)
             with agent_thread_lock:
                 agent_result["messages"] = next_messages
         except Exception as e:

minicode/tui/session_flow.py

@@ -67,6 +67,7 @@ def build_tty_runtime_state(
     permissions: PermissionManager,
     session: SessionData,
     memory_manager: Any | None = None,
+    context_manager: Any | None = None,
 ) -> tuple[TtyAppArgs, ScreenState]:
     args = TtyAppArgs(
         runtime=runtime,
@@ -76,6 +77,7 @@ def build_tty_runtime_state(
         cwd=cwd,
         permissions=permissions,
         memory_manager=memory_manager,
+        context_manager=context_manager,
     )
 
     state = ScreenState(

minicode/tui/state.py

@@ -21,6 +21,7 @@ class TtyAppArgs:
     cwd: str
     permissions: PermissionManager
     memory_manager: Any | None = None
+    context_manager: Any | None = None
 
 
 @dataclass

tests/test_tools.py

@@ -1,12 +1,16 @@
 from pathlib import Path
+import io
 import sys
+import tarfile
+import zipfile
 
 import pytest
 
 import minicode.tools.run_command as run_command_module
 from minicode.permissions import PermissionManager
 from minicode.tools.run_command import _build_execution_command, split_command_line
 from minicode.tools.patch_file import patch_file_tool
+from minicode.tools.archive_utils import tar_extract_tool, zip_extract_tool
 from minicode.tools.run_command import run_command_tool
 from minicode.tools.write_file import write_file_tool
 from minicode.tooling import ToolContext
@@ -141,6 +145,39 @@ def test_full_tool_registry_can_opt_into_utility_wrappers(tmp_path: Path) -> Non
     assert "csv_parse" in names
 
 
+def test_zip_extract_rejects_entries_that_escape_destination(tmp_path: Path) -> None:
+    archive = tmp_path / "evil.zip"
+    with zipfile.ZipFile(archive, "w") as zf:
+        zf.writestr("../escape.txt", "owned")
+
+    result = zip_extract_tool.run(
+        {"source": "evil.zip", "destination": "out"},
+        ToolContext(cwd=str(tmp_path), permissions=None),
+    )
+
+    assert result.ok is False
+    assert "escapes extraction destination" in result.output
+    assert not (tmp_path / "escape.txt").exists()
+
+
+def test_tar_extract_rejects_entries_that_escape_destination(tmp_path: Path) -> None:
+    archive = tmp_path / "evil.tar"
+    payload = b"owned"
+    info = tarfile.TarInfo("../escape.txt")
+    info.size = len(payload)
+    with tarfile.open(archive, "w") as tf:
+        tf.addfile(info, io.BytesIO(payload))
+
+    result = tar_extract_tool.run(
+        {"source": "evil.tar", "destination": "out"},
+        ToolContext(cwd=str(tmp_path), permissions=None),
+    )
+
+    assert result.ok is False
+    assert "escapes extraction destination" in result.output
+    assert not (tmp_path / "escape.txt").exists()
+
+
 def test_core_tool_registry_does_not_import_utility_modules(tmp_path: Path) -> None:
     utility_modules = [
         "minicode.tools.archive_utils",

tests/test_tty_app.py

@@ -7,7 +7,10 @@
     summarize_tool_input,
     summarize_tool_output,
 )
+import minicode.tui.input_handler as input_handler_module
+from minicode.context_manager import ContextManager
 from minicode.permissions import PermissionManager
+from minicode.tooling import ToolRegistry
 from minicode.tui.runtime_control import _ThrottledRenderer as RuntimeThrottledRenderer
 from minicode.tui.event_flow import _handle_event
 from minicode.tui.input_parser import KeyEvent
@@ -147,3 +150,36 @@ def handle_input(*_args, **_kwargs):
 
     assert "handle_input" not in calls
     assert state.input == ""
+
+
+def test_tty_input_passes_and_persists_context_manager(tmp_path, monkeypatch) -> None:
+    captured: dict = {}
+    saved: list[ContextManager] = []
+    context_manager = ContextManager(model="default", context_window=1000)
+
+    def fake_run_agent_turn(**kwargs):
+        captured.update(kwargs)
+        manager = kwargs["context_manager"]
+        manager.messages = list(kwargs["messages"])
+        return [*kwargs["messages"], {"role": "assistant", "content": "done"}]
+
+    monkeypatch.setattr(input_handler_module, "run_agent_turn", fake_run_agent_turn)
+    monkeypatch.setattr(input_handler_module, "save_context_state", saved.append, raising=False)
+
+    state = ScreenState(input="Please inspect context", cursor_offset=22)
+    args = TtyAppArgs(
+        runtime={"model": "default"},
+        tools=ToolRegistry([]),
+        model=object(),
+        messages=[{"role": "system", "content": "sys"}],
+        cwd=str(tmp_path),
+        permissions=PermissionManager(str(tmp_path)),
+        context_manager=context_manager,
+    )
+
+    assert input_handler_module._handle_input(args, state, lambda: None) is False
+    state.agent_thread.join(timeout=5)
+
+    assert captured["context_manager"] is context_manager
+    assert saved == [context_manager]
+    assert state.agent_result["messages"][-1] == {"role": "assistant", "content": "done"}