diff --git a/servers/filesystem/main.py b/servers/filesystem/main.py index 62facd2..b9a4987 100644 --- a/servers/filesystem/main.py +++ b/servers/filesystem/main.py @@ -12,6 +12,7 @@ import difflib import shutil from datetime import datetime, timezone, timedelta # Added timedelta import uuid # Added uuid +import json # Added json from config import ALLOWED_DIRECTORIES app = FastAPI( @@ -143,10 +144,58 @@ class GetMetadataRequest(BaseModel): # Global state for pending confirmations # ------------------------------------------------------------------------------ -# Store pending confirmations: {token: {"path": str, "recursive": bool, "expiry": datetime}} -pending_confirmations: Dict[str, Dict] = {} +# --- Confirmation Token State Management (using a file) --- +CONFIRMATION_FILE = pathlib.Path("./.pending_confirmations.json") CONFIRMATION_TTL_SECONDS = 60 # Token validity period +def load_confirmations() -> Dict[str, Dict]: + """Loads pending confirmations from the JSON file.""" + if not CONFIRMATION_FILE.exists(): + return {} + try: + with CONFIRMATION_FILE.open("r") as f: + data = json.load(f) + # Convert expiry string back to datetime object + now = datetime.now(timezone.utc) + valid_confirmations = {} + for token, details in data.items(): + try: + details["expiry"] = datetime.fromisoformat(details["expiry"]) + # Clean up expired tokens during load + if details["expiry"] > now: + valid_confirmations[token] = details + except (ValueError, TypeError, KeyError): + print(f"Warning: Skipping invalid confirmation data for token {token}") + continue # Skip invalid entries + return valid_confirmations + except (json.JSONDecodeError, IOError) as e: + print(f"Error loading confirmations file: {e}. Returning empty dict.") + return {} + +def save_confirmations(confirmations: Dict[str, Dict]): + """Saves pending confirmations to the JSON file.""" + try: + # Convert datetime objects to ISO strings for JSON serialization + serializable_confirmations = {} + for token, details in confirmations.items(): + serializable_details = details.copy() + serializable_details["expiry"] = details["expiry"].isoformat() + serializable_confirmations[token] = serializable_details + + with CONFIRMATION_FILE.open("w") as f: + json.dump(serializable_confirmations, f, indent=2) + except IOError as e: + print(f"Error saving confirmations file: {e}") + +# Clean up the file on startup if it exists from a previous run +if CONFIRMATION_FILE.exists(): + # print("Cleaning up stale confirmation file on startup.") + try: + CONFIRMATION_FILE.unlink() + except OSError as e: + # print(f"Warning: Could not delete stale confirmation file: {e}") # Removed print + pass # Silently ignore if cleanup fails, not critical + # ------------------------------------------------------------------------------ # Routes # ------------------------------------------------------------------------------ @@ -202,9 +251,6 @@ async def write_file(data: WriteFileRequest = Body(...)): except Exception as e: raise HTTPException(status_code=500, detail=f"Failed to write to {data.path}: {str(e)}") - -# Removed duplicate import - already added Union at the top - @app.post( "/edit_file", response_model=Union[SuccessResponse, DiffResponse], # Use Union for multiple response types @@ -356,12 +402,15 @@ async def delete_path(data: DeletePathRequest = Body(...)): Use 'recursive=True' to delete non-empty directories. """ + pending_confirmations = load_confirmations() # Load state from file path = normalize_path(data.path) now = datetime.now(timezone.utc) # --- Step 2: Confirmation Request --- if data.confirmation_token: + # print(f"Attempting confirmation with token: {data.confirmation_token}") # Removed print if data.confirmation_token not in pending_confirmations: + # print(f"Error: Token '{data.confirmation_token}' not found in pending_confirmations.") # Removed print raise HTTPException(status_code=400, detail="Invalid or expired confirmation token.") confirmation_data = pending_confirmations[data.confirmation_token] @@ -369,6 +418,7 @@ async def delete_path(data: DeletePathRequest = Body(...)): # Validate token expiry if now > confirmation_data["expiry"]: del pending_confirmations[data.confirmation_token] # Clean up expired token + save_confirmations(pending_confirmations) # Save updated state raise HTTPException(status_code=400, detail="Confirmation token has expired.") # Validate request parameters match @@ -380,6 +430,7 @@ async def delete_path(data: DeletePathRequest = Body(...)): # --- Parameters match and token is valid: Proceed with deletion --- del pending_confirmations[data.confirmation_token] # Consume the token + save_confirmations(pending_confirmations) # Save updated state try: if not path.exists(): @@ -427,6 +478,7 @@ async def delete_path(data: DeletePathRequest = Body(...)): "recursive": data.recursive, "expiry": expiry_time, } + save_confirmations(pending_confirmations) # Save updated state # Return confirmation required response return ConfirmationRequiredResponse( @@ -444,13 +496,13 @@ async def move_path(data: MovePathRequest = Body(...)): Returns JSON success message. """ source = normalize_path(data.source_path) - destination = normalize_path(data.destination_path) # Also normalize destination + destination = normalize_path(data.destination_path) try: if not source.exists(): raise HTTPException(status_code=404, detail=f"Source path not found: {data.source_path}") - shutil.move(str(source), str(destination)) # Raises FileNotFoundError, PermissionError etc. + shutil.move(str(source), str(destination)) return SuccessResponse(message=f"Successfully moved '{data.source_path}' to '{data.destination_path}'") except PermissionError: