审批工作流引擎模块

module-11-approval-workflow/
├── go.mod
├── deploy/
│   ├── approvalticket-crd.yaml     # ① 审批工单 CRD（流程实例载体）
│   ├── engine-rbac.yaml            # ② 引擎自身 RBAC + Webhook 接收 SA/ClusterRole
│   ├── executor-rbac.yaml          # ③ 审批通过后「执行用」受限 RBAC（仅租户 ns）
│   └── prometheus-rules.yaml       # ④ 审批延迟/超时/敏感操作 告警规则
├── config/
│   └── servicemonitor.yaml         # 联邦抓取配置（带 tenant 标签）
├── internal/
│   ├── apis/approval/v1/ticket.go  # ApprovalTicket 类型 + CRD 转换
│   ├── metrics/metrics.go          # Prometheus 指标（tenant/app/env/cluster）
│   ├── audit/audit.go              # WORM 不可变审计留痕
│   ├── notify/im.go                # 企业微信 / 钉钉 IM 通知
│   ├── engine/
│   │   ├── statemachine.go         # 状态机：多级 / 会签(AND) / 或签(OR)
│   │   ├── executor.go             # 审批通过后经 API 网关 client-go 执行 K8s 变更
│   │   └── engine.go               # 控制器 Reconcile（超时升级 + 落地执行）
│   └── gateway/token.go            # 向统一 K8s API 网关申请签名执行凭证
├── cmd/engine/main.go              # 引擎启动入口（含 /metrics）
└── runbook.md                      # 运维 SOP：建单/审批/查结果/绕行验证

# file: deploy/approvalticket-crd.yaml
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: approvaltickets.approval.paas.example.com
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
spec:
  group: approval.paas.example.com
  names:
    kind: ApprovalTicket
    listKind: ApprovalTicketList
    plural: approvaltickets
    singular: approvalticket
    shortNames: ["ap"]
  scope: Namespaced
  versions:
    - name: v1
      served: true
      storage: true
      subresources:
        status: {}
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              required: ["applicant", "tenant", "env", "cluster", "namespace", "action", "policy", "chain"]
              properties:
                applicant:
                  type: string
                  description: "申请人（租户内账号）"
                tenant:
                  type: string
                  description: "租户标识，用于 Namespace 隔离与指标标签"
                env:
                  type: string
                  enum: ["dev", "test", "stage", "prod"]
                cluster:
                  type: string
                  description: "目标物理隔离集群名"
                namespace:
                  type: string
                  description: "目标 Namespace（租户强隔离）"
                action:
                  type: string
                  description: "受控动作，如 deploy.apply / quota.update / secret.rotate"
                riskLevel:
                  type: string
                  enum: ["low", "mid", "high"]
                  default: mid
                policy:
                  type: string
                  enum: ["and", "or"]
                  description: "and=会签(全部同意) or=或签(任一同意)"
                chain:
                  type: array
                  description: "多级审批链，按顺序串行"
                  items:
                    type: object
                    required: ["level", "role"]
                    properties:
                      level:
                        type: integer
                        minimum: 1
                      role:
                        type: string
                        description: "该级审批人所需 RBAC 角色，如 tenant-admin"
                      slaSeconds:
                        type: integer
                        description: "该级 SLA，超时则升级"
                        default: 1800
                payloadRef:
                  type: string
                  description: "待执行 manifest 引用（加密存储于 Secret/对象存储）"
                summary:
                  type: string
                  description: "脱敏后的操作摘要，用于 IM 推送"
            status:
              type: object
              properties:
                phase:
                  type: string
                  enum: ["Submitted", "Approving", "Approved", "Rejected", "Executing", "Succeeded", "Failed"]
                currentLevel:
                  type: integer
                approvals:
                  type: array
                  items:
                    type: object
                    properties:
                      level:
                        type: integer
                      approver:
                        type: string
                      decision:
                        type: string
                        enum: ["approve", "reject"]
                      at:
                        type: string
                        format: date-time
                      deviceId:
                        type: string
                escalationCount:
                  type: integer
                execTokenRef:
                  type: string
                  description: "网关签发的执行凭证引用（不落明文）"
                execResult:
                  type: string
                auditRef:
                  type: string
                  description: "WORM 审计链尾 hash，用于防篡改校验"
      additionalPrinterColumns:
        - name: Phase
          type: string
          jsonPath: .status.phase
        - name: Env
          type: string
          jsonPath: .spec.env
        - name: Cluster
          type: string
          jsonPath: .spec.cluster
        - name: Action
          type: string
          jsonPath: .spec.action
        - name: Age
          type: date
          jsonPath: .metadata.creationTimestamp

# file: deploy/engine-rbac.yaml
apiVersion: v1
kind: ServiceAccount
metadata:
  name: approval-engine
  namespace: paas-approval
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: approval-engine-ticket-rw
  namespace: paas-approval
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
rules:
  - apiGroups: ["approval.paas.example.com"]
    resources: ["approvaltickets"]
    verbs: ["get", "list", "watch", "create", "update", "patch"]
  - apiGroups: ["approval.paas.example.com"]
    resources: ["approvaltickets/status"]
    verbs: ["get", "update", "patch"]
  - apiGroups: [""]
    resources: ["configmaps"]
    verbs: ["get", "list"]
    # 仅用于读取流程定义与 IM Webhook 配置，限定本命名空间
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: approval-engine-ticket-rw
  namespace: paas-approval
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
subjects:
  - kind: ServiceAccount
    name: approval-engine
    namespace: paas-approval
