from __future__ import annotations from pathlib import Path from langchain_core.tools import tool from app.utils.logger import get_logger logger = get_logger(__name__) ALLOWED_ROOT = Path.home() / "Desktop" / "AI-workingdir" def _resolve_safe_path(path: str) -> Path | None: try: ALLOWED_ROOT.mkdir(parents=True, exist_ok=True) target = (ALLOWED_ROOT / path).resolve() if not target.exists(): return None target2 = target.resolve() if target != target2 or (ALLOWED_ROOT.resolve() not in target2.parents and ALLOWED_ROOT.resolve() != target2): logger.warning("Path traversal detected: %s", path) return None return target2 except Exception: return None @tool("read_file") def read_file(path: str) -> str: """Read the contents of a file from the sandboxed working directory.""" target = _resolve_safe_path(path) if not target: return f"Access denied. Can only read from {ALLOWED_ROOT}" if not target.exists(): return "File not found." try: content = target.read_text(encoding="utf-8", errors="ignore") if len(content) > 15000: content = content[:15000] + "\n... (truncated)" return content except Exception: logger.exception("File read failed: %s", path) return "Read failed. Unable to read the file." @tool("list_files") def list_files(path: str = ".") -> str: """List files and directories in the sandboxed working directory.""" target = _resolve_safe_path(path) if not target: return f"Access denied. Can only list {ALLOWED_ROOT}" if not target.exists(): return "Path not found." if not target.is_dir(): return "Path is not a directory." try: items = [] for item in sorted(target.iterdir()): suffix = "/" if item.is_dir() else "" items.append(f"{item.name}{suffix}") if not items: return "(empty)" return "\n".join(items) except Exception: logger.exception("File list failed: %s", path) return "List failed. Unable to list the directory."