Documentation
¶
Overview ¶
Package kube 提供 K8s/GKE 安全检查能力。 本文件实现 GKE 托管层专项基线:通过 GCP Container API 拉取集群配置, 对 CIS GKE Benchmark 关注的托管面控制项(Shielded Nodes / Workload Identity / 私有集群 / Binary Authorization / Release Channel / 节点自动升级等)做评估。 这些控制项不在 K8s API 内,必须读 GCP 侧配置,是区分通用 CIS 与商业级 GKE 加固的关键。
Package kube — Pod Security Standards 检查器 (B10).
K8s 官方 3 个 Profile (PSS v1.27):
- Privileged: 完全不限制 (开发/测试)
- Baseline: 防最常见已知特权升级 (生产最低线)
- Restricted: 当前 PSS 最严 (CIS recommended)
本 checker 给定 PodSpec 返当前等级 + 违例列表.
参考: https://kubernetes.io/docs/concepts/security/pod-security-standards/
Index ¶
- Constants
- Variables
- func Check(spec PodSpec) (Profile, []Violation)
- func EvaluateGKEChecks(cfg *container.Cluster) map[string]string
- func FetchGKECluster(ctx context.Context, projectID, location, clusterName, saJSON string) (*container.Cluster, error)
- type AlarmNotifier
- type Container
- type DetectionRule
- type KubeAlarmService
- type KubeAuditProcessor
- type KubeDetector
- type KubeRuleEngine
- type PodSpec
- type Profile
- type Violation
- type Volume
Constants ¶
const ( GKEShieldedNodes = "CIS-GKE-001" GKEWorkloadIdentity = "CIS-GKE-002" GKEPrivateNodes = "CIS-GKE-003" GKEBinaryAuth = "CIS-GKE-004" GKEMasterAuthNets = "CIS-GKE-005" GKEReleaseChannel = "CIS-GKE-006" GKENodeAutoUpgrade = "CIS-GKE-007" GKENodeAutoRepair = "CIS-GKE-008" GKENetworkPolicy = "CIS-GKE-009" GKELegacyABAC = "CIS-GKE-010" GKEClientCertAuth = "CIS-GKE-011" GKECloudLogging = "CIS-GKE-012" GKESecretsEncryption = "CIS-GKE-013" )
GKE 专项检查项 CheckID(CIS GKE Benchmark 映射,与通用 CIS-K8S-* 区分)
Variables ¶
var GKECheckIDs = []string{ GKEShieldedNodes, GKEWorkloadIdentity, GKEPrivateNodes, GKEBinaryAuth, GKEMasterAuthNets, GKEReleaseChannel, GKENodeAutoUpgrade, GKENodeAutoRepair, GKENetworkPolicy, GKELegacyABAC, GKEClientCertAuth, GKECloudLogging, GKESecretsEncryption, }
GKECheckIDs 全部 GKE 专项检查 ID(供基线 checker 识别 + 结果合并)
Functions ¶
func Check ¶
Check 给定 PodSpec 返最高可达等级 + 所有违例.
算法: 假设 Restricted, 找所有违例; 没 Restricted 违例 → Restricted; 有 → 降到 Baseline, 再检; 仍有 → Privileged.
func EvaluateGKEChecks ¶
EvaluateGKEChecks 对 GKE 集群配置做托管层基线评估,返回 checkID -> "pass"/"fail"。 纯函数(无 IO),可单测。安全特性缺省/未启用一律判 fail(hardening 基线的安全默认)。
Types ¶
type AlarmNotifier ¶
AlarmNotifier 是 KubeAlarmService 向外发送通知的解耦接口。 实现方在 Manager biz 层 (NotificationService 包装), router 启动时注入,避免 engine/kube 循环依赖 biz。
type Container ¶
type Container struct {
Name string
Image string
Privileged bool
AllowPrivilegeEscalation *bool
RunAsUser *int64
RunAsNonRoot *bool
ReadOnlyRootFilesystem *bool
AddedCapabilities []string
DroppedCapabilities []string
SELinuxOptions map[string]string
SeccompProfile string // RuntimeDefault / Localhost / Unconfined / ""
HostPorts []int32
}
Container 安全字段.
type DetectionRule ¶
type DetectionRule struct {
ID string
Name string
Severity string
AlarmType model.KubeAlarmType
Description string
Remediation string // 处置建议
Match func(event *model.AuditEvent) bool
}
DetectionRule 检测规则定义
type KubeAlarmService ¶
type KubeAlarmService struct {
// contains filtered or unexported fields
}
KubeAlarmService 告警服务,负责告警创建、白名单过滤与通知派发。
通知派发通过 AlarmNotifier interface 解耦, 避免 engine/kube 包反向依赖 manager/biz。
func NewKubeAlarmService ¶
func NewKubeAlarmService(db *gorm.DB, logger *zap.Logger) *KubeAlarmService
NewKubeAlarmService 创建告警服务
func (*KubeAlarmService) BatchCreateAlarmsWithFilter ¶
func (s *KubeAlarmService) BatchCreateAlarmsWithFilter(alarms []model.KubeAlarm) (created int, filtered int, err error)
BatchCreateAlarmsWithFilter 批量创建告警(带白名单过滤)
func (*KubeAlarmService) CreateAlarmWithFilter ¶
func (s *KubeAlarmService) CreateAlarmWithFilter(alarm *model.KubeAlarm) (bool, error)
CreateAlarmWithFilter 创建告警前检查白名单,命中则跳过 命中同指纹 pending 告警则 UPSERT(count++、更新 last_seen_at),否则创建新告警 返回 (created bool, err error):created=true 表示新建了告警行(非重复告警)
func (*KubeAlarmService) SetNotifier ¶
func (s *KubeAlarmService) SetNotifier(n AlarmNotifier)
SetNotifier 注入 notifier (启动时调用一次)。
type KubeAuditProcessor ¶
type KubeAuditProcessor struct {
// contains filtered or unexported fields
}
KubeAuditProcessor 审计事件处理器(Webhook 和 Pub/Sub 共用)
func NewKubeAuditProcessor ¶
func NewKubeAuditProcessor(db *gorm.DB, logger *zap.Logger, alarmService *KubeAlarmService) *KubeAuditProcessor
NewKubeAuditProcessor 创建审计事件处理器
func (*KubeAuditProcessor) ProcessAuditEvents ¶
func (p *KubeAuditProcessor) ProcessAuditEvents(cluster model.KubeCluster, events []model.AuditEvent)
ProcessAuditEvents 处理审计事件列表:创建 KubeEvent 记录 + 规则引擎检测生成告警
type KubeDetector ¶
type KubeDetector struct {
// contains filtered or unexported fields
}
KubeDetector Audit Event 规则检测引擎
func NewKubeDetector ¶
func NewKubeDetector(db *gorm.DB, logger *zap.Logger, alarmService *KubeAlarmService) *KubeDetector
NewKubeDetector 创建检测引擎
func (*KubeDetector) DetectAuditEvent ¶
func (d *KubeDetector) DetectAuditEvent(clusterID uint, clusterName string, event *model.AuditEvent)
DetectAuditEvent 对单个 audit event 执行所有规则检测
type KubeRuleEngine ¶
type KubeRuleEngine struct {
// contains filtered or unexported fields
}
KubeRuleEngine CEL-based K8s 基线规则引擎
func NewKubeRuleEngine ¶
func NewKubeRuleEngine(logger *zap.Logger) (*KubeRuleEngine, error)
NewKubeRuleEngine 创建 K8s CEL 规则引擎
func (*KubeRuleEngine) CompileExpression ¶
func (e *KubeRuleEngine) CompileExpression(expression string) error
CompileExpression 编译并验证 CEL 表达式(供 API 层调用)
func (*KubeRuleEngine) EvaluateRule ¶
func (e *KubeRuleEngine) EvaluateRule(ctx context.Context, clientset *kubernetes.Clientset, config *model.KubeCheckConfig) (string, model.AffectedResources)
EvaluateRule 对指定集群执行 CEL 规则检查
type PodSpec ¶
type PodSpec struct {
HostNetwork bool
HostPID bool
HostIPC bool
HostUsers *bool // nil = 集群默认
Containers []Container
InitContainers []Container
Volumes []Volume
}
PodSpec 简化的 Pod 安全字段视图.
不引 k8s.io/api 重型包, 调用方自行从 corev1.Pod 抽取后传入.