roleRef:
  kind: Role
  name: approval-engine-ticket-rw
  apiGroup: rbac.authorization.k8s.io
---
# Webhook 接收器：IM 回调把审批决定 POST 进来，由专属 SA 写入工单
apiVersion: v1
kind: ServiceAccount
metadata:
  name: approval-webhook-receiver
  namespace: paas-approval
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: approval-webhook-receiver
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
rules:
  - apiGroups: ["approval.paas.example.com"]
    resources: ["approvaltickets"]
    verbs: ["get", "update", "patch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: approval-webhook-receiver
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
subjects:
  - kind: ServiceAccount
    name: approval-webhook-receiver
    namespace: paas-approval
roleRef:
  kind: ClusterRole
  name: approval-webhook-receiver
  apiGroup: rbac.authorization.k8s.io

# file: deploy/executor-rbac.yaml
# 该 Role 由统一 K8s API 网关在「执行凭证」scope 内临时绑定到 executor 身份，
# 仅作用于目标租户 Namespace，仅含最小动词。
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: approval-executor-deploy
  namespace: tenant-a-prod          # 由凭证 scope.tenant/namespace 动态渲染
  labels:
    tenant: tenant-a
    app: approval-engine
    env: prod
    cluster: prod-sh
spec:
  rules:
    # deploy.apply 仅允许对 Deployment 做 apply/scale
    - apiGroups: ["apps"]
      resources: ["deployments"]
      verbs: ["get", "list", "create", "update", "patch"]
    - apiGroups: [""]
      resources: ["configmaps"]
      verbs: ["get", "list", "create", "update", "patch"]
    # 明确拒绝 secrets/rbac/exec 等高危资源，证明「凭证越界即拒绝」
    # （高危资源不出现在 rules 中 = 网关侧 403）
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: approval-executor
  namespace: paas-gateway-proxy
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: approval-executor-deploy
  namespace: tenant-a-prod
  labels:
    tenant: tenant-a
    app: approval-engine
    env: prod
    cluster: prod-sh
subjects:
  - kind: ServiceAccount
    name: approval-executor
    namespace: paas-gateway-proxy
roleRef:
  kind: Role
  name: approval-executor-deploy
  apiGroup: rbac.authorization.k8s.io

// file: internal/apis/approval/v1/ticket.go
package v1

import (
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
)

// Group/Version/Resource —— 与 deploy/approvalticket-crd.yaml 完全一致
const (
	Group    = "approval.paas.example.com"
	Version  = "v1"
	Resource = "approvaltickets"
	Kind     = "ApprovalTicket"
)

// ApprovalTicket 对应 CRD 实例。
type ApprovalTicket struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec ApprovalTicketSpec `json:"spec,omitempty"`
	Status ApprovalTicketStatus `json:"status,omitempty"`
}

type ApprovalTicketSpec struct {
	Applicant  string           `json:"applicant"`
	Tenant     string           `json:"tenant"`
	Env        string           `json:"env"` // dev/test/stage/prod
	Cluster    string           `json:"cluster"`
	Namespace  string           `json:"namespace"`
	Action     string           `json:"action"` // deploy.apply / quota.update / secret.rotate
	RiskLevel  string           `json:"riskLevel,omitempty"`
	Policy     string           `json:"policy"` // and=会签 or=或签
	Chain      []ApprovalLevel  `json:"chain"`
	PayloadRef string           `json:"payloadRef,omitempty"`
	Summary    string           `json:"summary,omitempty"`
}

type ApprovalLevel struct {
	Level     int    `json:"level"`
	Role      string `json:"role"` // 该级审批人所需 RBAC 角色
	SLASeconds int   `json:"slaSeconds,omitempty"`
}

type ApprovalTicketStatus struct {
	Phase          string          `json:"phase"` // Submitted/Approving/Approved/Rejected/Executing/Succeeded/Failed
	CurrentLevel   int             `json:"currentLevel,omitempty"`
	Approvals      []ApprovalRecord `json:"approvals,omitempty"`
	EscalationCount int            `json:"escalationCount,omitempty"`
	ExecTokenRef   string          `json:"execTokenRef,omitempty"`
	ExecResult     string          `json:"execResult,omitempty"`
	AuditRef       string          `json:"auditRef,omitempty"`
}

type ApprovalRecord struct {
	Level    int       `json:"level"`
	Approver string    `json:"approver"`
	Decision string    `json:"decision"` // approve/reject
	At       time.Time `json:"at"`
	DeviceID string    `json:"deviceId,omitempty"`
}

func (t *ApprovalTicket) GetObjectKind() runtime.ObjectKind { return &t.TypeMeta }

// ToUnstructured 用于经 dynamic client 写回 apiserver。
func (t *ApprovalTicket) ToUnstructured() (*unstructured.Unstructured, error) {
	m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(t)
	if err != nil {
		return nil, err
	}
	return &unstructured.Unstructured{Object: m}, nil
}

// FromUnstructured 用于经 dynamic client 读回 apiserver 对象。
func FromUnstructured(u *unstructured.Unstructured) (*ApprovalTicket, error) {
	t := &ApprovalTicket{}
	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, t); err != nil {
		return nil, err
	}
	return t, nil
}

// file: internal/metrics/metrics.go
package metrics

import "github.com/prometheus/client_golang/prometheus"

// LabelKeys 全局统一标签，与 _SHARED.md 约束一致。
var LabelKeys = []string{"tenant", "app", "env", "cluster"}

var (
	// 申请量与通过/拒绝率
	ApprovalRequestTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "approval_request_total",
		Help: "Total approval tickets by action/env/result.",
	}, append([]string{"action", "result"}, LabelKeys...))

	// 各节点待审批时长（SLA 监控）
	ApprovalNodePendingSeconds = prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Name:    "approval_node_pending_seconds",
		Help:    "Seconds a node waits for approval.",
		Buckets: []float64{60, 300, 900, 1800, 3600, 7200},
	}, append([]string{"level"}, LabelKeys...))

	// 超时升级次数（告警关键指标）
	ApprovalEscalationTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "approval_escalation_total",
		Help: "Total escalations due to SLA timeout.",
	}, append([]string{"level"}, LabelKeys...))

	// 凭证签发到执行完成时延
	ApprovalExecutionLatencySeconds = prometheus.NewHistogramVec(prometheus.HistogramOpts{
		Name:    "approval_execution_latency_seconds",
		Help:    "Latency from token issue to execution done.",
		Buckets: []float64{5, 15, 30, 60, 120, 300},
	}, append([]string{"action"}, LabelKeys...))

	// 敏感操作（env=prod 的高危动作）计数，供告警
	SensitiveOpApprovedTotal = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "approval_sensitive_op_approved_total",
		Help: "Approved sensitive operations in prod, by action.",
	}, append([]string{"action", "risk_level"}, LabelKeys...))
)

func init() {
	prometheus.MustRegister(
		ApprovalRequestTotal,
		ApprovalNodePendingSeconds,
		ApprovalEscalationTotal,
		ApprovalExecutionLatencySeconds,
		SensitiveOpApprovedTotal,
	)
}

// file: internal/audit/audit.go
package audit

import (
	"bufio"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"os"
	"sync"
	"time"
)

// Event 是不可变审计记录。WORM：只追加、防篡改。
type Event struct {
	Seq       uint64    `json:"seq"`
	PrevHash  string    `json:"prev_hash"`
	Hash      string    `json:"hash"`
	Timestamp time.Time `json:"timestamp"`
	Operator  string    `json:"operator"`
	Role      string    `json:"role"`
	Action    string    `json:"action"`
	Tenant    string    `json:"tenant"`
	Env       string    `json:"env"`
	Cluster   string    `json:"cluster"`
	TicketID  string    `json:"ticket_id"`
	Decision  string    `json:"decision,omitempty"`
	Detail    string    `json:"detail,omitempty"`
}

// WORM 追加写审计，支持多副本共享同一后端（此处以本地文件演示，生产可换对象存储/S3 追加写）。
type WORM struct {
	mu       sync.Mutex
	w        *bufio.Writer
	f        *os.File
	prevHash string
	seq      uint64
}

// Open 打开（或创建）审计文件，并恢复上一事件 hash 以维持链。
func Open(path string) (*WORM, error) {
	f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0600)
	if err != nil {
		return nil, err
	}
	prev := ""
	var last Event
	// 简单回放恢复 prevHash 与 seq（生产环境建议存独立索引）
	sc := bufio.NewScanner(f)
	for sc.Scan() {
		var e Event
		if json.Unmarshal(sc.Bytes(), &e) == nil {
			last = e
		}
	}
	if last.Hash != "" {
		prev = last.Hash
	}
	if _, err := f.Seek(0, 2); err != nil { // 回到末尾追加
		return nil, err
	}
	w := &WORM{w: bufio.NewWriter(f), f: f, prevHash: prev, seq: last.Seq}
	return w, nil
}

// Record 追加一条事件并返回其 hash（用于写回 ApprovalTicket.status.auditRef）。
func (w *WORM) Record(e Event) (string, error) {
	w.mu.Lock()
	defer w.mu.Unlock()

	w.seq++
	e.Seq = w.seq
	e.PrevHash = w.prevHash
	b, _ := json.Marshal(struct {
		Operator  string `json:"operator"`
		Role      string `json:"role"`
		Action    string `json:"action"`
		Tenant    string `json:"tenant"`
		Env       string `json:"env"`
		Cluster   string `json:"cluster"`
		TicketID  string `json:"ticket_id"`
		Decision  string `json:"decision,omitempty"`
		Timestamp int64  `json:"timestamp"`
	}{e.Operator, e.Role, e.Action, e.Tenant, e.Env, e.Cluster, e.TicketID, e.Decision, e.Timestamp.UnixNano()})
	sum := sha256.Sum256(append([]byte(w.prevHash), b...))
	e.Hash = hex.EncodeToString(sum[:])
	line, _ := json.Marshal(e)
	if _, err := w.w.Write(append(line, '\n')); err != nil {
		return "", err
	}
	if err := w.w.Flush(); err != nil {
		return "", err
	}
	w.prevHash = e.Hash
	return e.Hash, nil
}

func (w *WORM) Close() error { return w.f.Close() }

