flowchart TD
    subgraph "认证层"
        CertAuth[证书认证器<br/>ClientCertAuthenticator]
        JWTAuth[JWT认证器<br/>JwtAuthenticator]
        XFCCAuth[XFCC认证器<br/>XfccAuthenticator]
        OIDCAuth[OIDC认证器<br/>OidcAuthenticator]
    end
    
    subgraph "证书管理层"
        CAServer[CA服务器<br/>Server]
        CertAuthority[证书颁发机构<br/>CertificateAuthority]
        KeyCertBundle[密钥证书捆绑<br/>KeyCertBundle]
        NodeAuth[节点授权器<br/>MulticlusterNodeAuthorizor]
    end
    
    subgraph "密钥分发层"
        SDSServer[SDS服务器<br/>Server]
        SDSService[SDS服务<br/>sdsservice]
        SecretManager[密钥管理器<br/>SecretManager]
        SecretCache[密钥缓存<br/>SecretCache]
    end
    
    subgraph "存储层"
        K8sSecrets[Kubernetes Secrets]
        FileSystem[文件系统证书]
        ExternalCA[外部CA系统]
    end
    
    subgraph "客户端层"
        Envoy[Envoy代理]
        Workload[工作负载]
        IstioAgent[Istio Agent]
    end
    
    %% 认证流程
    Envoy -->|证书请求| CertAuth
    Workload -->|JWT令牌| JWTAuth
    IstioAgent -->|XFCC头部| XFCCAuth
    
    CertAuth -->|认证结果| CAServer
    JWTAuth -->|认证结果| CAServer
    XFCCAuth -->|认证结果| CAServer
    OIDCAuth -->|认证结果| CAServer
    
    %% 证书签发流程
    CAServer -->|验证授权| NodeAuth
    CAServer -->|签发证书| CertAuthority
    CertAuthority -->|密钥操作| KeyCertBundle
    CertAuthority -->|证书存储| K8sSecrets
    
    %% 密钥分发流程
    SecretManager -->|监听变更| K8sSecrets
    SecretManager -->|加载证书| FileSystem
    SecretManager -->|获取证书| ExternalCA
    SecretManager -->|更新缓存| SecretCache
    SecretCache -->|提供密钥| SDSService
    SDSService -->|SDS协议| SDSServer
    SDSServer -->|推送证书| Envoy
    
    %% 证书轮换
    CAServer -.->|定时轮换| SecretManager
    SecretManager -.->|通知更新| SDSService
    SDSService -.->|推送更新| Envoy
    
    %% 样式定义
    classDef auth fill:#e3f2fd
    classDef ca fill:#f3e5f5
    classDef sds fill:#e8f5e8
    classDef storage fill:#fff3e0
    classDef client fill:#fce4ec
    
    class CertAuth,JWTAuth,XFCCAuth,OIDCAuth auth
    class CAServer,CertAuthority,KeyCertBundle,NodeAuth ca
    class SDSServer,SDSService,SecretManager,SecretCache sds
    class K8sSecrets,FileSystem,ExternalCA storage
    class Envoy,Workload,IstioAgent client

type Server struct {
    pb.UnimplementedIstioCertificateServiceServer
    monitoring     monitoringMetrics
    Authenticators []security.Authenticator
    ca             CertificateAuthority
    serverCertTTL  time.Duration
    nodeAuthorizer *MulticlusterNodeAuthorizor
}

type CertificateAuthority interface {
    Sign(csrPEM []byte, opts ca.CertOpts) ([]byte, error)
    SignWithCertChain(csrPEM []byte, opts ca.CertOpts) ([]string, error)
    GetCAKeyCertBundle() *util.KeyCertBundle
}

// 基本信息
// 名称: `CreateCertificate`
// 协议/方法: gRPC `IstioCertificateService.CreateCertificate`
// 幂等性: 否（每次调用生成不同的证书）

// 请求结构体
type IstioCertificateRequest struct {
    // 证书签名请求，PEM格式
    Csr string
    // 证书有效期（秒）
    ValidityDuration int64
    // 请求元数据，包含身份信息
    Metadata *google_protobuf.Struct
}

// 响应结构体
type IstioCertificateResponse struct {
    // 证书链，PEM格式数组
    CertChain []string
}

func (s *Server) CreateCertificate(ctx context.Context, request *pb.IstioCertificateRequest) (*pb.IstioCertificateResponse, error) {
    // 1) 客户端身份认证
    caller, err := security.Authenticate(ctx, s.Authenticators)
    if caller == nil || err != nil {
        s.monitoring.AuthnError.Increment()
        return nil, status.Error(codes.Unauthenticated, "request authenticate failure")
    }
    
    // 2) 处理身份冒充授权（如果存在）
    sans := caller.Identities
    impersonatedIdentity := request.Metadata.GetFields()[security.ImpersonatedIdentity].GetStringValue()
    if impersonatedIdentity != "" {
        if s.nodeAuthorizer == nil {
            return nil, status.Error(codes.Unauthenticated, "request impersonation authentication failure")
        }
        if err := s.nodeAuthorizer.authenticateImpersonation(ctx, caller.KubernetesInfo, impersonatedIdentity); err != nil {
            return nil, status.Error(codes.Unauthenticated, "request impersonation authentication failure")
        }
        sans = []string{impersonatedIdentity}
    }
    
    // 3) 构建证书选项
    certOpts := ca.CertOpts{
        SubjectIDs: sans,
        TTL:        time.Duration(request.ValidityDuration) * time.Second,
        ForCA:      false,
        CertSigner: request.Metadata.GetFields()[security.CertSigner].GetStringValue(),
    }
    
    // 4) 签发证书
    var signErr error
    var respCertChain []string
    if certOpts.CertSigner == "" {
        cert, signErr := s.ca.Sign([]byte(request.Csr), certOpts)
        if signErr == nil {
            respCertChain = []string{string(cert)}
        }
    } else {
        respCertChain, signErr = s.ca.SignWithCertChain([]byte(request.Csr), certOpts)
    }
    
    if signErr != nil {
        return nil, status.Errorf(signErr.(*caerror.Error).HTTPErrorCode(), "CSR signing error (%v)", signErr.(*caerror.Error))
    }
    
    // 5) 构建响应，扁平化证书链
    response := &pb.IstioCertificateResponse{}
    for _, pem := range respCertChain {
        for _, cert := range util.PemCertBytestoString([]byte(pem)) {
            response.CertChain = append(response.CertChain, cert+"\n")
        }
    }
    
    return response, nil
}

type sdsservice struct {
    st security.SecretManager
    stop       chan struct{}
    rootCaPath string
    pkpConf    *mesh.PrivateKeyProvider
    sync.Mutex
    clients    map[string]*Context
}

type Context struct {
    BaseConnection xds.Connection
    s              *sdsservice
    w              *Watch
}

// 基本信息
// 名称: `StreamSecrets`
// 协议/方法: gRPC Stream `SecretDiscoveryService.StreamSecrets`
// 幂等性: 是（相同请求返回相同证书）

// 请求处理核心逻辑
func (c *Context) Process(req *discovery.DiscoveryRequest) error {
    // 1) 判断是否需要响应
    shouldRespond, delta := xds.ShouldRespond(c.Watcher(), c.XdsConnection().ID(), req)
    if !shouldRespond {
        return nil
    }
    
    // 2) 确定需要生成的资源
    resources := req.ResourceNames
    if !delta.IsEmpty() {
        resources = delta.Subscribed.UnsortedList()
    }
    
    // 3) 生成SDS响应
    res, err := c.s.generate(resources)
    if err != nil {
        return err
    }
    
    // 4) 发送给客户端
    return xds.Send(c, res)
}

// 主动推送更新
func (c *Context) Push(ev any) error {
    secretName := ev.(string)
    
    // 检查客户端是否订阅了该证书
    if !c.w.requested(secretName) {
        return nil
    }
    
    // 生成并推送证书更新
    res, err := c.s.generate([]string{secretName})
    if err != nil {
        return err
    }
    return xds.Send(c, res)
}

type Authenticator interface {
    AuthenticatorType() string
    Authenticate(authCtx AuthContext) (*Caller, error)
}

type Caller struct {
    Identities        []string
    KubernetesInfo    KubernetesInfo  
    CredentialType    string
}

type ClientCertAuthenticator struct{}

func (cca *ClientCertAuthenticator) Authenticate(authCtx security.AuthContext) (*security.Caller, error) {
    if authCtx.GrpcContext != nil {
        return cca.authenticateGrpc(authCtx.GrpcContext)
    }
    if authCtx.Request != nil {
        return cca.authenticateHTTP(authCtx.Request)
    }
    return nil, nil
}

