Coverage for src/local_deep_research/security/file_write_verifier.py: 96%
63 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Security module for verified file write operations.
3This module ensures that file writes only occur when explicitly allowed by configuration,
4maintaining the encryption-at-rest security model.
5"""
7import json
8import os
9from pathlib import Path
10from typing import Any
12from loguru import logger
14# Platforms without O_NOFOLLOW retain their normal symlink-following behavior.
15_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
18def _open_nofollow(path: str, flags: int) -> int:
19 """Add supported leaf-symlink refusal to Python's normal open flags."""
20 return os.open(path, flags | _O_NOFOLLOW, 0o666)
23def _open_nofollow_with_permissions(
24 path: str, flags: int, permissions: int
25) -> int:
26 """Tighten an existing file before truncation, or create it privately."""
27 truncate = bool(flags & os.O_TRUNC)
28 fd = os.open(path, (flags & ~os.O_TRUNC) | _O_NOFOLLOW, permissions)
29 try:
30 os.fchmod(fd, permissions)
31 if truncate: 31 ↛ 36line 31 didn't jump to line 36 because the condition on line 31 was always true
32 os.ftruncate(fd, 0)
33 except BaseException:
34 os.close(fd)
35 raise
36 return fd
39# Keys that should never be written to disk in clear text
40SENSITIVE_KEYS = frozenset(
41 {
42 "password",
43 "api_key",
44 "apikey",
45 "api-key",
46 "secret",
47 "secret_key",
48 "secretkey",
49 "token",
50 "access_token",
51 "refresh_token",
52 "private_key",
53 "privatekey",
54 "credentials",
55 "auth",
56 "authorization",
57 }
58)
61def _sanitize_sensitive_data(data: Any) -> Any:
62 """Remove sensitive keys from data before writing to disk.
64 Args:
65 data: Data to sanitize (dict, list, or primitive)
67 Returns:
68 Sanitized copy of the data with sensitive keys redacted
69 """
70 if isinstance(data, dict):
71 result = {}
72 for key, value in data.items():
73 key_lower = key.lower() if isinstance(key, str) else key
74 if key_lower in SENSITIVE_KEYS:
75 result[key] = "[REDACTED]"
76 else:
77 result[key] = _sanitize_sensitive_data(value)
78 return result
79 if isinstance(data, list):
80 return [_sanitize_sensitive_data(item) for item in data]
81 return data
84class FileWriteSecurityError(Exception):
85 """Raised when a file write operation is not allowed by security settings."""
87 pass
90def write_file_verified(
91 filepath: str | Path,
92 content: str,
93 setting_name: str,
94 required_value: Any = True,
95 context: str = "",
96 mode: str = "w",
97 encoding: str = "utf-8",
98 settings_snapshot: dict = None,
99 permissions: int | None = None,
100) -> None:
101 """Write content to a file only if security settings allow it.
103 Args:
104 filepath: Path to the file to write
105 content: Content to write to the file
106 setting_name: Configuration setting name to check (e.g., "api.allow_file_output")
107 required_value: Required value for the setting (default: True)
108 context: Description of what's being written (for error messages)
109 mode: File open mode (default: "w")
110 encoding: File encoding (default: "utf-8")
111 settings_snapshot: Optional settings snapshot for programmatic mode
112 permissions: POSIX file mode for private output. Existing files are
113 tightened before their contents are replaced.
115 Raises:
116 FileWriteSecurityError: If the security setting doesn't match required value
117 OSError: errno ELOOP, if the destination leaf is a symlink and the
118 platform exposes O_NOFOLLOW
120 Example:
121 >>> write_file_verified(
122 ... "report.md",
123 ... markdown_content,
124 ... "api.allow_file_output",
125 ... context="API research report"
126 ... )
127 """
128 from ..config.search_config import get_setting_from_snapshot
130 try:
131 actual_value = get_setting_from_snapshot(
132 setting_name, settings_snapshot=settings_snapshot
133 )
134 except Exception:
135 # Setting doesn't exist - default deny
136 actual_value = None
138 if actual_value != required_value:
139 error_msg = (
140 f"File write not allowed: {context or 'file operation'}. "
141 f"Set '{setting_name}={required_value}' in config to enable this feature."
142 )
143 logger.warning(error_msg)
144 raise FileWriteSecurityError(error_msg)
146 # Where O_NOFOLLOW is available, refuse a symlink at the final path
147 # component. Parent-directory symlinks and a leaf already dereferenced
148 # by the caller's Path.resolve() are outside this check.
149 # Let open() validate modes, own the descriptor and handle text/binary
150 # translation. This remains an in-place write, not an atomic replacement.
151 opener = _open_nofollow
152 if permissions is not None:
153 if os.name != "posix" or not 0 <= permissions <= 0o777: 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true
154 raise ValueError("Private POSIX file permissions are unavailable")
156 def private_opener(path: str, flags: int) -> int:
157 return _open_nofollow_with_permissions(path, flags, permissions)
159 opener = private_opener
161 if "b" in mode:
162 with open(filepath, mode, opener=opener) as f: # nosec B603
163 f.write(content)
164 else:
165 with open(filepath, mode, encoding=encoding, opener=opener) as f: # nosec B603
166 f.write(content)
168 logger.debug(
169 f"Verified file write: {filepath} (setting: {setting_name}={required_value})"
170 )
173def write_json_verified(
174 filepath: str | Path,
175 data: dict | list,
176 setting_name: str,
177 required_value: Any = True,
178 context: str = "",
179 settings_snapshot: dict = None,
180 permissions: int | None = None,
181 **json_kwargs,
182) -> None:
183 """Write JSON data to a file only if security settings allow it.
185 Args:
186 filepath: Path to the file to write
187 data: Dictionary or list to serialize as JSON
188 setting_name: Configuration setting name to check
189 required_value: Required value for the setting (default: True)
190 context: Description of what's being written (for error messages)
191 settings_snapshot: Optional settings snapshot for programmatic mode
192 permissions: POSIX file mode for private output.
193 **json_kwargs: Additional keyword arguments to pass to json.dumps()
194 (e.g., indent=2, ensure_ascii=False, sort_keys=True, default=custom_serializer)
196 Raises:
197 FileWriteSecurityError: If the security setting doesn't match required value
199 Example:
200 >>> write_json_verified(
201 ... "results.json",
202 ... {"accuracy": 0.95},
203 ... "benchmark.allow_file_output",
204 ... context="benchmark results",
205 ... indent=2,
206 ... sort_keys=True
207 ... )
208 """
209 # Default to indent=2 if not specified for readability
210 if "indent" not in json_kwargs:
211 json_kwargs["indent"] = 2
213 # Sanitize sensitive data before writing to disk
214 sanitized_data = _sanitize_sensitive_data(data)
215 content = json.dumps(sanitized_data, **json_kwargs)
216 write_file_verified(
217 filepath,
218 content,
219 setting_name,
220 required_value,
221 context,
222 mode="w",
223 encoding="utf-8",
224 settings_snapshot=settings_snapshot,
225 permissions=permissions,
226 )