fix: harden validation container volume mounts and support private sigstore instances

internal/webhooks/pod_webhook.go

@@ -103,7 +103,7 @@
 	mv := &v1alpha1.ModelValidation{}
 	err := p.client.Get(ctx, client.ObjectKey{Name: modelValidationName, Namespace: pod.Namespace}, mv)
 	if err != nil {
 		logger.Error(err, "failed to get ModelValidation CR", "namespace", pod.Namespace, "modelValidation", modelValidationName)
 		return admission.Errored(http.StatusBadRequest, err) // Fail deployment if CR not found
 	}
 	// NOTE: check if validation sidecar is already injected. Then no action needed.
@@ -141,10 +141,8 @@
 		logger.Error(err, "failed to find TelemetryConfig, proceeding without telemetry")
 	}
 
-	vm := []corev1.VolumeMount{}
-	for _, c := range pod.Spec.Containers {
-		vm = append(vm, c.VolumeMounts...)
-	}
+	neededPaths := collectNeededPaths(mergedModel, mv.Spec.Config)
+	vm := filterVolumeMounts(pod.Spec.Containers, neededPaths)
 
 	continuousEnabled := mv.Spec.ContinuousValidation != nil && mv.Spec.ContinuousValidation.Enabled
 	useLegacySidecar := continuousEnabled && !p.nativeSidecarSupport
@@ -153,6 +151,17 @@
 		logger.Info("Using legacy sidecar for continuous validation (native sidecars not supported)")
 	}
 
+	if mv.Spec.Config.SigstoreConfig != nil {
+		const tufVolName = "sigstore-tuf-cache"
+		pp.Spec.Volumes = append(pp.Spec.Volumes, corev1.Volume{
+			Name: tufVolName,
+			VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{
+				SizeLimit: ptr.To(resource.MustParse("10Mi")),
+			}},
+		})
+		vm = append(vm, corev1.VolumeMount{Name: tufVolName, MountPath: "/.sigstore"})
+	}
+
 	container := buildValidationContainer(mv, args, vm, pp, tc, p.nativeSidecarSupport)
 	pp.Spec.InitContainers = append(pp.Spec.InitContainers, container)
 
@@ -473,6 +482,52 @@
 	return *merged
 }
 
+// collectNeededPaths returns file paths the validation agent needs access to.
+func collectNeededPaths(model v1alpha1.Model, cfg v1alpha1.ValidationConfig) []string {
+	paths := []string{model.Path}
+	if model.SignaturePath != "" {
+		paths = append(paths, model.SignaturePath)
+	}
+	if cfg.PkiConfig != nil && cfg.PkiConfig.CertificateAuthority != "" {
+		paths = append(paths, cfg.PkiConfig.CertificateAuthority)
+	}
+	if cfg.PublicKeyConfig != nil && cfg.PublicKeyConfig.KeyPath != "" {
+		paths = append(paths, cfg.PublicKeyConfig.KeyPath)
+	}
+	if cfg.ClientTrustConfig != nil && cfg.ClientTrustConfig.TrustConfigPath != "" {
+		paths = append(paths, cfg.ClientTrustConfig.TrustConfigPath)
+	}
+	return paths
+}
+
+// filterVolumeMounts returns only the mounts whose mountPath is a proper directory
+// prefix of a needed path, all forced read-only. Mounts at "/" are excluded to
+// prevent leaking the entire root filesystem into the validation container.
+func filterVolumeMounts(containers []corev1.Container, neededPaths []string) []corev1.VolumeMount {
+	seen := make(map[string]bool)
+	var out []corev1.VolumeMount
+	for _, c := range containers {
+		for _, m := range c.VolumeMounts {
+			if seen[m.MountPath] || m.MountPath == "/" {
+				continue
+			}
+			prefix := m.MountPath
+			if !strings.HasSuffix(prefix, "/") {
+				prefix += "/"
+			}
+			for _, p := range neededPaths {
+				if strings.HasPrefix(p, prefix) || p == m.MountPath {
+					m.ReadOnly = true
+					out = append(out, m)
+					seen[m.MountPath] = true
+					break
+				}
+			}
+		}
+	}
+	return out
+}
+
 func webhookResult(resp admission.Response) string {
 	if resp.Result == nil {
 		return "success"

internal/webhooks/pod_webhook_test.go

@@ -870,9 +870,9 @@
 			Expect(initContainer.SecurityContext.ReadOnlyRootFilesystem).ToNot(BeNil())
 			Expect(*initContainer.SecurityContext.ReadOnlyRootFilesystem).To(BeTrue(), "ReadOnlyRootFilesystem must be true")
 			Expect(initContainer.SecurityContext.AllowPrivilegeEscalation).ToNot(BeNil())
 			Expect(*initContainer.SecurityContext.AllowPrivilegeEscalation).To(BeFalse(), "AllowPrivilegeEscalation must be false")
 			Expect(initContainer.SecurityContext.Capabilities).ToNot(BeNil())
 			Expect(initContainer.SecurityContext.Capabilities.Drop).To(ContainElement(corev1.Capability("ALL")), "Must drop ALL capabilities")
 			Expect(initContainer.SecurityContext.SeccompProfile).ToNot(BeNil())
 			Expect(initContainer.SecurityContext.SeccompProfile.Type).To(Equal(corev1.SeccompProfileTypeRuntimeDefault), "Seccomp must be RuntimeDefault")
 
@@ -955,4 +955,79 @@
 			_ = k8sClient.Delete(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: contSecCtxTestNamespace}})
 		})
 	})
