Redact sensitive tool result values (stdout, stderr, content, text) in audit logs; Restrict audit log and run store file permissions to owner-only

- Introduce SENSITIVE_RESULT_KEYS frozenset with content, stderr, stdout, text
- Route result dict through redact_tool_result to strip file contents,
  shell output, and match text while preserving metadata like path,
  truncated, line, and exit_code
- Set 0700 on audit and run store directories and 0600 on individual
  log files via chmod to prevent unauthorized reads
- Apply owner-only permissions on creation of .teaagent/ dirs and when
  writing new audit/run store files

Author: johnteee

teaagent/audit.py

@@ -12,6 +12,8 @@
 AUDIT_REDACTED = '[redacted]'
 AUDIT_TRUNCATED = '[truncated]'
 MAX_AUDIT_STRING_LENGTH = 20_000
+AUDIT_DIR_MODE = 0o700
+AUDIT_FILE_MODE = 0o600
 SENSITIVE_KEY_PARTS = (
     'api_key',
     'authorization',
@@ -21,6 +23,7 @@
     'token',
 )
 SENSITIVE_ARGUMENT_KEYS = frozenset({'command', 'content', 'new', 'old'})
+SENSITIVE_RESULT_KEYS = frozenset({'content', 'stderr', 'stdout', 'text'})
 
 
 def utc_now() -> str:
@@ -58,6 +61,7 @@ def __init__(self, path: Optional[Path] = None) -> None:
         self._lock = threading.Lock()
         if self.path is not None:
             self.path.parent.mkdir(parents=True, exist_ok=True)
+            secure_audit_dir(self.path.parent)
 
     def add_sink(self, sink: Callable[[AuditEvent], None]) -> None:
         self._sinks.append(sink)
@@ -71,6 +75,7 @@ def record(self, event_type: str, run_id: str, **payload: Any) -> AuditEvent:
             if self.path is not None:
                 with self.path.open('a', encoding='utf-8') as handle:
                     handle.write(event.to_json() + '\n')
+                secure_audit_file(self.path)
             sinks = list(self._sinks)
         for sink in sinks:
             sink(event)
@@ -82,11 +87,21 @@ def redact_audit_payload(payload: dict[str, Any]) -> dict[str, Any]:
     for key, value in payload.items():
         if key == 'arguments' and isinstance(value, dict):
             redacted[key] = redact_tool_arguments(value)
+        elif key == 'result' and isinstance(value, dict):
+            redacted[key] = redact_tool_result(value)
         else:
             redacted[key] = redact_audit_value(key, value)
     return redacted
 
 