func (cca *ClientCertAuthenticator) authenticateGrpc(ctx context.Context) (*security.Caller, error) {
    peer, ok := peer.FromContext(ctx)
    if !ok {
        return nil, fmt.Errorf("no peer info")
    }
    
    if tlsInfo, ok := peer.AuthInfo.(credentials.TLSInfo); ok {
        // 验证证书链和提取身份信息
        chains := tlsInfo.State.VerifiedChains
        if len(chains) == 0 || len(chains[0]) == 0 {
            return nil, fmt.Errorf("no verified chain")
        }
        
        cert := chains[0][0]
        caller := &security.Caller{
            Identities: util.ExtractIDsFromCert(cert),
            CredentialType: "mtls",
        }
        
        return caller, nil
    }
    
    return nil, fmt.Errorf("no TLS info")
}

sequenceDiagram
    autonumber
    participant Envoy as Envoy代理
    participant IstioAgent as Istio Agent
    participant CAServer as CA服务器
    participant CertAuth as 证书认证器
    participant NodeAuth as 节点授权器
    participant CertAuthority as 证书颁发机构
    participant K8sAPI as Kubernetes API
    
    Note over Envoy,K8sAPI: 证书请求阶段
    Envoy->>IstioAgent: 请求工作负载证书
    IstioAgent->>IstioAgent: 生成CSR和私钥
    IstioAgent->>CAServer: 发送CreateCertificate请求
    
    Note over CAServer,NodeAuth: 身份认证和授权阶段
    CAServer->>CertAuth: 验证客户端证书
    CertAuth->>CertAuth: 提取客户端身份信息
    CertAuth->>CAServer: 返回认证结果
    
    alt 存在身份冒充
        CAServer->>NodeAuth: 验证冒充授权
        NodeAuth->>K8sAPI: 查询节点授权策略
        K8sAPI->>NodeAuth: 返回授权信息
        NodeAuth->>CAServer: 授权验证结果
    end
    
    Note over CAServer,K8sAPI: 证书签发阶段
    CAServer->>CertAuthority: 调用证书签发接口
    CertAuthority->>CertAuthority: 使用CA私钥签发证书
    CertAuthority->>CAServer: 返回签名证书链
    CAServer->>IstioAgent: 返回证书响应
    
    Note over IstioAgent,K8sAPI: 证书存储阶段
    IstioAgent->>IstioAgent: 验证证书有效性
    IstioAgent->>K8sAPI: 存储证书到Secret
    K8sAPI->>IstioAgent: 确认存储成功
    IstioAgent->>Envoy: 通知证书就绪

sequenceDiagram
    autonumber
    participant Envoy as Envoy代理
    participant SDSServer as SDS服务器
    participant SecretManager as 密钥管理器
    participant SecretCache as 密钥缓存
    participant K8sAPI as Kubernetes API
    
    Note over Envoy,K8sAPI: SDS连接建立阶段
    Envoy->>SDSServer: 建立gRPC流连接
    SDSServer->>SDSServer: 创建客户端上下文
    Envoy->>SDSServer: 发送DiscoveryRequest
    
    Note over SDSServer,K8sAPI: 证书获取阶段
    SDSServer->>SecretManager: 请求指定证书
    
    alt 缓存命中
        SecretManager->>SecretCache: 查询缓存
        SecretCache->>SecretManager: 返回缓存证书
    else 缓存未命中
        SecretManager->>K8sAPI: 从Secret获取证书
        K8sAPI->>SecretManager: 返回证书数据
        SecretManager->>SecretCache: 更新缓存
    end
    
    SecretManager->>SDSServer: 返回证书内容
    
    Note over SDSServer,Envoy: 证书推送阶段
    SDSServer->>SDSServer: 构建SDS响应
    SDSServer->>Envoy: 发送DiscoveryResponse
    Envoy->>SDSServer: 发送ACK确认
    
    Note over SecretManager,Envoy: 证书更新阶段
    K8sAPI->>SecretManager: 证书变更通知
    SecretManager->>SecretCache: 更新缓存条目
    SecretManager->>SDSServer: 触发推送事件
    SDSServer->>Envoy: 推送证书更新
    Envoy->>SDSServer: 确认更新接收

