json-proxy: add OpenJSONRPCFdPass to broker splitfdstream sockets

Add a new json-proxy method that returns a Unix socket file descriptor
to the client.  The client then speaks the jsonrpc-fdpass protocol
directly over that socket for splitfdstream operations, bypassing the
json-proxy for bulk data transfer.

Bump protocolVersion to "0.2.9".

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Giuseppe Scrivano <gscrivan@redhat.com>

Author: giuseppe

common/cmd/json-proxy-test-server/main.go

@@ -8,12 +8,22 @@ import (
 	"fmt"
 	"os"
 
-	jsonproxy "go.podman.io/common/pkg/json-proxy"
+	imgcopy "go.podman.io/image/v5/copy"
 	"go.podman.io/image/v5/signature"
+	istorage "go.podman.io/image/v5/storage"
+	"go.podman.io/image/v5/transports/alltransports"
 	"go.podman.io/image/v5/types"
+	"go.podman.io/storage"
+	"go.podman.io/storage/pkg/reexec"
+	storagetypes "go.podman.io/storage/types"
+
+	jsonproxy "go.podman.io/common/pkg/json-proxy"
 )
 
 func main() {
+	if reexec.Init() {
+		return
+	}
 	if err := run(); err != nil {
 		fmt.Fprintf(os.Stderr, "error: %v\n", err)
 		os.Exit(1)
@@ -24,10 +34,25 @@ func run() error {
 	sockfd := flag.Int("sockfd", -1, "socket file descriptor")
 	policyPath := flag.String("policy", "", "path to policy.json (default: system default)")
 	overrideArch := flag.String("override-arch", "", "override architecture for manifest list resolution")
+	graphRoot := flag.String("graph-root", "", "storage graph root")
+	runRoot := flag.String("run-root", "", "storage run root")
+	seedImage := flag.String("seed-image", "", "image to copy into local store")
 	flag.Parse()
 
 	if *sockfd < 0 {
-		return fmt.Errorf("usage: %s --sockfd <fd> [--policy <path>] [--override-arch <arch>]", os.Args[0])
+		return fmt.Errorf("usage: %s --sockfd <fd> [--policy <path>] [--override-arch <arch>] [--graph-root <path> --run-root <path> --seed-image <ref>]", os.Args[0])
+	}
+
+	if *graphRoot != "" {
+		ref, store, err := setupStore(*graphRoot, *runRoot, *seedImage)
+		if err != nil {
+			return fmt.Errorf("setting up store: %w", err)
+		}
+		defer func() {
+			_, _ = store.Shutdown(true)
+		}()
+		// Print the containers-storage:// reference for the test to read.
+		fmt.Fprintln(os.Stdout, ref)
 	}
 
 	manager, err := jsonproxy.NewManager(
@@ -58,3 +83,47 @@ func run() error {
 	defer manager.Close()
 	return manager.Serve(context.Background(), *sockfd)
 }
+
+func setupStore(graphRoot, runRoot, seedImage string) (string, storage.Store, error) {
+	store, err := storage.GetStore(storagetypes.StoreOptions{
+		GraphRoot:       graphRoot,
+		RunRoot:         runRoot,
+		GraphDriverName: "overlay",
+	})
+	if err != nil {
+		return "", nil, fmt.Errorf("creating store: %w", err)
+	}
+
+	ctx := context.Background()
+
+	srcRef, err := alltransports.ParseImageName(seedImage)
+	if err != nil {
+		return "", nil, fmt.Errorf("parsing seed image %q: %w", seedImage, err)
+	}
+
+	destRef, err := istorage.Transport.ParseStoreReference(store, "testimage:latest")
+	if err != nil {
+		return "", nil, fmt.Errorf("creating store reference: %w", err)
+	}
+
+	policy, err := signature.DefaultPolicy(nil)
+	if err != nil {
+		return "", nil, fmt.Errorf("getting default policy: %w", err)
+	}
+	pc, err := signature.NewPolicyContext(policy)
+	if err != nil {
+		return "", nil, fmt.Errorf("creating policy context: %w", err)
+	}
+	defer func() {
+		if err := pc.Destroy(); err != nil {
+			fmt.Fprintf(os.Stderr, "warning: destroying policy context: %v\n", err)
+		}
+	}()
+
+	_, err = imgcopy.Image(ctx, pc, destRef, srcRef, nil)
+	if err != nil {
+		return "", nil, fmt.Errorf("copying seed image: %w", err)
+	}
+
+	return "containers-storage:" + destRef.StringWithinTransport(), store, nil
+}

common/pkg/json-proxy/handler.go

@@ -28,9 +28,10 @@ type handler struct {
 	lock sync.Mutex
 
 	// Dependency injection functions.
-	getSystemContext func() (*types.SystemContext, error)
-	getPolicyContext func() (*signature.PolicyContext, error)
-	logger           logrus.FieldLogger
+	getSystemContext   func() (*types.SystemContext, error)
+	getPolicyContext   func() (*signature.PolicyContext, error)
+	splitFDStreamStore splitFDStreamStore
+	logger             logrus.FieldLogger
 
 	// Internal state.
 	sysctx      *types.SystemContext
@@ -141,6 +142,12 @@ func (h *handler) openImageImpl(ctx context.Context, args []any, allowNotFound b
 		return ret, err
 	}
 
+	if h.splitFDStreamStore == nil {
+		if sfds, ok := imgsrc.(splitFDStreamStore); ok {
+			h.splitFDStreamStore = sfds
+		}
+	}
+
 	unparsedTopLevel := image.UnparsedInstance(imgsrc, nil)
 	// Check the signature on the toplevel (possibly multi-arch) manifest, but we don't
 	// yet propagate the error here.
@@ -689,6 +696,31 @@ func (h *handler) FinishPipe(ctx context.Context, args []any) (replyBuf, error)
 	return ret, err
 }
 
+// OpenJSONRPCFdPass returns a socket FD over which the client can
+// speak the jsonrpc-fdpass protocol for splitfdstream operations.
+// The json-proxy does not interpret the protocol; it just brokers the socket.
+func (h *handler) OpenJSONRPCFdPass(ctx context.Context, args []any) (replyBuf, error) {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	var ret replyBuf
+
+	if h.splitFDStreamStore == nil {
+		return ret, errors.New("splitfdstream store not configured")
+	}
+	if len(args) != 0 {
+		return ret, fmt.Errorf("found %d args, expecting none", len(args))
+	}
+
+	sockFile, err := h.splitFDStreamStore.SplitFDStreamSocket()
+	if err != nil {
+		return ret, err
+	}
+
+	ret.fd = sockFile
+	return ret, nil
+}
+
 // processRequest dispatches a remote request.
 // replyBuf is the result of the invocation.
 // terminate should be true if processing of requests should halt.
@@ -728,6 +760,8 @@ func (h *handler) processRequest(ctx context.Context, readBytes []byte) (rb repl
 		rb, err = h.GetLayerInfoPiped(ctx, req.Args)
 	case "FinishPipe":
 		rb, err = h.FinishPipe(ctx, req.Args)
+	case "OpenJSONRPCFdPass":
+		rb, err = h.OpenJSONRPCFdPass(ctx, req.Args)
 	case "Shutdown":
 		terminate = true
 	// NOTE: If you add a method here, you should very likely be bumping the

common/pkg/json-proxy/proxy.go

@@ -4,11 +4,20 @@
 package jsonproxy
 
 import (
+	"os"
+
 	"github.com/sirupsen/logrus"
 	"go.podman.io/image/v5/signature"
 	"go.podman.io/image/v5/types"
 )
 
+// splitFDStreamStore is the subset of storage.SplitFDStreamStore needed
+// by the json-proxy.  Keeping a local interface avoids a hard dependency
+// on go.podman.io/storage for consumers that do not use splitfdstream.
+type splitFDStreamStore interface {
+	SplitFDStreamSocket() (*os.File, error)
+}
+
 // options holds the internal configuration for a Manager.
 type options struct {
 	getSystemContext func() (*types.SystemContext, error)

common/pkg/json-proxy/proxy_test.go

@@ -3,12 +3,14 @@
 package jsonproxy_test
 
 import (
+	"bufio"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net"
 	"os"
 	"os/exec"
+	"path/filepath"
 	"strings"
 	"sync"
 	"syscall"
@@ -560,3 +562,128 @@ func TestProxyPolicyVerification(t *testing.T) {
 		})
 	}
 }
+
+// newProxyWithStore spawns the test binary with a local containers-storage
+// store seeded with the given image. It returns the proxy and the
+// containers-storage:// reference string for the seeded image.
+func newProxyWithStore(t *testing.T, seedImage string) (*proxy, string) {
+	t.Helper()
+
+	proxyBinary := os.Getenv("JSON_PROXY_TEST_BINARY")
+	if proxyBinary == "" {
+		t.Skip("JSON_PROXY_TEST_BINARY is not set; skipping integration test")
+	}
+
+	wd := t.TempDir()
+	graphRoot := filepath.Join(wd, "root")
+	runRoot := filepath.Join(wd, "run")
+
+	fds, err := syscall.Socketpair(syscall.AF_LOCAL, syscall.SOCK_SEQPACKET, 0)
+	require.NoError(t, err)
+	myfd := os.NewFile(uintptr(fds[0]), "myfd")
+	defer myfd.Close()
+	theirfd := os.NewFile(uintptr(fds[1]), "theirfd")
+	defer theirfd.Close()
+
+	mysock, err := net.FileConn(myfd)
+	require.NoError(t, err)
+	unixConn, ok := mysock.(*net.UnixConn)
+	require.True(t, ok, "expected *net.UnixConn, got %T", mysock)
+
+	proc := exec.Command(proxyBinary, //nolint:gosec
+		"--sockfd", "3",
+		"--graph-root", graphRoot,
+		"--run-root", runRoot,
+		"--seed-image", seedImage,
+	)
+	proc.Stderr = os.Stderr
+	proc.ExtraFiles = append(proc.ExtraFiles, theirfd)
+
+	stdoutPipe, err := proc.StdoutPipe()
+	require.NoError(t, err)
+
+	err = proc.Start()
+	require.NoError(t, err)
+
+	// Read the containers-storage reference from stdout.
+	scanner := bufio.NewScanner(stdoutPipe)
+	require.True(t, scanner.Scan(), "expected storage reference on stdout")
+	storageRef := strings.TrimSpace(scanner.Text())
+	require.True(t, strings.HasPrefix(storageRef, "containers-storage:"), "unexpected ref: %s", storageRef)
+
+	p := &proxy{
+		c:    unixConn,
+		proc: proc,
+	}
+	t.Cleanup(p.close)
+
+	v, err := p.callNoFd("Initialize", nil)
+	require.NoError(t, err)
+	semver, ok := v.(string)
+	require.True(t, ok, "proxy Initialize: Unexpected value %T", v)
+	require.True(t, strings.HasPrefix(semver, expectedProxySemverMajor), "Unexpected semver %s", semver)
+
+	return p, storageRef
+}
+
+func TestOpenJSONRPCFdPass(t *testing.T) {
+	p, storageRef := newProxyWithStore(t, knownListImage)
+
+	// Open the containers-storage image to trigger auto-discovery.
+	imgidVal, err := p.callNoFd("OpenImage", []any{storageRef})
+	require.NoError(t, err)
+	imgid, ok := imgidVal.(float64)
+	require.True(t, ok)
+	require.NotZero(t, imgid)
+
+	// OpenJSONRPCFdPass should return a valid FD.
+	_, fd, err := p.call("OpenJSONRPCFdPass", nil)
+	require.NoError(t, err)
+	require.NotNil(t, fd, "expected an FD from OpenJSONRPCFdPass")
+
+	// Verify the received FD is a unix socket.
+	var stat syscall.Stat_t
+	err = syscall.Fstat(int(fd.datafd.Fd()), &stat)
+	require.NoError(t, err)
+	require.True(t, stat.Mode&syscall.S_IFMT == syscall.S_IFSOCK, "expected socket, got mode %o", stat.Mode)
+
+	// Validate the socket speaks the splitfdstream jsonrpc-fdpass protocol.
+	// Send a JSON-RPC request for a bogus method and expect a method-not-found error.
+	conn, err := net.FileConn(fd.datafd)
+	fd.datafd.Close()
+	require.NoError(t, err)
+	unixSock, ok := conn.(*net.UnixConn)
+	require.True(t, ok)
+	defer unixSock.Close()
+
+	rpcReq := []byte("{\"jsonrpc\":\"2.0\",\"method\":\"NoSuchMethod\",\"id\":1}\n")
+	_, err = unixSock.Write(rpcReq)
+	require.NoError(t, err)
+
+	respBuf := make([]byte, 4096)
+	n, err := unixSock.Read(respBuf)
+	require.NoError(t, err)
+	var rpcResp map[string]any
+	err = json.Unmarshal(respBuf[:n], &rpcResp)
+	require.NoError(t, err)
+	// A valid JSON-RPC server returns an error object for unknown methods.
+	rpcErr, ok := rpcResp["error"].(map[string]any)
+	require.True(t, ok, "expected JSON-RPC error object, got %v", rpcResp)
+	require.Contains(t, rpcErr["message"], "not found")
+
+	_, err = p.callNoFd("CloseImage", []any{imgid})
+	require.NoError(t, err)
+}
+
+func TestOpenJSONRPCFdPassNotAvailable(t *testing.T) {
+	p := newProxy(t)
+
+	// Open a docker:// image (no splitfdstream support).
+	_, err := p.callNoFd("OpenImage", []any{knownNotManifestListedImageX8664})
+	require.NoError(t, err)
+
+	// OpenJSONRPCFdPass should fail since no containers-storage source was opened.
+	_, _, err = p.call("OpenJSONRPCFdPass", nil)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "splitfdstream store not configured")
+}

common/pkg/json-proxy/types.go

@@ -17,7 +17,7 @@ import (
 // departure from the original code which used HTTP.
 //
 // When bumping this, please also update the man page.
-const protocolVersion = "0.2.8"
+const protocolVersion = "0.2.9"
 
 // maxMsgSize is the current limit on a packet size.
 // Note that all non-metadata (i.e. payload data) is sent over a pipe.