Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
13 changes: 9 additions & 4 deletions autobot-backend/api/chat_knowledge.py
Original file line number Diff line number Diff line change
Expand Up @@ -600,10 +600,15 @@ async def upload_file_to_chat(
):
"""Upload a file and associate it with a chat"""
try:
# Save uploaded file
file_path = os.path.join(
chat_knowledge_manager.storage_dir,
f"{chat_id}_{os.path.basename(file.filename)}",
# Save uploaded file (#1721)
from autobot_shared.security.path_validator import validate_relative_path

safe_name = f"{chat_id}_{os.path.basename(file.filename)}"
file_path = str(
validate_relative_path(
safe_name,
chat_knowledge_manager.storage_dir,
)
)

content = await file.read()
Expand Down
7 changes: 6 additions & 1 deletion autobot-backend/api/chat_sessions.py
Original file line number Diff line number Diff line change
Expand Up @@ -1196,7 +1196,12 @@ async def _cleanup_conversation_transcript(session_id: str) -> dict:
try:
if "/" in session_id or "\\" in session_id or ".." in session_id:
raise ValueError("Invalid session ID")
transcript_path = Path("data/conversation_transcripts") / f"{session_id}.json"
from autobot_shared.security.path_validator import validate_relative_path

transcript_path = validate_relative_path(
f"{session_id}.json",
Path("data/conversation_transcripts"),
)

if transcript_path.exists():
os.remove(transcript_path)
Expand Down
25 changes: 7 additions & 18 deletions autobot-backend/api/codebase_analytics/endpoints/environment.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,35 +45,24 @@ def _validate_env_path_security(path: str, project_root: str) -> Optional[JSONRe
Returns:
JSONResponse error if validation fails, None if valid
"""
from autobot_shared.security.path_validator import validate_path

try:
resolved_path = Path(path).resolve()
if not str(resolved_path).startswith(project_root):
logger.warning("Path traversal attempt blocked: %s", path)
return JSONResponse(
{
"status": "error",
"message": "Invalid path: must be within project root",
"total_hardcoded_values": 0,
"categories": {},
"recommendations_count": 0,
},
status_code=400,
)
except Exception as e:
logger.warning("Invalid path provided: %s - %s", path, e)
validate_path(path, allowed_roots=[project_root])
return None
except ValueError:
logger.warning("Path traversal attempt blocked: %s", path)
return JSONResponse(
{
"status": "error",
"message": "Invalid path",
"message": "Invalid path: must be within project root",
"total_hardcoded_values": 0,
"categories": {},
"recommendations_count": 0,
},
status_code=400,
)

return None


async def _run_environment_analysis(analyzer, path: str, pattern_list: list):
"""
Expand Down
35 changes: 14 additions & 21 deletions autobot-backend/api/codebase_analytics/endpoints/ownership.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,39 +85,32 @@ def _validate_path_security(path: str, project_root: str) -> Optional[JSONRespon
"""
Validate that path is within project root.

Uses shared path validator (#1721).

Returns:
JSONResponse error if validation fails, None if valid
"""
from autobot_shared.security.path_validator import validate_path

try:
resolved_path = Path(path).resolve()
if not str(resolved_path).startswith(project_root):
logger.warning("Path traversal attempt blocked: %s", path)
return JSONResponse(
{
"status": "error",
"message": "Invalid path: must be within project root",
"summary": {},
"file_ownership": [],
"directory_ownership": [],
"expertise_scores": [],
"knowledge_gaps": [],
"metrics": {},
},
status_code=400,
)
except Exception as e:
logger.warning("Invalid path provided: %s - %s", path, e)
validate_path(path, allowed_roots=[project_root])
return None
except ValueError:
logger.warning("Path traversal attempt blocked: %s", path)
return JSONResponse(
{
"status": "error",
"message": "Invalid path",
"message": "Invalid path: must be within project root",
"summary": {},
"file_ownership": [],
"directory_ownership": [],
"expertise_scores": [],
"knowledge_gaps": [],
"metrics": {},
},
status_code=400,
)

return None


async def _run_ownership_analysis(analyzer, path: str, pattern_list: list, days: int):
"""
Expand Down
20 changes: 12 additions & 8 deletions autobot-backend/api/data_storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
from utils.catalog_http_exceptions import raise_server_error

from autobot_shared.error_boundaries import ErrorCategory, with_error_handling
from autobot_shared.security.path_validator import validate_relative_path

router = APIRouter(prefix="/data-storage", tags=["Data Storage"])
logger = logging.getLogger(__name__)
Expand All @@ -31,12 +32,14 @@


def _safe_data_path(user_segment: str) -> Path:
"""Resolve a user-supplied path segment under DATA_DIR safely (#1733)."""
resolved = (DATA_DIR / user_segment).resolve()
data_root = DATA_DIR.resolve()
if not str(resolved).startswith(str(data_root) + "/") and resolved != data_root:
"""Resolve a user-supplied path segment under DATA_DIR safely.

Uses shared path validator (#1721).
"""
try:
return validate_relative_path(user_segment, DATA_DIR)
except ValueError:
raise HTTPException(status_code=400, detail="Invalid path")
return resolved


class StorageCategory(BaseModel):
Expand Down Expand Up @@ -557,8 +560,9 @@ async def delete_conversation(
files_deleted = []
errors = []

transcript_path = (
DATA_DIR / "conversation_transcripts" / f"{conversation_id}.json"
transcripts_base = DATA_DIR / "conversation_transcripts"
transcript_path = validate_relative_path(
f"{conversation_id}.json", transcripts_base
)
if transcript_path.exists():
try:
Expand All @@ -569,7 +573,7 @@ async def delete_conversation(

chats_dir = DATA_DIR / "chats"
for suffix in ["_chat.json", "_terminal_transcript.txt"]:
chat_file = chats_dir / f"{conversation_id}{suffix}"
chat_file = validate_relative_path(f"{conversation_id}{suffix}", chats_dir)
if chat_file.exists():
try:
chat_file.unlink()
Expand Down
40 changes: 16 additions & 24 deletions autobot-backend/api/files.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
from utils.paths_manager import ensure_data_directory, get_data_path

from autobot_shared.error_boundaries import ErrorCategory, with_error_handling
from autobot_shared.security.path_validator import validate_relative_path

router = APIRouter()
logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -291,19 +292,14 @@ def validate_and_resolve_path(path: str) -> Path:
status_code=400, detail="Invalid path: encoded traversal not allowed"
)

# Resolve the full path within sandbox
full_path = SANDBOXED_ROOT / clean_path

# Use realpath for canonical path resolution (prevents symlink attacks)
# Validate via shared security module (#1721)
try:
canonical_full = full_path.resolve(strict=False)
canonical_sandbox = SANDBOXED_ROOT.resolve()
canonical_full.relative_to(canonical_sandbox)
return validate_relative_path(clean_path, SANDBOXED_ROOT)
except ValueError:
raise HTTPException(status_code=400, detail="Path outside sandbox not allowed")

# Return the canonical (absolute) path to prevent CWD-related issues
return canonical_full
raise HTTPException(
status_code=400,
detail="Path outside sandbox not allowed",
)


def get_file_info(file_path: Path, relative_path: str) -> FileInfo:
Expand Down Expand Up @@ -1304,21 +1300,17 @@ def _validate_admin_path(path: str) -> Path:
"""Resolve and validate an absolute path for the SLM admin file browser.

Restricts access to pre-approved base directories (Issue #984).
Uses shared path validator (#1721).
"""
from autobot_shared.security.path_validator import validate_path

try:
resolved = Path(path).resolve()
except Exception:
raise HTTPException(status_code=400, detail="Invalid path")
for allowed in _ADMIN_ALLOWED_DIRS:
try:
resolved.relative_to(allowed)
return resolved
except ValueError:
continue
raise HTTPException(
status_code=403,
detail=f"Path outside allowed directories: {path}",
)
return validate_path(path, allowed_roots=list(_ADMIN_ALLOWED_DIRS))
except ValueError:
raise HTTPException(
status_code=403,
detail=f"Path outside allowed directories: {path}",
)


def _entry_to_file_item(entry: Path) -> dict:
Expand Down
Loading
Loading