func (sc *SecretCache) shouldRefresh(certItem *SecretItem) bool {
    // 获取证书过期时间
    cert, err := parseWorkloadCertificate(certItem.CertificateChain)
    if err != nil {
        return true // 解析失败，强制刷新
    }
    
    now := time.Now()
    // 证书在有效期50%时开始预轮换
    threshold := cert.NotBefore.Add(cert.NotAfter.Sub(cert.NotBefore) / 2)
    
    // 添加随机抖动，避免同时轮换
    jitter := time.Duration(rand.Int63n(int64(time.Hour)))
    return now.After(threshold.Add(-jitter))
}

// 证书轮换调度器
func (sc *SecretCache) startCertRotationScheduler() {
    ticker := time.NewTicker(10 * time.Minute) // 每10分钟检查一次
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            sc.rotateExpiredCertificates()
        case <-sc.stopCh:
            return
        }
    }
}

func (sc *SecretCache) rotateExpiredCertificates() {
    sc.mu.RLock()
    itemsToRotate := make([]*SecretItem, 0)
    
    for _, item := range sc.cache {
        if sc.shouldRefresh(item) {
            itemsToRotate = append(itemsToRotate, item)
        }
    }
    sc.mu.RUnlock()
    
    // 并发轮换多个证书
    var wg sync.WaitGroup
    semaphore := make(chan struct{}, 10) // 限制并发数
    
    for _, item := range itemsToRotate {
        wg.Add(1)
        go func(secretName string) {
            defer wg.Done()
            semaphore <- struct{}{}
            defer func() { <-semaphore }()
            
            // 重新生成证书
            sc.GenerateSecret(secretName)
        }(item.ResourceName)
    }
    
    wg.Wait()
}

func (s *sdsservice) calculateDelta(clientState map[string]string, currentSecrets map[string]*SecretItem) (*DeltaUpdate, error) {
    delta := &DeltaUpdate{
        Updated: make(map[string]*SecretItem),
        Deleted: make([]string, 0),
    }
    
    // 检查更新和新增的证书
    for name, secret := range currentSecrets {
        clientVersion, exists := clientState[name]
        if !exists || clientVersion != secret.Version {
            delta.Updated[name] = secret
        }
    }
    
    // 检查已删除的证书
    for name := range clientState {
        if _, exists := currentSecrets[name]; !exists {
            delta.Deleted = append(delta.Deleted, name)
        }
    }
    
    return delta, nil
}

// 增量推送实现
func (s *sdsservice) pushDeltaUpdate(ctx *Context, delta *DeltaUpdate) error {
    resources := make([]*any.Any, 0, len(delta.Updated))
    removedResources := delta.Deleted
    
    // 构建更新的资源
    for name, secret := range delta.Updated {
        tlsSecret := &tls.Secret{
            Name: name,
            Type: &tls.Secret_TlsCertificate{
                TlsCertificate: &tls.TlsCertificate{
                    CertificateChain: &core.DataSource{
                        Specifier: &core.DataSource_InlineBytes{
                            InlineBytes: secret.CertificateChain,
                        },
                    },
                    PrivateKey: &core.DataSource{
                        Specifier: &core.DataSource_InlineBytes{
                            InlineBytes: secret.PrivateKey,
                        },
                    },
                },
            },
        }
        
        resource, _ := anypb.New(tlsSecret)
        resources = append(resources, resource)
    }
    
    // 发送增量响应
    response := &discovery.DeltaDiscoveryResponse{
        SystemVersionInfo: version.GetString(),
        Resources:         resources, 
        RemovedResources:  removedResources,
        TypeUrl:          SecretTypeURL,
        Nonce:            generateNonce(),
    }
    
    return ctx.SendDelta(response)
}

type MultiClusterCertSync struct {
    clusters       map[string]*ClusterClient
    syncInterval   time.Duration
    conflictResolver CertConflictResolver
}

func (mcs *MultiClusterCertSync) syncCertificatesAcrossClusters() {
    ticker := time.NewTicker(mcs.syncInterval)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            mcs.performSync()
        }
    }
}

func (mcs *MultiClusterCertSync) performSync() {
    // 1) 收集所有集群的证书状态
    clusterStates := make(map[string]map[string]*CertInfo)
    
    for clusterID, client := range mcs.clusters {
        certs, err := client.ListCertificates()
        if err != nil {
            log.Warnf("无法获取集群 %s 的证书: %v", clusterID, err)
            continue
        }
        clusterStates[clusterID] = certs
    }
    
    // 2) 计算需要同步的证书
    syncPlan := mcs.calculateSyncPlan(clusterStates)
    
    // 3) 执行同步操作
    mcs.executeSyncPlan(syncPlan)
}

