85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from typing import Optional, Dict, Any
|
|
import uvicorn
|
|
from callback import call_action
|
|
from fastapi.responses import JSONResponse
|
|
import hashlib
|
|
import json
|
|
import time
|
|
|
|
PASSWORD = "<password>"
|
|
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
# Add CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"], # Allows all origins
|
|
allow_credentials=True,
|
|
allow_methods=["*"], # Allows all methods (GET, POST, etc.)
|
|
allow_headers=["*"], # Allows all headers
|
|
)
|
|
|
|
# Static files & UI server
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|
|
|
|
# Pydantic model for request payload
|
|
class CommandModel(BaseModel):
|
|
args: Dict[str, Any]
|
|
hash: str # Mandatory hash field
|
|
|
|
# Show main page
|
|
@app.get("/")
|
|
async def read_index():
|
|
with open("static/index.html", "r", encoding="utf-8") as f:
|
|
html_content = f.read()
|
|
return HTMLResponse(content=html_content, status_code=200)
|
|
|
|
# Process actions via POST
|
|
@app.post("/action/{name:path}")
|
|
async def handle_action(name: str, payload: CommandModel):
|
|
|
|
action_args = payload.args
|
|
hash_with_ts = payload.hash
|
|
if not hash_with_ts:
|
|
raise HTTPException(status_code=400, detail="Missing hash")
|
|
|
|
# Split hash and timestamp
|
|
try:
|
|
received_hash, received_ts = hash_with_ts.split(".")
|
|
received_ts = int(received_ts)
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="Invalid hash format")
|
|
|
|
# Check if token is within 5-second window
|
|
current_ts = int(time.time())
|
|
if abs(current_ts - received_ts) > 5:
|
|
raise HTTPException(status_code=401, detail="Token expired")
|
|
|
|
# Reconstruct hash
|
|
data_to_hash = PASSWORD + json.dumps(action_args, separators=(",", ":")) + str(received_ts)
|
|
computed_hash = hashlib.sha256(data_to_hash.encode("utf-8")).hexdigest()
|
|
|
|
# Verify hash
|
|
if computed_hash != received_hash:
|
|
raise HTTPException(status_code=401, detail="Invalid hash")
|
|
|
|
try:
|
|
# Вызываем функцию из callback.py, передавая имя действия и аргументы
|
|
result = await call_action(name, action_args or {})
|
|
return JSONResponse(content={"status": "success", "result": result, "command": name})
|
|
except ValueError as e:
|
|
# Если функция не найдена, возвращаем 404
|
|
raise HTTPException(status_code=404, detail=str(e))
|
|
except Exception as e:
|
|
# Обработка остальных ошибок
|
|
raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
|
|
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |