|
from fastapi import FastAPI, HTTPException |
|
from fastapi.middleware.cors import CORSMiddleware |
|
from fastapi.staticfiles import StaticFiles |
|
from fastapi.responses import FileResponse |
|
from pydantic import BaseModel |
|
import subprocess |
|
import shlex |
|
|
|
app = FastAPI() |
|
|
|
|
|
app.add_middleware( |
|
CORSMiddleware, |
|
allow_origins=["*"], |
|
allow_credentials=True, |
|
allow_methods=["*"], |
|
allow_headers=["*"], |
|
) |
|
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static") |
|
|
|
ALLOWED_COMMANDS = { |
|
'ls', 'cd', 'pwd', 'echo', 'cat', 'grep', 'find', 'touch', 'mkdir', 'rm', 'cp', 'mv' |
|
} |
|
|
|
class Command(BaseModel): |
|
command: str |
|
|
|
@app.post("/execute") |
|
async def execute_command(command: Command): |
|
try: |
|
|
|
base_command = shlex.split(command.command)[0] |
|
|
|
|
|
if base_command not in ALLOWED_COMMANDS: |
|
raise HTTPException(status_code=403, detail=f"Command '{base_command}' is not allowed") |
|
|
|
|
|
result = subprocess.run(command.command, shell=True, capture_output=True, text=True, timeout=5) |
|
return { |
|
"output": result.stdout, |
|
"error": result.stderr, |
|
"returncode": result.returncode |
|
} |
|
except subprocess.TimeoutExpired: |
|
raise HTTPException(status_code=408, detail="Command execution timed out") |
|
except Exception as e: |
|
raise HTTPException(status_code=500, detail=str(e)) |
|
|
|
@app.get("/") |
|
async def root(): |
|
return FileResponse('static/index.html') |