func (mcs *MultiClusterCertSync) calculateSyncPlan(states map[string]map[string]*CertInfo) *SyncPlan {
    plan := &SyncPlan{
        Create: make(map[string]map[string]*CertInfo),
        Update: make(map[string]map[string]*CertInfo),
        Delete: make(map[string][]string),
    }
    
    // 找到全局最新版本的证书
    globalLatest := make(map[string]*CertInfo)
    for _, clusterCerts := range states {
        for certName, certInfo := range clusterCerts {
            if existing, exists := globalLatest[certName]; !exists || certInfo.Version > existing.Version {
                globalLatest[certName] = certInfo
            }
        }
    }
    
    // 为每个集群计算需要的操作
    for clusterID, clusterCerts := range states {
        plan.Create[clusterID] = make(map[string]*CertInfo)
        plan.Update[clusterID] = make(map[string]*CertInfo)
        
        for certName, latestCert := range globalLatest {
            if clusterCert, exists := clusterCerts[certName]; exists {
                if clusterCert.Version < latestCert.Version {
                    plan.Update[clusterID][certName] = latestCert
                }
            } else {
                plan.Create[clusterID][certName] = latestCert
            }
        }
        
        // 检查需要删除的证书
        for certName := range clusterCerts {
            if _, exists := globalLatest[certName]; !exists {
                plan.Delete[clusterID] = append(plan.Delete[clusterID], certName)
            }
        }
    }
    
    return plan
}

// 自定义API密钥认证器
type APIKeyAuthenticator struct {
    validKeys map[string]*KeyInfo
    mutex     sync.RWMutex
}

type KeyInfo struct {
    Identity   string
    Namespace  string
    Expiry     time.Time
    Scopes     []string
}

func (aka *APIKeyAuthenticator) AuthenticatorType() string {
    return "APIKeyAuthenticator"
}

func (aka *APIKeyAuthenticator) Authenticate(authCtx security.AuthContext) (*security.Caller, error) {
    var apiKey string
    
    // 从gRPC元数据中提取API密钥
    if authCtx.GrpcContext != nil {
        if md, ok := metadata.FromIncomingContext(authCtx.GrpcContext); ok {
            if keys := md.Get("x-api-key"); len(keys) > 0 {
                apiKey = keys[0]
            }
        }
    }
    
    // 从HTTP头部中提取API密钥
    if authCtx.Request != nil {
        apiKey = authCtx.Request.Header.Get("X-API-Key")
    }
    
    if apiKey == "" {
        return nil, fmt.Errorf("missing API key")
    }
    
    // 验证API密钥
    aka.mutex.RLock()
    keyInfo, exists := aka.validKeys[apiKey]
    aka.mutex.RUnlock()
    
    if !exists {
        return nil, fmt.Errorf("invalid API key")
    }
    
    // 检查密钥是否过期
    if time.Now().After(keyInfo.Expiry) {
        return nil, fmt.Errorf("API key expired")
    }
    
    return &security.Caller{
        Identities: []string{keyInfo.Identity},
        KubernetesInfo: security.KubernetesInfo{
            Namespace: keyInfo.Namespace,
        },
        CredentialType: "api-key",
    }, nil
}

// 注册自定义认证器
func setupCustomAuthentication(caServer *ca.Server) {
    apiKeyAuth := &APIKeyAuthenticator{
        validKeys: loadAPIKeysFromConfig(),
    }
    
    // 将自定义认证器添加到CA服务器
    caServer.Authenticators = append(caServer.Authenticators, apiKeyAuth)
    
    // 定期刷新API密钥配置
    go func() {
        ticker := time.NewTicker(5 * time.Minute)
        defer ticker.Stop()
        
        for range ticker.C {
            newKeys := loadAPIKeysFromConfig()
            apiKeyAuth.mutex.Lock()
            apiKeyAuth.validKeys = newKeys
            apiKeyAuth.mutex.Unlock()
        }
    }()
}

// 证书健康检查器
type CertificateHealthChecker struct {
    secretManager security.SecretManager
    alertManager  AlertManager
    checkInterval time.Duration
}

func (chc *CertificateHealthChecker) StartMonitoring() {
    ticker := time.NewTicker(chc.checkInterval)
    defer ticker.Stop()
    
    for range ticker.C {
        chc.checkCertificateHealth()
    }
}

