mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-09-27 18:42:20 +02:00
Merge commit from fork
* fix(security): keep agent file tools out of the app state directory
The agent's read tools (read_file, grep, glob, ls) resolved model-supplied
paths against a root list whose first entry was the whole data directory.
That directory holds the session store, the auth database, the app
encryption key and the settings file, so prompt-injected content could ask
for any of them. No approval prompt stood in the way: reads are classified
read_workspace and pass the untrusted-context gate untouched, which is
correct for reading a workspace and wrong for reading the app's own state.
The agent gets data/agent_workspace/ instead, and the subprocess cwd and
HOME move with it so bash and read_file agree on where scratch files live.
The deny itself is a property of the path, not of the root it arrived
through, because three routes reach the same bytes and closing only the
first leaves the other two working:
- the default root list
- a workspace bound at or above the data directory, which vet_workspace
accepted and chat_routes auto-binds from a path named in the message
- a tool_path_extra_roots setting covering the data directory
_resolve_search_root also returned the workspace root unchecked when the
path was empty, so a bare ls enumerated the directory whatever the deny
list said. It now resolves that case through the same guards.
A containment rule rather than a filename deny list, so state files added
later are covered without anyone remembering to list them, and so a user's
own settings.json or app.db inside a real workspace is not caught.
Four directories of user content stay readable, because the application
hands their paths to the model and tells it to open them: the chat upload
manifest, downloaded mail attachments, personal docs (which covers the
runbook) and personal uploads.
* fix: enforce state deny during recursive file search
* fix: bound protected filesystem searches
* fix(security): reject inode aliases and workspace redirects
* fix(security): harden partitioned agent searches
* fix(security): report fallback worker exits promptly
* fix(security): clean up search readers and retain relative data roots
---------
Co-authored-by: RaresKeY <158580472+RaresKeY@users.noreply.github.com>
This commit is contained in:
nopoz
@@ -3,8 +3,8 @@ import json
|
||||
import os
|
||||
import re
|
||||
import difflib
|
||||
import fnmatch
|
||||
import shutil
|
||||
import time
|
||||
from typing import Optional, Dict, Any, Tuple, List
|
||||
|
||||
from src.constants import MAX_READ_CHARS, MAX_DIFF_LINES, MAX_OUTPUT_CHARS
|
||||
@@ -16,6 +16,8 @@ _CODENAV_SKIP_DIRS = frozenset({
|
||||
})
|
||||
_CODENAV_MAX_HITS = 200
|
||||
_CODENAV_MAX_LINE = 400
|
||||
_GREP_TIMEOUT_SECONDS = 20
|
||||
_GREP_STDERR_PREFIX = 20_000
|
||||
|
||||
|
||||
def _glob_to_regex(pat: str) -> "re.Pattern":
|
||||
@@ -42,6 +44,113 @@ def _glob_to_regex(pat: str) -> "re.Pattern":
|
||||
i += 1
|
||||
return re.compile("".join(out))
|
||||
|
||||
|
||||
def _python_grep_worker(payload: dict, output_queue) -> None:
|
||||
"""Spawn-safe fallback grep worker used when ripgrep is unavailable.
|
||||
|
||||
Keep this at module scope: a frozen Windows executable cannot safely be
|
||||
relaunched as ``sys.executable -c ...``, while multiprocessing can invoke a
|
||||
top-level target through its frozen-process bootstrap.
|
||||
"""
|
||||
try:
|
||||
flags = re.IGNORECASE if payload["ignore_case"] else 0
|
||||
try:
|
||||
regex = re.compile(payload["pattern"], flags)
|
||||
glob_regex = (
|
||||
_glob_to_regex(payload["glob"].replace("\\", "/"))
|
||||
if payload["glob"]
|
||||
else None
|
||||
)
|
||||
except re.error as exc:
|
||||
output_queue.put(("error", f"grep: bad pattern: {exc}"))
|
||||
return
|
||||
|
||||
requested_root = payload["root"]
|
||||
skip_dirs = set(payload["skip_dirs"])
|
||||
sensitive = {name.casefold() for name in payload["sensitive_names"]}
|
||||
max_hits = payload["max_hits"]
|
||||
hits = 0
|
||||
|
||||
def within(path: str, root: str) -> bool:
|
||||
try:
|
||||
return os.path.commonpath(
|
||||
[os.path.normcase(path), os.path.normcase(root)]
|
||||
) == os.path.normcase(root)
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
def safe_file(path: str, target: str) -> Optional[str]:
|
||||
if os.path.islink(path):
|
||||
return None
|
||||
canonical = os.path.realpath(path)
|
||||
if not within(canonical, requested_root) or not within(canonical, target):
|
||||
return None
|
||||
parts = [part.casefold() for part in canonical.split(os.sep)]
|
||||
if any(part in sensitive for part in parts):
|
||||
return None
|
||||
try:
|
||||
if not os.path.isfile(canonical) or os.stat(canonical).st_nlink > 1:
|
||||
return None
|
||||
except OSError:
|
||||
return None
|
||||
return canonical
|
||||
|
||||
for target in payload["targets"]:
|
||||
if hits >= max_hits:
|
||||
break
|
||||
if os.path.isfile(target):
|
||||
file_iter = iter((target,))
|
||||
else:
|
||||
def walk_files():
|
||||
for directory, dirnames, filenames in os.walk(
|
||||
target, followlinks=False
|
||||
):
|
||||
dirnames[:] = [
|
||||
name
|
||||
for name in dirnames
|
||||
if name not in skip_dirs
|
||||
and name.casefold() not in sensitive
|
||||
and not os.path.islink(os.path.join(directory, name))
|
||||
]
|
||||
for name in filenames:
|
||||
yield os.path.join(directory, name)
|
||||
|
||||
file_iter = walk_files()
|
||||
|
||||
for candidate in file_iter:
|
||||
path = safe_file(candidate, target)
|
||||
if path is None:
|
||||
continue
|
||||
relative = os.path.relpath(path, requested_root).replace(os.sep, "/")
|
||||
if glob_regex and not (
|
||||
glob_regex.fullmatch(relative)
|
||||
or glob_regex.fullmatch(os.path.basename(path))
|
||||
):
|
||||
continue
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8", errors="strict") as handle:
|
||||
for number, line in enumerate(handle, 1):
|
||||
if regex.search(line):
|
||||
output_queue.put((
|
||||
"match",
|
||||
path,
|
||||
number,
|
||||
line.rstrip()[:_CODENAV_MAX_LINE],
|
||||
))
|
||||
hits += 1
|
||||
if hits >= max_hits:
|
||||
break
|
||||
except (UnicodeDecodeError, OSError):
|
||||
continue
|
||||
if hits >= max_hits:
|
||||
break
|
||||
output_queue.put(("done",))
|
||||
except BaseException as exc:
|
||||
try:
|
||||
output_queue.put(("error", f"grep: fallback worker failed: {exc}"))
|
||||
except BaseException:
|
||||
pass
|
||||
|
||||
def _unified_diff(old: str, new: str, path: str) -> Optional[Dict[str, Any]]:
|
||||
if old == new:
|
||||
return None
|
||||
@@ -407,7 +516,11 @@ def _apply_patch_hunks(original: str, hunks: List[List[str]], label: str) -> str
|
||||
|
||||
class LsTool:
|
||||
async def execute(self, content: str, ctx: dict) -> dict:
|
||||
from src.tool_execution import _resolve_tool_path, _resolve_search_root, _truncate
|
||||
from src.tool_execution import (
|
||||
_is_denied_tool_path,
|
||||
_resolve_search_root,
|
||||
_truncate,
|
||||
)
|
||||
raw_path = ""
|
||||
_s = (content or "").strip()
|
||||
if _s.startswith("{"):
|
||||
@@ -431,6 +544,8 @@ class LsTool:
|
||||
for entry in it:
|
||||
if entry.name.startswith("."):
|
||||
continue
|
||||
if _is_denied_tool_path(os.path.realpath(entry.path)):
|
||||
continue
|
||||
try:
|
||||
is_dir = entry.is_dir(follow_symlinks=False)
|
||||
size = entry.stat(follow_symlinks=False).st_size if not is_dir else 0
|
||||
@@ -458,7 +573,8 @@ class GlobTool:
|
||||
async def execute(self, content: str, ctx: dict) -> dict:
|
||||
from src.tool_execution import (
|
||||
_SENSITIVE_BASENAMES,
|
||||
_is_sensitive_path,
|
||||
_can_traverse_tool_path,
|
||||
_is_denied_tool_path,
|
||||
_resolve_tool_path,
|
||||
_resolve_search_root,
|
||||
_truncate,
|
||||
@@ -507,7 +623,7 @@ class GlobTool:
|
||||
# .ssh/id_rsa, …) falls through to the walk, which skips it —
|
||||
# otherwise glob would surface secret paths that read_file /
|
||||
# grep already refuse to touch.
|
||||
if inside and os.path.exists(cand) and not _is_sensitive_path(cand):
|
||||
if inside and os.path.exists(cand) and not _is_denied_tool_path(cand):
|
||||
return [cand], None
|
||||
# Literal not at exact path — fall through to walk so
|
||||
# e.g. "foo.py" still matches at any depth (like rglob).
|
||||
@@ -517,13 +633,18 @@ class GlobTool:
|
||||
cap = _CODENAV_MAX_HITS * 5
|
||||
try:
|
||||
for dp, dns, fns in os.walk(base):
|
||||
if not _can_traverse_tool_path(os.path.realpath(dp)):
|
||||
dns[:] = []
|
||||
continue
|
||||
# Prune skipped dirs before descending (unlike rglob which
|
||||
# descends first then filters — fatal on large node_modules).
|
||||
# Sensitive dirs (.ssh, .gnupg, …) are pruned too so glob
|
||||
# never enumerates the keys/tokens inside them.
|
||||
dns[:] = [
|
||||
d for d in dns
|
||||
if d not in _CODENAV_SKIP_DIRS and d not in _SENSITIVE_BASENAMES
|
||||
if d not in _CODENAV_SKIP_DIRS
|
||||
and d not in _SENSITIVE_BASENAMES
|
||||
and _can_traverse_tool_path(os.path.realpath(os.path.join(dp, d)))
|
||||
]
|
||||
for name in fns + dns:
|
||||
full = os.path.join(dp, name)
|
||||
@@ -531,7 +652,7 @@ class GlobTool:
|
||||
if regex.fullmatch(rel) or regex.fullmatch(name):
|
||||
# Skip deny-listed sensitive files (.env, id_rsa,
|
||||
# known_hosts, …) the same way grep does.
|
||||
if _is_sensitive_path(os.path.realpath(full)):
|
||||
if _is_denied_tool_path(os.path.realpath(full)):
|
||||
continue
|
||||
try:
|
||||
mtime = os.stat(full).st_mtime
|
||||
@@ -558,9 +679,12 @@ class GlobTool:
|
||||
class GrepTool:
|
||||
async def execute(self, content: str, ctx: dict) -> dict:
|
||||
from src.tool_execution import (
|
||||
_SENSITIVE_BASENAMES,
|
||||
_SENSITIVE_FILE_PATTERNS,
|
||||
_agent_readable_data_subdirs,
|
||||
_is_denied_tool_path,
|
||||
_is_sensitive_path,
|
||||
_resolve_tool_path,
|
||||
_path_within,
|
||||
_resolve_search_root,
|
||||
_truncate,
|
||||
)
|
||||
@@ -589,64 +713,307 @@ class GrepTool:
|
||||
return {"error": f"grep: {e}", "exit_code": 1}
|
||||
|
||||
def _grep():
|
||||
import re as _re
|
||||
import shutil
|
||||
import multiprocessing
|
||||
import queue
|
||||
import subprocess
|
||||
import threading
|
||||
|
||||
from src.constants import DATA_DIR
|
||||
|
||||
rg = shutil.which("rg")
|
||||
if rg:
|
||||
cmd = [rg, "--line-number", "--no-heading", "--color=never",
|
||||
"--max-count", str(max_hits)]
|
||||
if ignore_case:
|
||||
cmd.append("--ignore-case")
|
||||
if glob_pat:
|
||||
cmd += ["--glob", glob_pat]
|
||||
# --iglob (not --glob) so the exclusion is case-insensitive:
|
||||
# on a case-insensitive filesystem "ID_RSA"/"Known_Hosts"
|
||||
# resolve to the same secret as their lowercase forms, and the
|
||||
# Python fallback below already folds case via _is_sensitive_path.
|
||||
for _pat in _SENSITIVE_FILE_PATTERNS:
|
||||
cmd += ["--iglob", f"!*{_pat}*"]
|
||||
for _d in _CODENAV_SKIP_DIRS:
|
||||
cmd += ["--glob", f"!**/{_d}/**"]
|
||||
cmd += ["--regexp", pattern, root]
|
||||
real_root = os.path.realpath(root)
|
||||
data_dir = os.path.realpath(DATA_DIR)
|
||||
spans_state = _path_within(data_dir, real_root)
|
||||
|
||||
def is_top_level_safe(path: str, *, partition_generated: bool) -> bool:
|
||||
lexical = os.path.abspath(path)
|
||||
if os.path.islink(lexical):
|
||||
return False
|
||||
canonical = os.path.realpath(lexical)
|
||||
if not _path_within(canonical, real_root):
|
||||
return False
|
||||
if partition_generated and os.path.basename(lexical) in _CODENAV_SKIP_DIRS:
|
||||
return False
|
||||
if _is_sensitive_path(canonical) or _is_denied_tool_path(canonical):
|
||||
return False
|
||||
return True
|
||||
|
||||
def safe_targets() -> tuple[list[str], Optional[str]]:
|
||||
candidates: list[tuple[str, bool]] = []
|
||||
if not spans_state:
|
||||
# Preserve direct-root compatibility: skip-directory policy
|
||||
# prunes descendants, but an explicitly requested allowed
|
||||
# root named node_modules remains searchable.
|
||||
candidates.append((real_root, False))
|
||||
else:
|
||||
current = real_root
|
||||
if current != data_dir:
|
||||
for part in os.path.relpath(data_dir, current).split(os.sep):
|
||||
try:
|
||||
with os.scandir(current) as entries:
|
||||
for entry in entries:
|
||||
if entry.name != part:
|
||||
# Reject a sibling link lexically before
|
||||
# canonicalizing or treating it as a target.
|
||||
if entry.is_symlink():
|
||||
continue
|
||||
candidates.append((entry.path, True))
|
||||
except OSError as exc:
|
||||
return [], f"grep: {exc}"
|
||||
current = os.path.join(current, part)
|
||||
for readable in _agent_readable_data_subdirs():
|
||||
if (
|
||||
_path_within(readable, data_dir)
|
||||
and _path_within(readable, real_root)
|
||||
and os.path.exists(readable)
|
||||
):
|
||||
candidates.append((readable, True))
|
||||
|
||||
targets: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for candidate, partition_generated in candidates:
|
||||
if not is_top_level_safe(
|
||||
candidate, partition_generated=partition_generated
|
||||
):
|
||||
continue
|
||||
canonical = os.path.realpath(candidate)
|
||||
if canonical not in seen:
|
||||
seen.add(canonical)
|
||||
targets.append(canonical)
|
||||
return targets, None
|
||||
|
||||
targets, target_error = safe_targets()
|
||||
if target_error:
|
||||
return None, target_error
|
||||
|
||||
base = real_root if os.path.isdir(real_root) else os.path.dirname(real_root)
|
||||
deadline = time.monotonic() + _GREP_TIMEOUT_SECONDS
|
||||
lines: list[str] = []
|
||||
|
||||
def parse_rg_result(raw: str) -> Optional[str]:
|
||||
try:
|
||||
import subprocess
|
||||
p = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
|
||||
lines = [ln for ln in (p.stdout or "").splitlines() if ln][:max_hits]
|
||||
return lines, None
|
||||
except subprocess.TimeoutExpired:
|
||||
return None, "grep: timed out"
|
||||
except Exception as _e:
|
||||
return None, f"grep: {_e}"
|
||||
try:
|
||||
rx = _re.compile(pattern, _re.IGNORECASE if ignore_case else 0)
|
||||
except _re.error as _e:
|
||||
return None, f"grep: bad pattern: {_e}"
|
||||
hits = []
|
||||
if os.path.isfile(root):
|
||||
file_iter = [root]
|
||||
else:
|
||||
file_iter = []
|
||||
for dp, dns, fns in os.walk(root):
|
||||
dns[:] = [d for d in dns if d not in _CODENAV_SKIP_DIRS]
|
||||
for fn in fns:
|
||||
if glob_pat and not fnmatch.fnmatch(fn, glob_pat):
|
||||
record = json.loads(raw)
|
||||
except (TypeError, json.JSONDecodeError):
|
||||
return None
|
||||
if record.get("type") != "match":
|
||||
return None
|
||||
data = record.get("data") or {}
|
||||
path = (data.get("path") or {}).get("text")
|
||||
text_value = (data.get("lines") or {}).get("text")
|
||||
number = data.get("line_number")
|
||||
if not isinstance(path, str) or not isinstance(text_value, str):
|
||||
return None
|
||||
absolute = path if os.path.isabs(path) else os.path.join(base, path)
|
||||
canonical = os.path.realpath(absolute)
|
||||
if not _path_within(canonical, real_root) or _is_denied_tool_path(canonical):
|
||||
return None
|
||||
return f"{os.path.abspath(absolute)}:{number}:{text_value.rstrip()[:_CODENAV_MAX_LINE]}"
|
||||
|
||||
def run_rg(cmd: list[str]) -> Optional[str]:
|
||||
try:
|
||||
process = subprocess.Popen(
|
||||
cmd,
|
||||
cwd=base,
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
bufsize=1,
|
||||
)
|
||||
except Exception as exc:
|
||||
return f"grep: {exc}"
|
||||
output: queue.Queue[Optional[str]] = queue.Queue(maxsize=max_hits + 2)
|
||||
stderr_prefix: list[str] = []
|
||||
stderr_size = 0
|
||||
stop_reader = threading.Event()
|
||||
|
||||
def enqueue_stdout(value: Optional[str]) -> bool:
|
||||
# The consumer stops at the result cap or deadline. Never
|
||||
# leave a producer blocked on its bounded queue afterward.
|
||||
while not stop_reader.is_set():
|
||||
try:
|
||||
output.put(value, timeout=0.05)
|
||||
return True
|
||||
except queue.Full:
|
||||
continue
|
||||
file_iter.append(os.path.join(dp, fn))
|
||||
for fp in file_iter:
|
||||
if len(hits) >= max_hits:
|
||||
break
|
||||
if _is_sensitive_path(os.path.realpath(fp)):
|
||||
continue
|
||||
return False
|
||||
|
||||
def read_stdout() -> None:
|
||||
assert process.stdout is not None
|
||||
try:
|
||||
for line in process.stdout:
|
||||
if not enqueue_stdout(line.rstrip("\n")):
|
||||
break
|
||||
finally:
|
||||
enqueue_stdout(None)
|
||||
|
||||
def read_stderr() -> None:
|
||||
nonlocal stderr_size
|
||||
assert process.stderr is not None
|
||||
while True:
|
||||
chunk = process.stderr.read(4096)
|
||||
if not chunk:
|
||||
break
|
||||
if stderr_size < _GREP_STDERR_PREFIX:
|
||||
kept = chunk[:_GREP_STDERR_PREFIX - stderr_size]
|
||||
stderr_prefix.append(kept)
|
||||
stderr_size += len(kept)
|
||||
|
||||
stdout_thread = threading.Thread(target=read_stdout, daemon=True)
|
||||
stderr_thread = threading.Thread(target=read_stderr, daemon=True)
|
||||
stdout_thread.start()
|
||||
stderr_thread.start()
|
||||
timed_out = False
|
||||
capped = False
|
||||
try:
|
||||
with open(fp, "r", encoding="utf-8", errors="strict") as f:
|
||||
for i, line in enumerate(f, 1):
|
||||
if rx.search(line):
|
||||
hits.append(f"{fp}:{i}:{line.rstrip()[:_CODENAV_MAX_LINE]}")
|
||||
if len(hits) >= max_hits:
|
||||
break
|
||||
except (UnicodeDecodeError, OSError):
|
||||
continue
|
||||
return hits, None
|
||||
while len(lines) < max_hits:
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
timed_out = True
|
||||
break
|
||||
try:
|
||||
raw = output.get(timeout=remaining)
|
||||
except queue.Empty:
|
||||
timed_out = True
|
||||
break
|
||||
if raw is None:
|
||||
break
|
||||
parsed = parse_rg_result(raw)
|
||||
if parsed and parsed not in lines:
|
||||
lines.append(parsed)
|
||||
capped = len(lines) >= max_hits
|
||||
finally:
|
||||
stop_reader.set()
|
||||
if (timed_out or capped) and process.poll() is None:
|
||||
process.terminate()
|
||||
try:
|
||||
remaining = max(0.01, deadline - time.monotonic())
|
||||
return_code = process.wait(timeout=min(1, remaining))
|
||||
except subprocess.TimeoutExpired:
|
||||
process.kill()
|
||||
return_code = process.wait()
|
||||
stdout_thread.join()
|
||||
stderr_thread.join()
|
||||
if timed_out:
|
||||
return "grep: timed out"
|
||||
if not capped and return_code not in (0, 1):
|
||||
detail = "".join(stderr_prefix).strip()
|
||||
return f"grep: {detail or f'process exited {return_code}'}"
|
||||
return None
|
||||
|
||||
if rg:
|
||||
# Validate even when policy filtering leaves no search targets.
|
||||
if not targets:
|
||||
error = run_rg([rg, "--json", "--no-config", "--regexp", pattern])
|
||||
return (None, error) if error else ([], None)
|
||||
relative_targets = [os.path.relpath(target, base) for target in targets]
|
||||
for offset in range(0, len(relative_targets), 128):
|
||||
if len(lines) >= max_hits:
|
||||
break
|
||||
cmd = [
|
||||
rg, "--json", "--no-config", "--no-follow",
|
||||
"--max-count", str(max_hits - len(lines)),
|
||||
"--max-columns", str(_CODENAV_MAX_LINE),
|
||||
"--max-columns-preview",
|
||||
]
|
||||
if ignore_case:
|
||||
cmd.append("--ignore-case")
|
||||
if glob_pat:
|
||||
cmd += ["--glob", glob_pat]
|
||||
for sensitive_pattern in _SENSITIVE_FILE_PATTERNS:
|
||||
cmd += ["--iglob", f"!{sensitive_pattern}"]
|
||||
for skipped_dir in _CODENAV_SKIP_DIRS:
|
||||
cmd += ["--glob", f"!**/{skipped_dir}/**"]
|
||||
cmd += ["--regexp", pattern, "--", *relative_targets[offset:offset + 128]]
|
||||
error = run_rg(cmd)
|
||||
if error:
|
||||
return None, error
|
||||
return lines, None
|
||||
|
||||
# This runs inside asyncio.to_thread(), so forking would clone a
|
||||
# multithreaded process and can deadlock. Spawn is platform-safe and
|
||||
# PyInstaller-compatible via launcher's early freeze_support().
|
||||
payload = {
|
||||
"root": real_root,
|
||||
"targets": targets,
|
||||
"pattern": pattern,
|
||||
"ignore_case": ignore_case,
|
||||
"glob": glob_pat,
|
||||
"max_hits": max_hits,
|
||||
"skip_dirs": tuple(_CODENAV_SKIP_DIRS),
|
||||
"sensitive_names": tuple(
|
||||
set(_SENSITIVE_BASENAMES) | set(_SENSITIVE_FILE_PATTERNS)
|
||||
),
|
||||
}
|
||||
try:
|
||||
context = multiprocessing.get_context("spawn")
|
||||
output_queue = context.Queue(maxsize=max_hits + 2)
|
||||
worker = context.Process(
|
||||
target=_python_grep_worker, args=(payload, output_queue)
|
||||
)
|
||||
worker.start()
|
||||
except Exception as exc:
|
||||
try:
|
||||
output_queue.close()
|
||||
except (NameError, OSError, ValueError):
|
||||
pass
|
||||
return None, f"grep: could not start fallback worker: {exc}"
|
||||
error = None
|
||||
completed = False
|
||||
try:
|
||||
while len(lines) < max_hits:
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
error = "grep: timed out"
|
||||
break
|
||||
try:
|
||||
# Keep queue waits short enough to observe a spawn
|
||||
# worker that dies during bootstrap/import before it
|
||||
# can enqueue either an error or the done sentinel.
|
||||
record = output_queue.get(timeout=min(0.05, remaining))
|
||||
except queue.Empty:
|
||||
if worker.is_alive():
|
||||
continue
|
||||
worker.join(timeout=0)
|
||||
try:
|
||||
# A multiprocessing queue's feeder can make the
|
||||
# final record visible at process-exit time. Give
|
||||
# that record precedence over the exit status.
|
||||
remaining = deadline - time.monotonic()
|
||||
record = output_queue.get(
|
||||
timeout=min(0.05, max(0, remaining))
|
||||
)
|
||||
except queue.Empty:
|
||||
error = f"grep: fallback worker exited {worker.exitcode}"
|
||||
break
|
||||
if record[0] == "done":
|
||||
completed = True
|
||||
break
|
||||
if record[0] == "error":
|
||||
error = record[1]
|
||||
break
|
||||
_, path, number, text_value = record
|
||||
canonical = os.path.realpath(path)
|
||||
if not _path_within(canonical, real_root) or _is_denied_tool_path(canonical):
|
||||
continue
|
||||
rendered = f"{path}:{number}:{text_value}"
|
||||
if rendered not in lines:
|
||||
lines.append(rendered)
|
||||
finally:
|
||||
if completed:
|
||||
worker.join(timeout=min(1, max(0.01, deadline - time.monotonic())))
|
||||
if worker.is_alive():
|
||||
worker.terminate()
|
||||
worker.join(timeout=1)
|
||||
if worker.is_alive():
|
||||
worker.kill()
|
||||
worker.join()
|
||||
output_queue.close()
|
||||
if error:
|
||||
return None, error
|
||||
if worker.exitcode not in (0, None) and len(lines) < max_hits:
|
||||
return None, f"grep: fallback worker exited {worker.exitcode}"
|
||||
return lines, None
|
||||
|
||||
lines, err = await asyncio.to_thread(_grep)
|
||||
if err:
|
||||
|
||||
Reference in New Issue
Block a user