云原生环境下服务网格的零信任安全架构

graph TD
A[工作负载] --> B[Sidecar代理]
B --> C[控制平面]
C --> D[安全策略引擎]subgraph 数据平面
B1[Envoy] --> B2[mTLS隧道]
B3[策略执行点] --> B4[审计日志]
endsubgraph 控制平面
C1[Istiod] --> C2[证书管理]
C3[策略分发] --> C4[拓扑发现]
endsubgraph 安全组件
D1[OPA策略引擎] --> D2[密钥管理]
D3[威胁情报] --> D4[行为分析]
end

package mainimport ("istio.io/istio/pilot/pkg/xds""istio.io/istio/security/pkg/nodeagent/cache"
)func setupAutoMTLS(workloadID string) {// 自动签发工作负载证书certManager := cache.NewSecretManager(xds.NewXDSClient(),"istio-system",)cert, key := certManager.GenerateCert(workloadID, 24*time.Hour)// Envoy配置注入envoyConfig := &envoy_bootstrap.Bootstrap{StaticResources: &envoy_config_bootstrap.Bootstrap_StaticResources{Clusters: []*envoy_config_cluster.Cluster{{Name: "istiod",TransportSocket: &envoy_config_core.TransportSocket{Name: "tls",ConfigType: &envoy_config_core.TransportSocket_TypedConfig{TypedConfig: util.MessageToAny(&envoy_extensions_transport_sockets_tls_v3.UpstreamTlsContext{CommonTlsContext: &envoy_extensions_transport_sockets_tls_v3.CommonTlsContext{TlsCertificates: []*envoy_extensions_transport_sockets_tls_v3.TlsCertificate{{CertificateChain: &envoy_config_core.DataSource{Specifier: &envoy_config_core.DataSource_InlineBytes{InlineBytes: cert,},},PrivateKey: &envoy_config_core.DataSource{Specifier: &envoy_config_core.DataSource_InlineBytes{InlineBytes: key,},},},},},}),},},},},},}applyEnvoyConfig(envoyConfig)
}

package istio.authzimport input.attributes.request.http as http_requestdefault allow = false# 允许同命名空间服务访问
allow {source.namespace == destination.namespacehttp_request.method == "GET"
}# 允许特定服务跨命名空间访问
allow {source.workload == "frontend"destination.workload == "payment-service"http_request.path == "/v1/charge"
}# 拒绝异常行为
deny {count(http_request.headers["x-forwarded-for"]) > 3  # 防止IP伪造
}deny {http_request.body_size > 1048576  # 阻止大文件上传
}

apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:name: topology-aware
spec:selector:matchLabels:app: databaseaction: ALLOWrules:- from:- source:namespaces: ["payment-ns"]to:- operation:ports: ["3306"]methods: ["SELECT"]when:- key: request.timevalues: ["09:00-18:00"]

from prometheus_api_client import PrometheusConnectdef auto_tune_policies():prom = PrometheusConnect(url="http://prometheus:9090")# 检测异常访问模式anomaly_query = 'rate(istio_request_count{response_code=~"5.."}[5m]) > 10'anomalies = prom.custom_query(anomaly_query)for anomaly in anomalies:src_svc = anomaly['metric']['source_app']dst_svc = anomaly['metric']['destination_app']# 自动添加临时拒绝规则apply_emergency_policy(source=src_svc,destination=dst_svc,action="DENY",duration="15m")# 触发安全审计trigger_audit(src_svc)

import tensorflow as tf
from tensorflow.keras.layers import LSTM, Dense# 构建LSTM异常检测模型
def build_behavior_model(input_dim):model = tf.keras.Sequential([LSTM(64, input_shape=(None, input_dim), return_sequences=True),LSTM(32),Dense(16, activation='relu'),Dense(1, activation='sigmoid')])model.compile(loss='binary_crossentropy', optimizer='adam')return model# 训练行为基线
def train_behavior_baseline(service_logs):# 提取特征：请求频率、响应大小、错误率等features = extract_features(service_logs)model = build_behavior_model(features.shape[1])model.fit(features, epochs=10)return model

func detectAnomalies(stream securityv1.SecurityService_StreamTelemetryServer) {baselineModel := loadModel("/models/behavior-baseline")for {telemetry, _ := stream.Recv()// 提取特征向量features := []float32{float32(telemetry.RequestCount),float32(telemetry.ErrorRate),float32(telemetry.ResponseSizeAvg),}// 模型预测prediction := baselineModel.Predict(features)if prediction > 0.85 { // 异常阈值triggerAlert(telemetry)enforceQuarantine(telemetry.SourceWorkload)}}
}