func (chc *CertificateHealthChecker) checkCertificateHealth() {
    // 检查工作负载证书
    workloadCert, err := chc.secretManager.GenerateSecret(security.WorkloadKeyCertResourceName)
    if err != nil {
        chc.alertManager.SendAlert(&Alert{
            Level:   "critical",
            Message: fmt.Sprintf("工作负载证书获取失败: %v", err),
            Type:    "certificate_error",
        })
        return
    }
    
    // 解析证书并检查有效期
    cert, err := parseWorkloadCertificate(workloadCert.CertificateChain)
    if err != nil {
        chc.alertManager.SendAlert(&Alert{
            Level:   "critical", 
            Message: fmt.Sprintf("证书解析失败: %v", err),
            Type:    "certificate_parse_error",
        })
        return
    }
    
    // 检查证书即将过期
    now := time.Now()
    timeToExpiry := cert.NotAfter.Sub(now)
    
    if timeToExpiry < 24*time.Hour {
        chc.alertManager.SendAlert(&Alert{
            Level:   "warning",
            Message: fmt.Sprintf("证书将在 %v 后过期", timeToExpiry),
            Type:    "certificate_expiring",
        })
    }
    
    // 检查根证书
    rootCert, err := chc.secretManager.GenerateSecret(security.RootCertReqResourceName)
    if err != nil {
        chc.alertManager.SendAlert(&Alert{
            Level:   "critical",
            Message: fmt.Sprintf("根证书获取失败: %v", err),
            Type:    "root_certificate_error", 
        })
    }
    
    // 记录健康指标
    certExpiryGauge.With(prometheus.Labels{
        "cert_type": "workload",
    }).Set(timeToExpiry.Seconds())
}

// Prometheus指标定义
var (
    certExpiryGauge = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "istio_cert_expiry_seconds",
            Help: "证书剩余有效时间（秒）",
        },
        []string{"cert_type"},
    )
    
    certRotationCounter = prometheus.NewCounterVec(
        prometheus.CounterOpts{
            Name: "istio_cert_rotations_total", 
            Help: "证书轮换总次数",
        },
        []string{"cert_type", "status"},
    )
)

// 高可用CA管理器
type HACAManager struct {
    nodes        []CANode
    leader       *CANode
    leaderElection *LeaderElection
    healthChecker *HealthChecker
}

type CANode struct {
    ID       string
    Address  string
    IsLeader bool
    IsHealthy bool
    LastSeen time.Time
}

func (ha *HACAManager) Start() error {
    // 1) 启动健康检查
    go ha.healthChecker.Start()
    
    // 2) 启动领导者选举
    go ha.leaderElection.Start()
    
    // 3) 监听角色变更
    go ha.handleRoleChanges()
    
    return nil
}

func (ha *HACAManager) handleRoleChanges() {
    for event := range ha.leaderElection.Events() {
        switch event.Type {
        case "leader_elected":
            ha.becomeLeader(event.NodeID)
        case "leader_lost":
            ha.becomeFollower()
        case "node_failed":
            ha.handleNodeFailure(event.NodeID)
        }
    }
}

func (ha *HACAManager) becomeLeader(nodeID string) {
    log.Infof("节点 %s 成为CA领导者", nodeID)
    
    if nodeID == ha.getCurrentNodeID() {
        // 当前节点成为领导者
        ha.startCAServices()
        
        // 同步其他节点的状态
        go ha.syncFollowerNodes()
    } else {
        // 其他节点成为领导者，当前节点转为跟随者
        ha.stopCAServices()
        ha.leader = ha.findNodeByID(nodeID)
    }
}

func (ha *HACAManager) startCAServices() {
    // 启动CA服务器
    go func() {
        caServer := ca.NewServer(/* 配置参数 */)
        if err := caServer.Start(); err != nil {
            log.Errorf("CA服务启动失败: %v", err)
        }
    }()
    
    // 启动SDS服务器  
    go func() {
        sdsServer := sds.NewServer(/* 配置参数 */)
        if err := sdsServer.Start(); err != nil {
            log.Errorf("SDS服务启动失败: %v", err)
        }
    }()
}