// VerifyChain 校验整条链未被篡改（审计存证）。
func VerifyChain(path string) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()
	sc := bufio.NewScanner(f)
	var prev string
	var seq uint64
	for sc.Scan() {
		var e Event
		if err := json.Unmarshal(sc.Bytes(), &e); err != nil {
			return fmt.Errorf("bad line: %w", err)
		}
		if e.Seq != seq+1 {
			return fmt.Errorf("seq gap at %d", e.Seq)
		}
		seq = e.Seq
		if e.PrevHash != prev {
			return fmt.Errorf("prev_hash broken at seq %d", e.Seq)
		}
		prev = e.Hash
	}
	return nil
}

// file: internal/notify/im.go
package notify

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"time"
)

// Notifier 统一的 IM 通知能力（待办/催办/结果）。
type Notifier interface {
	Notify(ctx context.Context, msg Message) error
}

type Message struct {
	TicketID  string // 工单号，如 AP-2026-0001
	Title     string // 待办/催办/结果
	Summary   string // 脱敏后的操作摘要
	Tenant    string
	Env       string
	Cluster   string
	Level     int
	Approvers []string
	IsUrgent  bool // 升级催办
}

// ---------- 企业微信 ----------
type WeComNotifier struct {
	Webhook string // https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=XXX
	client  *http.Client
}

func NewWeCom(webhook string) *WeComNotifier {
	return &WeComNotifier{Webhook: webhook, client: &http.Client{Timeout: 5 * time.Second}}
}

func (n *WeComNotifier) Notify(ctx context.Context, msg Message) error {
	markdown := fmt.Sprintf(
		"### [%s] %s\n>**工单**: %s\n>**租户/环境/集群**: %s / %s / %s\n>**级别**: L%d\n>**摘要**: %s\n>**审批人**: %v",
		map[bool]string{true: "催办", false: "待办"}[msg.IsUrgent], msg.Title,
		msg.TicketID, msg.Tenant, msg.Env, msg.Cluster, msg.Level, msg.Summary, msg.Approvers,
	)
	body, _ := json.Marshal(map[string]interface{}{
		"msgtype": "markdown",
		"markdown": map[string]string{"content": markdown},
	})
	return n.post(ctx, body)
}

// ---------- 钉钉 ----------
type DingTalkNotifier struct {
	Webhook string // https://oapi.dingtalk.com/robot/send?access_token=XXX
	client  *http.Client
}

func NewDingTalk(webhook string) *DingTalkNotifier {
	return &DingTalkNotifier{Webhook: webhook, client: &http.Client{Timeout: 5 * time.Second}}
}

func (n *DingTalkNotifier) Notify(ctx context.Context, msg Message) error {
	text := fmt.Sprintf(
		"[%s] %s\n工单: %s\n租户/环境/集群: %s/%s/%s\n级别: L%d\n摘要: %s\n审批人: %v",
		map[bool]string{true: "催办", false: "待办"}[msg.IsUrgent], msg.Title,
		msg.TicketID, msg.Tenant, msg.Env, msg.Cluster, msg.Level, msg.Summary, msg.Approvers,
	)
	body, _ := json.Marshal(map[string]interface{}{
		"msgtype": "text",
		"text":    map[string]string{"content": text},
	})
	return n.post(ctx, body)
}

func (n *WeComNotifier) post(ctx context.Context, body []byte) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.Webhook, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := n.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("wecom notify http %d", resp.StatusCode)
	}
	return nil
}

func (n *DingTalkNotifier) post(ctx context.Context, body []byte) error {
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, n.Webhook, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := n.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("dingtalk notify http %d", resp.StatusCode)
	}
	return nil
}

// Fanout 同时向企业微信与钉钉推送（按租户配置选择通道）。
func Fanout(ctx context.Context, notifiers []Notifier, msg Message) error {
	var firstErr error
	for _, n := range notifiers {
		if err := n.Notify(ctx, msg); err != nil && firstErr == nil {
			firstErr = err
		}
	}
	return firstErr
}

// file: internal/gateway/token.go
package gateway

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"os"
	"time"
)

// GatewayURL 取自环境变量，统一 K8s API 网关地址。
func GatewayURL() string {
	if v := os.Getenv("K8S_GATEWAY_URL"); v != "" {
		return v
	}
	return "https://k8s-gateway.paas.internal"
}

// ExecScope 描述执行凭证的作用范围与动作（与 deploy/executor-rbac.yaml 对应）。
type ExecScope struct {
	Tenant   string `json:"tenant"`
	Cluster  string `json:"cluster"`
	Namespace string `json:"namespace"`
	Action   string `json:"action"` // deploy.apply / quota.update ...
	TTL      int    `json:"ttl"`    // 秒，默认 1800
}

// TokenResponse 网关返回的执行凭证（仅引用，不落明文到审计）。
type TokenResponse struct {
	Token     string    `json:"token"`
	ExpiresAt time.Time `json:"expires_at"`
	Ref       string    `json:"ref"` // 凭证引用，写回 status.execTokenRef
}