+def secure_audit_dir(path: Path) -> None:
+    path.chmod(AUDIT_DIR_MODE)
+
+
+def secure_audit_file(path: Path) -> None:
+    path.chmod(AUDIT_FILE_MODE)
+
+
 def redact_tool_arguments(arguments: dict[str, Any]) -> dict[str, Any]:
     return {
         str(key): redact_tool_argument_value(str(key), value)
@@ -109,6 +124,28 @@ def redact_tool_argument_value(key: str, value: Any) -> Any:
     return redact_audit_value(key, value)
 
 
+def redact_tool_result(result: dict[str, Any]) -> dict[str, Any]:
+    return {
+        str(key): redact_tool_result_value(str(key), value)
+        for key, value in result.items()
+    }
+
+
+def redact_tool_result_value(key: str, value: Any) -> Any:
+    if is_sensitive_key(key) or key in SENSITIVE_RESULT_KEYS:
+        return AUDIT_REDACTED
+    if isinstance(value, dict):
+        return {
+            str(child_key): redact_tool_result_value(str(child_key), child_value)
+            for child_key, child_value in value.items()
+        }
+    if isinstance(value, list):
+        return [redact_tool_result_value('', item) for item in value]
+    if isinstance(value, tuple):
+        return [redact_tool_result_value('', item) for item in value]
+    return redact_audit_value(key, value)
+
+
 def redact_audit_value(key: str, value: Any) -> Any:
     if is_sensitive_key(key):
         return AUDIT_REDACTED

teaagent/run_store.py

@@ -6,7 +6,7 @@
 from typing import Any, Optional
 from uuid import uuid4
 
-from teaagent.audit import AuditLogger, utc_now
+from teaagent.audit import AuditLogger, secure_audit_dir, secure_audit_file, utc_now
 from teaagent.runner import RunResult
 
 
@@ -37,6 +37,8 @@ def __init__(self, root: str | Path = '.') -> None:
         self.root = Path(root).resolve()
         self.store_dir = self.root / '.teaagent' / 'runs'
         self.store_dir.mkdir(parents=True, exist_ok=True)
+        secure_audit_dir(self.root / '.teaagent')
+        secure_audit_dir(self.store_dir)
 
     def audit_logger(self, run_id: Optional[str] = None) -> AuditLogger:
         if run_id is None:
@@ -50,6 +52,7 @@ def logger_for_result(self, result: RunResult, audit: AuditLogger) -> None:
             return
         target = self.run_path(result.run_id)
         target.write_text(audit.path.read_text(encoding='utf-8'), encoding='utf-8')
+        secure_audit_file(target)
         audit.path.unlink(missing_ok=True)
 
     def run_path(self, run_id: str) -> Path:

tests/test_audit.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 import json
+import stat
 import tempfile
 import threading
 import unittest
 from dataclasses import FrozenInstanceError
 from pathlib import Path
 
 from teaagent.audit import (
+    AUDIT_DIR_MODE,
+    AUDIT_FILE_MODE,
     AUDIT_REDACTED,
     AUDIT_TRUNCATED,
     MAX_AUDIT_STRING_LENGTH,
@@ -17,6 +20,10 @@
 )
 
 
+def file_mode(path: Path) -> int:
+    return stat.S_IMODE(path.stat().st_mode)
+
+
 class AuditEventTests(unittest.TestCase):
     def test_event_has_default_fields(self) -> None:
         event = AuditEvent(
@@ -159,6 +166,32 @@ def test_record_preserves_non_argument_content(self) -> None:
 
         self.assertEqual(event.payload['content'], 'read result')
 
+    def test_record_redacts_sensitive_tool_result_values(self) -> None:
+        logger = AuditLogger()
+
+        event = logger.record(
+            'tool_call_completed',
+            'run-1',
+            tool_name='workspace_read_file',
+            result={
+                'path': 'file.txt',
+                'content': 'secret file body',
+                'truncated': False,
+                'matches': [{'line': 1, 'text': 'secret match'}],
+                'stdout': 'secret stdout',
+                'stderr': 'secret stderr',
+            },
+        )
+
+        result = event.payload['result']
+        self.assertEqual(result['path'], 'file.txt')
+        self.assertFalse(result['truncated'])
+        self.assertEqual(result['content'], AUDIT_REDACTED)
+        self.assertEqual(result['matches'][0]['line'], 1)
+        self.assertEqual(result['matches'][0]['text'], AUDIT_REDACTED)
+        self.assertEqual(result['stdout'], AUDIT_REDACTED)
+        self.assertEqual(result['stderr'], AUDIT_REDACTED)
+
     def test_record_truncates_large_strings(self) -> None:
         logger = AuditLogger()
 
@@ -178,6 +211,8 @@ def test_path_parent_dirs_are_created(self) -> None:
             logger.record('e', 'r')
 
             self.assertTrue(path.exists())
+            self.assertEqual(file_mode(path.parent), AUDIT_DIR_MODE)
+            self.assertEqual(file_mode(path), AUDIT_FILE_MODE)
 
     def test_in_memory_only_when_no_path(self) -> None:
         logger = AuditLogger()

tests/test_run_store.py

@@ -2,6 +2,7 @@
 
 import io
 import json
+import stat
 import tempfile
 import unittest
 from contextlib import redirect_stdout
@@ -13,6 +14,10 @@
 from teaagent.runner import RunResult
 
 
+def file_mode(path: Path) -> int:
+    return stat.S_IMODE(path.stat().st_mode)
+
+
 class RunStoreTests(unittest.TestCase):
     def test_run_store_persists_and_summarizes_audit_events(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
@@ -38,6 +43,29 @@ def test_run_store_persists_and_summarizes_audit_events(self) -> None:
             self.assertEqual(events[0]['event_type'], 'run_started')
             self.assertTrue((Path(tmp) / '.teaagent' / 'runs' / 'run-1.jsonl').exists())
 
+    def test_run_store_uses_owner_only_permissions(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            store = RunStore(tmp)
+            audit = store.audit_logger()
+            audit.record('run_started', 'secure-run', task='demo')
+            store.logger_for_result(
+                RunResult(
+                    run_id='secure-run',
+                    final_answer=None,
+                    iterations=1,
+                    tool_calls=0,
+                    status='completed',
+                ),
+                audit,
+            )
+
+            root = Path(tmp)
+            self.assertEqual(file_mode(root / '.teaagent'), 0o700)
+            self.assertEqual(file_mode(root / '.teaagent' / 'runs'), 0o700)
+            self.assertEqual(
+                file_mode(root / '.teaagent' / 'runs' / 'secure-run.jsonl'), 0o600
+            )
+
     def test_task_for_run_extracts_original_task(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             store = RunStore(tmp)
@@ -183,7 +211,8 @@ def test_observations_for_run_returns_completed_tool_calls(self) -> None:
             observations = store.observations_for_run('run-obs')
 
             self.assertEqual([obs['call_id'] for obs in observations], ['r1', 'r2'])
-            self.assertEqual(observations[0]['result']['content'], 'hi')
+            self.assertEqual(observations[0]['result']['path'], 'a.txt')
+            self.assertEqual(observations[0]['result']['content'], AUDIT_REDACTED)
 
     def test_pending_approval_for_run_returns_last_unresolved(self) -> None:
         with tempfile.TemporaryDirectory() as tmp: