我们的 MLOps 平台最近遇到了一个棘手的安全问题。数据科学团队提交了一个用于特征工程的 Pandas 脚本，附带一个 requirements.txt 文件。CI 管道按部就班地构建容器镜像，Argo CD 忠实地将其同步到生产环境。两天后，安全团队的例行扫描发现该容器内的一个依赖包存在已知的远程代码执行（RCE）漏洞。整个流程看起来无懈可击，自动化程度很高，但恰恰是这种“顺滑”的自动化，将一个定时炸弹悄无声息地部署到了核心数据处理环节。
这个事件暴露了一个核心矛盾：数据科学家关注的是算法和模型效果，他们倾向于快速迭代，频繁更换和测试各种Python库；而 MLOps 平台和运维团队关注的是稳定性、可靠性和安全性。一个简单的 pip install -r requirements.txt 命令，在生产环境中可能成为一个巨大的安全黑洞。
最初的构想是在 CI 阶段加入漏洞扫描，如果发现高危漏洞就直接失败构建。这是一个简单直接的方案，但很快我们发现这远远不够。在真实的项目中，这种“一刀切”的方式会带来几个问题：
- 缺乏信任传递： CI 任务成功了，但下游的部署系统（Argo CD）如何确信这个镜像是经过了扫描，并且是“干净”的？它只能盲目地相信推送到镜像仓库的最新
tag就是好的。 - 配置漂移风险： 如果有人绕过 CI，手动
docker push了一个有问题的镜像到同一个tag，Argo CD 会毫不犹豫地部署它。 - 审计困难： 当发生安全事件时，我们如何以一种加密可验证的方式，回溯某个正在运行的容器，证明它在构建时确实通过了所有质量门控？仅仅翻阅 CI/CD 的运行日志是不够的，日志可以被篡改或丢失。
我们需要的是一种机制，能够将构建过程中的安全检查结果，以一种防篡改、可验证的方式“附加”到最终的产物（容器镜像）上，并且在部署时强制校验这些“证明”。这正是软件供应链安全的核心思想，也是我们引入 in-toto 框架和相关工具链的原因。
复盘式构建日志：从不设防的管道到声明式安全
我们的目标是改造现有的 MLOps 流程，将安全检查从一个孤立的 CI 步骤，转变为贯穿整个软件交付生命周期的、可验证的“安全证明”。
阶段一：不设防的基线架构
在改造之前，我们的流程非常典型。
1. 应用代码 (process.py)
一个简单的 Pandas 脚本，读取数据，做一些转换，然后输出。
# app/process.py
import pandas as pd
import numpy as np
import logging
import os
import sys
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def process_data(input_path, output_path):
"""
A dummy data processing function using Pandas.
In a real scenario, this would involve complex feature engineering.
"""
try:
logging.info(f"Reading data from {input_path}...")
# Simulating reading a large dataset
df = pd.DataFrame(np.random.rand(10000, 5), columns=['A', 'B', 'C', 'D', 'E'])
logging.info("Performing data transformation...")
df['A_sqrt'] = np.sqrt(df['A'])
df['B_log'] = np.log1p(df['B'])
# A common source of vulnerabilities can be indirect dependencies
# pulled in by libraries used for more complex tasks.
if not os.path.exists(os.path.dirname(output_path)):
os.makedirs(os.path.dirname(output_path))
df.to_csv(output_path, index=False)
logging.info(f"Successfully processed data and saved to {output_path}")
except FileNotFoundError:
logging.error(f"Error: Input file not found at {input_path}")
sys.exit(1)
except Exception as e:
logging.error(f"An unexpected error occurred: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
# In a real MLOps pipeline, these paths would be managed by the orchestration system.
INPUT_FILE = "/data/input/sample.csv"
OUTPUT_FILE = "/data/output/processed.csv"
process_data(INPUT_FILE, OUTPUT_FILE)2. 依赖文件 (requirements.txt)
这里我们故意引入一个有漏洞的旧版本 numpy。
# app/requirements.txt
pandas==2.1.1
numpy==1.21.0
# This version of numpy has known vulnerabilities.3. Dockerfile
非常标准的 Dockerfile。
# app/Dockerfile
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
# A common mistake is not pinning versions or checking hashes.
RUN pip install --no-cache-dir -r requirements.txt
COPY process.py .
# This entrypoint assumes data volumes are mounted at /data.
CMD ["python", "process.py"]4. Argo CD 应用
GitOps 仓库中的部署清单。
# gitops/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: pandas-processor
spec:
replicas: 1
selector:
matchLabels:
app: pandas-processor
template:
metadata:
labels:
app: pandas-processor
spec:
containers:
- name: processor
# The image is pulled without any verification
image: your-registry/pandas-processor:v1.0.0
# In a real scenario, you'd have volume mounts for data I/O
# volumeMounts:
# - name: data-input
# mountPath: /data/input
# - name: data-output
# mountPath: /data/output这个流程的问题显而易见：从代码提交到生产部署，没有任何环节对软件物料的安全性进行验证和证明。
阶段二：引入基于 Tekton 和 Cosign 的安全证明 CI 管道
我们选择 Tekton 作为 CI/CD 引擎，因为它原生于 Kubernetes，其声明式的 Pipeline 和 Task 资源与我们的 GitOps 理念非常契合。我们的目标是构建一个管道，它不仅构建和扫描镜像，更重要的是，它要为构建过程的每个关键步骤生成可验证的 in-toto 证明（Attestation）。
一个 in-toto 证明本质上是一个经过签名的 JSON 文件，它声明了某个主语（Subject，例如一个容器镜像）具有某个谓词（Predicate，即描述性内容）。
我们的新 CI 流程如下：
graph TD
A[Git Push] --> B(Tekton PipelineRun);
B --> C{1. Clone Source};
C --> D{2. Scan Dependencies with Trivy};
D --> E{3. Build Image with Kaniko};
E --> F{4. Scan Image with Trivy};
F --> G{5. Generate in-toto Attestation};
G --> H{6. Sign Image & Attestation with Cosign};
H --> I[Update Image Tag in GitOps Repo];
关键实现细节：Tekton 任务
1. 漏洞扫描任务 (scan-task.yaml)
我们使用 Trivy 进行扫描。这个任务会输出一个结果文件。
apiVersion: tekton.dev/v1beta1
kind: Task
metadata:
name: trivy-scan
spec:
params:
- name: image-url
description: URL of the image to scan
workspaces:
- name: output
steps:
- name: scan
image: aquasec/trivy:0.45.1
script: |
#!/usr/bin/env sh
trivy image \
--format json \
--output $(workspaces.output.path)/scan-results.json \
--severity HIGH,CRITICAL \
--exit-code 0 \
$(params.image-url)
# We use exit-code 0 to ensure the pipeline continues,
# as the policy enforcement happens later.
# The scan result is what matters.2. 生成证明任务 (generate-attestation-task.yaml)
这是整个方案的核心。这个任务读取 Trivy 的扫描结果，并将其构造成一个符合 SLSA 规范的 in-toto 证明。
apiVersion: tekton.dev/v1beta1
kind: Task
metadata:
name: generate-in-toto-attestation
spec:
params:
- name: image-digest
description: The digest of the image being attested
workspaces:
- name: scan-results
- name: attestations
steps:
- name: generate
image: bash:5.2
script: |
#!/usr/bin/env bash
set -e
SCAN_RESULT_FILE=$(workspaces.scan-results.path)/scan-results.json
ATTESTATION_FILE=$(workspaces.attestations.path)/attestation.json
echo "Generating attestation for digest $(params.image-digest)..."
# Extract vulnerability summary from Trivy's JSON output using jq
# A common pitfall is handling the case where no vulnerabilities are found.
CRITICAL_COUNT=$(jq 'if .Results then map(select(.Vulnerabilities != null)) | map(.Vulnerabilities[]) | map(select(.Severity == "CRITICAL")) | length else 0 end' $SCAN_RESULT_FILE)
HIGH_COUNT=$(jq 'if .Results then map(select(.Vulnerabilities != null)) | map(.Vulnerabilities[]) | map(select(.Severity == "HIGH")) | length else 0 end' $SCAN_RESULT_FILE)
echo "Found ${CRITICAL_COUNT} CRITICAL and ${HIGH_COUNT} HIGH vulnerabilities."
# Construct the in-toto attestation predicate
PREDICATE=$(cat <<EOF
{
"builder": { "id": "tekton-pipeline-uri" },
"buildType": "custom-tekton-build",
"invocation": {
"configSource": {
"uri": "$(context.pipelineRun.annotations['git-url'])",
"digest": { "sha1": "$(context.pipelineRun.annotations['git-commit'])" },
"entryPoint": "tekton-pipeline.yaml"
}
},
"metadata": {
"buildStartedOn": "$(context.pipelineRun.creationTimestamp)",
"completeness": {
"parameters": true,
"environment": false,
"materials": false
}
},
"materials": [
{
"uri": "$(context.pipelineRun.annotations['git-url'])",
"digest": { "sha1": "$(context.pipelineRun.annotations['git-commit'])" }
}
],
"predicate": {
"scanner": "Trivy",
"scan_timestamp": "$(date -u +"%Y-%m-%dT%H:%M:%SZ")",
"vulnerabilities_summary": {
"critical": ${CRITICAL_COUNT},
"high": ${HIGH_COUNT}
}
}
}
EOF
)
# Construct the full in-toto statement
cat <<EOF > $ATTESTATION_FILE
{
"_type": "https://in-toto.io/Statement/v0.1",
"subject": [{
"name": "your-registry/pandas-processor",
"digest": {
"sha256": "$(params.image-digest)"
}
}],
"predicateType": "https://slsa.dev/provenance/v0.2",
"predicate": $PREDICATE
}
EOF
echo "Attestation generated successfully:"
cat $ATTESTATION_FILE- 注释解析: 这段脚本的核心是动态构建一个 JSON 对象。
predicate字段是自定义的，我们在这里嵌入了 Trivy 的扫描摘要。subject字段清晰地指明了这个证明是关于哪个具体镜像的（通过其 SHA256 摘要）。
3. 签名与附加证明任务 (cosign-sign-attest-task.yaml)
我们使用 Sigstore 项目的 Cosign 工具来完成签名和证明的附加。Cosign 会将证明作为一个独立的层（layer）推送到 OCI 兼容的镜像仓库中，并将其与目标镜像关联起来。
apiVersion: tekton.dev/v1beta1
kind: Task
metadata:
name: cosign-sign-attest
spec:
params:
- name: image-url-with-digest
- name: cosign-key-secret-name
workspaces:
- name: attestations
steps:
- name: sign-and-attest
image: gcr.io/projectsigstore/cosign/ci/cosign:v2.2.1
env:
- name: COSIGN_PASSWORD
valueFrom:
secretKeyRef:
name: $(params.cosign-key-secret-name)
key: password
script: |
#!/bin/sh
set -ex
# Cosign expects the private key mounted from a secret
# First, sign the container image itself
cosign sign --key /etc/secret-volume/cosign.key $(params.image-url-with-digest)
echo "Image signed successfully."
# Now, attach the attestation
cosign attest --key /etc/secret-volume/cosign.key \
--type slsaprovenance \
--predicate $(workspaces.attestations.path)/attestation.json \
$(params.image-url-with-digest)
echo "Attestation attached successfully."
volumeMounts:
- name: cosign-key
mountPath: /etc/secret-volume
readOnly: true
volumes:
- name: cosign-key
secret:
secretName: $(params.cosign-key-secret-name)- 注意: 在真实项目中，密钥管理是重中之重。这里我们从 Kubernetes Secret 中加载私钥。更安全的做法是使用 Vault 或者 Sigstore 的无密码 OIDC 流程（Fulcio）。
完整的 Tekton Pipeline
将以上任务串联起来，形成一个完整的构建与证明管道。
# tekton/pipeline.yaml
apiVersion: tekton.dev/v1beta1
kind: Pipeline
metadata:
name: secure-pandas-mlops-pipeline
spec:
workspaces:
- name: shared-data
params:
- name: git-repo-url
- name: image-reference
tasks:
# ... Git clone task ...
- name: build-image
taskRef: { name: kaniko }
# ... Kaniko parameters ...
- name: scan-image
runAfter: [build-image]
taskRef: { name: trivy-scan }
params:
- name: image-url
value: $(tasks.build-image.results.IMAGE_URL)
workspaces:
- name: output
workspace: shared-data
- name: generate-attestation
runAfter: [scan-image]
taskRef: { name: generate-in-toto-attestation }
params:
- name: image-digest
value: $(tasks.build-image.results.IMAGE_DIGEST)
workspaces:
- name: scan-results
workspace: shared-data
- name: attestations
workspace: shared-data
- name: sign-and-attest
runAfter: [generate-attestation]
taskRef: { name: cosign-sign-attest }
params:
- name: image-url-with-digest
value: "$(tasks.build-image.results.IMAGE_URL)@$(tasks.build-image.results.IMAGE_DIGEST)"
- name: cosign-key-secret-name
value: "cosign-keys"
workspaces:
- name: attestations
workspace: shared-data
# ... Task to update GitOps repo ...至此，每当数据科学家提交代码，我们的 Tekton 管道都会自动构建出一个带有签名和安全扫描证明的容器镜像。但仅仅生成证明是不够的，还需要一个“守门员”来强制执行。
阶段三：使用 Kyverno 作为部署阶段的守门员
Argo CD 的工作是同步 Git 仓库中的状态到 Kubernetes 集群。它本身不应该承担安全策略的决策。真正的安全门控应该在 Kubernetes 的 API Server 层实现，通过 Admission Controller 来完成。我们选择 Kyverno，因为它允许我们用简单的 YAML 来定义复杂的安全策略，这与 GitOps 的声明式理念完全一致。
我们需要创建一个 ClusterPolicy，它规定：任何带有 mlops.secure/verify: "true" 注解的 Pod，其容器镜像必须满足以下条件：
- 必须有合法的签名。
- 必须附带一个
slsaprovenance类型的 in-toto 证明。 - 证明中的
predicate.vulnerabilities_summary.critical字段值必须为 0。
Kyverno ClusterPolicy (policy.yaml)
apiVersion: kyverno.io/v1
kind: ClusterPolicy
metadata:
name: check-pandas-workload-attestation
annotations:
policies.kyverno.io/title: Verify Pandas Workload Image Attestation
policies.kyverno.io/category: Software Supply Chain Security
policies.kyverno.io/subject: Pod
policies.kyverno.io/description: >-
This policy checks that images for Pandas workloads have been signed
by Cosign and have an in-toto attestation from the secure CI pipeline.
It verifies the attestation to ensure there are no critical vulnerabilities.
spec:
validationFailureAction: Enforce
background: false
rules:
- name: verify-image-attestation
match:
any:
- resources:
kinds:
- Pod
annotations:
mlops.secure/verify: "true"
verifyImages:
- imageReferences:
- "your-registry/pandas-processor:*"
attestors:
- count: 1
entries:
- keys:
# The public key is stored in a ConfigMap or can be inline
publicKey: |
-----BEGIN PUBLIC KEY-----
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE...
-----END PUBLIC KEY-----
attestations:
- type: https://slsa.dev/provenance/v0.2
# This is the core of the policy logic
conditions:
- all:
- key: "{{predicate.vulnerabilities_summary.critical}}"
operator: Equals
# The crucial check: enforce zero critical vulnerabilities
value: 0-
validationFailureAction: Enforce: 这是最关键的配置，表示如果验证失败，API Server 将直接拒绝该资源的创建请求。 -
match: 策略仅对带有特定注解的 Pod 生效，这给了我们灵活性，可以逐步推广该策略。 verifyImages: Kyverno 的强大功能，它会：- 自动从 OCI 仓库拉取镜像的签名和证明。
- 使用提供的公钥验证签名。
- 解析证明的 JSON 内容。
- 使用
conditions中的 JMESPath 表达式 ({{...}}) 对证明内容进行断言。这里的key: "{{predicate.vulnerabilities_summary.critical}}"直接查询了我们之前在 Tekton 任务中注入的字段。
最终的工作流
现在，整个 MLOps 流程形成了一个闭环：
sequenceDiagram
participant DS as Data Scientist
participant Git
participant Tekton
participant Registry
participant ArgoCD
participant K8s_API as Kubernetes API Server
participant Kyverno
DS->>Git: Pushes Pandas script + requirements.txt
Git-->>Tekton: Triggers PipelineRun
Tekton->>Tekton: Clones, Builds, Scans
alt Scan finds CRITICAL vulnerability
Tekton->>Registry: Generates attestation (critical: 1)
Tekton->>Registry: Signs image & attaches attestation
else Scan is clean
Tekton->>Registry: Generates attestation (critical: 0)
Tekton->>Registry: Signs image & attaches attestation
end
Tekton->>Git: Updates image tag in GitOps repo
Git-->>ArgoCD: Detects change in manifest
ArgoCD->>K8s_API: Applies Deployment manifest (sync)
K8s_API->>Kyverno: Admission Webhook: Validate Pod creation
Kyverno->>Registry: Fetches signature & attestation for the image
Kyverno->>Kyverno: 1. Verifies signature with public key
Kyverno->>Kyverno: 2. Checks attestation predicate
alt Attestation shows critical: 1
Kyverno-->>K8s_API: DENY
K8s_API-->>ArgoCD: Fails to create Pod
ArgoCD->>ArgoCD: Sync Failed status
else Attestation shows critical: 0
Kyverno-->>K8s_API: ALLOW
K8s_API->>K8s_API: Creates Pod
ArgoCD->>ArgoCD: Sync OK status
end
现在，即使数据科学家提交了带有高危漏洞的依赖，或者有人恶意推送了被篡改的镜像，部署也会在最后一刻被 Kubernetes API Server 拒绝，Argo CD 的同步状态会明确显示失败。我们不仅阻止了不安全的部署，还拥有了完整的、可审计的加密证明链，可以准确回答“这个正在运行的容器，是谁、在何时、基于哪个代码版本、通过了哪些检查才构建出来的”。
遗留问题与未来迭代
这套方案虽然解决了核心的安全痛点，但在生产环境中推广，仍有一些工程问题需要考虑。
首先是密钥管理。示例中使用的静态密钥对不适用于大型团队，必须引入更成熟的方案，例如 HashiCorp Vault 进行密钥轮换和管理，或者全面拥抱 Sigstore 生态，使用 Fulcio 根证书颁发机构和 Rekor 透明日志，实现无密码的、基于 OIDC 的短时签名，这能极大地简化密钥分发的难题。
其次，证明的粒度可以更细化。当前我们的证明只包含了漏洞扫描结果。一个更成熟的供应链会包含更丰富的证明，例如：单元测试覆盖率证明、静态代码分析结果证明、软件物料清单（SBOM）证明。Kyverno 策略也可以相应地变得更复杂，例如，要求 SBOM 中不能包含未经许可的开源协议。
最后，开发者体验至关重要。当 Kyverno 拒绝一个部署时，返回给 Argo CD 或用户的错误信息必须清晰明了，让他们能快速定位到是哪个安全策略、哪个检查项失败了。这需要在 Kyverno 策略的 message 字段中精心设计，并确保 CI/CD 平台的日志能够清晰地反映整个证明生成和验证的过程。一个常见的错误是，安全系统过于黑盒，导致开发者为了绕过它而花费更多精力，反而降低了整体效率。
