Skip to content

fix(security): resolve file reader and summarizer path injection sinks #29

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
spelech merged 1 commit into from
Sep 27, 2026
Merged

fix(security): resolve file reader and summarizer path injection sinks #29

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion app/api/routers/files.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,9 +54,11 @@ async def api_summarize_file(payload: FileSummarizePayload):
raw_path = payload.path.strip() if payload.path else ""
if not raw_path or "\x00" in raw_path or any(part == ".." for part in raw_path.replace("\\", "/").split("/")):
return JSONResponse(status_code=400, content={"error": "Path traversal or invalid path detected."})
reader = get_file_reader_service()
safe_path, _ = reader.resolve_safe_path(raw_path, repo=payload.repo)
summarizer = get_summarizer_service()
summary_text = summarizer.get_or_create_summary(
filepath=raw_path,
filepath=safe_path,
repo=payload.repo,
force_refresh=payload.force_refresh
)
Expand Down
26 changes: 6 additions & 20 deletions app/services/file_reader.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -114,19 +114,17 @@ def resolve_safe_path(self, path: str, repo: Optional[str] = None) -> Tuple[str,
else:
for ip in matching_paths:
root = os.path.abspath(ip["path"])
root_prefix = root if root.endswith(os.sep) else root + os.sep
candidate = os.path.normpath(os.path.abspath(os.path.join(root, path)))
if (candidate.startswith(root_prefix) or candidate == root):
if candidate.startswith(root):
if os.path.lexists(candidate):
if not self._is_within_root(candidate, root):
raise ValueError("Path outside authorized roots")
return candidate, "indexed_path"

for ip in matching_paths:
root = os.path.abspath(ip["path"])
root_prefix = root if root.endswith(os.sep) else root + os.sep
candidate = os.path.normpath(os.path.abspath(os.path.join(root, path)))
if (candidate.startswith(root_prefix) or candidate == root) and self._is_within_root(candidate, root):
if candidate.startswith(root) and self._is_within_root(candidate, root):
return candidate, "indexed_path"
raise ValueError("Path outside authorized roots")

Expand All @@ -145,42 +143,35 @@ def resolve_safe_path(self, path: str, repo: Optional[str] = None) -> Tuple[str,

# Relative path without repo specified:
cand_storage = os.path.normpath(os.path.abspath(os.path.join(storage_root, path)))
storage_prefix = storage_root if storage_root.endswith(os.sep) else storage_root + os.sep
if (cand_storage.startswith(storage_prefix) or cand_storage == storage_root):
if cand_storage.startswith(storage_root):
if os.path.lexists(cand_storage):
if not self._is_within_root(cand_storage, storage_root):
raise ValueError("Path outside authorized roots")
return cand_storage, "local_storage"

for ip in indexed_paths:
root = os.path.abspath(ip["path"])
root_prefix = root if root.endswith(os.sep) else root + os.sep
cand_ip = os.path.normpath(os.path.abspath(os.path.join(root, path)))
if (cand_ip.startswith(root_prefix) or cand_ip == root):
if cand_ip.startswith(root):
if os.path.lexists(cand_ip):
if not self._is_within_root(cand_ip, root):
raise ValueError("Path outside authorized roots")
return cand_ip, "indexed_path"

# If not existing on disk, check if it falls inside valid storage root
if (cand_storage.startswith(storage_prefix) or cand_storage == storage_root) and self._is_within_root(cand_storage, storage_root):
if cand_storage.startswith(storage_root) and self._is_within_root(cand_storage, storage_root):
return cand_storage, "local_storage"

for ip in indexed_paths:
root = os.path.abspath(ip["path"])
root_prefix = root if root.endswith(os.sep) else root + os.sep
cand_ip = os.path.normpath(os.path.abspath(os.path.join(root, path)))
if (cand_ip.startswith(root_prefix) or cand_ip == root) and self._is_within_root(cand_ip, root):
if cand_ip.startswith(root) and self._is_within_root(cand_ip, root):
return cand_ip, "indexed_path"

raise ValueError("Path outside authorized roots")

def is_binary_file(self, abs_path: str) -> bool:
"""Detects binary files by checking for null bytes in the initial sample."""
root = os.path.abspath(os.path.sep)
root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
if not (abs_path.startswith(root_prefix) or abs_path == root):
raise ValueError(f"Invalid path: {abs_path}")
with open(abs_path, "rb") as f:
chunk = f.read(8192)
return b"\x00" in chunk
Expand All @@ -196,11 +187,6 @@ def read_file(
"""Reads a file with safe path resolution, binary checking, and line slicing."""
abs_path, source_type = self.resolve_safe_path(path, repo=repo)

root = os.path.abspath(os.path.sep)
root_prefix = root if root.endswith(os.path.sep) else root + os.path.sep
if not (abs_path.startswith(root_prefix) or abs_path == root):
raise ValueError(f"Invalid path: {path}")

if not os.path.exists(abs_path):
raise FileNotFoundError(f"File not found: {path}")
if os.path.isdir(abs_path):
Expand Down
43 changes: 17 additions & 26 deletions app/services/summarizer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,35 +243,26 @@
content = None

if content is None:
norm_fp = os.path.normpath(os.path.abspath(filepath))
root_dir = os.path.abspath(os.path.sep)
root_prefix = root_dir if root_dir.endswith(os.path.sep) else root_dir + os.path.sep
if not (norm_fp.startswith(root_prefix) or norm_fp == root_dir):
norm_fp = ""
# Fallback to direct read if safe file exists on disk
try:
if norm_fp and os.path.commonpath([norm_fp, root_dir]) != root_dir:
norm_fp = ""
except ValueError:
norm_fp = ""

if norm_fp and os.path.exists(norm_fp) and os.path.isfile(norm_fp):
try:
norm_fp = os.path.normpath(os.path.abspath(filepath))
if os.path.exists(norm_fp) and os.path.isfile(norm_fp):
with open(norm_fp, "r", encoding="utf-8", errors="replace") as f:
content = f.read()
except Exception as e:
logger.warning(f"Failed to read file from disk '{norm_fp}': {e}")
else:
try:
from app.services.local_storage import get_default_storage_path
storage_root = os.path.normpath(os.path.abspath(get_default_storage_path()))
storage_prefix = storage_root if storage_root.endswith(os.sep) else storage_root + os.sep
cleaned_fp = filepath.strip().replace("\\", "/").lstrip("/")
storage_cand = os.path.normpath(os.path.abspath(os.path.join(storage_root, cleaned_fp)))
if storage_cand.startswith(storage_prefix) and os.path.exists(storage_cand) and os.path.isfile(storage_cand):
with open(storage_cand, "r", encoding="utf-8", errors="replace") as f:
content = f.read()
except Exception:
pass
except Exception:
pass

if content is None:
# Fallback to local_storage
try:
from app.services.local_storage import LocalStorageService
ls = LocalStorageService()
cleaned_fp = filepath.strip().replace("\\", "/").lstrip("/")
res = ls.read_file_content(cleaned_fp)
if isinstance(res, dict) and "content" in res:
content = res["content"]
except Exception:
pass

if content is None:
raise FileNotFoundError(f"File not found on disk or storage: {filepath}")
Expand Down
Loading