fix(submit): 审阅绑定上下文 sha256,批准须带摘要,构建只从内部 API 取上下文并校验字节

This commit is contained in:
Lemon-miaow committed 2026-09-24 22:24:50 +08:00
1 parent 13b65e19ec
commit e521cf5976
30 files changed
+910 -98

No files matched your search

+5
View File
@@ -70,6 +70,11 @@ func withRecover(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
defer func() {
if rec := recover(); rec != nil {
if rec == http.ErrAbortHandler {
// A deliberate abort of a committed response (see
// streamSubmissionContext): let net/http cut the connection.
panic(rec)
}
log.Printf("api: %s %s: panic (request_id=%s): %v\n%s",
r.Method, r.URL.Path, requestIDFromContext(r.Context()), rec, debug.Stack())
writeError(w, r, newError(http.StatusInternalServerError, "panic", "internal error"))
+81 -19
View File
@@ -2,8 +2,11 @@ package api
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"io"
"log"
"net/http"
"felis.lolicon.best/internal/build"
@@ -39,7 +42,9 @@ type SubmissionService interface {
List(ctx context.Context) ([]submit.Submission, error)
// Approve is the admin gate: it claims pending_review -> approved (CAS) and the
// winner starts the SAME Trivy-gated build as an admin's direct build.
Approve(ctx context.Context, id, reviewedBy string) (*submit.Submission, error)
// expectedDigest is the context sha256 the reviewer inspected; a row whose
// context has since been replaced refuses with ErrContextChanged.
Approve(ctx context.Context, id, reviewedBy, expectedDigest string) (*submit.Submission, error)
// Reject is the admin's other verdict: pending_review -> rejected with a
// required reason; it starts no build.
Reject(ctx context.Context, id, reviewedBy, reason string) (*submit.Submission, error)
@@ -54,7 +59,9 @@ type SubmissionService interface {
// context-fetch route: the build Pod's initContainer cannot mount the uploads
// PVC across namespaces and holds no object-store credentials, so it streams
// the blob from the API over the service-token-gated internal face instead.
OpenContext(ctx context.Context, id string) (io.ReadCloser, error)
// The string is the sha256 the row records for the blob ("" when none was
// recorded); the routes refuse to finish a stream that does not match it.
OpenContext(ctx context.Context, id string) (io.ReadCloser, string, error)
}
// createSubmissionRequest is the POST /me/submissions body. The user
@@ -81,6 +88,18 @@ type rejectSubmissionRequest struct {
Reason string `json:"reason"`
}
// approveSubmissionRequest is the POST /submissions/{id}/approve body: the
// context digest the admin reviewed. The panel sends the digest of the bytes it
// downloaded (the X-Felis-Context-Sha256 header), or the listed one.
type approveSubmissionRequest struct {
ExpectedDigest string `json:"expected_digest"`
}
// contextDigestHeader carries the recorded sha256 on both context routes, so a
// reviewer can compare it with `sha256sum` and the panel can approve exactly the
// bytes it fetched.
const contextDigestHeader = "X-Felis-Context-Sha256"
// handleCreateSubmission records a new pending_review submission (app-tier). The
// submitter is the authenticated principal's id — never the body — so a user can
// only ever file an upload under their own identity.
@@ -269,7 +288,14 @@ func (a *API) handleApproveSubmission(w http.ResponseWriter, r *http.Request) {
}
p := principalFromContext(r.Context())
id := r.PathValue("id")
sub, err := a.Submissions.Approve(r.Context(), id, p.Email)
var body approveSubmissionRequest
if r.ContentLength != 0 {
if err := decodeJSON(w, r, &body); err != nil {
writeError(w, r, err)
return
}
}
sub, err := a.Submissions.Approve(r.Context(), id, p.Email, body.ExpectedDigest)
if err != nil {
writeSubmitError(w, r, err)
return
@@ -368,6 +394,9 @@ func writeSubmitError(w http.ResponseWriter, r *http.Request, err error) {
case errors.Is(err, submit.ErrAlreadyReviewed):
writeError(w, r, newError(http.StatusConflict, "already_reviewed",
"submission has already been reviewed"))
case errors.Is(err, submit.ErrContextChanged):
writeError(w, r, newError(http.StatusConflict, "context_changed",
"the build context was uploaded again after it was reviewed; review the new upload before approving"))
case errors.Is(err, submit.ErrQuotaExceeded):
writeError(w, r, newError(http.StatusForbidden, "submission_quota_exceeded",
"submission quota reached"))
@@ -390,11 +419,11 @@ func writeSubmitError(w http.ResponseWriter, r *http.Request, err error) {
// fetcher extracts it under a zip-slip guard, and Kaniko treats the result as
// hostile regardless (spec §16).
func (a *API) handleInternalSubmissionContext(w http.ResponseWriter, r *http.Request) {
rc, ok := a.openSubmissionContext(w, r)
rc, digest, ok := a.openSubmissionContext(w, r)
if !ok {
return
}
streamSubmissionContext(w, rc)
streamSubmissionContext(w, r, rc, digest)
}
// handleAdminSubmissionContext streams a submission's stored build-context
@@ -405,44 +434,77 @@ func (a *API) handleInternalSubmissionContext(w http.ResponseWriter, r *http.Req
// ones the build Pod fetches over the internal face; the attachment disposition
// makes the browser download the attacker-supplied archive, never render it.
func (a *API) handleAdminSubmissionContext(w http.ResponseWriter, r *http.Request) {
rc, ok := a.openSubmissionContext(w, r)
rc, digest, ok := a.openSubmissionContext(w, r)
if !ok {
return
}
w.Header().Set("Content-Disposition", `attachment; filename="context.tar.gz"`)
w.Header().Set("X-Content-Type-Options", "nosniff")
a.audit(r, "submission.context.download", r.PathValue("id"))
streamSubmissionContext(w, rc)
streamSubmissionContext(w, r, rc, digest)
}
// openSubmissionContext resolves the build-context blob named in the request
// path, mapping the submit-layer errors onto the shared submission statuses (a
// missing blob is 404, an unwired transport 503). On failure the error response
// is already written and the caller must return.
func (a *API) openSubmissionContext(w http.ResponseWriter, r *http.Request) (io.ReadCloser, bool) {
func (a *API) openSubmissionContext(w http.ResponseWriter, r *http.Request) (io.ReadCloser, string, bool) {
if a.Submissions == nil {
writeError(w, r, errSubmissionsUnavailable)
return nil, false
return nil, "", false
}
rc, err := a.Submissions.OpenContext(r.Context(), r.PathValue("id"))
rc, digest, err := a.Submissions.OpenContext(r.Context(), r.PathValue("id"))
if err != nil {
writeSubmitError(w, r, err)
return nil, false
return nil, "", false
}
return rc, true
return rc, digest, true
}
// streamSubmissionContext copies the blob to w verbatim and closes it. The
// caller must have set every header already: the copy commits the response, so
// a failure mid-stream can only truncate it.
func streamSubmissionContext(w http.ResponseWriter, rc io.ReadCloser) {
// streamSubmissionContext copies the blob to w and closes it. The caller must
// have set every header already: the copy commits the response.
//
// With a recorded digest the final chunk is held back until the whole blob has
// been hashed. Bytes that do not match the row (a re-upload landed between the
// row read and the open) or a failed read abort the response instead of ending
// it cleanly, so neither the reviewer's download nor the build's fetch can take
// a prefix or a different blob for the recorded one.
func streamSubmissionContext(w http.ResponseWriter, r *http.Request, rc io.ReadCloser, digest string) {
defer rc.Close()
w.Header().Set("Content-Type", "application/gzip")
if _, err := io.Copy(w, rc); err != nil {
// The status is already committed; the client sees a truncated stream and
// the fetch fails on size/extract, so there is nothing left to write here.
if digest == "" {
// A context uploaded before digests were kept: nothing to check against.
_, _ = io.Copy(w, rc)
return
}
w.Header().Set(contextDigestHeader, digest)
h := sha256.New()
buf := make([]byte, 32<<10)
var held []byte
for {
n, err := rc.Read(buf)
if n > 0 {
if len(held) > 0 {
if _, werr := w.Write(held); werr != nil {
return
}
}
held = append(held[:0], buf[:n]...)
h.Write(buf[:n])
}
if err == io.EOF {
break
}
if err != nil {
log.Printf("api: submission %s context read failed mid-stream: %v", r.PathValue("id"), err)
panic(http.ErrAbortHandler)
}
}
if got := hex.EncodeToString(h.Sum(nil)); got != digest {
log.Printf("api: submission %s context hashes to %s, the row records %s; response aborted", r.PathValue("id"), got, digest)
panic(http.ErrAbortHandler)
}
_, _ = w.Write(held)
}
// Compile-time proof that the production Manager satisfies the API interface.
+70 -5
View File
@@ -2,11 +2,14 @@ package api
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
@@ -46,6 +49,9 @@ type fakeSubmissions struct {
openedID string
openBody string
openErr error
approvedDigest string
openDigest string
}
func (f *fakeSubmissions) Create(_ context.Context, req submit.CreateRequest) (*submit.Submission, error) {
@@ -77,8 +83,8 @@ func (f *fakeSubmissions) List(_ context.Context) ([]submit.Submission, error) {
return f.listed, f.listErr
}
func (f *fakeSubmissions) Approve(_ context.Context, id, reviewedBy string) (*submit.Submission, error) {
f.approvedID, f.approvedBy = id, reviewedBy
func (f *fakeSubmissions) Approve(_ context.Context, id, reviewedBy, expectedDigest string) (*submit.Submission, error) {
f.approvedID, f.approvedBy, f.approvedDigest = id, reviewedBy, expectedDigest
if f.approveErr != nil {
return nil, f.approveErr
}
@@ -111,12 +117,12 @@ func (f *fakeSubmissions) Delete(_ context.Context, id string) (*submit.Submissi
// openErr injects the OpenContext outcome; the body recorder lets the internal
// route test assert byte-exact streaming and the 404 mapping.
func (f *fakeSubmissions) OpenContext(_ context.Context, id string) (io.ReadCloser, error) {
func (f *fakeSubmissions) OpenContext(_ context.Context, id string) (io.ReadCloser, string, error) {
f.openedID = id
if f.openErr != nil {
return nil, f.openErr
return nil, "", f.openErr
}
return io.NopCloser(strings.NewReader(f.openBody)), nil
return io.NopCloser(strings.NewReader(f.openBody)), f.openDigest, nil
}
// appSubAPI wires a submissions service behind an ordinary user principal (the
@@ -483,6 +489,31 @@ func TestApproveSubmission(t *testing.T) {
}
}
// The digest the reviewer inspected travels in the body, and a context replaced
// since then is a 409 the panel can act on.
func TestApproveSubmissionForwardsTheReviewedDigest(t *testing.T) {
fs := &fakeSubmissions{}
digest := strings.Repeat("ab", 32)
w := do(adminSubAPI(fs).ExternalHandler(), "POST", "/api/v1/submissions/sub-9/approve",
`{"expected_digest":"`+digest+`"}`, nil)
if w.Code != http.StatusOK || fs.approvedDigest != digest {
t.Fatalf("code = %d, forwarded digest %q (%s)", w.Code, fs.approvedDigest, w.Body.String())
}
fs = &fakeSubmissions{approveErr: submit.ErrContextChanged}
w = do(adminSubAPI(fs).ExternalHandler(), "POST", "/api/v1/submissions/sub-9/approve",
`{"expected_digest":"`+digest+`"}`, nil)
if w.Code != http.StatusConflict || decodeErr(t, w) != "context_changed" {
t.Fatalf("code = %d body %s, want 409 context_changed", w.Code, w.Body.String())
}
w = do(adminSubAPI(&fakeSubmissions{}).ExternalHandler(), "POST", "/api/v1/submissions/sub-9/approve",
`{"digest":"x"}`, nil)
if w.Code != http.StatusBadRequest {
t.Fatalf("unknown field code = %d, want 400", w.Code)
}
}
func TestApproveSubmissionAlreadyReviewedIs409(t *testing.T) {
fs := &fakeSubmissions{approveErr: submit.ErrAlreadyReviewed}
api := adminSubAPI(fs)
@@ -636,6 +667,40 @@ func TestAdminSubmissionContextRoute(t *testing.T) {
}
})
t.Run("names the recorded digest", func(t *testing.T) {
body := "\x1f\x8b\x08\x00blob"
sum := sha256.Sum256([]byte(body))
fs := &fakeSubmissions{openBody: body, openDigest: hex.EncodeToString(sum[:])}
w := do(adminSubAPI(fs).ExternalHandler(), "GET", "/api/v1/submissions/sub-7/context", "", nil)
if w.Code != http.StatusOK || w.Body.String() != body {
t.Fatalf("code = %d body %q", w.Code, w.Body.String())
}
if got := w.Header().Get("X-Felis-Context-Sha256"); got != fs.openDigest {
t.Fatalf("digest header = %q, want %q", got, fs.openDigest)
}
})
// Bytes that are not the recorded ones must not arrive as a complete download:
// the reviewer would inspect a blob the approval does not name.
t.Run("aborts a blob that does not match", func(t *testing.T) {
big := "\x1f\x8b" + strings.Repeat("x", 100<<10)
fs := &fakeSubmissions{openBody: big, openDigest: strings.Repeat("0", 64)}
srv := httptest.NewServer(adminSubAPI(fs).ExternalHandler())
defer srv.Close()
resp, err := http.Get(srv.URL + "/api/v1/submissions/sub-7/context")
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
got, err := io.ReadAll(resp.Body)
if err == nil {
t.Fatalf("read %d bytes cleanly; want the response aborted", len(got))
}
if len(got) >= len(big) {
t.Fatalf("received all %d bytes before the abort", len(got))
}
})
t.Run("missing blob is 404", func(t *testing.T) {
fs := &fakeSubmissions{openErr: fmt.Errorf("%w: gone", submit.ErrBlobNotFound)}
w := do(adminSubAPI(fs).ExternalHandler(), "GET", "/api/v1/submissions/sub-7/context", "", nil)
+16
View File
@@ -129,6 +129,11 @@ type Request struct {
// ContextRef locates the uploaded tar.gz context in object storage / a PVC
// (spec §17: Kaniko pulls it; Git context is intentionally not supported).
ContextRef string
// ContextDigest is the lowercase hex sha256 of the context tarball an admin
// approved. When set, the context-fetch initContainer refuses any other bytes,
// so a context replaced after review never reaches Kaniko. It needs an
// http(s) ContextRef: a ref Kaniko fetches itself has no fetch step to check.
ContextDigest string
// BaseImage is the resolved FROM, recorded for audit only — it is NOT a hard
// gate (spec §16: base FROM is not hard-gated; the scan + egress lock cover
// poisoned bases).
@@ -152,6 +157,10 @@ type Build struct {
Error string `json:"error,omitempty"`
CreatedAt time.Time `json:"created_at"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
// ContextDigest is Request.ContextDigest, kept on the row as the audit record
// of which bytes the build was allowed to consume.
ContextDigest string `json:"context_digest,omitempty"`
}
// Image mirrors an image_whitelist row (spec §6): the dynamic, auditable image
@@ -257,6 +266,11 @@ type Config struct {
// RuntimeClass runs build pods under a sandbox RuntimeClass (gVisor, Kata)
// when set. The class must exist on the cluster.
RuntimeClass string
// ContextOrigin is the scheme://host[:port] of the platform's internal API
// face, the only host an http(s) ContextRef may name: the fetch step presents
// the service token to it. Empty refuses every http(s) context.
ContextOrigin string
}
// Values of Config.UserNamespaces.
@@ -375,6 +389,7 @@ func (b *Builder) Submit(ctx context.Context, req Request) (*Build, error) {
RequestedBy: req.RequestedBy,
CreatedAt: now,
}
bld.ContextDigest = req.ContextDigest
if err := b.Store.CreateBuild(ctx, bld); err != nil {
return nil, err
}
@@ -408,6 +423,7 @@ func (b *Builder) jobParams(bld *Build, cfg Config) JobParams {
BuildID: bld.ID,
ImageRef: bld.ImageRef,
ContextRef: bld.ContextRef,
ContextDigest: bld.ContextDigest,
Namespace: cfg.Namespace,
ServiceAccount: cfg.ServiceAccount,
RegistryURL: cfg.RegistryURL,
+55
View File
@@ -221,6 +221,61 @@ func TestSubmitRejectsExternalRegistryTarget(t *testing.T) {
// The platform's own images and the scanner's DB mirrors live under felis/ and
// mirror/; the registry gate refuses the build principal there, and Validate turns
// that into a 400 before a Job spends minutes building an image it cannot push.
// A context digest must be a real sha256 and needs a fetch step to enforce it.
func TestValidateContextDigest(t *testing.T) {
cfg := Config{RegistryURL: "registry.felis.svc:5000", ContextOrigin: "http://felis-api-internal:8081"}
req := Request{ImageRef: "registry.felis.svc:5000/user-uploads/sub-1:latest", Dockerfile: "FROM scratch",
ContextRef: "http://felis-api-internal:8081/api/v1/internal/submissions/sub-1/context", ContextDigest: strings.Repeat("e", 64)}
if err := Validate(req, cfg); err != nil {
t.Fatalf("valid digest: %v", err)
}
bad := req
bad.ContextDigest = strings.Repeat("E", 64)
if err := Validate(bad, cfg); err == nil {
t.Fatal("an uppercase digest was accepted")
}
bad = req
bad.ContextRef = "s3://bucket/ctx.tar.gz"
if err := Validate(bad, cfg); err == nil {
t.Fatal("a digest on a context Kaniko fetches itself was accepted")
}
}
// The fetch step hands the service token to the context URL's host, so an
// http(s) context must be an upload on the internal face (build-supply-chain-13).
func TestValidateConfinesHTTPContextsToTheInternalFace(t *testing.T) {
cfg := Config{RegistryURL: "registry.felis.svc:5000", ContextOrigin: "http://felis-api-internal.felis.svc.cluster.local:8081/"}
req := goodRequest()
for _, ref := range []string{
"http://felis-api-internal.felis.svc.cluster.local:8081/api/v1/internal/submissions/sub-1/context",
"HTTP://Felis-API-Internal.felis.svc.cluster.local:8081/api/v1/internal/submissions/sub-1/context",
"tar://contexts/abc.tar.gz",
} {
req.ContextRef = ref
if err := Validate(req, cfg); err != nil {
t.Errorf("Validate(%q) = %v, want accepted", ref, err)
}
}
for _, ref := range []string{
"https://attacker.example/api/v1/internal/submissions/sub-1/context",
"http://felis-api-internal.felis.svc.cluster.local:8082/api/v1/internal/submissions/sub-1/context",
"https://felis-api-internal.felis.svc.cluster.local:8081/api/v1/internal/submissions/sub-1/context",
"http://felis-api-internal.felis.svc.cluster.local:8081/api/v1/internal/servers",
"http://felis-api-internal.felis.svc.cluster.local:8081/api/v1/internal/submissions/../x/context",
"http://felis-api-internal.felis.svc.cluster.local:8081/api/v1/internal/submissions/sub-1/context?next=https://x",
"http://u:[email protected]:8081/api/v1/internal/submissions/sub-1/context",
} {
req.ContextRef = ref
if err := Validate(req, cfg); !errors.Is(err, ErrInvalid) {
t.Errorf("Validate(%q) = %v, want ErrInvalid", ref, err)
}
}
req.ContextRef = "http://felis-api-internal.felis.svc.cluster.local:8081/api/v1/internal/submissions/sub-1/context"
if err := Validate(req, Config{RegistryURL: "registry.felis.svc:5000"}); !errors.Is(err, ErrInvalid) {
t.Errorf("without an internal face configured an http context was accepted: %v", err)
}
}
func TestValidateRejectsReservedRepos(t *testing.T) {
cfg := Config{RegistryURL: "registry.felis.svc:5000"}
for _, ref := range []string{
+20 -11
View File
@@ -100,9 +100,12 @@ const buildJobTTL = 7 * 24 * time.Hour
// Build + Config by the Builder; jobspec is a pure function of them so the
// security-critical Job shape is unit-tested without a cluster.
type JobParams struct {
BuildID string
ImageRef string
ContextRef string
BuildID string
ImageRef string
ContextRef string
// ContextDigest, when set, is passed to the fetch container, which refuses a
// context whose sha256 differs (see Request.ContextDigest).
ContextDigest string
Namespace string
ServiceAccount string
RegistryURL string
@@ -269,7 +272,7 @@ func BuildJob(p JobParams) (*batchv1.Job, error) {
SizeLimit: quantityPtr(imageSizeLimit),
}},
}}
if isHTTPContextRef(p.ContextRef) {
if IsHTTPContextRef(p.ContextRef) {
contextPath = contextMountPath
// The fetch container runs as root while Kaniko keeps the image default
// (also root): Kaniko re-copies the Dockerfile out of the context and
@@ -287,11 +290,7 @@ func BuildJob(p JobParams) (*batchv1.Job, error) {
fetch := corev1.Container{
Name: ContainerFetch,
Image: p.FelisImage,
Args: []string{
"fetch-context",
"--url=" + p.ContextRef,
"--out=" + contextMountPath,
},
Args: fetchArgs(p),
// The internal face is service-token gated, and the token is read from a
// Secret the installer materializes in THIS namespace (secretKeyRef is
// namespace-local). It is mounted into this initContainer only: the Kaniko
@@ -461,10 +460,20 @@ func withDisk(limits corev1.ResourceList, d diskBounds) corev1.ResourceRequireme
return corev1.ResourceRequirements{Limits: lim, Requests: req}
}
// isHTTPContextRef reports whether ref is an http(s) URL — the shape the submit
// fetchArgs is the context-fetch container's argv. The digest flag rides along
// only when the build pins one; admin builds from a URL they supplied have none.
func fetchArgs(p JobParams) []string {
args := []string{"fetch-context", "--url=" + p.ContextRef, "--out=" + contextMountPath}
if p.ContextDigest != "" {
args = append(args, "--sha256="+p.ContextDigest)
}
return args
}
// IsHTTPContextRef reports whether ref is an http(s) URL — the shape the submit
// lane derives when the API is the blob transport — i.e. a context only the
// fetch initContainer can turn into a local path for Kaniko.
func isHTTPContextRef(ref string) bool {
func IsHTTPContextRef(ref string) bool {
return strings.HasPrefix(ref, "http://") || strings.HasPrefix(ref, "https://")
}
+31
View File
@@ -527,6 +527,37 @@ func TestBuildJobGatesEgressFirst(t *testing.T) {
}
}
// An approved submission pins its context digest on the fetch container, which
// then refuses any other bytes; a build without one fetches unpinned.
func TestBuildJobPinsContextDigest(t *testing.T) {
p := sampleJobParams()
p.ContextRef = "http://felis-api-internal.felis.svc.cluster.local:8081/ctx"
fetchArgsOf := func(p JobParams) []string {
t.Helper()
job, err := BuildJob(p)
if err != nil {
t.Fatalf("BuildJob: %v", err)
}
for _, c := range job.Spec.Template.Spec.InitContainers {
if c.Name == ContainerFetch {
return c.Args
}
}
t.Fatal("no fetch container")
return nil
}
for _, a := range fetchArgsOf(p) {
if strings.HasPrefix(a, "--sha256") {
t.Fatalf("unpinned build carries %q", a)
}
}
p.ContextDigest = strings.Repeat("d", 64)
args := fetchArgsOf(p)
if args[len(args)-1] != "--sha256="+p.ContextDigest {
t.Fatalf("fetch args = %v, want the digest pinned", args)
}
}
// The pod runs under RuntimeDefault seccomp always, and in a user namespace or
// a sandbox runtime when the install asks for them.
func TestBuildJobSandboxing(t *testing.T) {
+11 -7
View File
@@ -21,17 +21,17 @@ func NewPGStore(db *sql.DB) *PGStore { return &PGStore{db: db} }
func (s *PGStore) CreateBuild(ctx context.Context, b *Build) error {
const q = `INSERT INTO image_builds
(id, image_ref, status, dockerfile, context_ref, base_image, requested_by, created_at)
VALUES ($1, $2, $3, $4, NULLIF($5, ''), NULLIF($6, ''), $7, $8)`
(id, image_ref, status, dockerfile, context_ref, base_image, requested_by, created_at, context_digest)
VALUES ($1, $2, $3, $4, NULLIF($5, ''), NULLIF($6, ''), $7, $8, NULLIF($9, ''))`
_, err := s.db.ExecContext(ctx, q,
b.ID, b.ImageRef, string(b.Status), b.Dockerfile, b.ContextRef, b.BaseImage,
b.RequestedBy, b.CreatedAt)
b.RequestedBy, b.CreatedAt, b.ContextDigest)
return err
}
func (s *PGStore) GetBuild(ctx context.Context, id string) (*Build, error) {
const q = `SELECT id, image_ref, status, dockerfile, context_ref, base_image,
requested_by, job_name, log_ref, error, created_at, finished_at
requested_by, job_name, log_ref, error, created_at, finished_at, context_digest
FROM image_builds WHERE id = $1`
return s.scanBuild(s.db.QueryRowContext(ctx, q, id))
}
@@ -42,9 +42,10 @@ func (s *PGStore) scanBuild(row *sql.Row) (*Build, error) {
status string
ctxRef, base, jobName, logRef, eMsg sql.NullString
finished sql.NullTime
digest sql.NullString
)
switch err := row.Scan(&b.ID, &b.ImageRef, &status, &b.Dockerfile, &ctxRef, &base,
&b.RequestedBy, &jobName, &logRef, &eMsg, &b.CreatedAt, &finished); {
&b.RequestedBy, &jobName, &logRef, &eMsg, &b.CreatedAt, &finished, &digest); {
case err == sql.ErrNoRows:
return nil, ErrNotFound
case err != nil:
@@ -56,6 +57,7 @@ func (s *PGStore) scanBuild(row *sql.Row) (*Build, error) {
b.JobName = jobName.String
b.LogRef = logRef.String
b.Error = eMsg.String
b.ContextDigest = digest.String
if finished.Valid {
t := finished.Time
b.FinishedAt = &t
@@ -91,7 +93,7 @@ func (s *PGStore) FinishBuild(ctx context.Context, id string, status Status, err
func (s *PGStore) ListUnfinishedBuilds(ctx context.Context) ([]Build, error) {
const q = `SELECT id, image_ref, status, dockerfile, context_ref, base_image,
requested_by, job_name, log_ref, error, created_at, finished_at
requested_by, job_name, log_ref, error, created_at, finished_at, context_digest
FROM image_builds WHERE status IN ('pending', 'building') ORDER BY created_at ASC`
rows, err := s.db.QueryContext(ctx, q)
if err != nil {
@@ -105,9 +107,10 @@ func (s *PGStore) ListUnfinishedBuilds(ctx context.Context) ([]Build, error) {
status string
ctxRef, base, jobName, logRef, eMsg sql.NullString
finished sql.NullTime
digest sql.NullString
)
if err := rows.Scan(&b.ID, &b.ImageRef, &status, &b.Dockerfile, &ctxRef, &base,
&b.RequestedBy, &jobName, &logRef, &eMsg, &b.CreatedAt, &finished); err != nil {
&b.RequestedBy, &jobName, &logRef, &eMsg, &b.CreatedAt, &finished, &digest); err != nil {
return nil, err
}
b.Status = Status(status)
@@ -116,6 +119,7 @@ func (s *PGStore) ListUnfinishedBuilds(ctx context.Context) ([]Build, error) {
b.JobName = jobName.String
b.LogRef = logRef.String
b.Error = eMsg.String
b.ContextDigest = digest.String
if finished.Valid {
t := finished.Time
b.FinishedAt = &t
+58
View File
@@ -2,6 +2,7 @@ package build
import (
"fmt"
"net/url"
"regexp"
"strings"
@@ -37,9 +38,66 @@ func Validate(req Request, cfg Config) error {
if strings.TrimSpace(req.ContextRef) == "" {
return invalidf("context reference is required")
}
if IsHTTPContextRef(req.ContextRef) {
if err := validateContextURL(req.ContextRef, cfg.ContextOrigin); err != nil {
return err
}
}
if req.ContextDigest != "" {
if !IsSHA256Hex(req.ContextDigest) {
return invalidf("context digest %q is not a lowercase hex sha256", req.ContextDigest)
}
if !IsHTTPContextRef(req.ContextRef) {
return invalidf("a context digest needs an http(s) context reference, whose fetch step checks it")
}
}
return nil
}
// internalContextPathRE is the one internal-face route a build fetches from.
var internalContextPathRE = regexp.MustCompile(`^/api/v1/internal/submissions/[A-Za-z0-9_-]+/context$`)
// validateContextURL admits an http(s) context only when it is an uploaded
// submission on the platform's internal face. The fetch step sends the service
// token to whatever host the URL names, so an admin-typed URL pointing anywhere
// else would hand that token to a stranger.
func validateContextURL(ref, origin string) error {
if origin == "" {
return invalidf("an http(s) context reference is fetched from the platform's internal API, which this builder is not configured with")
}
u, err := url.Parse(ref)
if err != nil || u.User != nil || u.RawQuery != "" || u.Fragment != "" ||
URLOrigin(ref) != URLOrigin(origin) || !internalContextPathRE.MatchString(u.Path) {
return invalidf("an http(s) context reference must be an uploaded submission on the internal API (%s/api/v1/internal/submissions/<id>/context)",
strings.TrimRight(origin, "/"))
}
return nil
}
// URLOrigin returns the lowercased scheme://host[:port] of raw, or "" when raw
// is not an absolute http(s) URL.
func URLOrigin(raw string) string {
u, err := url.Parse(strings.TrimSpace(raw))
if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
return ""
}
return strings.ToLower(u.Scheme + "://" + u.Host)
}
// IsSHA256Hex reports whether s is a lowercase hex-encoded sha256 digest, the
// form the submit lane records and the context fetcher compares against.
func IsSHA256Hex(s string) bool {
if len(s) != 64 {
return false
}
for _, c := range s {
if (c < '0' || c > '9') && (c < 'a' || c > 'f') {
return false
}
}
return true
}
// ValidateImageRef checks a bare image reference (used by external admission,
// where there is no registry-target constraint beyond well-formedness). A
// whitelist entry may be a tag wildcard ("registry/foo:*", a legitimate
+23 -4
View File
@@ -1153,14 +1153,31 @@ func TestSubmitStoreContract(t *testing.T) {
t.Fatalf("ListSubmissions: (%v, %v), want the submission present", all, err)
}
// The upload records its digest; the approval CAS names it, so an approval of
// a replaced context loses while the row stays pending.
reviewed, swapped := strings.Repeat("a", 64), strings.Repeat("b", 64)
if ok, err := s.SetContextDigest(ctx, id, reviewed); err != nil || !ok {
t.Fatalf("SetContextDigest = (%v, %v), want (true, nil)", ok, err)
}
if got, _ := s.GetSubmission(ctx, id); got.ContextSHA256 != reviewed {
t.Fatalf("context_sha256 = %q, want %q", got.ContextSHA256, reviewed)
}
if ok, err := s.ApproveSubmission(ctx, id, "[email protected]", "registry/x:1", swapped, now); err != nil || ok {
t.Fatalf("approve naming another digest = (%v, %v), want (false, nil)", ok, err)
}
// The approval CAS: exactly one winner, and only from pending_review.
ok, err := s.ApproveSubmission(ctx, id, "[email protected]", "registry.felis.svc:5000/user-uploads/"+id+":latest", now)
ok, err := s.ApproveSubmission(ctx, id, "[email protected]", "registry.felis.svc:5000/user-uploads/"+id+":latest", reviewed, now)
if err != nil || !ok {
t.Fatalf("ApproveSubmission = (%v, %v), want (true, nil)", ok, err)
}
if ok, err := s.ApproveSubmission(ctx, id, "[email protected]", "registry/x:1", now); err != nil || ok {
if ok, err := s.ApproveSubmission(ctx, id, "[email protected]", "registry/x:1", reviewed, now); err != nil || ok {
t.Fatalf("second approve = (%v, %v), want (false, nil)", ok, err)
}
// A reviewed row's digest is frozen: a late upload cannot rewrite it.
if ok, err := s.SetContextDigest(ctx, id, swapped); err != nil || ok {
t.Fatalf("SetContextDigest after approve = (%v, %v), want (false, nil)", ok, err)
}
if ok, err := s.RejectSubmission(ctx, id, "[email protected]", "no", now); err != nil || ok {
t.Fatalf("reject after approve = (%v, %v), want (false, nil)", ok, err)
}
@@ -1218,15 +1235,17 @@ func TestBuildStoreContract(t *testing.T) {
now := mustNow()
done := "bld-done-" + suffix(t)
digest := strings.Repeat("c", 64)
if err := s.CreateBuild(ctx, &build.Build{ID: done, ImageRef: "registry.felis.svc:5000/user-uploads/" + done + ":latest",
Status: build.StatusPending, RequestedBy: "pgint", Dockerfile: "FROM scratch\n", ContextRef: "http://api/x"}); err != nil {
Status: build.StatusPending, RequestedBy: "pgint", Dockerfile: "FROM scratch\n", ContextRef: "http://api/x",
ContextDigest: digest}); err != nil {
t.Fatalf("CreateBuild: %v", err)
}
got, err := s.GetBuild(ctx, done)
if err != nil {
t.Fatalf("GetBuild: %v", err)
}
if got.Status != build.StatusPending || got.JobName != "" {
if got.Status != build.StatusPending || got.JobName != "" || got.ContextDigest != digest {
t.Fatalf("fresh build = %+v", got)
}
if err := s.SetBuildJob(ctx, done, "job-"+suffix(t)); err != nil {
@@ -0,0 +1,12 @@
-- Binds a submission's review to the exact bytes that get built. The upload
-- records the sha256 of the stored context tarball; an admin approves naming
-- the digest they reviewed, and the approve CAS only wins while the row still
-- carries it. A re-upload while pending replaces both the blob and the digest,
-- so an approval issued against the old digest loses instead of building
-- content nobody saw. NULL marks a context uploaded before digests were kept
-- (or no upload yet); such a row cannot be approved until it is uploaded again.
ALTER TABLE image_submissions ADD COLUMN context_sha256 text;
-- The digest the build was allowed to consume. The context-fetch step refuses
-- any other bytes, so a context swapped after approval fails the build.
ALTER TABLE image_builds ADD COLUMN context_digest text;
+15 -5
View File
@@ -22,7 +22,7 @@ func NewPGStore(db *sql.DB) *PGStore { return &PGStore{db: db} }
var _ Store = (*PGStore)(nil)
const submissionColumns = `id, submitted_by, display_name, context_ref, status,
image_ref, build_id, reviewed_by, reject_reason, created_at, reviewed_at`
image_ref, build_id, reviewed_by, reject_reason, created_at, reviewed_at, context_sha256`
func (s *PGStore) CreateSubmission(ctx context.Context, sub *Submission) error {
const q = `INSERT INTO image_submissions
@@ -77,12 +77,20 @@ func (s *PGStore) cas(ctx context.Context, q string, args ...any) (bool, error)
}
// ApproveSubmission is the approve CAS: it flips the row only while it is still
// pending_review, so a concurrent reviewer cannot also win.
func (s *PGStore) ApproveSubmission(ctx context.Context, id, reviewedBy, imageRef string, at time.Time) (bool, error) {
// pending_review AND still carries the digest the reviewer approved, so neither
// a concurrent reviewer nor a re-upload after the review can let it win.
func (s *PGStore) ApproveSubmission(ctx context.Context, id, reviewedBy, imageRef, digest string, at time.Time) (bool, error) {
const q = `UPDATE image_submissions
SET status = 'approved', image_ref = $2, reviewed_by = $3, reviewed_at = $4
WHERE id = $1 AND status = 'pending_review' AND COALESCE(context_sha256, '') = $5`
return s.cas(ctx, q, id, imageRef, reviewedBy, at, digest)
}
// SetContextDigest records an upload's digest, only while the row is pending.
func (s *PGStore) SetContextDigest(ctx context.Context, id, digest string) (bool, error) {
const q = `UPDATE image_submissions SET context_sha256 = $2
WHERE id = $1 AND status = 'pending_review'`
return s.cas(ctx, q, id, imageRef, reviewedBy, at)
return s.cas(ctx, q, id, digest)
}
// RejectSubmission is the reject CAS, mirroring ApproveSubmission.
@@ -164,10 +172,11 @@ func scanSubmissionRows(row rowScanner) (*Submission, error) {
status string
imageRef, buildID, reviewedBy, rejectReason sql.NullString
reviewedAt sql.NullTime
digest sql.NullString
)
if err := row.Scan(
&sub.ID, &sub.SubmittedBy, &sub.DisplayName, &sub.ContextRef, &status,
&imageRef, &buildID, &reviewedBy, &rejectReason, &sub.CreatedAt, &reviewedAt,
&imageRef, &buildID, &reviewedBy, &rejectReason, &sub.CreatedAt, &reviewedAt, &digest,
); err != nil {
return nil, err
}
@@ -176,6 +185,7 @@ func scanSubmissionRows(row rowScanner) (*Submission, error) {
sub.BuildID = buildID.String
sub.ReviewedBy = reviewedBy.String
sub.RejectReason = rejectReason.String
sub.ContextSHA256 = digest.String
if reviewedAt.Valid {
t := reviewedAt.Time
sub.ReviewedAt = &t
+78 -13
View File
@@ -59,6 +59,7 @@ import (
"bufio"
"context"
"crypto/rand"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
@@ -98,6 +99,11 @@ var (
// an honest 503, never a 500, exactly as the restore executor does when its
// integration is not wired.
ErrUploadsUnavailable = errors.New("submit: context upload transport not configured")
// ErrContextChanged reports that the submission's context digest no longer
// matches the one the reviewer approved: the submitter uploaded again after
// the review began. The API maps it to 409 so the panel reloads and the admin
// reviews the new bytes.
ErrContextChanged = errors.New("submit: the build context changed since it was reviewed")
// ErrBlobNotFound reports that a submission has no stored context blob (or it
// was never uploaded). The internal context-fetch route maps it to 404, the
// same distinction Exists draws for Approve.
@@ -170,6 +176,11 @@ type Submission struct {
RejectReason string `json:"reject_reason,omitempty"`
CreatedAt time.Time `json:"created_at"`
ReviewedAt *time.Time `json:"reviewed_at,omitempty"`
// ContextSHA256 is the lowercase hex sha256 of the stored context tarball,
// recorded by the upload that wrote it; empty until one is uploaded. Approve
// must name it, and the build refuses any other bytes.
ContextSHA256 string `json:"context_sha256,omitempty"`
}
// Store is the business-layer persistence the Manager depends on. It is an
@@ -192,7 +203,14 @@ type Store interface {
// derived image_ref, the reviewer and reviewed_at. It reports whether THIS
// call won the transition: false means a concurrent review already moved the
// row, so the caller must NOT start a build.
ApproveSubmission(ctx context.Context, id, reviewedBy, imageRef string, at time.Time) (won bool, err error)
//
// digest is part of the predicate: the row's context_sha256 (NULL read as "")
// must equal it, so an approval of content that a re-upload has since
// replaced loses the CAS.
ApproveSubmission(ctx context.Context, id, reviewedBy, imageRef, digest string, at time.Time) (won bool, err error)
// SetContextDigest records the sha256 of the context an upload just stored,
// only while the row is still pending_review. Reports whether it did.
SetContextDigest(ctx context.Context, id, digest string) (bool, error)
// RejectSubmission atomically flips pending_review -> rejected, recording the
// reviewer, the reason and reviewed_at. Reports whether THIS call won.
RejectSubmission(ctx context.Context, id, reviewedBy, reason string, at time.Time) (won bool, err error)
@@ -371,14 +389,26 @@ func (m *Manager) deriveContextRef(id string) string {
}
// OpenContext returns the stored build context for id — the read path behind the
// internal context-fetch route. It requires the upload transport (Blobs): with no
// transport there is no blob to read, so it reports ErrUploadsUnavailable, the
// same honest 503 the upload endpoint gives.
func (m *Manager) OpenContext(ctx context.Context, id string) (io.ReadCloser, error) {
// internal context-fetch route and the reviewer's download — together with the
// sha256 the row records for it (empty for a context uploaded before digests
// were kept). The row is read first: a re-upload landing between the two reads
// shows up as bytes that do not match the digest, which the caller's verifying
// stream refuses. It requires the upload transport (Blobs): with no transport
// there is no blob to read, so it reports ErrUploadsUnavailable, the same honest
// 503 the upload endpoint gives.
func (m *Manager) OpenContext(ctx context.Context, id string) (io.ReadCloser, string, error) {
if m.Blobs == nil {
return nil, ErrUploadsUnavailable
return nil, "", ErrUploadsUnavailable
}
return m.Blobs.Open(ctx, id)
sub, err := m.Store.GetSubmission(ctx, id)
if err != nil {
return nil, "", err
}
rc, err := m.Blobs.Open(ctx, id)
if err != nil {
return nil, "", err
}
return rc, sub.ContextSHA256, nil
}
// auditDockerfile is the audit-archive Dockerfile recorded on the build row. It
@@ -466,7 +496,10 @@ func (m *Manager) Create(ctx context.Context, req CreateRequest) (*Submission, e
// refused as a spent allowance (403) before the excess is persisted.
//
// A re-upload while still pending atomically supersedes the previous blob, so a
// user can fix their pack before an admin reviews it.
// user can fix their pack before an admin reviews it. The upload hashes what it
// stores and records the digest on the row; Approve binds to that digest, so a
// re-upload after the admin looked makes the approval fail rather than build
// the new bytes unseen.
func (m *Manager) UploadContext(ctx context.Context, id, submittedBy string, r io.Reader) (*Submission, error) {
if strings.TrimSpace(submittedBy) == "" {
return nil, invalidf("submitter identity is required")
@@ -518,9 +551,21 @@ func (m *Manager) UploadContext(ctx context.Context, id, submittedBy string, r i
// refused as a spent allowance, never as a malformed request.
limit, over = remaining, errStorageQuota
}
if _, err := m.Blobs.Put(ctx, id, &cappedReader{r: br, left: limit, over: over}); err != nil {
h := sha256.New()
if _, err := m.Blobs.Put(ctx, id, io.TeeReader(&cappedReader{r: br, left: limit, over: over}, h)); err != nil {
return nil, err
}
digest := hex.EncodeToString(h.Sum(nil))
won, err := m.Store.SetContextDigest(ctx, id, digest)
if err != nil {
return nil, err
}
if !won {
// Reviewed while the bytes streamed in. The blob was replaced anyway, but
// the approved digest no longer matches it, so its build refuses them.
return nil, ErrAlreadyReviewed
}
sub.ContextSHA256 = digest
return sub, nil
}
@@ -611,10 +656,14 @@ func (c *cappedReader) Read(p []byte) (int, error) {
// one and the reason the error is differentiated. The alternative ordering
// (Submit-then-CAS) would either double-build under a concurrent approve or
// orphan a build on a lost race, both worse still.
func (m *Manager) Approve(ctx context.Context, id, reviewedBy string) (*Submission, error) {
func (m *Manager) Approve(ctx context.Context, id, reviewedBy, expectedDigest string) (*Submission, error) {
if strings.TrimSpace(reviewedBy) == "" {
return nil, invalidf("reviewer identity is required")
}
expectedDigest = strings.ToLower(strings.TrimSpace(expectedDigest))
if m.Blobs != nil && !build.IsSHA256Hex(expectedDigest) {
return nil, invalidf("expected_digest must be the sha256 of the context you reviewed (64 hex characters)")
}
sub, err := m.Store.GetSubmission(ctx, id)
if err != nil {
@@ -638,6 +687,12 @@ func (m *Manager) Approve(ctx context.Context, id, reviewedBy string) (*Submissi
if !ok {
return nil, invalidf("no build context has been uploaded for this submission")
}
if sub.ContextSHA256 == "" {
return nil, invalidf("this context was uploaded before digests were recorded; the submitter must upload it again")
}
if sub.ContextSHA256 != expectedDigest {
return nil, ErrContextChanged
}
}
imageRef := m.deriveImageRef(id)
@@ -648,19 +703,29 @@ func (m *Manager) Approve(ctx context.Context, id, reviewedBy string) (*Submissi
BaseImage: "", // declared inside the uploaded context, unknown here
RequestedBy: reviewedBy,
}
// Pin the build to the reviewed bytes: the fetch step refuses anything else,
// which covers an upload that was already streaming when the CAS won. Only an
// http(s) context has that step; the installed API always derives one.
if build.IsHTTPContextRef(sub.ContextRef) {
req.ContextDigest = sub.ContextSHA256
}
// Pre-validate against the SAME registry the Builder enforces, BEFORE the CAS,
// so a deterministic config error cannot strand the row in approved.
if err := build.Validate(req, build.Config{RegistryURL: m.Registry}); err != nil {
if err := build.Validate(req, build.Config{RegistryURL: m.Registry, ContextOrigin: m.ContextBaseURL}); err != nil {
return nil, err
}
now := m.now()
won, err := m.Store.ApproveSubmission(ctx, id, reviewedBy, imageRef, now)
won, err := m.Store.ApproveSubmission(ctx, id, reviewedBy, imageRef, sub.ContextSHA256, now)
if err != nil {
return nil, err
}
if !won {
// Lost the race to a concurrent approve/reject.
// Lost the race: a concurrent approve/reject moved the row, or a re-upload
// replaced the digest between the read above and the CAS.
if cur, err := m.Store.GetSubmission(ctx, id); err == nil && cur.Status == StatusPendingReview {
return nil, ErrContextChanged
}
return nil, ErrAlreadyReviewed
}
+133 -16
View File
@@ -3,6 +3,8 @@ package submit
import (
"bytes"
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"io"
@@ -103,6 +105,10 @@ type fakeStore struct {
deleteErr error
linked []string // "id=buildID" recorder
// beforeApprove runs between the Manager's read and its CAS, the window a
// concurrent re-upload lands in.
beforeApprove func()
}
func newFakeStore() *fakeStore { return &fakeStore{subs: map[string]*Submission{}} }
@@ -156,12 +162,15 @@ func (f *fakeStore) ListSubmissionsBy(_ context.Context, by string) ([]Submissio
return out, nil
}
func (f *fakeStore) ApproveSubmission(_ context.Context, id, reviewedBy, imageRef string, at time.Time) (bool, error) {
func (f *fakeStore) ApproveSubmission(_ context.Context, id, reviewedBy, imageRef, digest string, at time.Time) (bool, error) {
if f.beforeApprove != nil {
f.beforeApprove()
}
if f.approveErr != nil {
return false, f.approveErr
}
s, ok := f.subs[id]
if !ok || s.Status != StatusPendingReview {
if !ok || s.Status != StatusPendingReview || s.ContextSHA256 != digest {
return false, nil // CAS lost / nonexistent
}
s.Status = StatusApproved
@@ -172,6 +181,15 @@ func (f *fakeStore) ApproveSubmission(_ context.Context, id, reviewedBy, imageRe
return true, nil
}
func (f *fakeStore) SetContextDigest(_ context.Context, id, digest string) (bool, error) {
s, ok := f.subs[id]
if !ok || s.Status != StatusPendingReview {
return false, nil
}
s.ContextSHA256 = digest
return true, nil
}
func (f *fakeStore) RejectSubmission(_ context.Context, id, reviewedBy, reason string, at time.Time) (bool, error) {
if f.rejectErr != nil {
return false, f.rejectErr
@@ -307,14 +325,14 @@ func TestContextRefIsFetchURLAndOpenContextServesIt(t *testing.T) {
}
// Before any upload the read path reports not-found (the route's 404).
if _, err := m.OpenContext(ctx, sub.ID); !errors.Is(err, ErrBlobNotFound) {
if _, _, err := m.OpenContext(ctx, sub.ID); !errors.Is(err, ErrBlobNotFound) {
t.Fatalf("OpenContext before upload = %v, want ErrBlobNotFound", err)
}
payload := "\x1f\x8b\x08\x00payload"
if _, err := m.UploadContext(ctx, sub.ID, "user-1", strings.NewReader(payload)); err != nil {
t.Fatalf("UploadContext: %v", err)
}
rc, err := m.OpenContext(ctx, sub.ID)
rc, digest, err := m.OpenContext(ctx, sub.ID)
if err != nil {
t.Fatalf("OpenContext: %v", err)
}
@@ -323,13 +341,16 @@ func TestContextRefIsFetchURLAndOpenContextServesIt(t *testing.T) {
if string(got) != payload {
t.Fatalf("OpenContext served %q, want %q", got, payload)
}
if digest != sha256Hex(payload) {
t.Fatalf("OpenContext digest = %q, want the payload's sha256", digest)
}
}
// No upload transport ⇒ no readable blob: the route reports the same 503 the
// upload endpoint does, rather than a misleading 404.
func TestOpenContextWithoutTransportIsUnavailable(t *testing.T) {
m, _, _ := newManager()
if _, err := m.OpenContext(context.Background(), "sub-1"); !errors.Is(err, ErrUploadsUnavailable) {
if _, _, err := m.OpenContext(context.Background(), "sub-1"); !errors.Is(err, ErrUploadsUnavailable) {
t.Fatalf("OpenContext with nil Blobs = %v, want ErrUploadsUnavailable", err)
}
}
@@ -362,7 +383,7 @@ func TestApproveStartsExactlyOneBuildAndLinks(t *testing.T) {
m, st, bl := newManager()
seed, _ := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
sub, err := m.Approve(context.Background(), seed.ID, "[email protected]")
sub, err := m.Approve(context.Background(), seed.ID, "[email protected]", "")
if err != nil {
t.Fatalf("Approve: %v", err)
}
@@ -428,11 +449,11 @@ func TestApproveIsCASNoDoubleBuild(t *testing.T) {
m, _, bl := newManager()
seed, _ := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
if _, err := m.Approve(context.Background(), seed.ID, "admin@x"); err != nil {
if _, err := m.Approve(context.Background(), seed.ID, "admin@x", ""); err != nil {
t.Fatalf("first Approve: %v", err)
}
// A second approve loses the CAS and must NOT start another build.
_, err := m.Approve(context.Background(), seed.ID, "admin@x")
_, err := m.Approve(context.Background(), seed.ID, "admin@x", "")
if !errors.Is(err, ErrAlreadyReviewed) {
t.Fatalf("second Approve err = %v, want ErrAlreadyReviewed", err)
}
@@ -447,7 +468,7 @@ func TestApproveRejectedSubmission(t *testing.T) {
if _, err := m.Reject(context.Background(), seed.ID, "admin@x", "nope"); err != nil {
t.Fatalf("Reject: %v", err)
}
_, err := m.Approve(context.Background(), seed.ID, "admin@x")
_, err := m.Approve(context.Background(), seed.ID, "admin@x", "")
if !errors.Is(err, ErrAlreadyReviewed) {
t.Fatalf("Approve after reject err = %v, want ErrAlreadyReviewed", err)
}
@@ -458,7 +479,7 @@ func TestApproveRejectedSubmission(t *testing.T) {
func TestApproveUnknown(t *testing.T) {
m, _, _ := newManager()
_, err := m.Approve(context.Background(), "sub-nope", "admin@x")
_, err := m.Approve(context.Background(), "sub-nope", "admin@x", "")
if !errors.Is(err, ErrNotFound) {
t.Fatalf("err = %v, want ErrNotFound", err)
}
@@ -472,7 +493,7 @@ func TestApproveBuildFailureLeavesApprovedUnlinked(t *testing.T) {
bl.submitErr = errors.New("apiserver unreachable")
seed, _ := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
_, err := m.Approve(context.Background(), seed.ID, "admin@x")
_, err := m.Approve(context.Background(), seed.ID, "admin@x", "")
if err == nil {
t.Fatal("Approve should surface the build hand-off failure")
}
@@ -503,7 +524,7 @@ func TestApproveLinkFailureNamesRunningBuild(t *testing.T) {
st.linkErr = errors.New("db write timeout")
seed, _ := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
_, err := m.Approve(context.Background(), seed.ID, "admin@x")
_, err := m.Approve(context.Background(), seed.ID, "admin@x", "")
if err == nil {
t.Fatal("Approve must surface the link-write failure (a build is running)")
}
@@ -544,7 +565,7 @@ func TestApproveValidatesBeforeCAS(t *testing.T) {
m.Registry = "" // derived ref becomes "/user-uploads/...", not host-qualified
seed, _ := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
_, err := m.Approve(context.Background(), seed.ID, "admin@x")
_, err := m.Approve(context.Background(), seed.ID, "admin@x", "")
if !errors.Is(err, build.ErrInvalid) {
t.Fatalf("err = %v, want build.ErrInvalid (pre-CAS validation)", err)
}
@@ -1009,7 +1030,7 @@ func TestApproveRefusesMissingContext(t *testing.T) {
m.Blobs = newFakeBlobs() // empty → Exists=false
seed, _ := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
_, err := m.Approve(context.Background(), seed.ID, "admin@x")
_, err := m.Approve(context.Background(), seed.ID, "admin@x", sha256Hex("never uploaded"))
if !errors.Is(err, ErrInvalid) {
t.Fatalf("err = %v, want ErrInvalid (no context uploaded)", err)
}
@@ -1026,12 +1047,17 @@ func TestApproveProceedsWithUploadedContext(t *testing.T) {
// build through the SAME gated Builder.
m, _, bl := newManager()
m.Blobs = newFakeBlobs()
m.ContextBaseURL = "http://felis-api-internal.felis.svc.cluster.local:8081"
seed, _ := m.Create(context.Background(), CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
if _, err := m.UploadContext(context.Background(), seed.ID, "user-1", strings.NewReader(gzBody("mods"))); err != nil {
up, err := m.UploadContext(context.Background(), seed.ID, "user-1", strings.NewReader(gzBody("mods")))
if err != nil {
t.Fatalf("UploadContext: %v", err)
}
if up.ContextSHA256 != sha256Hex(gzBody("mods")) {
t.Fatalf("upload digest = %q, want the stored bytes' sha256", up.ContextSHA256)
}
sub, err := m.Approve(context.Background(), seed.ID, "admin@x")
sub, err := m.Approve(context.Background(), seed.ID, "admin@x", up.ContextSHA256)
if err != nil {
t.Fatalf("Approve: %v", err)
}
@@ -1041,6 +1067,97 @@ func TestApproveProceedsWithUploadedContext(t *testing.T) {
if bl.calls != 1 {
t.Fatalf("builds started = %d, want 1", bl.calls)
}
if bl.got[0].ContextDigest != up.ContextSHA256 {
t.Fatalf("build digest = %q, want the approved %q", bl.got[0].ContextDigest, up.ContextSHA256)
}
}
// Review binds to bytes (build-supply-chain-6): an approval names the digest
// the admin inspected, and a re-upload after that makes it fail instead of
// building content nobody saw.
func TestApproveBindsTheReviewedDigest(t *testing.T) {
ctx := context.Background()
m, st, bl := newManager()
m.Blobs = newFakeBlobs()
seed, _ := m.Create(ctx, CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
benign, evil := gzBody("benign"), gzBody("evil")
if _, err := m.UploadContext(ctx, seed.ID, "user-1", strings.NewReader(benign)); err != nil {
t.Fatal(err)
}
reviewed := sha256Hex(benign)
// The swap lands after the review.
if _, err := m.UploadContext(ctx, seed.ID, "user-1", strings.NewReader(evil)); err != nil {
t.Fatal(err)
}
if _, err := m.Approve(ctx, seed.ID, "admin@x", reviewed); !errors.Is(err, ErrContextChanged) {
t.Fatalf("approve after a swap = %v, want ErrContextChanged", err)
}
// The swap lands between the Manager's read and its CAS.
if _, err := m.UploadContext(ctx, seed.ID, "user-1", strings.NewReader(benign)); err != nil {
t.Fatal(err)
}
st.beforeApprove = func() { st.subs[seed.ID].ContextSHA256 = sha256Hex(evil) }
if _, err := m.Approve(ctx, seed.ID, "admin@x", reviewed); !errors.Is(err, ErrContextChanged) {
t.Fatalf("approve racing a swap = %v, want ErrContextChanged", err)
}
st.beforeApprove = nil
if st.subs[seed.ID].Status != StatusPendingReview || bl.calls != 0 {
t.Fatalf("status %q, builds %d; want still pending with nothing built", st.subs[seed.ID].Status, bl.calls)
}
for _, bad := range []string{"", "not-a-digest"} {
if _, err := m.Approve(ctx, seed.ID, "admin@x", bad); !errors.Is(err, ErrInvalid) {
t.Fatalf("approve with digest %q = %v, want ErrInvalid", bad, err)
}
}
// A context uploaded before digests were recorded must be uploaded again.
st.subs[seed.ID].ContextSHA256 = ""
if _, err := m.Approve(ctx, seed.ID, "admin@x", reviewed); !errors.Is(err, ErrInvalid) ||
!strings.Contains(err.Error(), "upload it again") {
t.Fatalf("approve of an undigested context = %v, want the re-upload instruction", err)
}
}
// An upload that finishes after the approval won cannot slip its digest in: the
// row keeps the approved one, which the build's fetch step then enforces, and
// the uploader hears that the submission was reviewed.
func TestUploadAfterApprovalKeepsTheApprovedDigest(t *testing.T) {
ctx := context.Background()
m, st, _ := newManager()
fb := newFakeBlobs()
m.Blobs = fb
seed, _ := m.Create(ctx, CreateRequest{DisplayName: "Pack", SubmittedBy: "user-1"})
up, _ := m.UploadContext(ctx, seed.ID, "user-1", strings.NewReader(gzBody("benign")))
// The second upload passed its pending check; the approval wins while its
// bytes are still streaming in.
m.Blobs = approvingBlobs{fakeBlobs: fb, approve: func() { st.subs[seed.ID].Status = StatusApproved }}
if _, err := m.UploadContext(ctx, seed.ID, "user-1", strings.NewReader(gzBody("evil"))); !errors.Is(err, ErrAlreadyReviewed) {
t.Fatalf("late upload = %v, want ErrAlreadyReviewed", err)
}
if st.subs[seed.ID].ContextSHA256 != up.ContextSHA256 {
t.Fatalf("row digest = %q, want the approved %q", st.subs[seed.ID].ContextSHA256, up.ContextSHA256)
}
}
// approvingBlobs flips the row to approved once Put has stored the bytes.
type approvingBlobs struct {
*fakeBlobs
approve func()
}
func (a approvingBlobs) Put(ctx context.Context, id string, r io.Reader) (int64, error) {
n, err := a.fakeBlobs.Put(ctx, id, r)
a.approve()
return n, err
}
func sha256Hex(s string) string {
sum := sha256.Sum256([]byte(s))
return hex.EncodeToString(sum[:])
}
// keysOf lists a map's keys for test diagnostics.