// RequestExecToken 向网关申请签名执行凭证。网关会回验审批链与 RBAC。
func RequestExecToken(ctx context.Context, ticketID string, scope ExecScope) (*TokenResponse, error) {
	body, _ := json.Marshal(map[string]interface{}{
		"ticket_id": ticketID,
		"scope":     scope,
	})
	url := GatewayURL() + "/v1/exec-token"
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")
	// 引擎自身身份（来自 engine SA 注入的 token），用于网关鉴别「谁在申请」
	req.Header.Set("Authorization", "Bearer "+os.Getenv("ENGINE_TOKEN"))

	client := &http.Client{Timeout: 10 * time.Second}
	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("gateway exec-token http %d (审批链可能不完整或被拒)", resp.StatusCode)
	}
	var tr TokenResponse
	if err := json.NewDecoder(resp.Body).Decode(&tr); err != nil {
		return nil, err
	}
	return &tr, nil
}

// file: internal/engine/statemachine.go
package engine

import (
	"errors"
	"fmt"

	"module-11-approval-workflow/internal/apis/approval/v1"
)

var (
	ErrRejected      = errors.New("ticket rejected")
	ErrNotEnough     = errors.New("not enough approvals for this level")
	ErrBadTransition = errors.New("invalid phase transition")
)

// HandleDecision 应用一次审批决定，返回是否推进到下一阶段。
// 会签(and)：该级需全部 approver 同意；或签(or)：该级任一人同意即过。
func HandleDecision(t *v1.ApprovalTicket, level int, approver, decision, deviceID string) error {
	if t.Status.Phase != "Approving" && t.Status.Phase != "Submitted" {
		return ErrBadTransition
	}
	// 阶段迁移：Submitted -> Approving
	if t.Status.Phase == "Submitted" {
		t.Status.Phase = "Approving"
		t.Status.CurrentLevel = 1
	}
	if level != t.Status.CurrentLevel {
		return fmt.Errorf("expecting level %d, got %d", t.Status.CurrentLevel, level)
	}
	t.Status.Approvals = append(t.Status.Approvals, v1.ApprovalRecord{
		Level:    level,
		Approver: approver,
		Decision: decision,
		At:       timeNow(),
		DeviceID: deviceID,
	})

	if decision == "reject" {
		t.Status.Phase = "Rejected"
		return ErrRejected
	}

	// 本级是否已满足策略
	levelApproved, err := levelSatisfied(t, level)
	if err != nil {
		return err
	}
	if !levelApproved {
		return ErrNotEnough // 等待会签中的其他人
	}

	// 推进到下一级或整体通过
	if t.Status.CurrentLevel >= len(t.Spec.Chain) {
		t.Status.Phase = "Approved"
		return nil
	}
	t.Status.CurrentLevel++
	return nil
}

// levelSatisfied 判定当前级是否满足会签/或签条件。
func levelSatisfied(t *v1.ApprovalTicket, level int) (bool, error) {
	var approvers []string
	for _, l := range t.Spec.Chain {
		if l.Level == level {
			approvers = append(approvers, l.Role) // 简化：以 role 数代表应到人数
		}
	}
	if len(approvers) == 0 {
		return false, fmt.Errorf("level %d not found in chain", level)
	}
	approved := 0
	for _, a := range t.Status.Approvals {
		if a.Level == level && a.Decision == "approve" {
			approved++
		}
	}
	if t.Spec.Policy == "or" {
		return approved >= 1, nil // 或签：一人即过
	}
	return approved >= len(approvers), nil // 会签：全部
}

// file: internal/engine/executor.go
package engine

import (
	"context"
	"fmt"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/rest"

	"module-11-approval-workflow/internal/apis/approval/v1"
	"module-11-approval-workflow/internal/gateway"
	"module-11-approval-workflow/internal/metrics"
)

// Executor 负责把已批准工单的 manifest 经网关落地到目标集群。
type Executor struct {
	GatewayCA []byte // 网关 mTLS CA，来自加密卷
}

// Execute 申请执行凭证并用 client-go 经网关 apply 暂存对象。
func (e *Executor) Execute(ctx context.Context, t *v1.ApprovalTicket, staged *unstructured.Unstructured) error {
	t.Status.Phase = "Executing"
	start := time.Now()

	// 1) 向统一 K8s API 网关申请签名执行凭证（有时效/范围）
	tr, err := gateway.RequestExecToken(ctx, t.Name, gateway.ExecScope{
		Tenant:    t.Spec.Tenant,
		Cluster:   t.Spec.Cluster,
		Namespace: t.Spec.Namespace,
		Action:    t.Spec.Action,
		TTL:       tokenTTL(t.Spec.Env),
	})
	if err != nil {
		t.Status.Phase = "Failed"
		t.Status.ExecResult = "token denied: " + err.Error()
		return err
	}
	t.Status.ExecTokenRef = tr.Ref

	// 2) 用执行凭证构造指向「网关」的 client-go 配置（绝不直接连 apiserver）
	cfg := &rest.Config{
		Host:        gateway.GatewayURL(),
		BearerToken: tr.Token,
		TLSClientConfig: rest.TLSClientConfig{
			CAData: e.GatewayCA,
		},
	}
	dyn, err := dynamic.NewForConfig(cfg)
	if err != nil {
		t.Status.Phase = "Failed"
		return err
	}

	// 3) 经网关把 manifest apply 到目标租户 Namespace（网关用 executor-rbac 校验 scope/action）
	gvr := staged.GroupVersionResource()
	_, err = dyn.Resource(gvr).Namespace(t.Spec.Namespace).Apply(
		ctx, staged.GetName(), staged, metav1.ApplyOptions{FieldManager: "approval-engine"},
	)
	if err != nil {
		t.Status.Phase = "Failed"
		t.Status.ExecResult = "apply failed: " + err.Error()
		metrics.ApprovalExecutionLatencySeconds.WithLabelValues(
			t.Spec.Action, t.Spec.Tenant, "approval-engine", t.Spec.Env, t.Spec.Cluster,
		).Observe(time.Since(start).Seconds())
		return fmt.Errorf("gateway apply rejected: %w", err) // 越界/过期 -> 403 已由网关拦截
	}

	t.Status.Phase = "Succeeded"
	t.Status.ExecResult = "applied via gateway"
	metrics.ApprovalExecutionLatencySeconds.WithLabelValues(
		t.Spec.Action, t.Spec.Tenant, "approval-engine", t.Spec.Env, t.Spec.Cluster,
	).Observe(time.Since(start).Seconds())
	if t.Spec.Env == "prod" && t.Spec.RiskLevel == "high" {
		metrics.SensitiveOpApprovedTotal.WithLabelValues(
			t.Spec.Action, t.Spec.RiskLevel, t.Spec.Tenant, "approval-engine", t.Spec.Env, t.Spec.Cluster,
		).Inc()
	}
	return nil
}

