diff --git a/integration/cloud_test.go b/integration/cloud_test.go index 96728c777..835774a31 100644 --- a/integration/cloud_test.go +++ b/integration/cloud_test.go @@ -96,6 +96,11 @@ func testSchema(t *testing.T, prefix string) (*sql.DB, *pgx.ConnConfig) { return pool, config } +// setGinTestModeOnce makes setup safe to call from parallel tests: gin.SetMode writes a +// package global, so concurrent setups would race under the race detector. The first call +// wins and every router is created after the mode is fixed. +var setGinTestModeOnce sync.Once + func setup(t *testing.T) *fixture { t.Helper() pool, config := testSchema(t, "test_") @@ -116,7 +121,7 @@ func setup(t *testing.T) *fixture { must(t, e) auth, e := core.NewAuthenticator("ora-cloud", credentials.Trust) must(t, e) - gin.SetMode(gin.TestMode) + setGinTestModeOnce.Do(func() { gin.SetMode(gin.TestMode) }) log, _ := zap.NewDevelopment() cloud := httptest.NewServer(router.NewDevelopment(store, auth, log)) t.Cleanup(cloud.Close) diff --git a/integration/workflow_asset_fetch_boundaries_test.go b/integration/workflow_asset_fetch_boundaries_test.go new file mode 100644 index 000000000..87f0e9b6a --- /dev/null +++ b/integration/workflow_asset_fetch_boundaries_test.go @@ -0,0 +1,347 @@ +package integration + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "io" + "math/big" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/wanglongan587/cloud/internal/egress" + "github.com/wanglongan587/cloud/internal/modelgateway" + "github.com/wanglongan587/cloud/internal/workflowasset" +) + +// boundaryMode selects one served package behavior. The modes map one-to-one onto the +// approved fetch bounds: connection establishment, response headers, the whole-package +// deadline, the streaming size bound, and the repaired package for the retry. +type boundaryMode string + +const ( + boundaryHoldTLS boundaryMode = "hold-tls" + boundaryNoHeaders boundaryMode = "no-headers" + boundaryDribble boundaryMode = "dribble" + boundaryOversize boundaryMode = "oversize" + boundaryGood boundaryMode = "good" +) + +// boundaryOversizePlanned is far past the 16 MiB streaming bound plus every socket buffer +// involved, so the server can never finish writing it after the client stops reading. +const boundaryOversizePlanned int64 = 128 << 20 + +// testUpstreamIdentity is the CA and leaf the boundary server presents. The fetcher's +// transport is constructed with the CA in its root pool, so the boundary tests verify +// upstream TLS identity exactly as production does, without touching platform roots. +type testUpstreamIdentity struct { + pool *x509.CertPool + leaf tls.Certificate +} + +func newUpstreamIdentity(t *testing.T) testUpstreamIdentity { + t.Helper() + caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + must(t, err) + caTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "ora-asset-boundary-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + IsCA: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + BasicConstraintsValid: true, + } + caDER, err := x509.CreateCertificate(rand.Reader, caTemplate, caTemplate, &caKey.PublicKey, caKey) + must(t, err) + ca, err := x509.ParseCertificate(caDER) + must(t, err) + pool := x509.NewCertPool() + pool.AddCert(ca) + + leafKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + must(t, err) + leafTemplate := &x509.Certificate{ + SerialNumber: big.NewInt(2), + Subject: pkix.Name{CommonName: "127.0.0.1"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + leafDER, err := x509.CreateCertificate(rand.Reader, leafTemplate, ca, &leafKey.PublicKey, caKey) + must(t, err) + return testUpstreamIdentity{pool: pool, leaf: tls.Certificate{Certificate: [][]byte{leafDER}, PrivateKey: leafKey}} +} + +// boundaryServer is the controlled HTTPS server the fetch-boundary tests point the catalog +// at. It owns one TCP port for its whole life so a catalog URL stays valid across the +// failure and the repaired retry. Every connection goes through the production transport's +// real checks: DNS is exempted only for the fixture host 127.0.0.1, TLS identity is +// verified against the generated CA, and the response behavior is exactly the selected +// mode. The mode switch is what models a publisher repairing a broken release. +type boundaryServer struct { + tlsCfg *tls.Config + ln net.Listener + mode atomic.Value + good []byte + + // written and interrupted report what the oversize stream actually reached: the byte + // count the socket accepted, and a one-time signal that a write failed before the + // planned end. They are the server-side half of the "interrupted before the end" + // evidence; the client-side half is the bounded failure code. + written atomic.Int64 + interrupted chan struct{} + failOnce sync.Once +} + +func newBoundaryServer(t *testing.T, identity testUpstreamIdentity, good []byte, mode boundaryMode) *boundaryServer { + t.Helper() + s := &boundaryServer{ + tlsCfg: &tls.Config{Certificates: []tls.Certificate{identity.leaf}}, + good: good, + interrupted: make(chan struct{}), + } + s.mode.Store(mode) + ln, err := net.Listen("tcp", "127.0.0.1:0") + must(t, err) + s.ln = ln + mux := http.NewServeMux() + mux.HandleFunc("/package.orax", s.serve) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go s.handle(conn, mux) + } + }() + t.Cleanup(func() { _ = ln.Close() }) + return s +} + +func (s *boundaryServer) url() string { return "https://" + s.ln.Addr().String() + "/package.orax" } +func (s *boundaryServer) switchTo(m boundaryMode) { s.mode.Store(m) } + +// handle routes one accepted connection. The hold-tls mode holds the raw connection +// without any TLS bytes, so the transport's 10-second connection deadline must cancel the +// fetch; every other mode completes the TLS handshake and serves real HTTP over it. +func (s *boundaryServer) handle(conn net.Conn, mux *http.ServeMux) { + if s.mode.Load() == boundaryHoldTLS { + // The client speaks first in TLS; drain its ClientHello and stay silent. The + // deadline bounds the hold so a lost client cannot leak the goroutine. + _ = conn.SetDeadline(time.Now().Add(2 * time.Minute)) + _, _ = io.Copy(io.Discard, conn) + _ = conn.Close() + return + } + server := &http.Server{Handler: mux, ReadHeaderTimeout: 10 * time.Second} + _ = server.Serve(&singleConnListener{conn: tls.Server(conn, s.tlsCfg)}) +} + +// singleConnListener hands exactly one connection to http.Server.Serve. Each accepted +// connection gets its own server instance, so nothing is shared between connections. +type singleConnListener struct { + conn net.Conn + once sync.Once +} + +func (l *singleConnListener) Accept() (net.Conn, error) { + var conn net.Conn + l.once.Do(func() { conn = l.conn }) + if conn == nil { + return nil, net.ErrClosed + } + return conn, nil +} + +func (l *singleConnListener) Close() error { return nil } +func (l *singleConnListener) Addr() net.Addr { return l.conn.LocalAddr() } + +func (s *boundaryServer) serve(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/octet-stream") + switch s.mode.Load() { + case boundaryNoHeaders: + // ResponseHeaderTimeout must be what fires, so nothing is written: no status, no + // headers. The wait ends when the cancelled client closes the connection. + select { + case <-r.Context().Done(): + case <-time.After(2 * time.Minute): + } + case boundaryDribble: + // Headers arrive immediately; the body trickles far slower than the 60-second + // whole-package deadline can consume, and never finishes inside it. + w.WriteHeader(http.StatusOK) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-r.Context().Done(): + return + case <-ticker.C: + if _, err := w.Write([]byte("x")); err != nil { + return + } + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + } + } + case boundaryOversize: + // No Content-Length: the declared-size refusal cannot fire, so only the streaming + // count can interrupt the download. The planned volume is far beyond every buffer + // in the path, so a client that stops reading at the bound leaves the server + // blocked and then failed before it can finish writing. + w.WriteHeader(http.StatusOK) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + chunk := make([]byte, 64<<10) + for s.written.Load() < boundaryOversizePlanned { + n, err := w.Write(chunk) + s.written.Add(int64(n)) + if err != nil { + s.failOnce.Do(func() { close(s.interrupted) }) + return + } + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + } + default: + w.WriteHeader(http.StatusOK) + _, _ = w.Write(s.good) + } +} + +// realAssetFetcher wires the fetcher over the production upstream transport, not the test +// server's own client: the 10-second connection deadline, the 30-second response-header +// deadline and the checked dialer all come from modelgateway exactly as cmd/server wires +// them. The fixture-host exemption admits only 127.0.0.1, the public deployment policy keeps +// every production gate on, and the injected root pool keeps TLS verification real. +func realAssetFetcher(t *testing.T, identity testUpstreamIdentity) *workflowasset.Fetcher { + t.Helper() + policy, err := egress.Parse([]byte(`{"schemaVersion":1,"mode":"public"}`)) + must(t, err) + client, err := modelgateway.NewUpstreamClientWithRoots(modelgateway.UpstreamConfig{ + DevelopmentHosts: []string{"127.0.0.1"}, + DNS: egress.DefaultDNSConfig(), + Policy: policy, + }, identity.pool) + must(t, err) + t.Cleanup(client.CloseIdleConnections) + return workflowasset.NewFetcher(nil, client.Transport) +} + +// TestWorkflowAssetImportCancelsSlowOrOversizePackages covers the fetch-boundary row of +// specs test-cases/cloud/plugin-marketplace/workflow-plugin-assets.md: slow or oversized +// packages are cancelled by the real transport at the approved 10/30/60-second deadlines +// and the 16 MiB streaming bound, the failure is package-level (no document rows, no +// workflows), and the next administrator trigger retries the same URL and succeeds once +// the publisher's storage is repaired. Each case runs through the production transport +// chain; the subtests share nothing (own schema, marketplace and server) and run in +// parallel, so the test costs its longest boundary rather than their sum. +func TestWorkflowAssetImportCancelsSlowOrOversizePackages(t *testing.T) { + cases := []struct { + name string + mode boundaryMode + code string + // min and max bracket the deadline that must have cancelled the fetch: the lower + // bound proves no earlier failure fired, the upper bound (with generous CI slack) + // distinguishes this deadline from the next-larger one. + min, max time.Duration + }{ + {"connection never established within ten seconds", boundaryHoldTLS, workflowasset.FailureUnavailable, 10 * time.Second, 30 * time.Second}, + {"response headers never arrive within thirty seconds", boundaryNoHeaders, workflowasset.FailureUnavailable, 30 * time.Second, 50 * time.Second}, + {"whole package never completes within sixty seconds", boundaryDribble, workflowasset.FailureUnavailable, 60 * time.Second, 80 * time.Second}, + {"streamed package interrupted past the sixteen MiB bound", boundaryOversize, workflowasset.FailureTooLarge, 0, 0}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + t.Parallel() + f := setup(t) + // Two documents, so a package-level failure must leave neither: even the + // document that would import fine proves the failure is not per-document. + good := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Release Kit", ""), + "deploy.json": executableDocument("Deploy Kit", ""), + }) + repo := workflowMarketplace(t, f.root) + identity := newUpstreamIdentity(t) + server := newBoundaryServer(t, identity, good, c.mode) + publishWorkflowRelease(t, repo, "slow-kit", "1.0.0", server.url(), good) + f.syncMarketplace(t, repo) + f.store.WorkflowAssets = realAssetFetcher(t, identity) + + sid := f.defaultSpaceID() + f.installWorkflowPlugin(t, sid, "official/slow-kit", "") + start := time.Now() + err := f.importWorkflowAssets(t) + if err == nil || !strings.Contains(err.Error(), c.code) { + t.Fatalf("slow package must surface its bounded code %s, got %v", c.code, err) + } + elapsed := time.Since(start) + if elapsed < c.min { + t.Fatalf("the fetch was cancelled early at %s, before its %s deadline", elapsed, c.min) + } + if c.max > 0 && elapsed > c.max { + t.Fatalf("the fetch took %s, past the slack for its %s deadline", elapsed, c.min) + } + item := f.assetImport(t, sid, "official/slow-kit") + if item == nil || item.S("state") != "failed" || item.S("failureCode") != c.code { + t.Fatalf("package failure = %v", item) + } + if n := f.scalar("SELECT count(*) FROM workflow_asset_documents"); n != 0 { + t.Fatalf("a package failure must not record document outcomes: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflows"); n != 0 { + t.Fatalf("a package failure must not create workflows: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots"); n != 0 { + t.Fatalf("a package failure must not publish snapshots: %d rows", n) + } + if c.mode == boundaryOversize { + // The client stopped reading past the bound; the server must observe the + // interrupted stream once the closed socket refuses further writes. + select { + case <-server.interrupted: + case <-time.After(10 * time.Second): + t.Fatal("the server finished writing; the client did not interrupt the stream") + } + if written := server.written.Load(); written < (16<<20)+1 || written >= boundaryOversizePlanned { + t.Fatalf("the server wrote %d bytes; the interruption must land past the bound and before the planned end", written) + } + } + + // The publisher repairs the same URL; the next administrator trigger retries + // and succeeds against the identical catalog selection. + server.switchTo(boundaryGood) + f.installWorkflowPlugin(t, sid, "official/slow-kit", "1.0.0") + must(t, f.importWorkflowAssets(t)) + item = f.assetImport(t, sid, "official/slow-kit") + if item == nil || item.S("state") != "completed" || item.S("failureCode") != "" { + t.Fatalf("retried package = %v", item) + } + documents := assetDocuments(item) + if len(documents) != 2 { + t.Fatalf("retried documents = %v", documents) + } + if live := f.liveWorkflows(t); len(live) != 2 || live["Release Kit"] == nil || live["Deploy Kit"] == nil { + t.Fatalf("retried import did not create the workflows: %v", live) + } + }) + } +} diff --git a/integration/workflow_asset_import_scope_test.go b/integration/workflow_asset_import_scope_test.go new file mode 100644 index 000000000..fea73cac6 --- /dev/null +++ b/integration/workflow_asset_import_scope_test.go @@ -0,0 +1,190 @@ +package integration + +import ( + "testing" + + "github.com/wanglongan587/cloud/internal/core" + "github.com/wanglongan587/cloud/internal/workflowasset" +) + +// TestWorkflowAssetImportDoesNotFanOutToWorkspaces covers the fan-out boundary of +// specs test-cases/cloud/plugin-marketplace/workflow-plugin-assets.md: the import is +// triggered by the space's desired state and stays scoped to it. The zero-Workspace case +// is already covered by TestWorkflowAssetImportFromInstalledPlugin; this is the missing +// complement — with several live Workspaces in the space, the Node-side install still +// fans out (one workspace_plugin_instances row per live runtime), but the import must not +// follow it: one source row, one workflow and one published snapshot per document. +func TestWorkflowAssetImportDoesNotFanOutToWorkspaces(t *testing.T) { + f := setup(t) + repo := workflowMarketplace(t, f.root) + pkg := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Release Kit", ""), + "deploy.json": executableDocument("Deploy Kit", ""), + }) + packages := newWorkflowPackages(t, map[string][]byte{"/packages/fanout-1.0.0.orax": pkg}) + publishWorkflowRelease(t, repo, "fanout-kit", "1.0.0", packages.server.URL+"/packages/fanout-1.0.0.orax", pkg) + f.syncMarketplace(t, repo) + f.store.WorkflowAssets = workflowasset.NewFetcher(nil, packages.server.Client().Transport) + + // Several live Workspaces in the one space: the project's main Workspace plus two + // isolated ones, all ready before the install. + created := f.create("fanout-project") + f.drain() + pid := created.O("resource").S("id") + for _, target := range []struct{ title, key string }{{"Fanout One", "fanout-ws-one"}, {"Fanout Two", "fanout-ws-two"}} { + out := f.call("POST", f.path("/projects/"+pid+"/workspaces"), core.Object{"title": target.title, "baseRef": "main"}, target.key, 202) + f.drain() + wid := out.O("resource").S("id") + if state := f.ws(wid).S("observedState"); state != "ready" { + t.Fatalf("workspace %s must be live before the install, got %q", wid, state) + } + } + + sid := f.defaultSpaceID() + f.installWorkflowPlugin(t, sid, "official/fanout-kit", "") + f.drain() + // The Node-side install fans out to every live runtime; the import must not follow it. + liveWorkspaces := f.scalar("SELECT count(*) FROM workspaces WHERE deleted_at IS NULL") + if liveWorkspaces < 2 { + t.Fatalf("the fixture expected at least two live workspaces, got %d", liveWorkspaces) + } + if n := f.scalar("SELECT count(*) FROM workspace_plugin_instances WHERE identifier='fanout-kit'"); n != liveWorkspaces { + t.Fatalf("the install must fan out to all %d live workspaces, got %d instances", liveWorkspaces, n) + } + + must(t, f.importWorkflowAssets(t)) + item := f.assetImport(t, sid, "official/fanout-kit") + if item == nil || item.S("state") != "completed" || item.S("packageVersion") != "1.0.0" { + t.Fatalf("package import = %v", item) + } + if documents := assetDocuments(item); len(documents) != 2 { + t.Fatalf("documents = %v", documents) + } + // One source row for the whole space, not one per workspace. + if n := f.scalar("SELECT count(*) FROM workflow_asset_imports WHERE identifier='fanout-kit'"); n != 1 { + t.Fatalf("the import fanned out to workspaces: %d source rows", n) + } + // One workflow and one published snapshot per document, despite several live runtimes. + if n := f.scalar("SELECT count(*) FROM workflow_asset_documents WHERE identifier='fanout-kit'"); n != 2 { + t.Fatalf("per-workspace document results appeared: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflows WHERE deleted_at IS NULL"); n != 2 { + t.Fatalf("workflows were created per workspace: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots"); n != 2 { + t.Fatalf("snapshots were published per workspace: %d rows", n) + } + if live := f.liveWorkflows(t); len(live) != 2 || live["Release Kit"] == nil || live["Deploy Kit"] == nil { + t.Fatalf("imported workflows = %v", live) + } + + // A restart-equivalent second pass still creates nothing new. + must(t, f.importWorkflowAssets(t)) + if n := f.scalar("SELECT count(*) FROM workflow_asset_documents"); n != 2 { + t.Fatalf("second pass changed document results: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflows"); n != 2 { + t.Fatalf("second pass created workflows: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots"); n != 2 { + t.Fatalf("second pass published snapshots: %d rows", n) + } +} + +// TestWorkflowAssetImportKeepsUserEditsAcrossUpgrade covers the copy-ownership boundary of +// specs test-cases/cloud/plugin-marketplace/workflow-plugin-assets.md: once a document has +// imported, the copy is user data. After the user edits the imported workflow and publishes +// their own snapshot, a package upgrade imports only paths that never succeeded; the user's +// live graph and every published snapshot keep their exact stored text, and the upgraded +// document's new content appears nowhere. +func TestWorkflowAssetImportKeepsUserEditsAcrossUpgrade(t *testing.T) { + f := setup(t) + repo := workflowMarketplace(t, f.root) + v1 := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Edit Kit", `"origin":"package-1.0.0"`), + "deploy.json": executableDocument("Deploy Kit", ""), + }) + v2 := workflowPackage(t, map[string]string{ + "release.json": executableDocument("Edit Kit", `"origin":"package-1.1.0"`), + "deploy.json": executableDocument("Deploy Kit", ""), + "hotfix.json": executableDocument("Hotfix Kit", ""), + }) + packages := newWorkflowPackages(t, map[string][]byte{ + "/packages/edit-kit-1.0.0.orax": v1, + "/packages/edit-kit-1.1.0.orax": v2, + }) + publishWorkflowRelease(t, repo, "edit-kit", "1.0.0", packages.server.URL+"/packages/edit-kit-1.0.0.orax", v1) + f.syncMarketplace(t, repo) + f.store.WorkflowAssets = workflowasset.NewFetcher(nil, packages.server.Client().Transport) + + sid := f.defaultSpaceID() + f.installWorkflowPlugin(t, sid, "official/edit-kit", "") + must(t, f.importWorkflowAssets(t)) + documents := assetDocuments(f.assetImport(t, sid, "official/edit-kit")) + editID := documents["assets/workflows/release.json"].S("workflowId") + if editID == "" || documents["assets/workflows/deploy.json"].S("workflowId") == "" { + t.Fatalf("documents were not imported: %v", documents) + } + + // The user edits the imported copy and publishes their own second snapshot. + edited := f.liveWorkflows(t)["Edit Kit"] + editedGraph := core.Object{"nodes": []any{map[string]any{"id": "user-start", "data": map[string]any{"kind": "start"}}}, "edges": []any{}} + f.call("PUT", f.path("/workflows/"+editID), core.Object{"graph": editedGraph, "version": edited.N("version")}, "edit-user-edit", 200) + f.call("POST", f.path("/workflows/"+editID+"/publish"), core.Object{}, "edit-user-publish", 200) + snapshotText := func(version int64) string { + t.Helper() + var text string + must(t, f.store.Pool.QueryRow("SELECT graph::text FROM workflow_snapshots WHERE workflow_id=$1 AND version=$2", editID, version).Scan(&text)) + return text + } + liveText := func() string { + t.Helper() + var text string + must(t, f.store.Pool.QueryRow("SELECT graph::text FROM workflows WHERE id=$1", editID).Scan(&text)) + return text + } + liveBefore, importSnapshotBefore, userSnapshotBefore := liveText(), snapshotText(1), snapshotText(2) + + // Upgrade the package: the already-imported path now carries different content. + publishWorkflowRelease(t, repo, "edit-kit", "1.1.0", packages.server.URL+"/packages/edit-kit-1.1.0.orax", v2) + f.syncMarketplace(t, repo) + f.installWorkflowPlugin(t, sid, "official/edit-kit", "1.1.0") + must(t, f.importWorkflowAssets(t)) + + documents = assetDocuments(f.assetImport(t, sid, "official/edit-kit")) + if len(documents) != 3 || documents["assets/workflows/hotfix.json"].S("outcome") != "imported" { + t.Fatalf("the upgrade must import only the new path: %v", documents) + } + // The imported path keeps pointing at the copy the user now owns. + if documents["assets/workflows/release.json"].S("workflowId") != editID { + t.Fatalf("the upgrade revisited an imported path: %v", documents["assets/workflows/release.json"]) + } + if live := f.liveWorkflows(t); len(live) != 3 || live["Hotfix Kit"] == nil { + t.Fatalf("workflow library after upgrade = %v", live) + } + // The user's live graph and both published snapshots keep their exact stored text. + if after := liveText(); after != liveBefore { + t.Fatalf("the upgrade overwrote the user's edit: %s -> %s", liveBefore, after) + } + if after := snapshotText(1); after != importSnapshotBefore { + t.Fatalf("the upgrade rewrote the imported snapshot: %s -> %s", importSnapshotBefore, after) + } + if after := snapshotText(2); after != userSnapshotBefore { + t.Fatalf("the upgrade rewrote the user's snapshot: %s -> %s", userSnapshotBefore, after) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots WHERE workflow_id=$1", editID); n != 2 { + t.Fatalf("the upgrade must not add snapshots: %d rows", n) + } + // The upgraded document's new content appears nowhere. + if n := f.scalar("SELECT count(*) FROM workflows WHERE graph::text LIKE '%package-1.1.0%'"); n != 0 { + t.Fatalf("package upgrade content leaked into a live graph: %d rows", n) + } + if n := f.scalar("SELECT count(*) FROM workflow_snapshots WHERE graph::text LIKE '%package-1.1.0%'"); n != 0 { + t.Fatalf("package upgrade content leaked into a snapshot: %d rows", n) + } + // The untouched sibling keeps exactly its own one snapshot. + deployID := documents["assets/workflows/deploy.json"].S("workflowId") + if n := f.scalar("SELECT count(*) FROM workflow_snapshots WHERE workflow_id=$1", deployID); n != 1 { + t.Fatalf("the sibling's snapshots changed: %d rows", n) + } +} diff --git a/internal/modelgateway/transport.go b/internal/modelgateway/transport.go index d0db9bf3d..e3c6c8a19 100644 --- a/internal/modelgateway/transport.go +++ b/internal/modelgateway/transport.go @@ -2,6 +2,8 @@ package modelgateway import ( "context" + "crypto/tls" + "crypto/x509" "errors" "net" "net/http" @@ -83,6 +85,21 @@ type UpstreamConfig struct { // NewUpstreamClient disables environmental proxies and redirects, and enforces the deployment // policy on every request and new connection. Exact fixture host exemptions are deployment-owned, development only. func NewUpstreamClient(config UpstreamConfig) (*http.Client, error) { + return newUpstreamClient(config, nil) +} + +// NewUpstreamClientWithRoots builds the same client but verifies upstream identity against roots +// instead of the platform pool. Hermetic transport tests use it against a controlled HTTPS server +// and keep every other check exactly as deployed; production wiring only calls NewUpstreamClient, +// so upstream identity always comes from the deployment's system roots. +func NewUpstreamClientWithRoots(config UpstreamConfig, roots *x509.CertPool) (*http.Client, error) { + if roots == nil { + return nil, errors.New("upstream TLS trust requires an explicit root pool") + } + return newUpstreamClient(config, roots) +} + +func newUpstreamClient(config UpstreamConfig, roots *x509.CertPool) (*http.Client, error) { if config.Policy.Enterprise() && len(config.DevelopmentHosts) != 0 { return nil, errors.New("enterprise egress conflicts with development hosts") } @@ -103,6 +120,9 @@ func NewUpstreamClient(config UpstreamConfig) (*http.Client, error) { } d := &secureDialer{resolver: resolver, fixtures: net.DefaultResolver, dialer: net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}, allowed: allowed, policy: config.Policy} transport := &http.Transport{Proxy: nil, DialContext: d.dial, TLSHandshakeTimeout: 10 * time.Second, ResponseHeaderTimeout: 30 * time.Second, IdleConnTimeout: 60 * time.Second, MaxIdleConns: 64, MaxConnsPerHost: 32, MaxResponseHeaderBytes: 64 << 10, ForceAttemptHTTP2: !config.Policy.Enterprise()} + if roots != nil { + transport.TLSClientConfig = &tls.Config{RootCAs: roots} + } return &http.Client{Transport: &upstreamTransport{Transport: transport, resolver: resolver, policy: config.Policy}, CheckRedirect: func(*http.Request, []*http.Request) error { return endpointRedirectForbidden }}, nil }