feat: 模块化命令框架 + runbook 严格断言与查询结果回传

- core/module: 命令模板、门控(feature flag/confirm/dry-run)、ecode 错误标签、ResultFormatter
- runbook executor: http.get/http.post/assert.json(expect_success 严格断言)、目标主机管理、任务落库、最终 stdout 直接回传
- cf 模块: /cf dns list 支持域名自动解析 zone_id;dns list/zones/workers list 格式化回传;API success:false 判真失败
- 全部 CF/CPA runbook 接入严格断言,杜绝假阳性 success
- 新增 cpa/cfdav/mail/sub 模块与 runbooks;QQ 订阅到期提醒调度
- Web 后台: 模块设置(CF/CPA/cfdav/AI)、OPS 任务、渠道管理、审计页
This commit is contained in:
2026-09-13 07:42:17 +08:00
parent 9085c490f3
commit 81939825d1
86 changed files with 6885 additions and 467 deletions
+363 -20
View File
@@ -1,10 +1,15 @@
package runbook
import (
"archive/tar"
"archive/zip"
"bytes"
"compress/gzip"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"os"
"os/exec"
"path/filepath"
@@ -80,6 +85,7 @@ func (e *Executor) RunWithInputsAndMeta(commandText, runbookName string, operato
outputs := map[string]string{}
ctx := map[string]string{}
dataOut := "ok"
jobCtx, jobCancel := context.WithCancel(context.Background())
registerJobCancel(job.ID, jobCancel)
@@ -174,10 +180,13 @@ func (e *Executor) RunWithInputsAndMeta(commandText, runbookName string, operato
_ = e.db.Save(&step).Error
outputs[rendered.ID] = stdout
ctx["steps."+rendered.ID+".output"] = stdout
if strings.TrimSpace(stdout) != "" && rendered.Action != "assert.json" {
dataOut = stdout
}
}
e.finishJob(&job, "success", "ok")
return job.ID, "ok", nil
return job.ID, dataOut, nil
}
func (e *Executor) execStep(parent context.Context, st Step, outputs map[string]string, timeout time.Duration) (int, string, string, error) {
@@ -230,6 +239,125 @@ func (e *Executor) execStep(parent context.Context, st Step, outputs map[string]
}
return rc, strings.TrimSpace(outb.String()), strings.TrimSpace(errb.String()), err
case "http.get":
url := strings.TrimSpace(fmt.Sprintf("%v", st.With["url"]))
if url == "" {
return 1, "", "missing url", fmt.Errorf("missing url")
}
ctx, cancel := context.WithTimeout(parent, timeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
return 1, "", err.Error(), err
}
applyHTTPHeaders(req, st.With["headers"])
resp, err := http.DefaultClient.Do(req)
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
return 1, "", "http step timeout", fmt.Errorf(ecode.Tag(ecode.ErrStepTimeout, "http step timeout"))
}
return 1, "", err.Error(), err
}
defer resp.Body.Close()
body, _ := io.ReadAll(io.LimitReader(resp.Body, 2*1024*1024))
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return 1, strings.TrimSpace(string(body)), resp.Status, fmt.Errorf("http status: %d", resp.StatusCode)
}
return 0, strings.TrimSpace(string(body)), "", nil
case "http.post":
url := strings.TrimSpace(fmt.Sprintf("%v", st.With["url"]))
if url == "" {
return 1, "", "missing url", fmt.Errorf("missing url")
}
bodyStr := ""
if b, ok := st.With["body"]; ok {
switch t := b.(type) {
case string:
bodyStr = t
default:
if jb, err := json.Marshal(t); err == nil {
bodyStr = string(jb)
}
}
}
ctx, cancel := context.WithTimeout(parent, timeout)
defer cancel()
req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, strings.NewReader(bodyStr))
if err != nil {
return 1, "", err.Error(), err
}
hasCT := applyHTTPHeaders(req, st.With["headers"])
if bodyStr != "" && !hasCT {
req.Header.Set("Content-Type", "application/json")
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
if ctx.Err() == context.DeadlineExceeded {
return 1, "", "http step timeout", fmt.Errorf(ecode.Tag(ecode.ErrStepTimeout, "http step timeout"))
}
return 1, "", err.Error(), err
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 2*1024*1024))
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return 1, strings.TrimSpace(string(respBody)), resp.Status, fmt.Errorf("http status: %d", resp.StatusCode)
}
return 0, strings.TrimSpace(string(respBody)), "", nil
case "file.upload":
source := strings.TrimSpace(fmt.Sprintf("%v", st.With["source"]))
stagingDir := strings.TrimSpace(fmt.Sprintf("%v", st.With["staging_dir"]))
filename := strings.TrimSpace(fmt.Sprintf("%v", st.With["filename"]))
if source == "" {
return 1, "", "missing source", fmt.Errorf("missing source")
}
if stagingDir == "" {
stagingDir = "/tmp/ops-staging"
}
if filename == "" {
filename = filepath.Base(source)
}
if err := os.MkdirAll(stagingDir, 0o755); err != nil {
return 1, "", err.Error(), err
}
dest := filepath.Join(stagingDir, filepath.Base(filename))
if err := copyFile(source, dest); err != nil {
return 1, "", err.Error(), err
}
return 0, dest, "", nil
case "file.extract":
archivePath := strings.TrimSpace(fmt.Sprintf("%v", st.With["archive"]))
toDir := strings.TrimSpace(fmt.Sprintf("%v", st.With["to_dir"]))
if archivePath == "" {
return 1, "", "missing archive", fmt.Errorf("missing archive")
}
if toDir == "" {
toDir = "/tmp/ops-staging/extracted"
}
if err := os.MkdirAll(toDir, 0o755); err != nil {
return 1, "", err.Error(), err
}
lower := strings.ToLower(archivePath)
switch {
case strings.HasSuffix(lower, ".zip"):
if err := extractZip(archivePath, toDir); err != nil {
return 1, "", err.Error(), err
}
case strings.HasSuffix(lower, ".tar.gz") || strings.HasSuffix(lower, ".tgz"):
if err := extractTarGz(archivePath, toDir); err != nil {
return 1, "", err.Error(), err
}
case strings.HasSuffix(lower, ".tar"):
if err := extractTar(archivePath, toDir); err != nil {
return 1, "", err.Error(), err
}
default:
return 1, "", "unsupported archive format", fmt.Errorf("unsupported archive format: %s", archivePath)
}
return 0, toDir, "", nil
case "assert.json":
sourceStep := strings.TrimSpace(fmt.Sprintf("%v", st.With["source_step"]))
if sourceStep == "" {
@@ -245,15 +373,27 @@ func (e *Executor) execStep(parent context.Context, st Step, outputs map[string]
return 1, "", "invalid json", err
}
rules := parseRequiredPaths(st.With["required_paths"])
if len(rules) == 0 {
return 1, "", "required_paths empty", fmt.Errorf("required_paths empty")
expectSuccess := parseExpectSuccess(st.With["expect_success"])
if expectSuccess {
v, ok := lookupPath(payload, "success")
if !ok {
return 1, "", "json path not found: success", fmt.Errorf("json path not found: success")
}
if !truthyValue(v) {
errSummary := summarizeJSONErrors(payload)
return 1, "", "api success=false: " + errSummary, fmt.Errorf("api success=false: %s", errSummary)
}
}
rules := parseRequiredPaths(st.With["required_paths"])
for _, p := range rules {
if _, ok := lookupPath(payload, p); !ok {
return 1, "", "json path not found: " + p, fmt.Errorf("json path not found: %s", p)
}
}
if len(rules) == 0 && !expectSuccess {
return 1, "", "required_paths empty", fmt.Errorf("required_paths empty")
}
return 0, "assert ok", "", nil
case "sleep":
@@ -292,27 +432,33 @@ func renderStep(st Step, ctx map[string]string) Step {
}
m := make(map[string]any, len(out.With))
for k, v := range out.With {
switch t := v.(type) {
case string:
m[k] = renderString(t, ctx)
case []any:
arr := make([]any, 0, len(t))
for _, it := range t {
if s, ok := it.(string); ok {
arr = append(arr, renderString(s, ctx))
} else {
arr = append(arr, it)
}
}
m[k] = arr
default:
m[k] = v
}
m[k] = renderAny(v, ctx)
}
out.With = m
return out
}
func renderAny(v any, ctx map[string]string) any {
switch t := v.(type) {
case string:
return renderString(t, ctx)
case []any:
arr := make([]any, 0, len(t))
for _, it := range t {
arr = append(arr, renderAny(it, ctx))
}
return arr
case map[string]any:
m := make(map[string]any, len(t))
for k, v := range t {
m[k] = renderAny(v, ctx)
}
return m
default:
return v
}
}
func renderString(s string, ctx map[string]string) string {
res := s
for k, v := range ctx {
@@ -321,6 +467,158 @@ func renderString(s string, ctx map[string]string) string {
return res
}
func applyHTTPHeaders(req *http.Request, raw any) bool {
hasContentType := false
m, ok := raw.(map[string]any)
if !ok {
return hasContentType
}
for k, v := range m {
key := strings.TrimSpace(k)
if key == "" {
continue
}
val := strings.TrimSpace(fmt.Sprintf("%v", v))
if val == "" {
continue
}
req.Header.Set(key, val)
if strings.EqualFold(key, "Content-Type") {
hasContentType = true
}
}
return hasContentType
}
func copyFile(src, dst string) error {
in, err := os.Open(src)
if err != nil {
return err
}
defer in.Close()
fi, err := in.Stat()
if err != nil {
return err
}
if !fi.Mode().IsRegular() {
return fmt.Errorf("source is not regular file")
}
out, err := os.Create(dst)
if err != nil {
return err
}
defer out.Close()
if _, err := io.Copy(out, in); err != nil {
return err
}
return out.Close()
}
func extractZip(src, dst string) error {
zr, err := zip.OpenReader(src)
if err != nil {
return err
}
defer zr.Close()
for _, f := range zr.File {
clean := filepath.Clean(f.Name)
if strings.HasPrefix(clean, "..") || filepath.IsAbs(clean) {
return fmt.Errorf("invalid zip entry path: %s", f.Name)
}
target := filepath.Join(dst, clean)
if f.FileInfo().IsDir() {
if err := os.MkdirAll(target, 0o755); err != nil {
return err
}
continue
}
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return err
}
rc, err := f.Open()
if err != nil {
return err
}
out, err := os.Create(target)
if err != nil {
rc.Close()
return err
}
_, cpErr := io.Copy(out, rc)
rc.Close()
if err := out.Close(); err != nil {
return err
}
if cpErr != nil {
return cpErr
}
}
return nil
}
func extractTar(src, dst string) error {
f, err := os.Open(src)
if err != nil {
return err
}
defer f.Close()
return untar(f, dst)
}
func extractTarGz(src, dst string) error {
f, err := os.Open(src)
if err != nil {
return err
}
defer f.Close()
gz, err := gzip.NewReader(f)
if err != nil {
return err
}
defer gz.Close()
return untar(gz, dst)
}
func untar(r io.Reader, dst string) error {
tr := tar.NewReader(r)
for {
hdr, err := tr.Next()
if err == io.EOF {
break
}
if err != nil {
return err
}
clean := filepath.Clean(hdr.Name)
if strings.HasPrefix(clean, "..") || filepath.IsAbs(clean) {
return fmt.Errorf("invalid tar entry path: %s", hdr.Name)
}
target := filepath.Join(dst, clean)
switch hdr.Typeflag {
case tar.TypeDir:
if err := os.MkdirAll(target, 0o755); err != nil {
return err
}
case tar.TypeReg, tar.TypeRegA:
if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
return err
}
out, err := os.Create(target)
if err != nil {
return err
}
if _, err := io.Copy(out, tr); err != nil {
out.Close()
return err
}
if err := out.Close(); err != nil {
return err
}
}
}
return nil
}
func parseRequiredPaths(v any) []string {
res := []string{}
switch t := v.(type) {
@@ -363,6 +661,51 @@ func lookupPath(root any, path string) (any, bool) {
return cur, true
}
func parseExpectSuccess(v any) bool {
switch t := v.(type) {
case bool:
return t
case string:
return strings.EqualFold(strings.TrimSpace(t), "true")
default:
return false
}
}
func truthyValue(v any) bool {
switch t := v.(type) {
case bool:
return t
case string:
return strings.EqualFold(strings.TrimSpace(t), "true")
default:
return false
}
}
// summarizeJSONErrors 从 Cloudflare 风格的 {"success":false,"errors":[...]} 中提取可读错误摘要
func summarizeJSONErrors(payload any) string {
msgs := []string{}
if m, ok := payload.(map[string]any); ok {
if errs, ok := m["errors"].([]any); ok {
for _, e := range errs {
switch ev := e.(type) {
case map[string]any:
code := fmt.Sprintf("%v", ev["code"])
msg := fmt.Sprintf("%v", ev["message"])
msgs = append(msgs, code+" "+msg)
default:
msgs = append(msgs, fmt.Sprintf("%v", ev))
}
}
}
}
if len(msgs) == 0 {
return "unknown error"
}
return strings.Join(msgs, "; ")
}
func (e *Executor) finishJob(job *models.OpsJob, status, summary string) {
job.Status = status
job.Summary = summary