// tokenTTL 生产收紧（默认 30min，可配 10min），测试放宽。
func tokenTTL(env string) int {
	if env == "prod" {
		return 1800
	}
	return 7200
}

// file: internal/engine/engine.go
package engine

import (
	"context"
	"fmt"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"

	"module-11-approval-workflow/internal/apis/approval/v1"
	"module-11-approval-workflow/internal/audit"
	"module-11-approval-workflow/internal/metrics"
	"module-11-approval-workflow/internal/notify"
)

var ticketGVR = schema.GroupVersionResource{
	Group:    v1.Group,
	Version:  v1.Version,
	Resource: v1.Resource,
}

// Engine 审批工作流引擎控制器。
type Engine struct {
	Dyn      dynamic.Interface
	Executor *Executor
	Audit    *audit.WORM
	Notifier []notify.Notifier
}

// timeNow 便于测试替换。
var timeNow = time.Now

// Run 简易轮询 Reconcile（生产建议用 informer + workqueue）。
func (e *Engine) Run(ctx context.Context, namespace string, interval time.Duration) error {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		if err := e.reconcileAll(ctx, namespace); err != nil {
			fmt.Printf("reconcile error: %v\n", err)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-ticker.C:
		}
	}
}

func (e *Engine) reconcileAll(ctx context.Context, ns string) error {
	list, err := e.Dyn.Resource(ticketGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
	if err != nil {
		return err
	}
	for _, item := range list.Items {
		t, err := v1.FromUnstructured(&item)
		if err != nil {
			continue
		}
		if err := e.Reconcile(ctx, t); err != nil {
			fmt.Printf("ticket %s reconcile: %v\n", t.Name, err)
		}
		// 写回 status
		u, _ := t.ToUnstructured()
		_, err = e.Dyn.Resource(ticketGVR).Namespace(ns).Update(ctx, u, metav1.UpdateOptions{})
		if err != nil && !apierrors.IsConflict(err) {
			fmt.Printf("update ticket %s: %v\n", t.Name, err)
		}
	}
	return nil
}

// Reconcile 处理单条工单：超时升级 / 全通过后执行。
func (e *Engine) Reconcile(ctx context.Context, t *v1.ApprovalTicket) error {
	switch t.Status.Phase {
	case "Approving":
		e.checkSLA(ctx, t)
	case "Approved":
		// 全通过 -> 申请凭证并执行
		staged, err := e.loadStaged(ctx, t)
		if err != nil {
			t.Status.Phase = "Failed"
			t.Status.ExecResult = "load staged failed: " + err.Error()
			return err
		}
		metrics.ApprovalRequestTotal.WithLabelValues(
			t.Spec.Action, "approved", t.Spec.Tenant, "approval-engine", t.Spec.Env, t.Spec.Cluster,
		).Inc()
		if err := e.Executor.Execute(ctx, t, staged); err != nil {
			return err
		}
		// 执行结果入 WORM 审计
		hash, _ := e.Audit.Record(audit.Event{
			Timestamp: timeNow(), Operator: "approval-engine", Role: "platform-ops",
			Action: t.Spec.Action, Tenant: t.Spec.Tenant, Env: t.Spec.Env,
			Cluster: t.Spec.Cluster, TicketID: t.Name, Decision: "executed",
			Detail: t.Status.ExecResult,
		})
		t.Status.AuditRef = hash
	}
	return nil
}

// checkSLA 超时升级：超过本级 SLA 则催办 + 升级上级，记 escalation 指标。
func (e *Engine) checkSLA(ctx context.Context, t *v1.ApprovalTicket) {
	level := currentLevel(t.Spec.Chain, t.Status.CurrentLevel)
	if level == nil {
		return
	}
	sla := time.Duration(level.SLASeconds) * time.Second
	// 本级最近一次审批时间；无记录则用工单创建时间
	since := timeNow().Sub(latestLevelActivity(t))
	if since <= sla {
		return
	}
	t.Status.EscalationCount++
	metrics.ApprovalEscalationTotal.WithLabelValues(
		fmt.Sprintf("%d", level.Level), t.Spec.Tenant, "approval-engine", t.Spec.Env, t.Spec.Cluster,
	).Inc()
	_ = notify.Fanout(ctx, e.Notifier, notify.Message{
		TicketID: t.Name, Title: "审批超时升级催办", Summary: t.Spec.Summary,
		Tenant: t.Spec.Tenant, Env: t.Spec.Env, Cluster: t.Spec.Cluster,
		Level: level.Level, Approvers: []string{level.Role}, IsUrgent: true,
	})
}

// loadStaged 读取执行 manifest（生产从加密 Secret/对象存储取，此处演示从 ConfigMap 引用）。
func (e *Engine) loadStaged(ctx context.Context, t *v1.ApprovalTicket) (*unstructured.Unstructured, error) {
	// 真实实现：从 payloadRef 指向的加密存储解析出 unstructured 对象。
	// 为演示自洽，返回一个最小 Deployment 对象。
	return &unstructured.Unstructured{Object: map[string]interface{}{
		"apiVersion": "apps/v1",
		"kind":       "Deployment",
		"metadata": map[string]interface{}{
			"name":      "demo-app",
			"namespace": t.Spec.Namespace,
			"labels": map[string]interface{}{
				"tenant":  t.Spec.Tenant,
				"env":     t.Spec.Env,
				"cluster": t.Spec.Cluster,
			},
		},
		"spec": map[string]interface{}{
			"replicas": int64(2),
			"selector": map[string]interface{}{
				"matchLabels": map[string]interface{}{"app": "demo-app"},
			},
			"template": map[string]interface{}{
				"metadata": map[string]interface{}{
					"labels": map[string]interface{}{"app": "demo-app"},
				},
				"spec": map[string]interface{}{
					"containers": []interface{}{map[string]interface{}{
						"name":  "app",
						"image": "registry.paas.internal/demo-app:approved",
					}},
				},
			},
		},
	}}, nil
}

func currentLevel(chain []v1.ApprovalLevel, lvl int) *v1.ApprovalLevel {
	for i := range chain {
		if chain[i].Level == lvl {
			return &chain[i]
		}
	}
	return nil
}

func latestLevelActivity(t *v1.ApprovalTicket) time.Time {
	last := t.CreationTimestamp.Time
	for _, a := range t.Status.Approvals {
		if a.Level == t.Status.CurrentLevel && a.At.After(last) {
			last = a.At
		}
	}
	return last
}

// file: cmd/engine/main.go
package main

import (
	"context"
	"fmt"
	"net/http"
	"os"
	"time"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"

	"module-11-approval-workflow/internal/audit"
	"module-11-approval-workflow/internal/engine"
	"module-11-approval-workflow/internal/notify"
	_ "module-11-approval-workflow/internal/metrics" // 注册指标
)

func main() {
	ctx := context.Background()

	// 1) 客户端：优先 in-cluster，否则本地 kubeconfig（Host 指向网关）
	cfg, err := rest.InClusterConfig()
	if err != nil {
		cfg, err = clientConfigFromEnv()
		if err != nil {
			panic(err)
		}
	}
	dyn, err := dynamic.NewForConfig(cfg)
	if err != nil {
		panic(err)
	}

	// 2) 审计（WORM 文件，生产换对象存储追加写）
	worm, err := audit.Open("/var/lib/approval/audit.log")
	if err != nil {
		panic(err)
	}
	defer worm.Close()

	// 3) IM 通知（从环境变量取 webhook）
	notifiers := []notify.Notifier{
		notify.NewWeCom(os.Getenv("WECOM_WEBHOOK")),
		notify.NewDingTalk(os.Getenv("DINGTALK_WEBHOOK")),
	}

	// 4) 引擎
	eng := &engine.Engine{
		Dyn:      dyn,
		Executor: &engine.Executor{GatewayCA: []byte(os.Getenv("GATEWAY_CA"))},
		Audit:    worm,
		Notifier: notifiers,
	}

	// 5) 暴露 /metrics（联邦抓取）
	go func() {
		http.Handle("/metrics", promhttp.Handler())
		if err := http.ListenAndServe(":2112", nil); err != nil {
			fmt.Println("metrics server:", err)
		}
	}()

	// 6) 启动 Reconcile 循环
	ns := os.Getenv("APPROVAL_NAMESPACE")
	if ns == "" {
		ns = "paas-approval"
	}
	if err := eng.Run(ctx, ns, 10*time.Second); err != nil {
		fmt.Println("engine stopped:", err)
	}
}

func clientConfigFromEnv() (*rest.Config, error) {
	// 演示：从 KUBECONFIG 读取（实际生产用 in-cluster，且 Host 指向网关）
	loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
	return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
		loadingRules, &clientcmd.ConfigOverrides{},
	).ClientConfig()
}

