AdamWen95
298 lines
14 KiB
Python
298 lines
14 KiB
Python
"""LAN-only HTTP interface; does not spawn processes or access Docker."""
|
|
import hashlib
|
|
import ipaddress
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
from urllib.parse import urlsplit
|
|
import uuid
|
|
from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFile
|
|
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from . import catalog
|
|
from . import test_catalog
|
|
from . import native_plans, native_io
|
|
from .settings import Settings
|
|
from .store import Store, TERMINAL, now
|
|
|
|
|
|
class ConfigPatch(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
args: list[str] | None = None
|
|
cwd: str | None = None
|
|
output_paths: list[str] | None = None
|
|
timeout_seconds: Annotated[int, Field(strict=True)] | None = None
|
|
native_options: dict[str, str | int | float | bool] | None = None
|
|
|
|
|
|
class ScriptPatch(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
name: str | None = Field(default=None, min_length=1, max_length=120)
|
|
description: str | None = Field(default=None, max_length=4000)
|
|
enabled: bool | None = None
|
|
config: ConfigPatch | None = None
|
|
|
|
|
|
class RunRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
request_id: uuid.UUID
|
|
plan_token: str | None = Field(default=None, max_length=64)
|
|
impact_acknowledged: bool = False
|
|
|
|
|
|
class NativeInput(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
text: str = Field(max_length=1024)
|
|
|
|
|
|
def create_app(settings=None):
|
|
settings = settings or Settings.from_env()
|
|
store = Store(settings)
|
|
app = FastAPI(title="XAS 测试控制台", docs_url=None, redoc_url=None, openapi_url=None)
|
|
app.state.store = store
|
|
static = Path(__file__).parent / "static"
|
|
|
|
@app.middleware("http")
|
|
async def boundaries(request, call_next):
|
|
host = request.headers.get("host", "")
|
|
try:
|
|
address = ipaddress.ip_address(request.client.host)
|
|
allowed = any(address in ipaddress.ip_network(n) for n in settings.allowed_networks)
|
|
except ValueError:
|
|
allowed = False
|
|
if host not in settings.allowed_hosts or not allowed:
|
|
return JSONResponse({"detail": "仅允许公司内网访问"}, status_code=403)
|
|
if request.method not in {"GET", "HEAD", "OPTIONS"}:
|
|
origin = request.headers.get("origin")
|
|
valid_origin = not origin or origin == f"{request.url.scheme}://{host}"
|
|
if (not valid_origin or request.headers.get("x-xas-console") != "1"
|
|
or request.headers.get("sec-fetch-site") == "cross-site"):
|
|
return JSONResponse({"detail": "请求来源校验失败，请从控制台页面操作"}, status_code=403)
|
|
try:
|
|
size = int(request.headers.get("content-length", "0"))
|
|
except ValueError:
|
|
size = -1
|
|
if size < 0 or size > settings.max_upload + 65536 or "transfer-encoding" in request.headers:
|
|
return JSONResponse({"detail": "请求超过大小限制"}, status_code=413)
|
|
response = await call_next(request)
|
|
response.headers["Cache-Control"] = "no-store"
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["Referrer-Policy"] = "same-origin"
|
|
response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
|
|
return response
|
|
|
|
@app.exception_handler(ValueError)
|
|
async def value_error(request, exc):
|
|
return JSONResponse({"detail": str(exc)}, status_code=400)
|
|
|
|
@app.exception_handler(KeyError)
|
|
async def missing(request, exc):
|
|
return JSONResponse({"detail": str(exc.args[0])}, status_code=404)
|
|
|
|
@app.get("/api/summary")
|
|
def summary():
|
|
inventory, runs = store.scripts(), store.runs(limit=10000)
|
|
scripts = test_catalog.tests(settings, inventory)
|
|
# No directory recursion on every page refresh; size derives from recorded artifacts.
|
|
size = sum(a["size"] for r in runs for a in r.get("artifacts", []))
|
|
return {"scripts": len(scripts), "source_files": len(inventory),
|
|
"examples": len(test_catalog.tests(settings, inventory, True)) - len(scripts),
|
|
"ready": sum(s["enabled"] for s in scripts), "runs": len(runs),
|
|
"active": sum(r["state"] in {"queued", "running"} for r in runs), "worker": store.health(),
|
|
"storage": {"run_bytes": size}, "limits": {"upload_bytes": settings.max_upload,
|
|
"log_bytes": settings.max_log, "concurrency": 1, "timeout_seconds": 7200}}
|
|
|
|
@app.get("/api/scripts")
|
|
def scripts(include_examples: bool = False):
|
|
return {"items": test_catalog.tests(settings, store.scripts(), include_examples)}
|
|
|
|
@app.post("/api/scripts/upload")
|
|
async def new_script(file: UploadFile, name: Annotated[str, Form()] = "", description: Annotated[str, Form()] = ""):
|
|
try:
|
|
item = catalog.upload(store, await file.read(settings.max_upload + 1), file.filename, name, description)
|
|
return store.script(item["id"], versions=True)
|
|
finally:
|
|
await file.close()
|
|
|
|
@app.get("/api/scripts/{identity}")
|
|
def script(identity: str):
|
|
return test_catalog.describe(settings, store.script(identity, versions=True), store.scripts())
|
|
|
|
@app.get("/api/scripts/{identity}/files")
|
|
def script_files(identity: str):
|
|
return {"items": test_catalog.members(settings, store.script(identity), store.scripts())}
|
|
|
|
@app.post("/api/scripts/{identity}/versions")
|
|
async def new_version(identity: str, file: UploadFile):
|
|
try:
|
|
catalog.upload(store, await file.read(settings.max_upload + 1), file.filename, identity=identity)
|
|
return store.script(identity, versions=True)
|
|
finally:
|
|
await file.close()
|
|
|
|
def source_version(identity, version):
|
|
item = store.script(identity)
|
|
if item["kind"] == "managed" and version:
|
|
v = store.version(identity, version)
|
|
item.update(source_path=v["path"], sha256=v["sha256"])
|
|
elif version and version != item["version"]:
|
|
raise ValueError("原位置脚本仅展示当前文件，历史运行请查看运行归档")
|
|
return item
|
|
|
|
@app.get("/api/scripts/{identity}/code")
|
|
def code(identity: str, version: int | None = None):
|
|
return catalog.read_code(settings, source_version(identity, version))
|
|
|
|
@app.get("/api/scripts/{identity}/download")
|
|
def download_source(identity: str, version: int | None = None):
|
|
item = source_version(identity, version)
|
|
source = catalog.read_code(settings, item)
|
|
if source["redacted"]:
|
|
raise ValueError("包含疑似凭据的原文件不能下载")
|
|
return FileResponse(catalog.script_path(settings, item), filename=source["filename"], media_type="text/plain")
|
|
|
|
@app.patch("/api/scripts/{identity}")
|
|
def update_script(identity: str, patch: ScriptPatch):
|
|
# Serialize with version uploads to avoid restoring an obsolete current-version pointer.
|
|
import json
|
|
with store.connect(write=True) as db:
|
|
row = db.execute("SELECT data FROM scripts WHERE id=?", (identity,)).fetchone()
|
|
if not row:
|
|
raise KeyError("脚本不存在")
|
|
item = json.loads(row[0])
|
|
if not test_catalog.is_entry(settings, item):
|
|
raise ValueError("此文件属于内部源码，请从测试项目入口配置和运行")
|
|
values = patch.model_dump(exclude_none=True)
|
|
if "config" in values:
|
|
config_patch = values.pop("config")
|
|
if native_plans.profile_for(settings, item) and set(config_patch) - {"native_options", "timeout_seconds"}:
|
|
raise ValueError("原环境入口、目录和输出规则由已登记方案管理")
|
|
item["config"] = catalog.validate_config(settings, item, {**item["config"], **config_patch})
|
|
if values.get("enabled"):
|
|
native = native_plans.profile_for(settings, item)
|
|
blocked = not native.get("available", True) if native else item.get("blocked")
|
|
if blocked or item["status"] == "missing":
|
|
raise ValueError(item["reason"] or "此脚本暂不能启用")
|
|
catalog.script_path(settings, item)
|
|
item["config"] = catalog.validate_config(settings, item, item["config"])
|
|
item.update(values, updated_at=now())
|
|
item["status"] = "ready" if item["enabled"] else "needs_configuration"
|
|
item["reason"] = "" if item["enabled"] else (item.get("reason") or "已停用")
|
|
db.execute("UPDATE scripts SET data=? WHERE id=?", (json.dumps(item), identity))
|
|
return store.script(identity, versions=True)
|
|
|
|
@app.post("/api/discover")
|
|
def discover():
|
|
result = catalog.discover(store)
|
|
result["tests"] = len(test_catalog.tests(settings, store.scripts()))
|
|
return result
|
|
|
|
@app.post("/api/scripts/{identity}/runs")
|
|
def run_script(identity: str, body: RunRequest, request: Request):
|
|
script = test_catalog.describe(settings, store.script(identity), store.scripts())
|
|
if not script["is_test"]:
|
|
raise ValueError("此文件属于内部源码，不能作为独立测试运行")
|
|
if not script["enabled"] or script["status"] != "ready":
|
|
raise ValueError(script["reason"] or "请先配置并启用脚本")
|
|
if native_plans.profile_for(settings, script):
|
|
return native_io.request(settings, {"op": "queue", "script_id": identity,
|
|
"request_id": str(body.request_id), "plan_token": body.plan_token,
|
|
"impact_acknowledged": body.impact_acknowledged, "source_ip": request.client.host})
|
|
source = catalog.script_path(settings, script).read_bytes()
|
|
if hashlib.sha256(source).hexdigest() != script["sha256"]:
|
|
raise ValueError("原脚本已变动，请刷新目录并重新确认配置")
|
|
if catalog.SECRET.search(catalog.clean_source(source)):
|
|
raise ValueError("源文件包含疑似凭据，暂不执行")
|
|
return store.queue(script, str(body.request_id), request.client.host)
|
|
|
|
@app.post("/api/scripts/{identity}/preflight")
|
|
def preflight(identity: str):
|
|
item = store.script(identity)
|
|
if not native_plans.profile_for(settings, item):
|
|
raise ValueError("此项目使用隔离容器，无原环境启动方案")
|
|
return native_io.request(settings, {"op": "preflight", "script_id": identity})
|
|
|
|
@app.post("/api/runs/{identity}/input")
|
|
def native_input(identity: str, body: NativeInput):
|
|
return native_io.request(settings, {"op": "input", "run_id": identity, "text": body.text})
|
|
|
|
@app.get("/api/runs")
|
|
def runs(script_id: str | None = None, limit: int = Query(default=100, ge=1, le=500)):
|
|
return {"items": store.runs(script_id, limit)}
|
|
|
|
@app.get("/api/runs/{identity}")
|
|
def run(identity: str):
|
|
return store.run(identity)
|
|
|
|
@app.post("/api/runs/{identity}/cancel")
|
|
def cancel(identity: str):
|
|
return store.cancel(identity)
|
|
|
|
def run_path(identity, filename):
|
|
store.run(identity)
|
|
root = catalog.safe_path(settings.runs / identity, settings.runs, must_exist=False)
|
|
return catalog.safe_path(root / filename, root, must_exist=False)
|
|
|
|
@app.get("/api/runs/{identity}/log")
|
|
def log(identity: str, offset: int = Query(default=0, ge=0)):
|
|
item = store.run(identity)
|
|
path = run_path(identity, "console.log")
|
|
data = b""
|
|
at_eof = True
|
|
if path.exists():
|
|
with path.open("rb") as handle:
|
|
handle.seek(min(offset, path.stat().st_size))
|
|
offset = handle.tell()
|
|
data = handle.read(65536)
|
|
at_eof = offset + len(data) >= path.stat().st_size
|
|
# Keep any incomplete UTF-8 tail for the next poll.
|
|
import codecs
|
|
decoder = codecs.getincrementaldecoder("utf-8")("replace")
|
|
text = decoder.decode(data, final=item["state"] in TERMINAL and at_eof)
|
|
pending = len(decoder.getstate()[0])
|
|
else:
|
|
text, pending = "", 0
|
|
return {"text": text, "next_offset": offset + len(data) - pending,
|
|
"done": item["state"] in TERMINAL and len(data) < 65536,
|
|
"truncated": path.exists() and path.stat().st_size >= settings.max_log}
|
|
|
|
@app.get("/api/runs/{identity}/log/download")
|
|
def log_download(identity: str):
|
|
path = run_path(identity, "console.log")
|
|
if not path.is_file():
|
|
raise KeyError("日志尚未生成")
|
|
return FileResponse(path, filename=f"{identity}.log", media_type="text/plain")
|
|
|
|
def artifact_path(identity, artifact_id):
|
|
item = store.run(identity)
|
|
artifact = next((a for a in item.get("artifacts", []) if a["id"] == artifact_id), None)
|
|
if not artifact:
|
|
raise KeyError("结果文件不存在")
|
|
return run_path(identity, "artifacts/" + artifact["id"]), artifact
|
|
|
|
@app.get("/api/runs/{identity}/artifacts/{artifact_id}/preview")
|
|
def artifact_preview(identity: str, artifact_id: str):
|
|
path, artifact = artifact_path(identity, artifact_id)
|
|
if artifact["kind"] not in {"text", "json"}:
|
|
raise ValueError("此文件只支持下载")
|
|
with path.open("rb") as handle:
|
|
data = handle.read(262145)
|
|
return {"text": data[:262144].decode("utf-8", errors="replace"), "truncated": len(data) > 262144}
|
|
|
|
@app.get("/api/runs/{identity}/artifacts/{artifact_id}")
|
|
def artifact_download(identity: str, artifact_id: str):
|
|
path, artifact = artifact_path(identity, artifact_id)
|
|
return FileResponse(path, filename=artifact["name"], media_type="application/octet-stream")
|
|
|
|
@app.get("/")
|
|
def index():
|
|
return FileResponse(static / "index.html")
|
|
|
|
@app.get("/project/xas-test-library/{legacy_path:path}")
|
|
def old_bookmark(legacy_path: str):
|
|
return RedirectResponse("/#/scripts", status_code=307)
|
|
|
|
if static.exists():
|
|
app.mount("/static", StaticFiles(directory=static), name="static")
|
|
return app
|