// 证书状态同步
func (ha *HACAManager) syncCertificateState() {
    if !ha.isLeader() {
        return
    }
    
    // 同步证书到所有跟随者节点
    for _, node := range ha.nodes {
        if node.ID == ha.getCurrentNodeID() || !node.IsHealthy {
            continue
        }
        
        go func(targetNode CANode) {
            client := ha.createNodeClient(targetNode)
            defer client.Close()
            
            // 获取本地证书列表
            localCerts := ha.getLocalCertificates()
            
            // 同步到目标节点
            for _, cert := range localCerts {
                if err := client.SyncCertificate(cert); err != nil {
                    log.Warnf("向节点 %s 同步证书失败: %v", targetNode.ID, err)
                }
            }
        }(node)
    }
}

// 零信任策略执行器
type ZeroTrustPolicyEnforcer struct {
    ca            *ca.Server
    policyStore   PolicyStore
    metricsCollector MetricsCollector
}

func (zte *ZeroTrustPolicyEnforcer) EnforceCertificatePolicy(ctx context.Context, request *pb.IstioCertificateRequest) (*pb.IstioCertificateResponse, error) {
    // 1) 提取请求者身份
    caller, err := security.Authenticate(ctx, zte.ca.Authenticators)
    if err != nil {
        zte.metricsCollector.RecordAuthFailure("authentication", err.Error())
        return nil, status.Error(codes.Unauthenticated, "authentication failed")
    }
    
    // 2) 获取适用的零信任策略
    policies := zte.policyStore.GetPoliciesForIdentity(caller.Identities[0])
    
    // 3) 验证请求是否符合策略
    for _, policy := range policies {
        if err := zte.validateAgainstPolicy(request, caller, policy); err != nil {
            zte.metricsCollector.RecordAuthFailure("policy_violation", err.Error())
            return nil, status.Error(codes.PermissionDenied, err.Error())
        }
    }
    
    // 4) 动态调整证书属性
    adjustedRequest := zte.adjustCertificateRequest(request, policies)
    
    // 5) 调用原始证书创建逻辑
    response, err := zte.ca.CreateCertificate(ctx, adjustedRequest)
    if err != nil {
        return nil, err
    }
    
    // 6) 记录审计日志
    zte.auditCertificateIssuance(caller, adjustedRequest, response)
    
    return response, nil
}

func (zte *ZeroTrustPolicyEnforcer) validateAgainstPolicy(request *pb.IstioCertificateRequest, caller *security.Caller, policy *ZeroTrustPolicy) error {
    // 验证时间窗口
    if policy.TimeRestrictions != nil {
        now := time.Now()
        if now.Before(policy.TimeRestrictions.NotBefore) || now.After(policy.TimeRestrictions.NotAfter) {
            return fmt.Errorf("请求超出允许的时间窗口")
        }
    }
    
    // 验证证书有效期限制
    maxTTL := time.Duration(policy.MaxCertificateTTL) * time.Second
    requestedTTL := time.Duration(request.ValidityDuration) * time.Second
    if requestedTTL > maxTTL {
        return fmt.Errorf("请求的证书有效期 %v 超过策略允许的最大值 %v", requestedTTL, maxTTL)
    }
    
    // 验证请求频率限制
    if policy.RateLimit != nil {
        key := fmt.Sprintf("cert_requests:%s", caller.Identities[0])
        if !zte.checkRateLimit(key, policy.RateLimit) {
            return fmt.Errorf("请求频率超过限制")
        }
    }
    
    // 验证网络位置限制
    if policy.NetworkRestrictions != nil {
        clientIP := security.GetConnectionAddress(context.Background()) // 简化实现
        if !zte.isAllowedNetwork(clientIP, policy.NetworkRestrictions) {
            return fmt.Errorf("请求来源网络不在允许范围内")
        }
    }
    
    return nil
}

// 证书使用监控
func (zte *ZeroTrustPolicyEnforcer) monitorCertificateUsage() {
    // 定期检查证书使用情况
    ticker := time.NewTicker(1 * time.Hour)
    defer ticker.Stop()
    
    for range ticker.C {
        // 分析证书使用模式
        usagePatterns := zte.analyzeCertificateUsage()
        
        // 检测异常使用模式
        for identity, pattern := range usagePatterns {
            if zte.isAnomalousUsage(pattern) {
                alert := &SecurityAlert{
                    Level:     "high",
                    Type:      "anomalous_certificate_usage",
                    Identity:  identity,
                    Details:   pattern,
                    Timestamp: time.Now(),
                }
                zte.sendSecurityAlert(alert)
            }
        }
        
        // 更新风险评分
        zte.updateRiskScores(usagePatterns)
    }
}