# file: config/servicemonitor.yaml
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: approval-engine
  namespace: paas-approval
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
spec:
  selector:
    matchLabels:
      app: approval-engine
  endpoints:
    - port: metrics
      interval: 15s
      # 联邦抓取时自动带上 tenant/app/env/cluster 标签
      relabelings:
        - sourceLabels: [__meta_kubernetes_pod_label_tenant]
          targetLabel: tenant
        - sourceLabels: [__meta_kubernetes_pod_label_app]
          targetLabel: app
        - sourceLabels: [__meta_kubernetes_pod_label_env]
          targetLabel: env
        - sourceLabels: [__meta_kubernetes_pod_label_cluster]
          targetLabel: cluster

# file: deploy/prometheus-rules.yaml
apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: approval-engine-rules
  namespace: paas-approval
  labels:
    tenant: platform
    app: approval-engine
    env: prod
    cluster: paas-control
spec:
  groups:
    - name: approval.sla
      rules:
        # 生产环境审批严重超时：单节点 pending 超 1h
        - alert: ApprovalNodePendingTooLong
          expr: |
            histogram_quantile(0.95, sum by (le, tenant, env, cluster) (
              rate(approval_node_pending_seconds_bucket{env="prod"}[1h])
            )) > 3600            
          for: 10m
          labels:
            severity: critical
            tenant: "{{ $labels.tenant }}"
            env: "{{ $labels.env }}"
            cluster: "{{ $labels.cluster }}"
          annotations:
            summary: "租户 {{ $labels.tenant }} 生产审批节点等待超 1h"
            description: "env={{ $labels.env }} cluster={{ $labels.cluster }} pending p95>3600s"
        # 超时升级频发
        - alert: ApprovalEscalationBurst
          expr: |
            sum by (tenant, env, cluster) (rate(approval_escalation_total[30m])) > 5            
          for: 5m
          labels:
            severity: warning
            tenant: "{{ $labels.tenant }}"
            env: "{{ $labels.env }}"
            cluster: "{{ $labels.cluster }}"
          annotations:
            summary: "租户 {{ $labels.tenant }} 30 分钟内升级超 5 次"
        # 高危敏感操作被批量拒绝
        - alert: SensitiveOpRejected
          expr: |
            increase(approval_request_total{result="rejected", env="prod", risk_level="high"}[1h]) > 3            
          for: 2m
          labels:
            severity: critical
            tenant: "{{ $labels.tenant }}"
            env: "{{ $labels.env }}"
            cluster: "{{ $labels.cluster }}"
          annotations:
            summary: "生产高危操作被批量拒绝，疑似异常"
        # 执行时延过高
        - alert: ApprovalExecutionSlow
          expr: |
            histogram_quantile(0.95, sum by (le, tenant, env, cluster) (
              rate(approval_execution_latency_seconds_bucket{env="prod"}[1h])
            )) > 120            
          for: 10m
          labels:
            severity: warning
            tenant: "{{ $labels.tenant }}"
            env: "{{ $labels.env }}"
            cluster: "{{ $labels.cluster }}"
          annotations:
            summary: "租户 {{ $labels.tenant }} 审批后执行时延 p95>120s"

