from fastapi import FastAPI, Request, UploadFile, File, Form, HTTPException from fastapi.responses import HTMLResponse, FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates import os import subprocess import datetime import uuid from pathlib import Path import logging import shutil # Setup logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger("file-test-app") app = FastAPI(title="File System Test App") # Setup templates and static files templates = Jinja2Templates(directory="templates") # Create files directory if it doesn't exist os.makedirs("templates", exist_ok=True) os.makedirs("files", exist_ok=True) os.makedirs("static", exist_ok=True) # Mount files directory as static app.mount("/files", StaticFiles(directory="files"), name="files") app.mount("/static", StaticFiles(directory="static"), name="static") @app.get("/", response_class=HTMLResponse) async def home(request: Request): """Home page""" # Log system information try: whoami = subprocess.run(["whoami"], capture_output=True, text=True).stdout.strip() file_perms = subprocess.run(["ls", "-la", "/app/files"], capture_output=True, text=True).stdout disk_info = subprocess.run(["df", "-h"], capture_output=True, text=True).stdout logger.info(f"Running as user: {whoami}") logger.info(f"Files directory permissions:\n{file_perms}") logger.info(f"Disk space:\n{disk_info}") except Exception as e: logger.error(f"Error getting system info: {str(e)}") return templates.TemplateResponse("index.html", {"request": request}) @app.post("/save-file") async def save_file(content: str = Form(...), filename: str = Form(...)): """Save content to a file""" try: # Create a safe filename safe_filename = filename.replace(" ", "_") if not safe_filename.endswith(".txt"): safe_filename += ".txt" file_path = os.path.join("files", safe_filename) # Write the file with open(file_path, "w") as f: f.write(content) # Get file size file_size = os.path.getsize(file_path) return JSONResponse({ "success": True, "message": "File saved successfully", "file_path": file_path, "file_size": file_size, "url": f"/files/{safe_filename}" }) except Exception as e: logger.error(f"Error saving file: {str(e)}") return JSONResponse({ "success": False, "message": f"Error saving file: {str(e)}" }, status_code=500) @app.post("/upload-file") async def upload_file(file: UploadFile = File(...)): """Upload a file""" try: # Create a safe filename safe_filename = file.filename.replace(" ", "_") file_path = os.path.join("files", safe_filename) # Save the file with open(file_path, "wb") as f: shutil.copyfileobj(file.file, f) # Get file size file_size = os.path.getsize(file_path) return JSONResponse({ "success": True, "message": "File uploaded successfully", "file_path": file_path, "file_size": file_size, "url": f"/files/{safe_filename}" }) except Exception as e: logger.error(f"Error uploading file: {str(e)}") return JSONResponse({ "success": False, "message": f"Error uploading file: {str(e)}" }, status_code=500) @app.get("/list-files") async def list_files(): """List all files in the files directory""" try: files_dir = Path("files") files = [] for f in files_dir.glob("*"): if f.is_file(): files.append({ "name": f.name, "path": str(f), "size": os.path.getsize(f), "size_readable": f"{os.path.getsize(f) / 1024:.2f} KB", "modified": datetime.datetime.fromtimestamp(os.path.getmtime(f)).strftime("%Y-%m-%d %H:%M:%S"), "url": f"/files/{f.name}" }) return JSONResponse({ "success": True, "files": files }) except Exception as e: logger.error(f"Error listing files: {str(e)}") return JSONResponse({ "success": False, "message": f"Error listing files: {str(e)}" }, status_code=500) @app.delete("/delete-file/{filename}") async def delete_file(filename: str): """Delete a file""" try: file_path = os.path.join("files", filename) if not os.path.exists(file_path): return JSONResponse({ "success": False, "message": "File not found" }, status_code=404) os.remove(file_path) return JSONResponse({ "success": True, "message": "File deleted successfully" }) except Exception as e: logger.error(f"Error deleting file: {str(e)}") return JSONResponse({ "success": False, "message": f"Error deleting file: {str(e)}" }, status_code=500) @app.get("/system-info") async def system_info(): """Get system information""" try: whoami = subprocess.run(["whoami"], capture_output=True, text=True).stdout.strip() file_perms = subprocess.run(["ls", "-la", "/app/files"], capture_output=True, text=True).stdout disk_info = subprocess.run(["df", "-h"], capture_output=True, text=True).stdout env_vars = {k: v for k, v in os.environ.items()} # Try to create a large test file test_file_path = os.path.join("files", "large_test_file.txt") try: with open(test_file_path, "w") as f: # Write a 10MB file with repeated text text_block = "This is a test line to check file system write capabilities.\n" * 100 for _ in range(1000): # Write 10,000 lines f.write(text_block) large_file_size = os.path.getsize(test_file_path) large_file_status = f"Created large test file: {large_file_size / (1024*1024):.2f} MB" except Exception as e: large_file_status = f"Failed to create large test file: {str(e)}" return JSONResponse({ "success": True, "whoami": whoami, "file_permissions": file_perms, "disk_info": disk_info, "large_file_test": large_file_status, "env_vars": env_vars, "python_version": subprocess.run(["python", "--version"], capture_output=True, text=True).stdout.strip(), "pwd": subprocess.run(["pwd"], capture_output=True, text=True).stdout.strip(), "ls_app": subprocess.run(["ls", "-la", "/app"], capture_output=True, text=True).stdout }) except Exception as e: logger.error(f"Error getting system info: {str(e)}") return JSONResponse({ "success": False, "message": f"Error getting system info: {str(e)}" }, status_code=500) @app.get("/test-binary-file") async def test_binary_file(): """Create a test binary file""" try: file_path = os.path.join("files", "binary_test.bin") # Create binary data binary_data = bytearray(os.urandom(1024 * 1024)) # 1MB of random data # Write binary file with open(file_path, "wb") as f: f.write(binary_data) file_size = os.path.getsize(file_path) return JSONResponse({ "success": True, "message": "Binary file created successfully", "file_path": file_path, "file_size": file_size, "url": f"/files/binary_test.bin" }) except Exception as e: logger.error(f"Error creating binary file: {str(e)}") return JSONResponse({ "success": False, "message": f"Error creating binary file: {str(e)}" }, status_code=500) # Testing command to simulate pg_dump @app.get("/test-command-output") async def test_command_output(): """Test running a command and saving its output""" try: output_file = os.path.join("files", "command_output.txt") # Run a command and redirect output to a file cmd = ["ls", "-la", "/"] # Method 1: Direct redirection with open(output_file, "w") as f: subprocess.run(cmd, stdout=f, stderr=subprocess.PIPE, text=True) # Method 2: Capture then write output_file2 = os.path.join("files", "command_output2.txt") result = subprocess.run(cmd, capture_output=True, text=True) with open(output_file2, "w") as f: f.write(result.stdout) return JSONResponse({ "success": True, "message": "Command output saved", "file1_path": output_file, "file1_size": os.path.getsize(output_file), "file1_url": f"/files/command_output.txt", "file2_path": output_file2, "file2_size": os.path.getsize(output_file2), "file2_url": f"/files/command_output2.txt", }) except Exception as e: logger.error(f"Error running command: {str(e)}") return JSONResponse({ "success": False, "message": f"Error running command: {str(e)}" }, status_code=500)