SLSA（软件工件供应链级别）

来源证明胜过口头承诺

构建可以可复现，但如果流水线被攻破，结果仍不可信。SLSA 把问题从“工件能否运行”转为“我能否验证它是如何产生的”。答案是一份签名的来源证明，把工件摘要绑定到源仓库、提交和构建系统配置。

等级如何工作

构建等级 1 表示构建已脚本化且存在来源证明，但可能不完整或未签名。等级 2 要求托管构建服务、签名来源证明，且证明由该服务而非用户作业生成。等级 3 增加加固：隔离构建、用户不能通过步骤改变输出、证明不可伪造。

验证发生在消费端。部署流水线下载工件和证明，用受信任的构建者身份校验签名，比对工件摘要和源 URI 与策略是否一致；任何一环失败就拒绝该工件。

何时使用与取舍

对外发布的工件、容器镜像和分发的包都适合采用 SLSA。被广泛使用的开源项目和构建系统易受攻击的组织收益最大。先从生成并在内部验证来源证明开始，再要求供应商达到某个等级。

等级越高，对构建方式约束越大：作业里不能有自定义签名步骤，需要隔离的 runner 和更严格的身份控制。来源证明增加存储和部署验证步骤。该框架覆盖构建完整性，不覆盖代码质量、依赖风险或运行时安全，必须与 SBOM 和扫描配合。

示例与常见误解

发布流水线下载二进制和来源证明，验证签名身份是项目的托管构建者，并确认摘要与工件、源提交一致。有人手工上传替换的二进制时摘要校验失败，被阻止部署。

常见误解是 SLSA 证明工件没有恶意代码。它证明工件与声明的构建一致；恶意提交仍能产生有效的来源证明。相关概念：软件物料清单、供应链安全、CI/CD。

代码示例

package main

import (
	"crypto/sha256"
	"crypto/subtle"
	"fmt"
)

func ConstantTimeEqual(a, b string) bool {
	left := sha256.Sum256([]byte(a))
	right := sha256.Sum256([]byte(b))
	return subtle.ConstantTimeCompare(left[:], right[:]) == 1
}

func main() {
	secret := "s3cr3t-token"
	fmt.Println(ConstantTimeEqual(secret, "s3cr3t-token"))
	fmt.Println(ConstantTimeEqual(secret, "s3cr3t-tokeX"))
	fmt.Println(ConstantTimeEqual(secret, "short"))
}

// SSRF guard: validate every serverSide request target before dialing
package main

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"strings"
	"time"
)

var blockedZones = []string{
	"127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12",
	"192.168.0.0/16", "169.254.0.0/16", "::1/128", "fc00::/7",
}

var allowedHosts = map[string]bool{
	"api.partner.com": true, "cdn.partner.net": true, "hooks.vendor.io": true,
}

func validateTarget(raw string) (*url.URL, error) {
	parsed, err := url.Parse(raw)
	if err != nil {
		return nil, fmt.Errorf("parse url: %w", err)
	}

	if parsed.Scheme != "https" {
		return nil, fmt.Errorf("scheme %q not allowed", parsed.Scheme)
	}
	if !allowedHosts[parsed.Hostname()] {
		return nil, fmt.Errorf("host %q is not allowlisted", parsed.Hostname())
	}
	return parsed, nil
}

func dialGuard(ctx context.Context, network, addr string) (net.Conn, error) {
	host, port, err := net.SplitHostPort(addr)
	if err != nil {
		return nil, err
	}

	ips, err := net.DefaultResolver.LookupIPAddr(ctx, host)
	if err != nil {
		return nil, err
	}
	for _, ip := range ips {
		if !isPublic(ip.IP) {
			return nil, fmt.Errorf("resolved %s to private address %s", host, ip.IP)
		}
	}

	dialer := &net.Dialer{Timeout: 5 * time.Second}
	return dialer.DialContext(ctx, network, net.JoinHostPort(ips[0].IP.String(), port))
}

func isPublic(ip net.IP) bool {
	for _, zone := range blockedZones {
		_, network, _ := net.ParseCIDR(zone)
		if network.Contains(ip) {
			return false
		}
	}
	return !ip.IsLoopback() && !ip.IsUnspecified() && !ip.IsLinkLocalUnicast()
}

func guardedClient() *http.Client {
	transport := &http.Transport{
		DialContext: dialGuard,
		// 重定向必须重走校验：禁止自动跟随
		DisableKeepAlives: true,
	}
	return &http.Client{
		Transport:     transport,
		CheckRedirect: func(req *http.Request, via []*http.Request) error {
			return validateSameHost(via[0], req.URL)
		},
		Timeout: 8 * time.Second,
	}
}

func validateSameHost(orig, next *url.URL) error {
	if strings.EqualFold(orig.Hostname(), next.Hostname()) {
		return nil
	}
	return fmt.Errorf("redirect to non-allowlisted host %q", next.Hostname())
}

func main() {
	client := guardedClient()
	if _, err := validateTarget("https://api.partner.com/v1/data"); err == nil {
		resp, err := client.Get("https://api.partner.com/v1/data")
		fmt.Println(resp, err)
	}

	_, err := validateTarget("http://169.254.169.254/latest/meta-data")
	fmt.Println("metadata endpoint blocked:", err)
}