import os
import pty
import asyncio
import re
import secrets
import shutil
from pydantic import BaseModel
# 引入 Cookie, Form 和 RedirectResponse
from fastapi import FastAPI, WebSocket, UploadFile, Request, Response, Depends, HTTPException, status, Form, Cookie
from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse
import uvicorn
app = FastAPI()
PASSWORD = os.getenv("PASSWORD", "123456")
# ==========================================
# 权限校验 (改用 Cookie 校验，避开 Header 冲突)
# ==========================================
def verify_auth(auth_token: str = Cookie(default=None)):
if not auth_token or not secrets.compare_digest(auth_token, PASSWORD):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="无权访问，请先登录")
return auth_token
# ==========================================
# 自定义暗黑风登录页面
# ==========================================
LOGIN_PAGE = """
Admin Login
"""
# ==========================================
# 前端 HTML (内置三栏布局、CodeMirror、文件管理器)
# ==========================================
HTML_PAGE = """
HF Web Shell (Pro)
"""
# ==========================================
# 公共路由：复盘报告服务器相关配置
# ==========================================
MEMORY_CACHE = {}
INDEX_FILE = "/app/index.html"
REPORT_DIR = "/app/reviewreports"
os.makedirs(REPORT_DIR, exist_ok=True)
@app.get("/")
async def index(request: Request):
host = request.headers.get("x-proxy-host") or request.headers.get("host", "")
if "online.4z.autos" in host:
if 'index' not in MEMORY_CACHE:
if os.path.exists(INDEX_FILE):
with open(INDEX_FILE, 'rb') as f:
MEMORY_CACHE['index'] = f.read()
else:
return Response("Index File Not Found", status_code=404)
return Response(content=MEMORY_CACHE['index'], media_type="text/html", headers={'Cache-Control': 'public, max-age=3600'})
return Response("Access Denied", status_code=403)
@app.get("/report/{filename}")
async def get_report(filename: str):
if filename.endswith(".json"): filename = filename[:-5]
if not re.match(r'^[a-zA-Z0-9]+$', filename): return Response("Bad Request", status_code=400)
filepath = os.path.join(REPORT_DIR, f"{filename}.json")
if not os.path.exists(filepath): return Response("File Not Found", status_code=404)
return FileResponse(filepath, media_type='application/json', headers={'Access-Control-Allow-Origin': '*','Cache-Control': 'public, max-age=31536000, immutable'})
# ==========================================
# 私密路由：后台登录与验证体系
# ==========================================
@app.get("/admin")
async def admin_shell(request: Request):
# 检查 Cookie 是否有效
token = request.cookies.get("auth_token")
if token and secrets.compare_digest(token, PASSWORD):
return HTMLResponse(HTML_PAGE)
# 无效则展示我们自己写的登录页面
return HTMLResponse(LOGIN_PAGE)
@app.post("/login")
async def do_login(password: str = Form(...)):
# 校验表单密码
if secrets.compare_digest(password, PASSWORD):
# 密码正确，重定向回 /admin，并种下有效时间为 1 天的 Cookie
response = RedirectResponse(url="/admin", status_code=302)
response.set_cookie(key="auth_token", value=PASSWORD, httponly=True, max_age=86400)
return response
# 密码错误
return HTMLResponse("密码错误，请返回重试
", status_code=401)
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
# WebSocket 拦截校验 Cookie
token = websocket.cookies.get("auth_token")
if not token or not secrets.compare_digest(token, PASSWORD):
await websocket.close(code=1008)
return
await websocket.accept()
pid, fd = pty.fork()
if pid == 0:
os.environ["TERM"] = "xterm-256color"
os.chdir("/app")
os.execvp("bash", ["bash"])
else:
loop = asyncio.get_running_loop()
async def read_pty():
while True:
try:
data = await loop.run_in_executor(None, os.read, fd, 1024)
if not data: break
await websocket.send_text(data.decode('utf-8', 'replace'))
except Exception:
break
async def read_ws():
while True:
try:
data = await websocket.receive_text()
os.write(fd, data.encode('utf-8'))
except Exception:
break
await asyncio.gather(read_pty(), read_ws())
# --- 以下所有 API 接口均受到 Depends(verify_auth) 保护 ---
class FsAction(BaseModel):
action: str
src: str
dest: str
class FileContent(BaseModel):
path: str
content: str
@app.get("/api/fs/list")
def fs_list(path: str = "/app", username: str = Depends(verify_auth)):
try:
items = []
for f in os.scandir(path):
items.append({"name": f.name, "is_dir": f.is_dir(), "size": f.stat().st_size})
items.sort(key=lambda x: (not x['is_dir'], x['name'].lower()))
return {"status": "ok", "path": path, "items": items}
except Exception as e:
return {"status": "error", "msg": str(e)}
@app.post("/api/fs/action")
def fs_action(req: FsAction, username: str = Depends(verify_auth)):
try:
if req.action == 'delete':
if os.path.isdir(req.src): shutil.rmtree(req.src)
else: os.remove(req.src)
elif req.action == 'copy':
if os.path.isdir(req.src): shutil.copytree(req.src, req.dest)
else: shutil.copy2(req.src, req.dest)
return {"status": "ok"}
except Exception as e:
return {"status": "error", "msg": str(e)}
@app.post("/upload")
async def upload(path: str, files: list[UploadFile], username: str = Depends(verify_auth)):
for file in files:
file_path = os.path.join(path, file.filename)
with open(file_path, "wb") as f:
f.write(await file.read())
return {"status": "ok"}
@app.get("/download")
def download(path: str, username: str = Depends(verify_auth)):
if os.path.exists(path): return FileResponse(path)
return {"error": "File not found"}
@app.get("/api/fs/read")
def fs_read(path: str, username: str = Depends(verify_auth)):
try:
with open(path, 'r', encoding='utf-8') as f:
return {"status": "ok", "content": f.read()}
except UnicodeDecodeError:
return {"status": "error", "msg": "不是文本文件，无法编辑"}
except Exception as e:
return {"status": "error", "msg": str(e)}
@app.post("/api/fs/write")
def fs_write(req: FileContent, username: str = Depends(verify_auth)):
try:
with open(req.path, 'w', encoding='utf-8') as f:
f.write(req.content)
return {"status": "ok"}
except Exception as e:
return {"status": "error", "msg": str(e)}