From cbadc5ea29ca99de8b95349808ca71faae60ecb9 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Thu, 29 Oct 2026 03:52:38 +0800 Subject: [PATCH] E1.3: verify owned mapped waiters before one-shot release Add invocation-owned inhibited Podman waiter startup with exact immutable-image, owner/peer, runtime and kernel process checks. Reuse the existing native admission writer before a single bounded framed stdin handoff; retain uncertain ownership and clean up only the exact created instance. Preserve application stdin after the frame, keep admission deadlines separate from application lifetime, and reject missing proof, stale process evidence, unsupported streams and uncertain release. Add framing, real process/namespace and held-operation-lock callback tests. Public mapped execution remains inhibited; the qualified final exec transition and production storage/network admission remain E2/F gates. --- .../unreleased/+mapped-waiter-admission.yaml | 2 + internal/dockerdeploy/mapped_process_linux.go | 188 +++++ .../dockerdeploy/mapped_process_linux_test.go | 354 +++++++++ .../dockerdeploy/mapped_process_other_test.go | 16 + internal/dockerdeploy/podman_mapped_waiter.go | 279 +++++++ .../dockerdeploy/podman_mapped_waiter_test.go | 693 ++++++++++++++++++ .../dockerdeploy/podman_mapping_peer_other.go | 6 + internal/mappedlaunch/release.go | 136 ++++ internal/mappedlaunch/release_test.go | 174 +++++ internal/probe/main.go | 9 + internal/probe/mapped_wait.go | 112 +++ internal/probe/mapped_wait_other_test.go | 18 + internal/probe/mapped_wait_test.go | 230 ++++++ internal/probe/sandbox_exec_other.go | 9 +- 14 files changed, 2225 insertions(+), 1 deletion(-) create mode 100644 .changes/unreleased/+mapped-waiter-admission.yaml create mode 100644 internal/dockerdeploy/mapped_process_linux.go create mode 100644 internal/dockerdeploy/mapped_process_linux_test.go create mode 100644 internal/dockerdeploy/mapped_process_other_test.go create mode 100644 internal/dockerdeploy/podman_mapped_waiter.go create mode 100644 internal/dockerdeploy/podman_mapped_waiter_test.go create mode 100644 internal/mappedlaunch/release.go create mode 100644 internal/mappedlaunch/release_test.go create mode 100644 internal/probe/mapped_wait.go create mode 100644 internal/probe/mapped_wait_other_test.go create mode 100644 internal/probe/mapped_wait_test.go diff --git a/.changes/unreleased/+mapped-waiter-admission.yaml b/.changes/unreleased/+mapped-waiter-admission.yaml new file mode 100644 index 00000000..55592e56 --- /dev/null +++ b/.changes/unreleased/+mapped-waiter-admission.yaml @@ -0,0 +1,2 @@ +kind: Added +body: Add internal owned-waiter observations and a bounded one-shot stdin release protocol for mapped containers. Missing storage/network proof blocks release, and mapped application execution remains unavailable pending final startup qualification. diff --git a/internal/dockerdeploy/mapped_process_linux.go b/internal/dockerdeploy/mapped_process_linux.go new file mode 100644 index 00000000..bdb14bc8 --- /dev/null +++ b/internal/dockerdeploy/mapped_process_linux.go @@ -0,0 +1,188 @@ +//go:build linux + +package dockerdeploy + +import ( + "bytes" + "fmt" + "io" + "os" + "reflect" + "strconv" + "strings" + "syscall" + + "github.com/omry/reploy/internal/identitymapping" + "golang.org/x/sys/unix" +) + +type mappedProcessSnapshotV1 struct { + start string + namespaces [5]string + device uint64 + inode uint64 +} + +type mappedProcessV1 struct { + directory *os.File + pidfd int + active identitymapping.Geometry + argv []byte + initial mappedProcessSnapshotV1 +} + +func pinMappedProcessV1(pid int, active identitymapping.Geometry, argv []string) (mappedWaiterProcessV1, error) { + active, err := identitymapping.ValidateGeometry(active) + if err != nil || len(argv) == 0 { + return nil, fmt.Errorf("mapped waiter requires exact active geometry and launch arguments") + } + if pid <= 0 { + return nil, fmt.Errorf("mapped waiter requires a live host PID") + } + pidfd, err := unix.PidfdOpen(pid, 0) + if err != nil { + return nil, fmt.Errorf("pin mapped waiter: %w", err) + } + p := &mappedProcessV1{pidfd: pidfd, active: active, argv: []byte(strings.Join(argv, "\x00") + "\x00")} + fd, err := unix.Open("/proc/"+strconv.Itoa(pid), unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0) + if err == nil { + p.directory = os.NewFile(uintptr(fd), "mapped-waiter-proc") + p.initial, err = p.observe() + } + if err != nil { + p.close() + return nil, err + } + return p, nil +} + +func (p *mappedProcessV1) close() { + if p.directory != nil { + _ = p.directory.Close() + } + _ = unix.Close(p.pidfd) +} + +func (p *mappedProcessV1) fresh() error { + current, err := p.observe() + if err == nil && current != p.initial { + err = fmt.Errorf("mapped waiter process or namespace identity changed") + } + return err +} + +func (p *mappedProcessV1) observe() (mappedProcessSnapshotV1, error) { + var result mappedProcessSnapshotV1 + poll := []unix.PollFd{{Fd: int32(p.pidfd), Events: unix.POLLIN}} + if n, err := unix.Poll(poll, 0); err != nil || n != 0 { + return result, fmt.Errorf("mapped waiter exited or its process pin is unavailable") + } + root := "/proc/self/fd/" + strconv.FormatUint(uint64(p.directory.Fd()), 10) + read := func(name string) ([]byte, error) { + f, err := os.Open(root + "/" + name) + if err != nil { + return nil, err + } + defer f.Close() + data, err := io.ReadAll(io.LimitReader(f, (64<<10)+1)) + if len(data) > 64<<10 { + return nil, fmt.Errorf("mapped process observation exceeds its limit") + } + return data, err + } + before, err := read("stat") + if err != nil { + return result, err + } + result.start, err = mappedProcessStartV1(before) + if err != nil { + return result, err + } + actual := identitymapping.Geometry{Profile: p.active.Profile} + for i, name := range []string{"uid_map", "gid_map"} { + data, err := read(name) + if err != nil { + return result, err + } + rows, err := mappedKernelRowsV1(data) + if err != nil { + return result, err + } + if i == 0 { + actual.UIDs = rows + } else { + actual.GIDs = rows + } + } + actual, err = identitymapping.ValidateGeometry(actual) + if err != nil || !reflect.DeepEqual(actual, p.active) { + return result, fmt.Errorf("complete kernel maps differ from admitted active geometry") + } + var owner unix.Stat_t + if err := unix.Fstat(int(p.directory.Fd()), &owner); err != nil || p.active.UIDs[0].ContainerStart != 0 || p.active.GIDs[0].ContainerStart != 0 || owner.Uid != p.active.UIDs[0].HostStart || owner.Gid != p.active.GIDs[0].HostStart { + return result, fmt.Errorf("mapped waiter kernel owner differs from mapped setup root") + } + for i, name := range []string{"user", "pid", "mnt", "net", "ipc"} { + link, err := os.Readlink(root + "/ns/" + name) + host, hostErr := os.Readlink("/proc/self/ns/" + name) + if err != nil || hostErr != nil || link == host || !strings.HasPrefix(link, name+":[") { + return result, fmt.Errorf("mapped waiter lacks isolated %s namespace evidence", name) + } + result.namespaces[i] = link + } + executable, err := os.Stat(root + "/exe") + trusted, trustedErr := os.Stat(root + "/root/reploy-probe") + if err != nil || trustedErr != nil || !executable.Mode().IsRegular() || !os.SameFile(executable, trusted) { + return result, fmt.Errorf("mapped waiter is not the immutable trusted startup executable") + } + metadata := executable.Sys().(*syscall.Stat_t) + result.device, result.inode = uint64(metadata.Dev), metadata.Ino + command, err := read("cmdline") + if err != nil || !bytes.Equal(command, p.argv) { + return result, fmt.Errorf("mapped waiter command differs from its exact trusted launch") + } + after, err := read("stat") + start, parseErr := mappedProcessStartV1(after) + if err != nil || parseErr != nil || start != result.start { + return result, fmt.Errorf("mapped waiter process changed during observation") + } + if n, err := unix.Poll(poll, 0); err != nil || n != 0 { + return result, fmt.Errorf("mapped waiter exited during observation") + } + return result, nil +} + +func mappedProcessStartV1(data []byte) (string, error) { + // comm may contain spaces and closing parentheses; fields after its last + // closing parenthesis start at field 3. starttime is field 22. + i := bytes.LastIndexByte(data, ')') + fields := strings.Fields(string(data[i+1:])) + if i < 0 || len(fields) < 20 || fields[0] == "Z" || fields[0] == "X" { + return "", fmt.Errorf("invalid or exited mapped process stat") + } + n, err := strconv.ParseUint(fields[19], 10, 64) + if err != nil || n == 0 { + return "", fmt.Errorf("invalid mapped process start identity") + } + return fields[19], nil +} + +func mappedKernelRowsV1(data []byte) ([]identitymapping.Interval, error) { + var rows []identitymapping.Interval + for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") { + fields := strings.Fields(line) + if len(fields) != 3 { + return nil, fmt.Errorf("invalid complete kernel map row") + } + var values [3]uint32 + for i, s := range fields { + n, err := strconv.ParseUint(s, 10, 32) + if err != nil { + return nil, fmt.Errorf("invalid complete kernel map number") + } + values[i] = uint32(n) + } + rows = append(rows, identitymapping.Interval{ContainerStart: values[0], HostStart: values[1], Count: values[2]}) + } + return rows, nil +} diff --git a/internal/dockerdeploy/mapped_process_linux_test.go b/internal/dockerdeploy/mapped_process_linux_test.go new file mode 100644 index 00000000..daef3f39 --- /dev/null +++ b/internal/dockerdeploy/mapped_process_linux_test.go @@ -0,0 +1,354 @@ +//go:build linux + +package dockerdeploy + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "syscall" + "testing" + "time" + + "github.com/omry/reploy/internal/identitymapping" + "golang.org/x/sys/unix" +) + +// This filesystem fixture exercises the observer's complete comparisons. It is +// not kernel namespace provenance or a production qualification receipt. +func mappedProcessFixtureV1(t *testing.T) (*mappedProcessV1, string) { + t.Helper() + root := t.TempDir() + write := func(name, body string) { + t.Helper() + if err := os.WriteFile(filepath.Join(root, name), []byte(body), 0600); err != nil { + t.Fatal(err) + } + } + fields := append([]string{"S"}, make([]string, 19)...) + for i := 1; i < len(fields); i++ { + fields[i] = "0" + } + fields[19] = "12345" + write("stat", "123 (trusted ) waiter) "+strings.Join(fields, " ")) + write("uid_map", fmt.Sprintf("0 %d 1\n200 100200 1\n", os.Getuid())) + write("gid_map", fmt.Sprintf("0 %d 1\n200 100200 1\n", os.Getgid())) + write("cmdline", "/reploy-probe\x00mapped-wait-v1\x00") + for _, name := range []string{"ns", "root"} { + if err := os.Mkdir(filepath.Join(root, name), 0700); err != nil { + t.Fatal(err) + } + } + for _, name := range []string{"user", "pid", "mnt", "net", "ipc"} { + if err := os.Symlink(name+":[12345]", filepath.Join(root, "ns", name)); err != nil { + t.Fatal(err) + } + } + write("exe", "synthetic trusted executable") + if err := os.Link(filepath.Join(root, "exe"), filepath.Join(root, "root/reploy-probe")); err != nil { + t.Fatal(err) + } + directory, err := os.Open(root) + if err != nil { + t.Fatal(err) + } + fd, err := unix.PidfdOpen(os.Getpid(), 0) + if err != nil { + t.Fatal(err) + } + p := &mappedProcessV1{directory: directory, pidfd: fd, active: identitymapping.Geometry{Profile: "exact", UIDs: []identitymapping.Interval{{HostStart: uint32(os.Getuid()), Count: 1}, {ContainerStart: 200, HostStart: 100200, Count: 1}}, GIDs: []identitymapping.Interval{{HostStart: uint32(os.Getgid()), Count: 1}, {ContainerStart: 200, HostStart: 100200, Count: 1}}}, argv: []byte("/reploy-probe\x00mapped-wait-v1\x00")} + t.Cleanup(p.close) + p.initial, err = p.observe() + if err != nil { + t.Fatal("observer fixture", err) + } + return p, root +} + +func TestMappedProcessChecksEveryObservationV1(t *testing.T) { + if os.Getuid() == 0 || os.Getgid() == 0 { + t.Skip("mapped host observations require non-root owner") + } + for _, mode := range []string{"unchanged", "start", "missing stat", "exited stat", "map extra", "map missing", "map overflow", "map zero", "map parse error", "missing map", "oversize map", "uid owner", "gid owner", "no zero", "namespace", "ipc namespace", "missing namespace", "missing ipc namespace", "changed ipc namespace", "wrong executable", "missing executable", "command", "missing command", "dead pin", "closed directory"} { + t.Run(mode, func(t *testing.T) { + p, root := mappedProcessFixtureV1(t) + write := func(name, body string) { + t.Helper() + if err := os.WriteFile(filepath.Join(root, name), []byte(body), 0600); err != nil { + t.Fatal(err) + } + } + switch mode { + case "start": + data, _ := os.ReadFile(filepath.Join(root, "stat")) + write("stat", strings.Replace(string(data), "12345", "54321", 1)) + case "missing stat": + _ = os.Remove(filepath.Join(root, "stat")) + case "exited stat": + write("stat", "123 (dead) Z") + case "map extra": + write("uid_map", fmt.Sprintf("0 %d 1\n200 100200 1\n300 100300 1\n", os.Getuid())) + case "map missing": + write("gid_map", fmt.Sprintf("0 %d 1\n", os.Getgid())) + case "map overflow": + write("uid_map", "0 4294967295 2\n") + case "map zero": + write("uid_map", "0 100000 0\n") + case "map parse error": + write("uid_map", "0 100000\n") + case "missing map": + _ = os.Remove(filepath.Join(root, "gid_map")) + case "oversize map": + write("uid_map", strings.Repeat("x", (64<<10)+1)) + case "uid owner": + p.active.UIDs[0].HostStart++ + write("uid_map", fmt.Sprintf("0 %d 1\n200 100200 1\n", p.active.UIDs[0].HostStart)) + case "gid owner": + p.active.GIDs[0].HostStart++ + write("gid_map", fmt.Sprintf("0 %d 1\n200 100200 1\n", p.active.GIDs[0].HostStart)) + case "no zero": + p.active.UIDs[0].ContainerStart = 1 + write("uid_map", fmt.Sprintf("1 %d 1\n200 100200 1\n", os.Getuid())) + case "namespace": + _ = os.Remove(filepath.Join(root, "ns/net")) + target, _ := os.Readlink("/proc/self/ns/net") + _ = os.Symlink(target, filepath.Join(root, "ns/net")) + case "ipc namespace": + _ = os.Remove(filepath.Join(root, "ns/ipc")) + target, _ := os.Readlink("/proc/self/ns/ipc") + _ = os.Symlink(target, filepath.Join(root, "ns/ipc")) + case "missing namespace": + _ = os.Remove(filepath.Join(root, "ns/pid")) + case "missing ipc namespace": + _ = os.Remove(filepath.Join(root, "ns/ipc")) + case "changed ipc namespace": + _ = os.Remove(filepath.Join(root, "ns/ipc")) + if err := os.Symlink("ipc:[11111]", filepath.Join(root, "ns/ipc")); err != nil { + t.Fatal(err) + } + initial, err := p.observe() + if err != nil { + t.Fatal(err) + } + p.initial = initial + if err := os.Remove(filepath.Join(root, "ns/ipc")); err != nil { + t.Fatal(err) + } + if err := os.Symlink("ipc:[22222]", filepath.Join(root, "ns/ipc")); err != nil { + t.Fatal(err) + } + case "wrong executable": + _ = os.Remove(filepath.Join(root, "root/reploy-probe")) + write("root/reploy-probe", "substitution") + case "missing executable": + _ = os.Remove(filepath.Join(root, "exe")) + case "command": + write("cmdline", "/reploy-probe\x00sandbox-exec\x00") + case "missing command": + _ = os.Remove(filepath.Join(root, "cmdline")) + case "dead pin": + _ = unix.Close(p.pidfd) + case "closed directory": + _ = p.directory.Close() + } + err := p.fresh() + if (err == nil) != (mode == "unchanged") { + t.Fatalf("%s: %v", mode, err) + } + }) + } +} + +func TestMappedProcessChangesDuringObservationV1(t *testing.T) { + if os.Getuid() == 0 || os.Getgid() == 0 { + t.Skip("mapped host observations require non-root owner") + } + for _, mode := range []string{"start changed", "process pin lost"} { + t.Run(mode, func(t *testing.T) { + p, root := mappedProcessFixtureV1(t) + name := filepath.Join(root, "cmdline") + if err := os.Remove(name); err != nil { + t.Fatal(err) + } + if err := unix.Mkfifo(name, 0600); err != nil { + t.Fatal(err) + } + done := make(chan error, 1) + go func() { done <- p.fresh() }() + // Opening the FIFO writer synchronizes after the first stat/maps/ns + // reads and before the final stat/liveness fences. + writer, err := os.OpenFile(name, os.O_WRONLY, 0600) + if err != nil { + t.Fatal(err) + } + if mode == "start changed" { + data, err := os.ReadFile(filepath.Join(root, "stat")) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(root, "stat"), bytes.Replace(data, []byte("12345"), []byte("54321"), 1), 0600); err != nil { + t.Fatal(err) + } + } else { + _ = unix.Close(p.pidfd) + } + _, err = writer.Write(p.argv) + _ = writer.Close() + if err != nil { + t.Fatal(err) + } + if err := <-done; err == nil { + t.Fatal("process changed during observation accepted") + } + }) + } +} + +func TestMappedKernelParsersRejectIncompleteAndOverflowV1(t *testing.T) { + for _, s := range []string{"", "0 1", "0 1 1 extra", "-1 100000 1", "0 4294967296 1", "0 1 nope"} { + if _, err := mappedKernelRowsV1([]byte(s)); err == nil { + t.Fatalf("map %q accepted", s) + } + } + rows, err := mappedKernelRowsV1([]byte(" 0 100000 1\n200 100200 1\n")) + if err != nil || len(rows) != 2 || rows[1].ContainerStart != 200 { + t.Fatalf("complete rows: %+v %v", rows, err) + } + for _, s := range []string{"", "123 missing", "123 (x) Z", "123 (x) X", "123 (x) S"} { + if _, err := mappedProcessStartV1([]byte(s)); err == nil { + t.Fatalf("stat %q accepted", s) + } + } + fields := make([]string, 20) + for i := range fields { + fields[i] = "0" + } + fields[0] = "S" + for _, start := range []string{"0", "bad", "18446744073709551616"} { + fields[19] = start + if _, err := mappedProcessStartV1([]byte("1 (a b)) " + strings.Join(fields, " "))); err == nil { + t.Fatalf("start %q accepted", start) + } + } + fields[19] = "18446744073709551615" + if start, err := mappedProcessStartV1([]byte("1 (a b)) " + strings.Join(fields, " "))); err != nil || start != fields[19] { + t.Fatalf("start identity: %q %v", start, err) + } +} + +func TestMappedProcessPinRefusesUnqualifiedHostProcessV1(t *testing.T) { + active := mappedGeometryFixtureV1("exact") + for _, pid := range []int{0, -1, 2147483647, os.Getpid()} { + if p, err := pinMappedProcessV1(pid, active, []string{"/reploy-probe"}); err == nil { + p.close() + t.Fatalf("unqualified PID %d accepted", pid) + } + } + if _, err := pinMappedProcessV1(os.Getpid(), identitymapping.Geometry{}, nil); err == nil { + t.Fatal("invalid plan pinned") + } + // A real pidfd is polled above; fake /proc fixture identities never enter + // the production pin constructor or satisfy this host-process negative. +} + +func TestMappedKernelOwnedProcessHelperV1(t *testing.T) { + if os.Getenv("REPLOY_TEST_MAPPED_PROCESS") != "child" { + return + } + var release [1]byte + if _, err := io.ReadFull(os.Stdin, release[:]); err != nil { + os.Exit(2) + } + os.Exit(0) +} + +// A disposable namespace/chroot fixture checks the real pidfd, proc directory, +// maps, executable and namespace readers. Its self-map is not a product mapping +// or F proof; it neither uses an engine nor runs application code. +func TestMappedKernelOwnedProcessObservationV1(t *testing.T) { + if os.Getuid() == 0 || os.Getgid() == 0 { + t.Skip("requires non-root disposable namespace owner") + } + self, err := os.Executable() + if err != nil { + t.Fatal(err) + } + root := t.TempDir() + copyFile := func(source, target string) { + t.Helper() + data, err := os.ReadFile(source) + if err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Dir(target), 0700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(target, data, 0700); err != nil { + t.Fatal(err) + } + } + copyFile(self, filepath.Join(root, "reploy-probe")) + // A cgo-linked test executable needs its existing loader/libraries. Copy + // only absolute dependencies reported for this exact fixture executable. + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + output, lddErr := exec.CommandContext(ctx, "ldd", self).CombinedOutput() + if lddErr != nil && !bytes.Contains(output, []byte("not a dynamic executable")) && !bytes.Contains(output, []byte("statically linked")) { + t.Fatalf("inspect fixture dependencies: %v %s", lddErr, output) + } + for _, field := range strings.Fields(string(output)) { + if filepath.IsAbs(field) { + copyFile(field, filepath.Join(root, strings.TrimPrefix(field, "/"))) + } + } + cmd := exec.CommandContext(ctx, "/reploy-probe", "-test.run=^TestMappedKernelOwnedProcessHelperV1$") + cmd.Env = append(os.Environ(), "REPLOY_TEST_MAPPED_PROCESS=child") + cmd.SysProcAttr = &syscall.SysProcAttr{Chroot: root, Cloneflags: unix.CLONE_NEWUSER | unix.CLONE_NEWNS | unix.CLONE_NEWPID | unix.CLONE_NEWNET | unix.CLONE_NEWIPC, UidMappings: []syscall.SysProcIDMap{{ContainerID: 0, HostID: os.Getuid(), Size: 1}}, GidMappings: []syscall.SysProcIDMap{{ContainerID: 0, HostID: os.Getgid(), Size: 1}}, GidMappingsEnableSetgroups: false, Credential: &syscall.Credential{Uid: 0, Gid: 0}} + input, err := cmd.StdinPipe() + if err != nil { + t.Fatal(err) + } + defer input.Close() + var stderr bytes.Buffer + cmd.Stderr = &stderr + if err := cmd.Start(); err != nil { + if errors.Is(err, syscall.EPERM) || errors.Is(err, syscall.EINVAL) { + t.Skipf("private kernel-observation fixture unavailable; no qualification: %v", err) + } + t.Fatalf("start private kernel-observation fixture: %v %s", err, stderr.String()) + } + reaped := false + defer func() { + if !reaped { + _ = cmd.Process.Kill() + _ = cmd.Wait() + } + }() + active := identitymapping.Geometry{Profile: "exact", UIDs: []identitymapping.Interval{{HostStart: uint32(os.Getuid()), Count: 1}}, GIDs: []identitymapping.Interval{{HostStart: uint32(os.Getgid()), Count: 1}}} + p, err := pinMappedProcessV1(cmd.Process.Pid, active, cmd.Args) + if err != nil { + t.Fatalf("pin real isolated waiter: %v %s", err, stderr.String()) + } + defer p.close() + if err := p.fresh(); err != nil { + t.Fatal("refresh real isolated waiter", err) + } + if _, err := input.Write([]byte{1}); err != nil { + t.Fatal(err) + } + _ = input.Close() + err = cmd.Wait() + reaped = true + if err != nil { + t.Fatalf("retire fixture: %v %s", err, stderr.String()) + } + if err := p.fresh(); err == nil { + t.Fatal("retired process retained liveness") + } +} diff --git a/internal/dockerdeploy/mapped_process_other_test.go b/internal/dockerdeploy/mapped_process_other_test.go new file mode 100644 index 00000000..f1ce6fb3 --- /dev/null +++ b/internal/dockerdeploy/mapped_process_other_test.go @@ -0,0 +1,16 @@ +//go:build !linux + +package dockerdeploy + +import ( + "strings" + "testing" + + "github.com/omry/reploy/internal/identitymapping" +) + +func TestMappedProcessPinRequiresLinuxV1(t *testing.T) { + if p, err := pinMappedProcessV1(1, identitymapping.Geometry{}, nil); p != nil || err == nil || !strings.Contains(err.Error(), "Linux") { + t.Fatalf("unsupported process observer: %v %v", p, err) + } +} diff --git a/internal/dockerdeploy/podman_mapped_waiter.go b/internal/dockerdeploy/podman_mapped_waiter.go new file mode 100644 index 00000000..c65270ff --- /dev/null +++ b/internal/dockerdeploy/podman_mapped_waiter.go @@ -0,0 +1,279 @@ +package dockerdeploy + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "slices" + "strings" + "sync" + "time" + + "github.com/omry/reploy/internal/canonical" + "github.com/omry/reploy/internal/deploy" + "github.com/omry/reploy/internal/identitymapping" + "github.com/omry/reploy/internal/mappedlaunch" +) + +type mappedWaiterProcessV1 interface { + fresh() error + close() +} + +type mappedWaiterBackendV1 struct { + run commandRunner + pin func(int, identitymapping.Geometry, []string) (mappedWaiterProcessV1, error) + observe func(context.Context) (*podmanHealthcheckClient, string, error) + create func(*preparedMappedPodmanCreateV1, context.Context, commandRunner) (string, error) + admit func(deploy.IdentityMappingOperationsV1, context.Context, deploy.StateV1, string) (deploy.StateV1, error) +} + +// ownedMappedWaiterV1 owns an invocation-local attempt, not an admission token. +// Its caller retains the lock through release, then owns lifetime/cleanup (E3). +type ownedMappedWaiterV1 struct { + prepared *preparedMappedPodmanCreateV1 + admission context.Context + binding mappedlaunch.Binding + container string + pid int + process mappedWaiterProcessV1 + input *io.PipeWriter + reader *io.PipeReader + cancel context.CancelFunc + done chan error + backend mappedWaiterBackendV1 + mu sync.Mutex + consumed bool + closed bool +} + +// startOwnedMappedWaiterV1 is internal plumbing. Existing public guards remain; +// no production F verifier or E2 transition is supplied by this constructor. +func startOwnedMappedWaiterV1(ctx context.Context, p *preparedMappedPodmanCreateV1, binding mappedlaunch.Binding, options RunOptions) (*ownedMappedWaiterV1, error) { + return startMappedWaiterWithBackendV1(ctx, p, binding, options, mappedWaiterBackendV1{ + run: runCommand, pin: pinMappedProcessV1, observe: p.observe, + create: (*preparedMappedPodmanCreateV1).create, + admit: deploy.IdentityMappingOperationsV1.AdmitNativeExecution, + }) +} + +func startMappedWaiterWithBackendV1(ctx context.Context, p *preparedMappedPodmanCreateV1, b mappedlaunch.Binding, options RunOptions, backend mappedWaiterBackendV1) (*ownedMappedWaiterV1, error) { + if ctx == nil || options.Context == nil || p == nil || backend.run == nil || backend.pin == nil || backend.observe == nil || backend.create == nil || backend.admit == nil { + return nil, fmt.Errorf("mapped waiter requires its owned preparation and complete backend") + } + deadline, bounded := ctx.Deadline() + if !bounded || b.Validate() != nil || ctx.Err() != nil || options.Context.Err() != nil { + return nil, fmt.Errorf("mapped waiter requires a valid binding and admission deadline") + } + m := p.state.IdentityMapping + body, name, err := podmanApplicationCreateRequest(p.spec, p.namespace) + if err != nil || body.Tty || len(body.Cmd) == 0 || body.Cmd[0] != "sandbox-exec" { + return nil, fmt.Errorf("mapped waiter requires qualified non-TTY sandbox stdin transport") + } + if m == nil || m.PendingOperation == nil || m.PendingOperation.Native == nil || p.state.Current == nil || + b.Installation != m.InstallationID || b.Operation != p.operation || b.Generation != p.state.Current.BuildLockDigest || b.Revision != m.Revision || + m.PendingOperation.Native.Engine != p.endpoint || m.PendingOperation.Native.Process != name { + return nil, fmt.Errorf("mapped waiter differs from its exact native journal owner") + } + client, namespace, err := backend.observe(ctx) + if err != nil { + return nil, err + } + defer client.http.CloseIdleConnections() + image, err := p.verifyImage(ctx, client) + if err != nil || namespace != p.namespace || client.version != p.version || string(b.Image) != image { + return nil, fmt.Errorf("mapped waiter image or engine binding changed: %v", err) + } + encoded, _ := canonical.Marshal(b) // Binding.Validate permits canonical values only. + copy := *p + copy.spec.Args = slices.Clone(p.spec.Args) + entry := slices.Index(copy.spec.Args, "--entrypoint") + prefix := slices.Clone(copy.spec.Args[:entry]) + if !body.OpenStdin { + prefix = append(prefix, "--interactive") + } + for key, value := range map[string]string{"io.reploy.native-operation": b.Operation, "io.reploy.native-installation": b.Installation} { + if _, exists := body.Labels[key]; exists { + return nil, fmt.Errorf("mapped waiter owner label already supplied") + } + prefix = append(prefix, "--label", key+"="+value) + } + copy.spec.Args = append(prefix, copy.spec.Args[entry:entry+3]...) + command := append([]string{mappedlaunch.Command, string(encoded), deadline.UTC().Format(time.RFC3339Nano)}, body.Cmd...) + copy.spec.Args = append(copy.spec.Args, command...) + runContext, cancel := context.WithCancel(options.Context) + r, w := io.Pipe() + a := &ownedMappedWaiterV1{prepared: ©, admission: ctx, binding: b, input: w, reader: r, cancel: cancel, backend: backend, done: make(chan error, 1)} + a.container, err = backend.create(©, ctx, backend.run) + if err != nil { + cancel() + _ = r.Close() + _ = w.Close() + return nil, err // C2 retains uncertain create ownership; never retry. + } + start := copy.spec + start.Args = []string{"start", "--attach", "--interactive", a.container} + options.Context = runContext + if options.Stdin == nil { + options.Stdin = bytes.NewReader(nil) + } + options.Stdin = io.MultiReader(r, options.Stdin) + go func() { + err := backend.run(start, options) + _ = r.CloseWithError(err) + a.done <- err + }() + for { + pid, err := a.observeContainer(ctx, client) + if err != nil { + return nil, errors.Join(err, a.close()) + } + if pid > 0 { + a.pid = pid + a.process, err = backend.pin(pid, p.active, append([]string{"/reploy-probe"}, command...)) + if err != nil { + return nil, errors.Join(err, a.close()) + } + return a, nil + } + select { + case err := <-a.done: + return nil, errors.Join(fmt.Errorf("mapped waiter exited before observation: %v", err), a.close()) + case <-ctx.Done(): + return nil, errors.Join(ctx.Err(), a.close()) + case <-time.After(10 * time.Millisecond): + } + } +} + +func (a *ownedMappedWaiterV1) observeContainer(ctx context.Context, client *podmanHealthcheckClient) (int, error) { + data, err := client.request(ctx, http.MethodGet, "/v1.44/containers/"+a.container+"/json", nil) + var object struct { + ID, Image string + RestartCount *int + Config struct { + Labels map[string]string + Entrypoint, Cmd []string + Tty *bool + } + State struct { + Running bool + Pid int + } + HostConfig struct{ RestartPolicy *struct{ Name string } } + } + entry := slices.Index(a.prepared.spec.Args, "--entrypoint") + if err != nil || json.Unmarshal(data, &object) != nil || object.ID != a.container || strings.TrimPrefix(object.Image, "sha256:") != strings.TrimPrefix(string(a.binding.Image), "sha256:") || object.Config.Tty == nil || *object.Config.Tty || + object.Config.Labels["io.reploy.native-operation"] != a.binding.Operation || object.Config.Labels["io.reploy.native-installation"] != a.binding.Installation || + !slices.Equal(object.Config.Entrypoint, []string{"/reploy-probe"}) || !slices.Equal(object.Config.Cmd, a.prepared.spec.Args[entry+3:]) || + object.RestartCount == nil || *object.RestartCount != 0 || object.HostConfig.RestartPolicy == nil || object.HostConfig.RestartPolicy.Name != "" && object.HostConfig.RestartPolicy.Name != "no" { + return 0, fmt.Errorf("mapped waiter runtime identity or inhibited command changed") + } + if !object.State.Running { + return 0, nil + } + if object.State.Pid <= 0 { + return 0, fmt.Errorf("running mapped waiter has no exact host PID") + } + return object.State.Pid, nil +} + +type mappedWaiterEvidenceV1 struct { + deploy.IdentityMappingEvidenceV1 + verify func(context.Context) error +} + +func (e mappedWaiterEvidenceV1) Verify(ctx context.Context, state deploy.StateV1) error { + // Run the potentially slow storage/network check first, then refresh the + // actual inhibited process immediately before the existing atomic writer. + if err := e.IdentityMappingEvidenceV1.Verify(ctx, state); err != nil { + return err + } + // This is the existing trusted effect-owner/F proof seam, not a boolean + // qualification flag. No implementation is manufactured by E1.3. + return e.verify(ctx) +} + +func (a *ownedMappedWaiterV1) release() (acceptedState deploy.StateV1, failure error) { + ctx := a.admission + a.mu.Lock() + cleanupOnFailure := !a.consumed && !a.closed + defer func() { + a.mu.Unlock() + if failure != nil && cleanupOnFailure { + failure = errors.Join(failure, a.close()) + } + }() + if ctx == nil || a.consumed || a.closed || a.process == nil || a.prepared.ops.Evidence == nil { + return deploy.StateV1{}, fmt.Errorf("mapped release requires its unconsumed owned waiter and effect proof") + } + a.consumed = true // Even a failed/uncertain admission cannot resend on this attempt. + p := a.prepared + client, namespace, err := a.backend.observe(ctx) + if err != nil { + return deploy.StateV1{}, err + } + defer client.http.CloseIdleConnections() + image, err := p.verifyImage(ctx, client) + if err != nil || image != string(a.binding.Image) || namespace != p.namespace || client.version != p.version { + return deploy.StateV1{}, fmt.Errorf("mapped release image or engine changed: %v", err) + } + verify := func(ctx context.Context) error { + pid, err := a.observeContainer(ctx, client) + if err != nil || pid != a.pid { + return fmt.Errorf("mapped waiter is no longer the pinned process: %v", err) + } + return a.process.fresh() + } + ops := p.ops + ops.Evidence = mappedWaiterEvidenceV1{ops.Evidence, verify} + accepted, err := a.backend.admit(ops, ctx, p.state, p.operation) + if err != nil { + return deploy.StateV1{}, err + } + if err := ctx.Err(); err != nil { + return accepted, err + } + if err := ops.Operation.RequireHeld(); err != nil { + return accepted, err + } + if err := a.process.fresh(); err != nil { + return accepted, err + } + frame, err := mappedlaunch.Frame(mappedlaunch.Request{Binding: a.binding, Container: a.container}) + if err != nil { + return accepted, err + } + stop := context.AfterFunc(ctx, func() { _ = a.input.CloseWithError(ctx.Err()) }) + defer stop() + if n, err := a.input.Write(frame); err != nil || n != len(frame) { + return accepted, fmt.Errorf("mapped release is uncertain: %w", err) + } + // No control parser remains after this one frame. Later stdin is ordinary + // application data; copied frames have no control meaning. + return accepted, a.input.Close() +} + +func (a *ownedMappedWaiterV1) close() error { + a.mu.Lock() + defer a.mu.Unlock() + if a.closed { + return nil + } + a.closed = true + a.cancel() + _ = a.reader.Close() + _ = a.input.Close() + if a.process != nil { + a.process.close() + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + // Only this invocation's exact successful create ID is ever removed. Errors + // retain the native journal and reservations; absence is not recovery proof. + return removeCreatedPodmanContainer(ctx, a.prepared.spec, a.container, a.backend.run) +} diff --git a/internal/dockerdeploy/podman_mapped_waiter_test.go b/internal/dockerdeploy/podman_mapped_waiter_test.go new file mode 100644 index 00000000..8165b789 --- /dev/null +++ b/internal/dockerdeploy/podman_mapped_waiter_test.go @@ -0,0 +1,693 @@ +package dockerdeploy + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "os/exec" + "path/filepath" + "slices" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/omry/reploy/internal/deploy" + "github.com/omry/reploy/internal/identitymapping" + "github.com/omry/reploy/internal/mappedlaunch" + "github.com/omry/reploy/internal/providers" +) + +type waiterEffectFixtureV1 struct { + fail bool + calls int +} + +func (*waiterEffectFixtureV1) Plan(context.Context, deploy.StateV1, identitymapping.Geometry) ([]string, error) { + return nil, nil +} +func (*waiterEffectFixtureV1) Unused(context.Context, deploy.StateV1) error { + return errors.New("not an unused-intent proof") +} +func (e *waiterEffectFixtureV1) Verify(context.Context, deploy.StateV1) error { + e.calls++ + if e.fail { + return errors.New("F storage/network proof missing") + } + return nil +} + +type waiterProcessFixtureV1 struct { + fail bool + checks, closed int +} + +func (p *waiterProcessFixtureV1) fresh() error { + p.checks++ + if p.fail { + return errors.New("kernel process fence changed") + } + return nil +} +func (p *waiterProcessFixtureV1) close() { p.closed++ } + +// Only orchestration is injected. The test retains real image/selected-state +// verification and operation lock. E1.2 separately tests the actual atomic +// admission writer; these injected successes grant no runtime/F qualification. +type waiterFixtureV1 struct { + prepared *preparedMappedPodmanCreateV1 + binding mappedlaunch.Binding + backend mappedWaiterBackendV1 + process *waiterProcessFixtureV1 + effect *waiterEffectFixtureV1 + client *podmanHealthcheckClient + ctx context.Context + cancelAdmission context.CancelFunc + options RunOptions + created, removed, admissions int + mu sync.Mutex + body podmanApplicationCreate + id string + pid int + objectChange func(map[string]any) + inspectFailure, stopped, createFailure, cleanupFailure, pinFailure bool + observeFailure bool + observes int + application atomic.Int32 + expectedInput string + frame chan mappedlaunch.Request + startErr chan error +} + +type trackedWaiterInputV1 struct { + f *waiterFixtureV1 + io.Reader +} + +func (r trackedWaiterInputV1) Read(b []byte) (int, error) { + r.f.application.Add(1) + return r.Reader.Read(b) +} + +func newWaiterFixtureV1(t *testing.T) *waiterFixtureV1 { + t.Helper() + dir, operation, _, lock, state := currentBuildFixture(t, true) + t.Cleanup(func() { _ = operation.Unlock() }) + f := &waiterFixtureV1{id: strings.Repeat("b", 64), pid: 1234, expectedInput: "application-input\n", effect: &waiterEffectFixtureV1{}, process: &waiterProcessFixtureV1{}, frame: make(chan mappedlaunch.Request, 1), startErr: make(chan error, 1)} + f.client = newPodmanApplicationClient(t, func(w http.ResponseWriter, r *http.Request) { + f.mu.Lock() + defer f.mu.Unlock() + if strings.HasPrefix(r.URL.Path, "/v1.44/images/") { + _ = json.NewEncoder(w).Encode(mappedImageInspectionFixtureV1(string(lock.FinalImage.ConfigDigest))) + return + } + if f.inspectFailure { + w.WriteHeader(500) + return + } + object := map[string]any{"Id": f.id, "Image": f.binding.Image, "RestartCount": 0, "Config": map[string]any{"Labels": f.body.Labels, "Entrypoint": f.body.Entrypoint, "Cmd": f.body.Cmd, "Tty": f.body.Tty}, "State": map[string]any{"Running": !f.stopped, "Pid": f.pid}, "HostConfig": map[string]any{"RestartPolicy": map[string]any{"Name": "no"}}} + if f.objectChange != nil { + f.objectChange(object) + } + _ = json.NewEncoder(w).Encode(object) + }) + state.Staging = &deploy.StagingStateV1{Schema: deploy.StagingStateSchemaV1} + state.BlueprintSource = "private E1.3 orchestration fixture; not qualified runtime evidence" + spec := sameIDApplicationSpec(t) + spec.Args = slices.DeleteFunc(spec.Args, func(s string) bool { return s == "--tty" }) + entry := slices.Index(spec.Args, "--entrypoint") + spec.Args[entry+2] = string(lock.FinalImage.ConfigDigest) + name := spec.Args[slices.Index(spec.Args, "--name")+1] + empty := []deploy.IdentityMappingIntervalV1{} + geometry := deploy.IdentityMappingGeometryV1{Profile: "exact", UIDs: []deploy.IdentityMappingIntervalV1{{ContainerStart: "0", HostStart: "100000", Count: "1"}}, GIDs: []deploy.IdentityMappingIntervalV1{{ContainerStart: "0", HostStart: "100000", Count: "1"}}} + state.IdentityMapping = &deploy.IdentityMappingStateV1{Schema: deploy.IdentityMappingStateSchemaV2, ProcessModel: deploy.IdentityMappingNativeProcessModelV1, InstallationID: "installation", OwnerDir: dir, Revision: "1", OperatorUID: "1000", OperatorGID: "1000", Profile: "exact", AcceptedUIDs: &empty, AcceptedGIDs: &empty, RetainedStorage: []string{}, PendingOperation: &deploy.IdentityMappingOperationV1{OperationID: "operation", Kind: "native-execute", ExpectedRevision: "1", ExpectedGeneration: state.Current, Candidate: geometry, OwnedObjects: []string{}, Phase: "effects-pending", Inhibited: true, Native: &deploy.NativeIdentityOperationV1{Active: geometry, Engine: f.client.endpoint, Process: name, RuntimeUsers: []string{}, RuntimeGroups: []string{}}}} + encoded, err := deploy.EncodeStateV1(state) + if err != nil { + t.Fatal("native fixture", err) + } + if err := os.WriteFile(filepath.Join(filepath.Dir(operation.Path()), "state.json"), encoded, 0600); err != nil { + t.Fatal(err) + } + f.prepared = &preparedMappedPodmanCreateV1{ops: deploy.IdentityMappingOperationsV1{Operation: operation, Evidence: f.effect}, state: state, active: identitymapping.Geometry{Profile: "exact", UIDs: []identitymapping.Interval{{HostStart: 100000, Count: 1}}, GIDs: []identitymapping.Interval{{HostStart: 100000, Count: 1}}}, operation: "operation", spec: spec, endpoint: f.client.endpoint, version: f.client.version, namespace: "fixture-namespace"} + f.binding = mappedlaunch.Binding{Schema: "mapped-release-v1", Installation: "installation", Operation: "operation", Generation: state.Current.BuildLockDigest, Revision: "1", Image: lock.FinalImage.ConfigDigest, Role: "app", Plan: rendererDigest("c")} + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + t.Cleanup(cancel) + f.ctx = ctx + f.cancelAdmission = cancel + f.options = RunOptions{Context: t.Context(), Stdin: trackedWaiterInputV1{f, strings.NewReader("application-input\n")}} + f.backend = mappedWaiterBackendV1{ + observe: func(ctx context.Context) (*podmanHealthcheckClient, string, error) { + f.observes++ + if f.observeFailure { + return nil, "", errors.New("engine observation changed") + } + return f.client, f.prepared.namespace, ctx.Err() + }, + create: func(p *preparedMappedPodmanCreateV1, ctx context.Context, _ commandRunner) (string, error) { + f.created++ + if f.createFailure { + return "", errors.New("uncertain create; journal retained") + } + body, _, err := podmanApplicationCreateRequest(p.spec, p.namespace) + if err != nil { + return "", err + } + f.mu.Lock() + f.body = body + f.mu.Unlock() + return f.id, ctx.Err() + }, + pin: func(pid int, active identitymapping.Geometry, argv []string) (mappedWaiterProcessV1, error) { + if f.pinFailure { + return nil, errors.New("kernel observation unproved") + } + if pid != f.pid || !slices.Equal(argv[1:], f.body.Cmd) || argv[0] != "/reploy-probe" || !slices.Equal(active.UIDs, f.prepared.active.UIDs) { + return nil, errors.New("wrong trusted process plan") + } + return f.process, nil + }, + admit: func(ops deploy.IdentityMappingOperationsV1, ctx context.Context, s deploy.StateV1, id string) (deploy.StateV1, error) { + if id != "operation" || f.application.Load() != 0 { + return deploy.StateV1{}, errors.New("wrong operation or early application stdin") + } + if err := ops.Operation.RequireHeld(); err != nil { + return deploy.StateV1{}, err + } + if err := ops.Evidence.Verify(ctx, s); err != nil { + return deploy.StateV1{}, err + } + f.admissions++ + return s, nil + }, + run: func(spec CommandSpec, options RunOptions) error { + if slices.Equal(spec.Args, []string{"container", "rm", "--force", f.id}) { + f.removed++ + if options.Context.Err() != nil { + return errors.New("cleanup lost independent context") + } + if f.cleanupFailure { + return errors.New("owned cleanup failed") + } + return nil + } + if !slices.Equal(spec.Args, []string{"start", "--attach", "--interactive", f.id}) { + return fmt.Errorf("unexpected command %v", spec.Args) + } + request, err := mappedlaunch.Read(options.Stdin, f.binding) + if err == nil { + f.frame <- request + var data []byte + data, err = io.ReadAll(options.Stdin) + if err == nil && !bytes.Equal(data, []byte(f.expectedInput)) { + err = fmt.Errorf("application stdin changed: %q", data) + } + } + f.startErr <- err + return err + }, + } + return f +} + +func (f *waiterFixtureV1) start(t *testing.T) *ownedMappedWaiterV1 { + t.Helper() + a, err := startMappedWaiterWithBackendV1(f.ctx, f.prepared, f.binding, f.options, f.backend) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = a.close() }) + return a +} + +func TestOwnedMappedWaiterReleasesExactlyOnceAfterProofV1(t *testing.T) { + f := newWaiterFixtureV1(t) + a := f.start(t) + if f.application.Load() != 0 || f.admissions != 0 || f.removed != 0 { + t.Fatal("application or admission ran before release") + } + if _, err := a.release(); err != nil { + t.Fatal(err) + } + select { + case request := <-f.frame: + if request.Container != f.id || request.Binding != f.binding { + t.Fatalf("wrong frame %+v", request) + } + case <-f.ctx.Done(): + t.Fatal("release not received") + } + select { + case err := <-f.startErr: + if err != nil { + t.Fatal(err) + } + case <-f.ctx.Done(): + t.Fatal("application stdin not delivered") + } + if f.effect.calls != 1 || f.admissions != 1 || f.process.checks != 2 || f.application.Load() == 0 { + t.Fatal("release bypassed proof or stdin handoff") + } + if _, err := a.release(); err == nil { + t.Fatal("duplicate release permitted") + } + if f.removed != 0 { + t.Fatal("duplicate request cleaned a successfully released instance") + } + if err := a.close(); err != nil { + t.Fatal(err) + } + if err := a.close(); err != nil { + t.Fatal(err) + } + if f.created != 1 || f.removed != 1 || f.process.closed != 1 { + t.Fatal("owned resources not closed exactly once") + } +} + +func TestOwnedMappedWaiterHeldOperationLockEvidenceV1(t *testing.T) { + if os.Getenv("REPLOY_MAPPED_WAITER_HELD_LOCK_HELPER") == "1" { + for _, cancel := range []bool{false, true} { + t.Run(map[bool]string{false: "completion", true: "cancelled"}[cancel], func(t *testing.T) { + f := newWaiterFixtureV1(t) + a := f.start(t) + _, _, _, heldLock, _ := newPreparedAPTGraphReuseFixture(t) + var callbackSeen atomic.Int32 + a.backend.admit = func(ops deploy.IdentityMappingOperationsV1, ctx context.Context, state deploy.StateV1, id string) (deploy.StateV1, error) { + if id != "operation" || state.Current == nil { + return deploy.StateV1{}, errors.New("held-lock fixture received the wrong admission") + } + validate := func(providers.RequirementProfile) error { + callbackSeen.Add(1) + if cancel { + f.cancelAdmission() + } + return ops.Evidence.Verify(ctx, state) + } + if _, err := ops.Operation.PublishBuildLock(heldLock, validate); err != nil { + return deploy.StateV1{}, fmt.Errorf("held-lock build validation: %w", err) + } + return state, nil + } + released := make(chan error, 1) + go func() { _, err := a.release(); released <- err }() + select { + case err := <-released: + if cancel { + if err == nil { + t.Fatal("cancelled held-lock admission succeeded") + } + if f.removed != 1 || f.process.closed != 1 || f.application.Load() != 0 { + t.Fatalf("cancelled admission was not failure-closed: err=%v removed=%d closed=%d application=%d", err, f.removed, f.process.closed, f.application.Load()) + } + select { + case <-f.frame: + t.Fatal("cancelled held-lock admission delivered a frame") + default: + } + } else if err != nil { + t.Fatal(err) + } + case <-time.After(2 * time.Second): + t.Fatal("held-lock evidence callback did not complete boundedly") + } + if callbackSeen.Load() == 0 { + t.Fatal("held-lock build validator did not invoke evidence") + } + if !cancel { + select { + case <-f.frame: + case <-time.After(2 * time.Second): + t.Fatal("successful held-lock admission did not send a frame") + } + } + }) + } + return + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + command := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestOwnedMappedWaiterHeldOperationLockEvidenceV1$", "-test.v") + command.Env = append(os.Environ(), "REPLOY_MAPPED_WAITER_HELD_LOCK_HELPER=1") + output, err := command.CombinedOutput() + if ctx.Err() != nil { + t.Fatalf("held-lock regression exceeded its bound: %v\n%s", ctx.Err(), output) + } + if err != nil { + t.Fatalf("held-lock regression helper failed: %v\n%s", err, output) + } +} + +func TestOwnedMappedWaiterKeepsOriginalAdmissionContextV1(t *testing.T) { + f := newWaiterFixtureV1(t) + deadline, _ := f.ctx.Deadline() + original := f.backend.admit + f.backend.admit = func(ops deploy.IdentityMappingOperationsV1, ctx context.Context, state deploy.StateV1, id string) (deploy.StateV1, error) { + actual, bounded := ctx.Deadline() + if !bounded || !actual.Equal(deadline) { + return deploy.StateV1{}, errors.New("original admission deadline lost") + } + return original(ops, ctx, state, id) + } + a := f.start(t) + if _, err := a.release(); err != nil { + t.Fatal(err) + } +} + +func TestOwnedMappedWaiterCancelsBlockedFrameWithOriginalContextV1(t *testing.T) { + f := newWaiterFixtureV1(t) + partial := make(chan struct{}) + original := f.backend.run + f.backend.run = func(spec CommandSpec, options RunOptions) error { + if spec.Args[0] != "start" { + return original(spec, options) + } + if _, err := io.ReadFull(options.Stdin, make([]byte, 7)); err != nil { + return err + } + close(partial) + <-options.Context.Done() + return options.Context.Err() + } + a := f.start(t) + released := make(chan error, 1) + go func() { _, err := a.release(); released <- err }() + select { + case <-partial: + case <-f.ctx.Done(): + t.Fatal("attach never consumed the partial release frame") + } + // The application lifetime context remains live. Only cancelling the + // original admission context can unblock this incomplete frame write. + f.cancelAdmission() + select { + case err := <-released: + if err == nil || !strings.Contains(err.Error(), "uncertain") { + t.Fatalf("partial release cancellation: %v", err) + } + case <-time.After(3 * time.Second): + _ = a.reader.Close() + <-released + t.Fatal("original cancellation did not unblock the release frame") + } + if f.admissions != 1 || f.removed != 1 || f.application.Load() != 0 { + t.Fatal("partial cancelled release lost ownership or sent application stdin") + } +} + +func TestOwnedMappedWaiterRejectsBeforeCreateV1(t *testing.T) { + for _, mode := range []string{"nil context", "nil preparation", "nil lifetime", "no deadline", "bad binding", "TTY", "not sandbox", "installation", "operation", "generation", "revision", "image", "engine owner", "process owner", "owner label", "missing native", "observe", "engine version", "namespace", "backend", "create"} { + t.Run(mode, func(t *testing.T) { + f := newWaiterFixtureV1(t) + ctx := f.ctx + p := f.prepared + b := f.binding + options := f.options + backend := f.backend + switch mode { + case "nil context": + ctx = nil + case "nil preparation": + p = nil + case "nil lifetime": + options.Context = nil + case "no deadline": + ctx = t.Context() + case "bad binding": + b.Schema = "wrong" + case "TTY": + entry := slices.Index(p.spec.Args, "--entrypoint") + p.spec.Args = append(append(slices.Clone(p.spec.Args[:entry]), "--tty"), p.spec.Args[entry:]...) + case "not sandbox": + entry := slices.Index(p.spec.Args, "--entrypoint") + p.spec.Args[entry+3] = "verify-exec" + case "installation": + b.Installation = "other" + case "operation": + b.Operation = "other" + case "generation": + b.Generation = rendererDigest("d") + case "revision": + b.Revision = "2" + case "image": + b.Image = rendererDigest("d") + case "engine owner": + p.state.IdentityMapping.PendingOperation.Native.Engine = "other" + case "process owner": + p.state.IdentityMapping.PendingOperation.Native.Process = "other" + case "missing native": + p.state.IdentityMapping.PendingOperation.Native = nil + case "owner label": + entry := slices.Index(p.spec.Args, "--entrypoint") + p.spec.Args = append(append(slices.Clone(p.spec.Args[:entry]), "--label", "io.reploy.native-operation=operation"), p.spec.Args[entry:]...) + case "observe": + f.observeFailure = true + case "engine version": + p.version = "other" + case "namespace": + p.namespace = "other" + backend.observe = func(context.Context) (*podmanHealthcheckClient, string, error) { + return f.client, "fixture-namespace", nil + } + case "backend": + backend.pin = nil + case "create": + f.createFailure = true + } + a, err := startMappedWaiterWithBackendV1(ctx, p, b, options, backend) + if err == nil { + _ = a.close() + t.Fatal("unproved waiter created") + } + if mode != "create" && f.created != 0 { + t.Fatal("rejected plan reached create") + } + if f.admissions != 0 || f.application.Load() != 0 { + t.Fatal("rejected plan released payload") + } + }) + } +} + +func TestOwnedMappedWaiterReleaseFailuresRetainJournalAndCleanupV1(t *testing.T) { + for _, mode := range []string{"F missing", "no effect owner", "process changed", "runtime PID changed", "runtime stopped", "image changed", "state changed", "engine unavailable", "admission failed", "released lock", "cancelled", "disconnect", "disconnect after header", "cleanup failed", "cancel after admission", "unlock after admission", "process changes after admission"} { + t.Run(mode, func(t *testing.T) { + f := newWaiterFixtureV1(t) + if mode == "disconnect after header" { + original := f.backend.run + f.backend.run = func(s CommandSpec, o RunOptions) error { + if s.Args[0] == "start" { + _, err := io.ReadFull(o.Stdin, make([]byte, 7)) + if err != nil { + return err + } + return errors.New("attach disconnected after partial frame") + } + return original(s, o) + } + } + a := f.start(t) + switch mode { + case "F missing", "cleanup failed": + f.effect.fail = true + f.cleanupFailure = mode == "cleanup failed" + case "no effect owner": + a.prepared.ops.Evidence = nil + case "process changed": + f.process.fail = true + case "runtime PID changed": + f.mu.Lock() + f.pid++ + f.mu.Unlock() + case "runtime stopped": + f.mu.Lock() + f.stopped = true + f.mu.Unlock() + case "image changed": + a.binding.Image = rendererDigest("d") + case "state changed": + a.prepared.state.BlueprintSource = "stale" + case "engine unavailable": + f.observeFailure = true + case "admission failed": + a.backend.admit = func(deploy.IdentityMappingOperationsV1, context.Context, deploy.StateV1, string) (deploy.StateV1, error) { + return deploy.StateV1{}, errors.New("uncertain atomic write") + } + case "released lock": + _ = f.prepared.ops.Operation.Unlock() + case "cancelled": + f.cancelAdmission() + case "disconnect": + _ = a.reader.Close() + case "cancel after admission", "unlock after admission", "process changes after admission": + original := a.backend.admit + cancel := f.cancelAdmission + a.backend.admit = func(ops deploy.IdentityMappingOperationsV1, ctx context.Context, state deploy.StateV1, id string) (deploy.StateV1, error) { + result, err := original(ops, ctx, state, id) + if err != nil { + return result, err + } + switch mode { + case "cancel after admission": + cancel() + case "unlock after admission": + _ = ops.Operation.Unlock() + case "process changes after admission": + f.process.fail = true + } + return result, nil + } + } + if _, err := a.release(); err == nil { + t.Fatal("failed fence or uncertain release succeeded") + } + if f.removed != 1 || f.process.closed != 1 { + t.Fatal("failed release did not clean exact owned resources") + } + if f.application.Load() != 0 { + t.Fatal("failed release sent application input") + } + select { + case <-f.frame: + t.Fatal("failed release delivered frame") + default: + } + if _, err := a.release(); err == nil { + t.Fatal("failed attempt retried") + } + }) + } +} + +func TestOwnedMappedWaiterRejectsMissingAndChangedRuntimeEvidenceV1(t *testing.T) { + for _, mode := range []string{"wrong ID", "wrong image", "missing TTY", "TTY", "wrong label", "wrong install", "wrong entrypoint", "wrong command", "restarted", "missing restart count", "restart policy", "missing restart policy", "missing PID", "inspect failure", "pin failure"} { + t.Run(mode, func(t *testing.T) { + f := newWaiterFixtureV1(t) + f.inspectFailure = mode == "inspect failure" + f.pinFailure = mode == "pin failure" + f.objectChange = func(o map[string]any) { + config := o["Config"].(map[string]any) + host := o["HostConfig"].(map[string]any) + switch mode { + case "wrong ID": + o["Id"] = strings.Repeat("c", 64) + case "wrong image": + o["Image"] = rendererDigest("d") + case "missing TTY": + delete(config, "Tty") + case "TTY": + config["Tty"] = true + case "wrong label": + config["Labels"] = map[string]string{"io.reploy.native-operation": "foreign"} + case "wrong install": + config["Labels"] = map[string]string{"io.reploy.native-operation": "operation", "io.reploy.native-installation": "foreign"} + case "wrong entrypoint": + config["Entrypoint"] = []string{"/bin/sh"} + case "wrong command": + config["Cmd"] = []string{"sandbox-exec"} + case "restarted": + o["RestartCount"] = 1 + case "missing restart count": + delete(o, "RestartCount") + case "restart policy": + host["RestartPolicy"] = map[string]string{"Name": "always"} + case "missing restart policy": + delete(host, "RestartPolicy") + case "missing PID": + o["State"].(map[string]any)["Pid"] = 0 + } + } + a, err := startMappedWaiterWithBackendV1(f.ctx, f.prepared, f.binding, f.options, f.backend) + if err == nil { + _ = a.close() + t.Fatal("unproved runtime observation accepted") + } + if f.created != 1 || f.removed != 1 || f.admissions != 0 || f.application.Load() != 0 { + t.Fatal("observation failure lost owned cleanup or released payload") + } + }) + } +} + +func TestOwnedMappedWaiterWaitsBoundedlyAndSupportsConstructedStdinV1(t *testing.T) { + for _, mode := range []string{"no original stdin", "no original interactive", "unprefixed image", "not running", "start exited", "admission deadline distinct from lifetime"} { + t.Run(mode, func(t *testing.T) { + f := newWaiterFixtureV1(t) + switch mode { + case "no original stdin": + f.options.Stdin = nil + f.expectedInput = "" + case "no original interactive": + f.prepared.spec.Args = slices.DeleteFunc(f.prepared.spec.Args, func(s string) bool { return s == "--interactive" }) + case "unprefixed image": + f.objectChange = func(o map[string]any) { o["Image"] = strings.TrimPrefix(string(f.binding.Image), "sha256:") } + case "not running": + f.stopped = true + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + f.ctx = ctx + case "start exited": + original := f.backend.run + f.stopped = true + f.backend.run = func(s CommandSpec, o RunOptions) error { + if s.Args[0] == "start" { + return errors.New("attach disconnected") + } + return original(s, o) + } + case "admission deadline distinct from lifetime": + original := f.backend.run + f.backend.run = func(s CommandSpec, o RunOptions) error { + if s.Args[0] == "start" { + if _, bounded := o.Context.Deadline(); bounded { + return errors.New("admission deadline leaked into app lifetime") + } + } + return original(s, o) + } + } + a, err := startMappedWaiterWithBackendV1(f.ctx, f.prepared, f.binding, f.options, f.backend) + if mode == "not running" || mode == "start exited" { + if err == nil { + _ = a.close() + t.Fatal("wait failure accepted") + } + if f.removed != 1 { + t.Fatal("wait failure lost cleanup") + } + return + } + if err != nil { + t.Fatal(err) + } + defer a.close() + if _, err := a.release(); err != nil { + t.Fatal(err) + } + select { + case <-f.frame: + case <-f.ctx.Done(): + t.Fatal("missing frame") + } + { + select { + case err := <-f.startErr: + if err != nil { + t.Fatal(err) + } + case <-f.ctx.Done(): + t.Fatal("stream did not complete") + } + } + }) + } + if _, err := startOwnedMappedWaiterV1(context.Background(), nil, mappedlaunch.Binding{}, RunOptions{}); err == nil { + t.Fatal("production constructor accepted missing ownership") + } +} diff --git a/internal/dockerdeploy/podman_mapping_peer_other.go b/internal/dockerdeploy/podman_mapping_peer_other.go index 13b533fc..7472fbc0 100644 --- a/internal/dockerdeploy/podman_mapping_peer_other.go +++ b/internal/dockerdeploy/podman_mapping_peer_other.go @@ -6,8 +6,14 @@ import ( "context" "fmt" "net" + + "github.com/omry/reploy/internal/identitymapping" ) func dialMappedPodmanV1(context.Context, string) (net.Conn, error) { return nil, fmt.Errorf("mapped Podman creation requires local Linux") } + +func pinMappedProcessV1(int, identitymapping.Geometry, []string) (mappedWaiterProcessV1, error) { + return nil, fmt.Errorf("mapped waiter observations require local Linux") +} diff --git a/internal/mappedlaunch/release.go b/internal/mappedlaunch/release.go new file mode 100644 index 00000000..26f4cc5d --- /dev/null +++ b/internal/mappedlaunch/release.go @@ -0,0 +1,136 @@ +// Package mappedlaunch defines the single stdin handoff between the trusted +// host admission owner and a fresh, inhibited startup process. +package mappedlaunch + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "fmt" + "io" + "strconv" + "strings" + "unicode/utf8" + + "github.com/omry/reploy/internal/canonical" +) + +const MaxRequest = 64 << 10 +const Command = "mapped-wait-v1" + +// Binding is fixed in trusted create arguments before the engine assigns an +// ID. The host binds that dynamic ID through its exact owned observation; the +// waiter does not discover identity from application output or engine access. +type Binding struct { + Schema string `json:"schema"` + Installation string `json:"installation"` + Operation string `json:"operation"` + Generation canonical.Digest `json:"generation"` + Revision string `json:"revision"` + Image canonical.Digest `json:"image"` + Role string `json:"role"` + Plan canonical.Digest `json:"plan"` +} + +type Request struct { + Binding Binding `json:"binding"` + Container string `json:"container"` +} + +func (b Binding) Validate() error { + n, err := strconv.ParseUint(b.Revision, 10, 64) + if b.Schema != "mapped-release-v1" || err != nil || n == 0 || strconv.FormatUint(n, 10) != b.Revision { + return fmt.Errorf("invalid mapped release schema or revision") + } + for _, s := range []string{b.Installation, b.Operation, b.Role} { + if s == "" || len(s) > 1024 || !utf8.ValidString(s) || strings.ContainsAny(s, "\x00\r\n") { + return fmt.Errorf("invalid mapped release identity") + } + } + for _, d := range []canonical.Digest{b.Generation, b.Image, b.Plan} { + if err := d.Validate(); err != nil { + return err + } + } + return nil +} + +// DecodeBinding accepts only the canonical closed shape, including unique +// keys. Re-encoding also rejects duplicate keys that encoding/json accepts. +func DecodeBinding(data []byte) (Binding, error) { + var b Binding + if err := decode(data, &b); err != nil { + return b, err + } + return b, b.Validate() +} + +func decode(data []byte, value any) error { + if len(data) == 0 || len(data) > MaxRequest || !utf8.Valid(data) { + return fmt.Errorf("invalid mapped release JSON length or UTF-8") + } + d := json.NewDecoder(bytes.NewReader(data)) + d.DisallowUnknownFields() + if err := d.Decode(value); err != nil { + return fmt.Errorf("invalid mapped release JSON") + } + encoded, err := canonical.Marshal(value) + if err != nil || !bytes.Equal(data, encoded) { + return fmt.Errorf("mapped release JSON is not canonical and closed") + } + return nil +} + +func (r Request) validate() error { + if err := r.Binding.Validate(); err != nil { + return err + } + // Container IDs use the engine's full lowercase 256-bit identity. + _, err := canonical.ParseDigest("sha256:" + r.Container) + return err +} + +func Frame(r Request) ([]byte, error) { + if err := r.validate(); err != nil { + return nil, err + } + data, err := canonical.Marshal(r) + if err != nil || len(data) > MaxRequest { + return nil, fmt.Errorf("mapped release exceeds the bounded request") + } + frame := make([]byte, 4+len(data)) + binary.BigEndian.PutUint32(frame, uint32(len(data))) + copy(frame[4:], data) + return frame, nil +} + +// Read consumes exactly one frame. Its reader must be the unbuffered stdin +// descriptor: bytes beyond the declared body belong exclusively to the app. +func Read(input io.Reader, expected Binding) (Request, error) { + var r Request + if err := expected.Validate(); err != nil { + return r, err + } + var header [4]byte + if _, err := io.ReadFull(input, header[:]); err != nil { + return r, fmt.Errorf("read mapped release header: %w", err) + } + n := binary.BigEndian.Uint32(header[:]) + if n == 0 || n > MaxRequest { + return r, fmt.Errorf("invalid mapped release length") + } + data := make([]byte, n) + if _, err := io.ReadFull(input, data); err != nil { + return r, fmt.Errorf("read mapped release body: %w", err) + } + if err := decode(data, &r); err != nil { + return r, err + } + if err := r.validate(); err != nil { + return r, err + } + if r.Binding != expected { + return r, fmt.Errorf("mapped release differs from trusted launch arguments") + } + return r, nil +} diff --git a/internal/mappedlaunch/release_test.go b/internal/mappedlaunch/release_test.go new file mode 100644 index 00000000..29bbb0b5 --- /dev/null +++ b/internal/mappedlaunch/release_test.go @@ -0,0 +1,174 @@ +package mappedlaunch + +import ( + "bytes" + "encoding/binary" + "io" + "strings" + "testing" + + "github.com/omry/reploy/internal/canonical" +) + +func bindingFixture() Binding { + d := canonical.Digest("sha256:" + strings.Repeat("a", 64)) + return Binding{"mapped-release-v1", "installation", "operation", d, "12", d, "app", d} +} + +func rawFrame(data []byte) []byte { + frame := make([]byte, 4) + binary.BigEndian.PutUint32(frame, uint32(len(data))) + return append(frame, data...) +} + +type fragmentedReader struct{ io.Reader } + +func (r fragmentedReader) Read(b []byte) (int, error) { + return r.Reader.Read(b[:min(len(b), 1)]) +} + +func TestReleaseFramePreservesEveryApplicationByte(t *testing.T) { + b := bindingFixture() + r := Request{b, strings.Repeat("b", 64)} + frame, err := Frame(r) + if err != nil { + t.Fatal(err) + } + // A duplicate frame after the accepted frame is application input, with no + // remaining parser. Include NULs and non-UTF8 bytes in the app canary. + canary := append([]byte{0, 255, '\n'}, frame...) + for _, fragmented := range []bool{false, true} { + input := bytes.NewReader(append(slicesClone(frame), canary...)) + var reader io.Reader = input + if fragmented { + reader = fragmentedReader{input} + } + got, err := Read(reader, b) + if err != nil || got != r { + t.Fatalf("read: %+v %v", got, err) + } + remaining, err := io.ReadAll(reader) + if err != nil || !bytes.Equal(remaining, canary) { + t.Fatalf("stdin consumed: %x %v", remaining, err) + } + } + for n := range len(frame) { + if _, err := Read(bytes.NewReader(frame[:n]), b); err == nil { + t.Fatalf("partial frame length %d accepted", n) + } + } +} + +func slicesClone(b []byte) []byte { return append([]byte(nil), b...) } + +func TestReleaseRejectsClosedShapeAndBindingAttacks(t *testing.T) { + b := bindingFixture() + valid, _ := canonical.Marshal(Request{b, strings.Repeat("b", 64)}) + for name, data := range map[string][]byte{ + "duplicate outer": bytes.Replace(valid, []byte(`"container":`), []byte(`"container":"`+strings.Repeat("b", 64)+`","container":`), 1), + "duplicate nested": bytes.Replace(valid, []byte(`"role":"app"`), []byte(`"role":"app","role":"app"`), 1), + "unknown": append([]byte(`{"unknown":true,`), valid[1:]...), + "unknown nested": bytes.Replace(valid, []byte(`"binding":{`), []byte(`"binding":{"unknown":"x",`), 1), + "trailer": append(slicesClone(valid), []byte(`{}`)...), + "space": append(slicesClone(valid), ' '), + "null": []byte(`null`), + "array": []byte(`[]`), + "empty": nil, + "invalid JSON": []byte(`{`), + "invalid UTF8": {255}, + "bad container": bytes.Replace(valid, []byte(strings.Repeat("b", 64)), []byte(strings.Repeat("B", 64)), 1), + "escaped alias": bytes.Replace(valid, []byte(`"app"`), []byte(`"\u0061pp"`), 1), + } { + t.Run(name, func(t *testing.T) { + if _, err := Read(bytes.NewReader(rawFrame(data)), b); err == nil { + t.Fatal("malformed request accepted") + } + }) + } + for _, n := range []uint32{0, MaxRequest + 1, ^uint32(0)} { + var header [4]byte + binary.BigEndian.PutUint32(header[:], n) + if _, err := Read(bytes.NewReader(header[:]), b); err == nil { + t.Fatalf("length %d accepted", n) + } + } + for _, field := range []string{"schema", "installation", "operation", "generation", "revision", "image", "role", "plan"} { + t.Run("mismatch "+field, func(t *testing.T) { + other := b + switch field { + case "schema": + other.Schema = "v2" + case "installation": + other.Installation = "other" + case "operation": + other.Operation = "other" + case "generation": + other.Generation = canonical.Digest("sha256:" + strings.Repeat("c", 64)) + case "revision": + other.Revision = "13" + case "image": + other.Image = canonical.Digest("sha256:" + strings.Repeat("c", 64)) + case "role": + other.Role = "other" + case "plan": + other.Plan = canonical.Digest("sha256:" + strings.Repeat("c", 64)) + } + if _, err := Read(bytes.NewReader(rawFrame(valid)), other); err == nil { + t.Fatal("binding mismatch accepted") + } + }) + } +} + +func TestBindingValidationAndLaunchArguments(t *testing.T) { + valid := bindingFixture() + encoded, _ := canonical.Marshal(valid) + if got, err := DecodeBinding(encoded); err != nil || got != valid { + t.Fatalf("binding roundtrip: %+v %v", got, err) + } + for _, data := range [][]byte{nil, []byte(`{}`), []byte(`null`), []byte(strings.Repeat("x", MaxRequest+1)), append(slicesClone(encoded), ' ')} { + if _, err := DecodeBinding(data); err == nil { + t.Fatal("invalid binding accepted") + } + } + for _, revision := range []string{"", "0", "01", "-1", "18446744073709551616"} { + b := valid + b.Revision = revision + if b.Validate() == nil { + t.Fatalf("revision %q accepted", revision) + } + } + for _, value := range []string{"", strings.Repeat("x", 1025), "a\x00b", "a\nb", "a\rb", string([]byte{255})} { + for field := range 3 { + b := valid + switch field { + case 0: + b.Installation = value + case 1: + b.Operation = value + case 2: + b.Role = value + } + if b.Validate() == nil { + t.Fatalf("identity %q accepted", value) + } + } + } + for field := range 3 { + b := valid + switch field { + case 0: + b.Generation = "bad" + case 1: + b.Image = "bad" + case 2: + b.Plan = "bad" + } + if _, err := Frame(Request{b, strings.Repeat("b", 64)}); err == nil { + t.Fatal("bad digest framed") + } + } + if _, err := Frame(Request{valid, "short"}); err == nil { + t.Fatal("short ID framed") + } +} diff --git a/internal/probe/main.go b/internal/probe/main.go index 90d62586..f2166000 100644 --- a/internal/probe/main.go +++ b/internal/probe/main.go @@ -8,9 +8,18 @@ import ( "syscall" "github.com/omry/reploy/internal/canonical" + "github.com/omry/reploy/internal/mappedlaunch" ) func Main(args []string, stdin io.Reader, stdout io.Writer, stderr io.Writer) int { + if len(args) > 0 && args[0] == mappedlaunch.Command { + input, _ := stdin.(*os.File) + if err := mappedWaitV1(args[1:], input); err != nil { + _, _ = fmt.Fprintf(stderr, "reploy-probe: mapped startup: %v\n", err) + return 1 + } + return 0 + } if len(args) == 1 && args[0] == "application-launch-proof-v1" { if err := waitForApplicationLaunchV1(); err != nil { _, _ = fmt.Fprintf(stderr, "reploy-probe: application startup: %v\n", err) diff --git a/internal/probe/mapped_wait.go b/internal/probe/mapped_wait.go new file mode 100644 index 00000000..411f0bf2 --- /dev/null +++ b/internal/probe/mapped_wait.go @@ -0,0 +1,112 @@ +//go:build linux + +package probe + +import ( + "context" + "fmt" + "io" + "os" + "runtime" + "time" + + "github.com/omry/reploy/internal/mappedlaunch" + "golang.org/x/sys/unix" +) + +// Poll the inherited descriptor directly: os.Stdin need not be registered with +// Go's deadline poller. Do not buffer or alter flags on application stdin. +func waitMappedReleaseV1(ctx context.Context, input *os.File, binding mappedlaunch.Binding) error { + deadline, ok := ctx.Deadline() + if !ok || input == nil { + return fmt.Errorf("mapped waiter requires bounded descriptor input") + } + raw, err := input.SyscallConn() + if err != nil { + return err + } + var fd int + if err := raw.Control(func(value uintptr) { fd = int(value) }); err != nil { + return err + } + var status unix.Stat_t + if err := unix.Fstat(fd, &status); err != nil || status.Mode&unix.S_IFMT != unix.S_IFIFO && status.Mode&unix.S_IFMT != unix.S_IFSOCK { + return fmt.Errorf("mapped waiter requires non-TTY pipe or socket stdin") + } + _, err = mappedlaunch.Read(mappedWaitInputV1{ctx, fd}, binding) + runtime.KeepAlive(input) + if err == nil && !time.Now().Before(deadline) { + err = context.DeadlineExceeded + } + if err == nil { + err = ctx.Err() + } + return err +} + +type mappedWaitInputV1 struct { + ctx context.Context + fd int +} + +func (r mappedWaitInputV1) Read(data []byte) (int, error) { + deadline, _ := r.ctx.Deadline() + for { + if err := r.ctx.Err(); err != nil { + return 0, err + } + remaining := time.Until(deadline) + if remaining <= 0 { + return 0, context.DeadlineExceeded + } + poll := []unix.PollFd{{Fd: int32(r.fd), Events: unix.POLLIN}} + n, err := unix.Poll(poll, int((min(remaining, 100*time.Millisecond)+time.Millisecond-1)/time.Millisecond)) + if err == unix.EINTR { + continue + } + if err != nil { + return 0, err + } + if n == 0 { + continue + } + if poll[0].Revents&(unix.POLLERR|unix.POLLNVAL) != 0 { + return 0, fmt.Errorf("mapped stdin disconnected") + } + // The trusted waiter is the sole reader before handoff. Read only the + // requested header/body bytes; no app callback can consume readiness. + n, err = unix.Read(r.fd, data) + if err == unix.EINTR || err == unix.EAGAIN { + continue + } + if n == 0 && err == nil { + err = io.EOF + } + return n, err + } +} + +func mappedWaitV1(args []string, input *os.File) error { + if len(args) < 3 || args[2] != "sandbox-exec" { + return fmt.Errorf("mapped waiter requires fixed binding and admission deadline") + } + if _, err := parseSandboxExecPlanV1(args[3:], true); err != nil { + return err + } + binding, err := mappedlaunch.DecodeBinding([]byte(args[0])) + if err != nil { + return err + } + deadline, err := time.Parse(time.RFC3339Nano, args[1]) + if err != nil { + return fmt.Errorf("invalid mapped admission deadline") + } + ctx, cancel := context.WithDeadline(context.Background(), deadline) + defer cancel() + if err := waitMappedReleaseV1(ctx, input, binding); err != nil { + return err + } + // E2 owns the qualified credential/descriptor/direct-exec tail. An accepted + // frame alone cannot substitute for that proof or execute an ordinary path. + return fmt.Errorf("mapped startup requires a qualified final exec transition") +} diff --git a/internal/probe/mapped_wait_other_test.go b/internal/probe/mapped_wait_other_test.go new file mode 100644 index 00000000..6e25e54c --- /dev/null +++ b/internal/probe/mapped_wait_other_test.go @@ -0,0 +1,18 @@ +//go:build !linux + +package probe + +import ( + "bytes" + "strings" + "testing" + + "github.com/omry/reploy/internal/mappedlaunch" +) + +func TestMappedWaitRequiresLinuxV1(t *testing.T) { + var output bytes.Buffer + if status := Main([]string{mappedlaunch.Command}, nil, &output, &output); status != 1 || !strings.Contains(output.String(), "Linux") { + t.Fatalf("unsupported waiter: %d %s", status, output.String()) + } +} diff --git a/internal/probe/mapped_wait_test.go b/internal/probe/mapped_wait_test.go new file mode 100644 index 00000000..8e87082f --- /dev/null +++ b/internal/probe/mapped_wait_test.go @@ -0,0 +1,230 @@ +//go:build linux + +package probe + +import ( + "bytes" + "context" + "errors" + "io" + "os" + "os/exec" + "strings" + "testing" + "time" + + "github.com/omry/reploy/internal/canonical" + "github.com/omry/reploy/internal/mappedlaunch" + "golang.org/x/sys/unix" +) + +func mappedWaitBindingV1() mappedlaunch.Binding { + d := canonical.Digest("sha256:" + strings.Repeat("a", 64)) + return mappedlaunch.Binding{Schema: "mapped-release-v1", Installation: "installation", Operation: "operation", Generation: d, Revision: "1", Image: d, Role: "app", Plan: d} +} + +func mappedWaitPipeV1(t *testing.T) (*os.File, *os.File) { + t.Helper() + r, w, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = r.Close(); _ = w.Close() }) + return r, w +} + +func TestMappedWaitHandoffLeavesStdinAndClearsDeadlineV1(t *testing.T) { + r, w := mappedWaitPipeV1(t) + binding := mappedWaitBindingV1() + frame, err := mappedlaunch.Frame(mappedlaunch.Request{Binding: binding, Container: strings.Repeat("b", 64)}) + if err != nil { + t.Fatal(err) + } + canary := []byte("application\x00stdin\n") + go func() { _, _ = w.Write(append(frame, canary...)); _ = w.Close() }() + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + if err := waitMappedReleaseV1(ctx, r, binding); err != nil { + t.Fatal(err) + } + got, err := io.ReadAll(r) + if err != nil || !bytes.Equal(got, canary) { + t.Fatalf("stdin: %q %v", got, err) + } +} + +func TestMappedWaitAbortsBeforeContinuationV1(t *testing.T) { + for _, mode := range []string{"unbounded", "nil", "closed", "cancelled", "EOF", "partial", "malformed", "deadline", "cancel-during-read", "regular file"} { + t.Run(mode, func(t *testing.T) { + r, w := mappedWaitPipeV1(t) + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + input := r + switch mode { + case "unbounded": + ctx = t.Context() + case "nil": + input = nil + case "closed": + _ = input.Close() + case "cancelled": + cancel() + case "EOF": + _ = w.Close() + case "partial": + go func() { _, _ = w.Write([]byte{0, 0}); _ = w.Close() }() + case "malformed": + go func() { _, _ = w.Write([]byte{0, 0, 0, 0}) }() + case "deadline": + ctx, cancel = context.WithTimeout(t.Context(), 10*time.Millisecond) + defer cancel() + case "cancel-during-read": + go func() { _, _ = w.Write([]byte{0}); cancel() }() + case "regular file": + f, err := os.CreateTemp(t.TempDir(), "input") + if err != nil { + t.Fatal(err) + } + defer f.Close() + input = f + } + if err := waitMappedReleaseV1(ctx, input, mappedWaitBindingV1()); err == nil { + t.Fatal("failed input released waiter") + } + }) + } +} + +func TestMappedWaitRawDescriptorErrorsAndSocketInputV1(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_CLOEXEC|unix.SOCK_NONBLOCK, 0) + if err != nil { + t.Fatal(err) + } + input := os.NewFile(uintptr(fds[0]), "mapped-stdin-socket") + defer input.Close() + defer unix.Close(fds[1]) + b := mappedWaitBindingV1() + flags, err := unix.FcntlInt(uintptr(fds[0]), unix.F_GETFL, 0) + if err != nil { + t.Fatal(err) + } + frame, _ := mappedlaunch.Frame(mappedlaunch.Request{Binding: b, Container: strings.Repeat("b", 64)}) + go func() { _, _ = unix.Write(fds[1], frame) }() + if err := waitMappedReleaseV1(ctx, input, b); err != nil { + t.Fatal("socket stdin handoff", err) + } + after, err := unix.FcntlInt(uintptr(fds[0]), unix.F_GETFL, 0) + if err != nil || after != flags { + t.Fatal("handoff changed application stdin flags", after, flags, err) + } + fd, err := unix.Dup(fds[0]) + if err != nil { + t.Fatal(err) + } + _ = unix.Close(fd) + if _, err := (mappedWaitInputV1{ctx, fd}).Read(make([]byte, 1)); err == nil { + t.Fatal("closed raw stdin descriptor accepted") + } +} + +func TestMappedWaitFarFutureDeadlineStillObservesCancellationV1(t *testing.T) { + r, w := mappedWaitPipeV1(t) + ctx, cancel := context.WithDeadline(t.Context(), time.Now().AddDate(500, 0, 0)) + defer cancel() + done := make(chan error, 1) + go func() { done <- waitMappedReleaseV1(ctx, r, mappedWaitBindingV1()) }() + time.AfterFunc(10*time.Millisecond, cancel) + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("far-future cancellation: %v", err) + } + case <-time.After(2 * time.Second): + _ = w.Close() + <-done + t.Fatal("deadline arithmetic overflow blocked cancellation") + } +} + +func TestMappedProbeModeRetainsE2QualificationRefusalV1(t *testing.T) { + b := mappedWaitBindingV1() + encoded, _ := canonical.Marshal(b) + deadline := time.Now().Add(time.Second).UTC().Format(time.RFC3339Nano) + args := []string{string(encoded), deadline, "sandbox-exec", "--uid", "200", "--gid", "200", "--public", "deny", "--local", "deny", "--ambiguous", "require-both", "--", "/bin/true"} + for _, mode := range []string{"valid frame", "missing args", "bad binding", "bad deadline", "bad sandbox", "wrong command", "non-descriptor stdin"} { + t.Run(mode, func(t *testing.T) { + r, w := mappedWaitPipeV1(t) + a := append([]string(nil), args...) + var input io.Reader = r + switch mode { + case "missing args": + a = nil + case "bad binding": + a[0] = "{}" + case "bad deadline": + a[1] = "bad" + case "bad sandbox": + a[3] = "--unknown" + case "wrong command": + a[2] = "verify-exec" + case "non-descriptor stdin": + input = bytes.NewReader(nil) + } + if mode == "valid frame" { + frame, _ := mappedlaunch.Frame(mappedlaunch.Request{Binding: b, Container: strings.Repeat("b", 64)}) + go func() { _, _ = w.Write(frame); _ = w.Close() }() + } + var output bytes.Buffer + if status := Main(append([]string{mappedlaunch.Command}, a...), input, &output, &output); status != 1 { + t.Fatalf("unqualified mode status %d", status) + } + if mode == "valid frame" && !strings.Contains(output.String(), "qualified final exec transition") { + t.Fatalf("unexpected refusal: %s", output.String()) + } + }) + } +} + +func TestMappedProbeInheritedStdinV1(t *testing.T) { + if os.Getenv("REPLOY_TEST_MAPPED_STDIN") == "child" { + os.Exit(Main(os.Args[3:], os.Stdin, os.Stdout, os.Stderr)) + } + b := mappedWaitBindingV1() + encoded, _ := canonical.Marshal(b) + frame, _ := mappedlaunch.Frame(mappedlaunch.Request{Binding: b, Container: strings.Repeat("b", 64)}) + for _, mode := range []string{"valid", "partial", "deadline"} { + t.Run(mode, func(t *testing.T) { + deadline := time.Now().Add(5 * time.Second) + if mode == "deadline" { + deadline = time.Now().Add(50 * time.Millisecond) + } + args := []string{"-test.run=^TestMappedProbeInheritedStdinV1$", "--", mappedlaunch.Command, string(encoded), deadline.UTC().Format(time.RFC3339Nano), "sandbox-exec", "--uid", "200", "--gid", "200", "--public", "deny", "--local", "deny", "--ambiguous", "require-both", "--", "/bin/true"} + cmd := exec.CommandContext(t.Context(), os.Args[0], args...) + cmd.Env = append(os.Environ(), "REPLOY_TEST_MAPPED_STDIN=child") + var output bytes.Buffer + cmd.Stdout = &output + cmd.Stderr = &output + var hold *os.File + switch mode { + case "valid": + cmd.Stdin = bytes.NewReader(frame) + case "partial": + cmd.Stdin = bytes.NewReader(frame[:7]) + case "deadline": + r, w := mappedWaitPipeV1(t) + cmd.Stdin = r + hold = w + } + err := cmd.Run() + if hold != nil { + _ = hold.Close() + } + want := map[string]string{"valid": "qualified final exec transition", "partial": "body", "deadline": "deadline exceeded"}[mode] + if err == nil || !strings.Contains(output.String(), want) { + t.Fatalf("inherited stdin %s: %v %s", mode, err, output.String()) + } + }) + } +} diff --git a/internal/probe/sandbox_exec_other.go b/internal/probe/sandbox_exec_other.go index 5643acec..14ecfe78 100644 --- a/internal/probe/sandbox_exec_other.go +++ b/internal/probe/sandbox_exec_other.go @@ -2,8 +2,15 @@ package probe -import "fmt" +import ( + "fmt" + "os" +) func sandboxAndExecApplicationV1(sandboxExecPlanV1) error { return fmt.Errorf("application sandbox setup is supported only in Linux containers") } + +func mappedWaitV1([]string, *os.File) error { + return fmt.Errorf("mapped waiter requires a qualified local Linux runtime") +}