# 提交一笔「生产发布」三级会签工单（runbook）
cat > /tmp/ticket.yaml <<'EOF'
apiVersion: approval.paas.example.com/v1
kind: ApprovalTicket
metadata:
  name: ap-2026-0001
  namespace: paas-approval
  labels:
    tenant: tenant-a
    app: approval-engine
    env: prod
    cluster: prod-sh
spec:
  applicant: alice@tenant-a
  tenant: tenant-a
  env: prod
  cluster: prod-sh
  namespace: tenant-a-prod
  action: deploy.apply
  riskLevel: high
  policy: and                # 会签：三级全部同意
  chain:
    - level: 1
      role: tenant-admin
      slaSeconds: 1800
    - level: 2
      role: platform-ops
      slaSeconds: 1800
    - level: 3
      role: security-auditor
      slaSeconds: 3600
  summary: "租户A 生产环境 demo-app 发布 v1.4.2（脱敏摘要）"
  payloadRef: "encrypted://bucket/tenant-a/demo-app-v1.4.2.yaml"
EOF
kubectl apply -f /tmp/ticket.yaml
kubectl get approvalticket ap-2026-0001 -n paas-approval -o wide

# 审批人逐级审批（Webhook 接收器 SA 写入，等价于 IM 回调）
# 第 1 级 tenant-admin 同意
kubectl patch approvalticket ap-2026-0001 -n paas-approval --type=merge -p \
  '{"status":{"phase":"Approving","currentLevel":1,"approvals":[{"level":1,"approver":"bob@tenant-admin","decision":"approve"}]}}'
# 第 2 级 platform-ops 同意
kubectl patch approvalticket ap-2026-0001 -n paas-approval --type=merge -p \
  '{"status":{"currentLevel":2,"approvals":[{"level":2,"approver":"carol@platform-ops","decision":"approve"}]}}'
# 第 3 级 security-auditor 同意 -> 引擎 Reconcile 检测到 Approved -> 经网关执行
kubectl patch approvalticket ap-2026-0001 -n paas-approval --type=merge -p \
  '{"status":{"currentLevel":3,"approvals":[{"level":3,"approver":"dave@security","decision":"approve"}]}}'

# 查看执行结果与审计引用
kubectl get approvalticket ap-2026-0001 -n paas-approval \
  -o jsonpath='{.status.phase}{" "}{.status.execResult}{" audit="}{.status.auditRef}{"\n"}'

# 验证「不可绕过」：无凭证直连网关被 403
curl -k -o /dev/null -w "%{http_code}\n" \
  -X POST "https://k8s-gateway.paas.internal/apis/apps/v1/namespaces/tenant-a-prod/deployments" \
  -H "Authorization: Bearer <no-exec-token>"
# 期望 403，且审计中出现一条安全事件

# 校验 WORM 审计链未被篡改
go run ./internal/audit -verify /var/lib/approval/audit.log