+
+	Context("filterVolumeMounts", func() {
+		containers := func(mounts ...corev1.VolumeMount) []corev1.Container {
+			return []corev1.Container{{VolumeMounts: mounts}}
+		}
+		mount := func(name, path string) corev1.VolumeMount {
+			return corev1.VolumeMount{Name: name, MountPath: path}
+		}
+
+		It("should match mounts whose path is a prefix of a needed path", func() {
+			result := filterVolumeMounts(
+				containers(mount("data", "/data"), mount("config", "/config")),
+				[]string{"/data/model.onnx"},
+			)
+			Expect(result).To(HaveLen(1))
+			Expect(result[0].Name).To(Equal("data"))
+			Expect(result[0].ReadOnly).To(BeTrue())
+		})
+
+		It("should reject root mountPath /", func() {
+			result := filterVolumeMounts(
+				containers(mount("root-vol", "/"), mount("data", "/data")),
+				[]string{"/data/model.onnx"},
+			)
+			Expect(result).To(HaveLen(1))
+			Expect(result[0].Name).To(Equal("data"))
+		})
+
+		It("should not match partial directory names", func() {
+			result := filterVolumeMounts(
+				containers(mount("dat", "/dat")),
+				[]string{"/data/model.onnx"},
+			)
+			Expect(result).To(BeEmpty())
+		})
+
+		It("should match exact mountPath equal to needed path", func() {
+			result := filterVolumeMounts(
+				containers(mount("model", "/data")),
+				[]string{"/data"},
+			)
+			Expect(result).To(HaveLen(1))
+			Expect(result[0].Name).To(Equal("model"))
+		})
+
+		It("should deduplicate mounts across containers", func() {
+			result := filterVolumeMounts(
+				[]corev1.Container{
+					{VolumeMounts: []corev1.VolumeMount{mount("a", "/data")}},
+					{VolumeMounts: []corev1.VolumeMount{mount("b", "/data")}},
+				},
+				[]string{"/data/model.onnx"},
+			)
+			Expect(result).To(HaveLen(1))
+			Expect(result[0].Name).To(Equal("a"))
+		})
+
+		It("should return empty for no matching paths", func() {
+			result := filterVolumeMounts(
+				containers(mount("logs", "/var/log")),
+				[]string{"/data/model.onnx"},
+			)
+			Expect(result).To(BeEmpty())
+		})
+
+		It("should match multiple needed paths to multiple mounts", func() {
+			result := filterVolumeMounts(
+				containers(mount("data", "/data"), mount("trust", "/trust"), mount("logs", "/var/log")),
+				[]string{"/data/model.onnx", "/trust/config.json"},
+			)
+			Expect(result).To(HaveLen(2))
+			names := []string{result[0].Name, result[1].Name}
+			Expect(names).To(ContainElements("data", "trust"))
+		})
+	})
 })