| 1 | //go:build linux && rootrunner |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "context" |
| 8 | "database/sql" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "fmt" |
| 12 | "os" |
| 13 | "os/exec" |
| 14 | "os/signal" |
| 15 | "path/filepath" |
| 16 | "reflect" |
| 17 | "strconv" |
| 18 | "strings" |
| 19 | "syscall" |
| 20 | "testing" |
| 21 | "time" |
| 22 | |
| 23 | "github.com/carlosframework/platform/internal/hostbox" |
| 24 | "github.com/carlosframework/platform/internal/hostclaim" |
| 25 | "github.com/carlosframework/platform/internal/hostguard" |
| 26 | "github.com/carlosframework/platform/internal/instancectl" |
| 27 | ) |
| 28 | |
| 29 | func TestRootServingParkStoppedWriterHandover(t *testing.T) { |
| 30 | runServingParkAuthenticatedPeer(t, true, true) |
| 31 | } |
| 32 | |
| 33 | // Only the fixture unit starts this helper. Keep SQLite open until SIGTERM; |
| 34 | // the root stop verifier must independently observe descriptor loss. |
| 35 | func TestRootServingParkWriterPayload(t *testing.T) { |
| 36 | path, readyPath := os.Getenv("SERVING_WRITER_DB"), os.Getenv("SERVING_WRITER_READY") |
| 37 | if path == "" || readyPath == "" { |
| 38 | t.Fatal("invoke through TestRootServingParkStoppedWriterHandover") |
| 39 | } |
| 40 | stop := make(chan os.Signal, 1) |
| 41 | signal.Notify(stop, syscall.SIGTERM) |
| 42 | defer signal.Stop(stop) |
| 43 | db, err := sql.Open("sqlite", path) |
| 44 | if err != nil { |
| 45 | t.Fatal(err) |
| 46 | } |
| 47 | defer db.Close() |
| 48 | conn, err := db.Conn(t.Context()) |
| 49 | if err != nil { |
| 50 | t.Fatal(err) |
| 51 | } |
| 52 | defer conn.Close() |
| 53 | for _, query := range []string{"PRAGMA journal_mode=WAL", "CREATE TABLE serving_witness(value TEXT)", "INSERT INTO serving_witness VALUES ('tenant committed bytes')"} { |
| 54 | if _, err := conn.ExecContext(t.Context(), query); err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | } |
| 58 | start, err := claimConstructionProcessStart() |
| 59 | if err != nil { |
| 60 | t.Fatal(err) |
| 61 | } |
| 62 | body, err := json.Marshal(StopPeer{PID: os.Getpid(), ProcessStart: start}) |
| 63 | if err != nil { |
| 64 | t.Fatal(err) |
| 65 | } |
| 66 | if err := os.WriteFile(readyPath, body, 0600); err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | select { |
| 70 | case <-stop: |
| 71 | case <-t.Context().Done(): |
| 72 | t.Fatal(t.Context().Err()) |
| 73 | } |
| 74 | } |
| 75 | |
| 76 | func runServingPeerStoppedWriter(t *testing.T, f *servingRootFixture, binary string, launch hostbox.LaunchInventory, peer StopPeer, command func(servingPeerCommand, bool) servingPeerReply, kernelResults <-chan error) { |
| 77 | t.Helper() |
| 78 | ctx, cancel := context.WithTimeout(t.Context(), 45*time.Second) |
| 79 | defer cancel() |
| 80 | unitName := instancectl.UnitName(instancectl.RoleRun, f.current.Host) |
| 81 | slices := []string{instancectl.SliceName(f.current.Host), "carlos-instances.slice", "carlos.slice"} |
| 82 | probe, probeErr := exec.CommandContext(ctx, "systemctl", "show", unitName, "-p", "LoadState", "--value").Output() |
| 83 | if strings.TrimSpace(string(probe)) != "not-found" { |
| 84 | t.Fatalf("refusing existing/unreadable writer unit: %q %v", probe, probeErr) |
| 85 | } |
| 86 | for _, name := range slices { |
| 87 | out, err := exec.CommandContext(ctx, "systemctl", "show", name, "-p", "ActiveState", "--value").Output() |
| 88 | if strings.TrimSpace(string(out)) != "inactive" { |
| 89 | t.Fatalf("refusing existing/unreadable fixture unit %s: %q %v", name, out, err) |
| 90 | } |
| 91 | } |
| 92 | executable, err := stopIdentity(binary) |
| 93 | if err != nil { |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | reservation := hostbox.LaunchReservation{ID: "writer:main", Role: "main", InvocationID: "writer", Inputs: "sqlite-witness", Target: launch.Target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(f.current.Host)}, Executable: hostbox.LaunchFile(executable)} |
| 97 | // Fixture unit creation and launch journal writes are explicit seams. The |
| 98 | // identity observations, stop verifier and stop transport are production. |
| 99 | if err := f.f.journal.IntendUnit(ctx, reservation); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | readyPath := filepath.Join(filepath.Dir(binary), "writer-ready") |
| 103 | content := fmt.Sprintf("[Unit]\nDescription=Serving park SQLite witness\n[Service]\nType=simple\nUser=root\nSlice=%s\nExecStart=%s -test.run=^TestRootServingParkWriterPayload$ -test.v -test.timeout=60s\nEnvironment=SERVING_WRITER_DB=%s SERVING_WRITER_READY=%s\nIPAddressDeny=any\nTimeoutStopSec=10s\n", slices[0], binary, f.db, readyPath) |
| 104 | unitPath := filepath.Join("/run/systemd/system", unitName) |
| 105 | unit, err := os.OpenFile(unitPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0644) |
| 106 | if err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | t.Cleanup(func() { |
| 110 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 111 | defer cancel() |
| 112 | for _, name := range append([]string{unitName}, slices...) { |
| 113 | if out, err := exec.CommandContext(ctx, "systemctl", "stop", name).CombinedOutput(); err != nil { |
| 114 | t.Errorf("stop fixture %s: %v %s", name, err, out) |
| 115 | } |
| 116 | } |
| 117 | if err := os.Remove(unitPath); err != nil { |
| 118 | t.Error(err) |
| 119 | } |
| 120 | if out, err := exec.CommandContext(ctx, "systemctl", "daemon-reload").CombinedOutput(); err != nil { |
| 121 | t.Errorf("reload fixture: %v %s", err, out) |
| 122 | } |
| 123 | }) |
| 124 | _, writeErr := unit.WriteString(content) |
| 125 | if err := errors.Join(writeErr, unit.Close()); err != nil { |
| 126 | t.Fatal(err) |
| 127 | } |
| 128 | servingRunnerSystemctl(t, "daemon-reload") |
| 129 | servingRunnerSystemctl(t, "start", unitName) |
| 130 | var process StopPeer |
| 131 | for { |
| 132 | body, err := os.ReadFile(readyPath) |
| 133 | if err == nil && json.Unmarshal(body, &process) == nil { |
| 134 | break |
| 135 | } |
| 136 | select { |
| 137 | case <-ctx.Done(): |
| 138 | t.Fatal("writer readiness", ctx.Err()) |
| 139 | case <-time.After(20 * time.Millisecond): |
| 140 | } |
| 141 | } |
| 142 | properties := func(name string) string { |
| 143 | t.Helper() |
| 144 | out, err := exec.CommandContext(ctx, "systemctl", "show", unitName, "-p", name, "--value").Output() |
| 145 | if err != nil { |
| 146 | t.Fatal(err) |
| 147 | } |
| 148 | return strings.TrimSpace(string(out)) |
| 149 | } |
| 150 | pid, err := strconv.Atoi(properties("MainPID")) |
| 151 | if err != nil || pid != process.PID || pid <= 0 { |
| 152 | t.Fatal("writer MainPID", pid, process.PID, err) |
| 153 | } |
| 154 | start, err := claimLaunchProcessStart("/proc", pid) |
| 155 | if err != nil || process.ProcessStart != start { |
| 156 | t.Fatalf("writer identity: %+v %v", process, err) |
| 157 | } |
| 158 | group := reservation.Group |
| 159 | group.InvocationID = properties("InvocationID") |
| 160 | if properties("ControlGroup") != "/"+group.RelativePath || group.InvocationID == "" { |
| 161 | t.Fatal("writer unit identity") |
| 162 | } |
| 163 | groupID, err := stopIdentity(filepath.Join("/sys/fs/cgroup", group.RelativePath)) |
| 164 | if err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | group.Device, group.Inode = groupID.Device, groupID.Inode |
| 168 | if err := f.f.journal.StartedUnit(ctx, reservation, group, pid, start); err != nil { |
| 169 | t.Fatal(err) |
| 170 | } |
| 171 | id, err := stopIdentity(f.db) |
| 172 | if err != nil { |
| 173 | t.Fatal(err) |
| 174 | } |
| 175 | if err := f.f.journal.RecordFile(ctx, launch.Target, hostbox.LaunchFile(id)); err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | if err := os.WriteFile(f.db+".gen", []byte(f.current.Generation), 0600); err != nil { |
| 179 | t.Fatal(err) |
| 180 | } |
| 181 | current := f.f.read() |
| 182 | req := StopRequest{Region: f.actor.Region, Host: current.Host, Incarnation: f.current.Incarnation, InstallationStamp: f.current.Stamp, WriterSession: f.current.Session, Box: f.actor.Box, Enrollment: f.actor.Enrollment, BootID: peer.BootID, ControllerID: peer.ControllerID, ExpectedClaimETag: current.ETag, ExpectedDirectoryRevision: "7", RosterRevision: 7} |
| 183 | stop := servingPeerCommand{Action: "stop", Stop: req} |
| 184 | checkNoReceipt := func() { |
| 185 | t.Helper() |
| 186 | receipts, err := f.f.guard.CompletedServingStops(ctx, current.Host, hostguard.Captured{Incarnation: f.current.Incarnation, Stamp: f.current.Stamp}, f.current.Session, f.actor.Box) |
| 187 | if !errors.Is(err, hostguard.ErrNoFenceReceipt) || len(receipts) != 0 { |
| 188 | t.Fatal("root trusted a drain acknowledgement while a writer remained") |
| 189 | } |
| 190 | } |
| 191 | command(stop, true) |
| 192 | select { |
| 193 | case err := <-kernelResults: |
| 194 | if err == nil { |
| 195 | t.Fatal("live writer escaped kernel verification") |
| 196 | } |
| 197 | default: |
| 198 | t.Fatal("stop refused before reaching the production kernel verifier") |
| 199 | } |
| 200 | checkNoReceipt() |
| 201 | servingRunnerSystemctl(t, "stop", unitName) |
| 202 | if properties("MainPID") != "0" || properties("ActiveState") != "inactive" || properties("Result") != "success" { |
| 203 | t.Fatal("writer unit did not stop cleanly") |
| 204 | } |
| 205 | if _, err := claimLaunchProcessStart("/proc", pid); !errors.Is(err, os.ErrNotExist) { |
| 206 | t.Fatalf("writer process still observable: %v", err) |
| 207 | } |
| 208 | if err := f.f.journal.SettleUnit(ctx, launch.Target, reservation.ID); err != nil { |
| 209 | t.Fatal(err) |
| 210 | } |
| 211 | // Even an empty group and settled journal cannot hide an independent FD. |
| 212 | openDB, err := os.Open(f.db) |
| 213 | if err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | defer openDB.Close() |
| 217 | command(stop, true) |
| 218 | select { |
| 219 | case err := <-kernelResults: |
| 220 | if err == nil || !strings.Contains(err.Error(), "descriptor remains") { |
| 221 | t.Fatalf("independent FD refusal: %v", err) |
| 222 | } |
| 223 | default: |
| 224 | t.Fatal("FD refusal did not reach the production kernel verifier") |
| 225 | } |
| 226 | checkNoReceipt() |
| 227 | if err := openDB.Close(); err != nil { |
| 228 | t.Fatal(err) |
| 229 | } |
| 230 | stopped := command(stop, false).Stopped |
| 231 | select { |
| 232 | case err := <-kernelResults: |
| 233 | if err != nil { |
| 234 | t.Fatal("stopped writer kernel verification", err) |
| 235 | } |
| 236 | default: |
| 237 | t.Fatal("stop succeeded without production kernel verification") |
| 238 | } |
| 239 | if err := verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: stopped.Target, Bytes: stopped.Bytes, EvidenceAt: stopped.EvidenceAt}); err != nil { |
| 240 | t.Fatal("independent stop receipt", err) |
| 241 | } |
| 242 | if replay := command(stop, false).Stopped; !reflect.DeepEqual(replay, stopped) { |
| 243 | t.Fatal("stop replay changed receipt") |
| 244 | } |
| 245 | before, err := os.ReadFile(f.db) |
| 246 | if err != nil { |
| 247 | t.Fatal(err) |
| 248 | } |
| 249 | previous := f.f.read() |
| 250 | h := previous.Claim.Holder |
| 251 | admitted, err := hostclaim.Update(ctx, f.f.claims, previous, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: f.actor.Box, Region: f.actor.Region, PredecessorRegion: f.actor.Region, Mode: "cold", NewGeneration: "G3"})) |
| 252 | if err != nil { |
| 253 | t.Fatal(err) |
| 254 | } |
| 255 | prior := f.current |
| 256 | f.current, err = servingWriterIdentity(admitted, f.actor.Box, f.actor.Region) |
| 257 | if err != nil { |
| 258 | t.Fatal(err) |
| 259 | } |
| 260 | command(servingPeerCommand{Action: "handover", Previous: previous, Admitted: admitted, Actor: f.actor, Target: stopped.Target}, false) |
| 261 | command(servingPeerCommand{Action: "prelaunch", Target: recoveryTuple(f.f.read(), claimStopLocal{Actor: f.actor})}, false) |
| 262 | q := servingPeerCommand{Action: "park", Current: f.current, Prior: prior} |
| 263 | parked := command(q, false).Authorization |
| 264 | f.tags(f.current.Stamp) |
| 265 | command(servingPeerCommand{Action: "complete", Current: f.current, ID: parked.ID}, false) |
| 266 | record, err := f.p.ledger.read(parked.ID) |
| 267 | if err != nil || !record.Complete || record.Disposition != servingParkRetain || len(record.Receipts) != 1 || !bytes.Equal(record.Receipts[0].Bytes, stopped.Bytes) { |
| 268 | t.Fatalf("park did not use independent stop: %+v %v", record, err) |
| 269 | } |
| 270 | parkPath := filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", parked.ID, filepath.Base(f.db)) |
| 271 | after, err := os.ReadFile(parkPath) |
| 272 | if err != nil || !bytes.Equal(before, after) { |
| 273 | t.Fatal("park changed committed SQLite bytes", err) |
| 274 | } |
| 275 | if _, err := os.Lstat(f.db); !errors.Is(err, os.ErrNotExist) { |
| 276 | t.Fatal("park left live database", err) |
| 277 | } |
| 278 | if replay := command(q, false).Authorization; replay != parked { |
| 279 | t.Fatal("stopped-writer park replay changed authority") |
| 280 | } |
| 281 | for name, want := range map[string]string{"files/upload": "tenant upload", "env": "tenant configuration"} { |
| 282 | body, err := os.ReadFile(filepath.Join(filepath.Dir(f.db), name)) |
| 283 | if err != nil || string(body) != want { |
| 284 | t.Fatal("writer retirement changed preserved state", name, err) |
| 285 | } |
| 286 | } |
| 287 | if err := f.p.ledger.fs.CheckParked(&record); err != nil { |
| 288 | t.Fatal(err) |
| 289 | } |
| 290 | servingRestoreAfterStoppedWriter(t, f, servingPeerRetirement{command: command}) |
| 291 | check, err := f.p.ledger.read(parked.ID) |
| 292 | if err != nil || !reflect.DeepEqual(record, check) { |
| 293 | t.Fatal("stopped-writer park replay changed record", err) |
| 294 | } |
| 295 | } |
| 296 | |
| 297 | // Adapt the fixture's control connection to the production client in the edge |
| 298 | // process. Root never receives a retirement request from the test controller. |
| 299 | type servingPeerRetirement struct { |
| 300 | command func(servingPeerCommand, bool) servingPeerReply |
| 301 | } |
| 302 | |
| 303 | func (p servingPeerRetirement) Park(ctx context.Context, current, prior servingArtifactIdentity) (servingParkAuthorization, error) { |
| 304 | if err := ctx.Err(); err != nil { |
| 305 | return servingParkAuthorization{}, err |
| 306 | } |
| 307 | return p.command(servingPeerCommand{Action: "park", Current: current, Prior: prior}, false).Authorization, nil |
| 308 | } |
| 309 | |
| 310 | func (p servingPeerRetirement) Complete(ctx context.Context, id string, current servingArtifactIdentity) error { |
| 311 | if err := ctx.Err(); err != nil { |
| 312 | return err |
| 313 | } |
| 314 | p.command(servingPeerCommand{Action: "complete", Current: current, ID: id}, false) |
| 315 | return nil |
| 316 | } |
| 317 | |
| 318 | func (servingPeerRetirement) ParkLegacy(context.Context, servingArtifactIdentity, servingLegacyPredecessor) (servingParkAuthorization, error) { |
| 319 | return servingParkAuthorization{}, hostbox.ErrUnsupported |
| 320 | } |
| 321 | |
amadan