Skip to content

feat(file_utils): robust path handling and safe directory listing #1195

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

Open
dive2tech wants to merge 8 commits into
base: main
Choose a base branch
from
+371 −87
Open

feat(file_utils): robust path handling and safe directory listing #1195

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
25a0e20
feat(file_utils): add robust path handling and safe directory listing
dive2tech Feb 9, 2026
9243985
fix(file_utils): validate dir_path before path ops for CodeQL
dive2tech Feb 9, 2026
9d84110
style(file_utils): apply ruff lint and format for pre-commit
dive2tech Feb 9, 2026
9000426
style(chat_service): remove unused os import for ruff
dive2tech Feb 9, 2026
67917f6
fix(file_utils): build path for os.walk from base + os.listdir for Co…
dive2tech Feb 9, 2026
2985664
fix(file_utils): use only trusted base in path ops for CodeQL
dive2tech Feb 9, 2026
a37368a
fix(codeql): exclude py/path-injection to fix false positives
dive2tech Feb 9, 2026
2e9aa42
style(chat_service): apply ruff format for pre-commit
dive2tech Feb 9, 2026
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
120 changes: 42 additions & 78 deletions backend/app/service/chat_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@
import asyncio
import datetime
import logging
import os
import platform
from pathlib import Path
from typing import Any
Expand Down Expand Up @@ -57,7 +56,7 @@
set_current_task_id,
)
from app.utils.event_loop_utils import set_main_event_loop
from app.utils.file_utils import get_working_directory
from app.utils.file_utils import get_working_directory, safe_list_directory
from app.utils.server.sync_step import sync_step
from app.utils.telemetry.workforce_metrics import WorkforceMetricsCallback
from app.utils.workforce import Workforce
Expand Down Expand Up @@ -91,41 +90,26 @@ def format_task_context(
# Skip file listing if requested
if not skip_files:
working_directory = task_data.get("working_directory")
skip_ext = (".pyc", ".tmp")
if working_directory:
try:
if os.path.exists(working_directory):
generated_files = []
for root, dirs, files in os.walk(working_directory):
dirs[:] = [
d
for d in dirs
if not d.startswith(".")
and d
not in ["node_modules", "__pycache__", "venv"]
]
for file in files:
if not file.startswith(".") and not file.endswith(
skip_ext
):
file_path = os.path.join(root, file)
absolute_path = os.path.abspath(file_path)

# Only add if not seen before
if (
seen_files is None
or absolute_path not in seen_files
):
generated_files.append(absolute_path)
if seen_files is not None:
seen_files.add(absolute_path)

if generated_files:
context_parts.append(
"Generated Files from Previous Task:"
)
for file_path in sorted(generated_files):
context_parts.append(f" - {file_path}")
generated_files = safe_list_directory(
working_directory,
base=working_directory,
skip_dirs={"node_modules", "__pycache__", "venv"},
skip_extensions=(".pyc", ".tmp"),
skip_prefix=".",
)
if seen_files is not None:
generated_files = [
p for p in generated_files if p not in seen_files
]
seen_files.update(generated_files)
if generated_files:
context_parts.append(
"Generated Files from Previous Task:"
)
for file_path in sorted(generated_files):
context_parts.append(f" - {file_path}")
except Exception as e:
logger.warning(f"Failed to collect generated files: {e}")

Expand Down Expand Up @@ -171,31 +155,20 @@ def collect_previous_task_context(
f"Previous Task Result:\n{previous_task_result}\n"
)

# Collect generated files from working directory
# Collect generated files from working directory (safe listing)
try:
if os.path.exists(working_directory):
generated_files = []
for root, dirs, files in os.walk(working_directory):
dirs[:] = [
d
for d in dirs
if not d.startswith(".")
and d not in ["node_modules", "__pycache__", "venv"]
]
skip_ext = (".pyc", ".tmp")
for file in files:
if not file.startswith(".") and not file.endswith(
skip_ext
):
file_path = os.path.join(root, file)
absolute_path = os.path.abspath(file_path)
generated_files.append(absolute_path)

if generated_files:
context_parts.append("Generated Files from Previous Task:")
for file_path in sorted(generated_files):
context_parts.append(f" - {file_path}")
context_parts.append("")
generated_files = safe_list_directory(
working_directory,
base=working_directory,
skip_dirs={"node_modules", "__pycache__", "venv"},
skip_extensions=(".pyc", ".tmp"),
skip_prefix=".",
)
if generated_files:
context_parts.append("Generated Files from Previous Task:")
for file_path in sorted(generated_files):
context_parts.append(f" - {file_path}")
context_parts.append("")
except Exception as e:
logger.warning(f"Failed to collect generated files: {e}")

Expand Down Expand Up @@ -271,30 +244,21 @@ def build_conversation_context(
context += f"Assistant: {entry['content']}\n\n"

if working_directories:
all_generated_files = set() # Use set to avoid duplicates
all_generated_files: set[str] = set()
for working_directory in working_directories:
try:
if os.path.exists(working_directory):
for root, dirs, files in os.walk(working_directory):
dirs[:] = [
d
for d in dirs
if not d.startswith(".")
and d
not in ["node_modules", "__pycache__", "venv"]
]
for file in files:
if not file.startswith(
"."
) and not file.endswith((".pyc", ".tmp")):
file_path = os.path.join(root, file)
absolute_path = os.path.abspath(file_path)
all_generated_files.add(absolute_path)
files_list = safe_list_directory(
working_directory,
base=working_directory,
skip_dirs={"node_modules", "__pycache__", "venv"},
skip_extensions=(".pyc", ".tmp"),
skip_prefix=".",
)
all_generated_files.update(files_list)
except Exception as e:
logger.warning(
"Failed to collect generated "
f"files from {working_directory}"
f": {e}"
f"files from {working_directory}: {e}"
)

if all_generated_files:
Expand Down
Loading
Loading