fix(workloads): world executors run as root so game-image worlds are readable

A live backup drill on test-one failed: 'tar walk: open
/world/world/level.dat: permission denied'. The world volume belongs to
the game image's own UID (root for every Paper image we ship), and Paper
saves level.dat mode 0600 — a fixed uid-1000 executor can neither read
it (backup/reaper archive) nor overwrite it (restore). The same identity
silently broke on-demand backups, restores, and the reaper for every
server that had saved once.

Run the backup Job, restore Job, file Job, and the reaper pod as root
with DAC_OVERRIDE on top of drop-ALL — the same owner-matching precedent
as the operator's forwarding-init container; DAC_OVERRIDE extends it to
game images whose UID is neither root nor ours. FSGroup is omitted when
zero so a root executor never chgrps the world volume. Shape tests
updated for the new identity.

internal/restore/jobspec.go

@@ -125,7 +125,14 @@ func RestoreJob(p JobParams) (*batchv1.Job, error) {
 			Privileged:               boolPtr(false),
 			AllowPrivilegeEscalation: boolPtr(false),
 			ReadOnlyRootFilesystem:   boolPtr(true),
-			Capabilities:             &corev1.Capabilities{Drop: []corev1.Capability{"ALL"}},
+			// Root + DAC_OVERRIDE (see restore.Config.RunAsUser): the world is
+			// owned by the game image's UID and Paper's files are mode 0600, so
+			// the restore must bypass file modes to overwrite what the server
+			// wrote — otherwise level.dat is un-restorable.
+			Capabilities: &corev1.Capabilities{
+				Drop: []corev1.Capability{"ALL"},
+				Add:  []corev1.Capability{"DAC_OVERRIDE"},
+			},
 		},
 	}
 
@@ -148,13 +155,8 @@ func RestoreJob(p JobParams) (*batchv1.Job, error) {
 					RestartPolicy:                corev1.RestartPolicyNever,
 					ServiceAccountName:           p.ServiceAccount,
 					AutomountServiceAccountToken: boolPtr(false),
-					SecurityContext: &corev1.PodSecurityContext{
-						RunAsNonRoot: boolPtr(true),
-						RunAsUser:    int64Ptr(p.RunAsUser),
-						RunAsGroup:   int64Ptr(p.RunAsGroup),
-						FSGroup:      int64Ptr(p.FSGroup),
-					},
-					Containers: []corev1.Container{container},
+					SecurityContext:              restorePodSecurityContext(p),
+					Containers:                   []corev1.Container{container},
 					Volumes: []corev1.Volume{
 						{
 							Name: worldVolume,
@@ -221,6 +223,22 @@ func resourceLimits(cpu, mem string) (corev1.ResourceList, error) {
 	}, nil
 }
 
-func boolPtr(b bool) *bool    { return &b }
+func boolPtr(b bool) *bool { return &b }
+
+// restorePodSecurityContext pins the Pod identity. Root by default — the world
+// volume is owned by the game image's UID and Paper writes mode-0600 files, so a
+// fixed non-root executor could neither read nor replace them. FSGroup is only
+// rendered when configured: a root executor must not chgrp the world volume.
+func restorePodSecurityContext(p JobParams) *corev1.PodSecurityContext {
+	sc := &corev1.PodSecurityContext{
+		RunAsNonRoot: boolPtr(false),
+		RunAsUser:    int64Ptr(p.RunAsUser),
+		RunAsGroup:   int64Ptr(p.RunAsGroup),
+	}
+	if p.FSGroup > 0 {
+		sc.FSGroup = int64Ptr(p.FSGroup)
+	}
+	return sc
+}
 func int32Ptr(i int32) *int32 { return &i }
 func int64Ptr(i int64) *int64 { return &i }

internal/restore/jobspec_test.go

@@ -23,9 +23,9 @@ func sampleJobParams() JobParams {
 		Deadline:         30 * time.Minute,
 		CPULimit:         "1",
 		MemLimit:         "1Gi",
-		RunAsUser:        1000,
-		RunAsGroup:       1000,
-		FSGroup:          1000,
+		RunAsUser:        0,
+		RunAsGroup:       0,
+		FSGroup:          0,
 		TTLAfterFinished: 10 * time.Minute,
 	}
 }
@@ -161,19 +161,24 @@ func TestRestoreJobIsBoundedOneShotAndSelfCleaning(t *testing.T) {
 	}
 }
 
-// The container must be non-root, non-privileged, escalation-proof, read-only
-// root, drop ALL caps, and carry resource limits.
+// The Pod runs as root (the world volume belongs to the game image's UID — see
+// restore.Config.RunAsUser), and the container stays non-privileged,
+// escalation-proof, read-only root, ALL caps dropped except DAC_OVERRIDE, with
+// resource limits.
 func TestRestoreJobContainerIsHardened(t *testing.T) {
 	job, err := RestoreJob(sampleJobParams())
 	if err != nil {
 		t.Fatalf("RestoreJob: %v", err)
 	}
 	pod := job.Spec.Template.Spec
-	if pod.SecurityContext == nil || pod.SecurityContext.RunAsNonRoot == nil || !*pod.SecurityContext.RunAsNonRoot {
-		t.Error("pod must set runAsNonRoot=true")
+	if pod.SecurityContext == nil || pod.SecurityContext.RunAsNonRoot == nil || *pod.SecurityContext.RunAsNonRoot {
+		t.Error("pod must NOT require non-root: root is the owner-matching default for game-image worlds")
 	}
-	if pod.SecurityContext == nil || pod.SecurityContext.FSGroup == nil || *pod.SecurityContext.FSGroup != 1000 {
-		t.Error("pod must set an fsGroup so restored files are group-owned by the server identity")
+	if pod.SecurityContext == nil || pod.SecurityContext.RunAsUser == nil || *pod.SecurityContext.RunAsUser != 0 {
+		t.Error("pod must run as uid 0 by default")
+	}
+	if pod.SecurityContext == nil || pod.SecurityContext.FSGroup != nil {
+		t.Error("fsGroup must stay unset when zero (a root executor must not chgrp the world volume)")
 	}
 	c := singleContainer(t, job)
 	sc := c.SecurityContext
@@ -192,6 +197,9 @@ func TestRestoreJobContainerIsHardened(t *testing.T) {
 	if sc.Capabilities == nil || len(sc.Capabilities.Drop) == 0 || string(sc.Capabilities.Drop[0]) != "ALL" {
 		t.Errorf("container must drop ALL capabilities, got %v", sc.Capabilities)
 	}
+	if len(sc.Capabilities.Add) != 1 || sc.Capabilities.Add[0] != "DAC_OVERRIDE" {
+		t.Errorf("container must add exactly DAC_OVERRIDE, got %v", sc.Capabilities.Add)
+	}
 	if c.Resources.Limits.Cpu().IsZero() || c.Resources.Limits.Memory().IsZero() {
 		t.Error("container must carry CPU+memory limits")
 	}

internal/restore/restore.go

@@ -86,11 +86,13 @@ type Config struct {
 	// CPULimit / MemLimit cap the restore container.
 	CPULimit string
 	MemLimit string
-	// RunAsUser / RunAsGroup / FSGroup are the Pod's runtime identity. FSGroup in
-	// particular MUST match the operator StatefulSet's runtime group so the files
-	// the restore Pod writes are readable by the minecraft server that later
-	// mounts the same world PVC. The default matches the conventional minecraft
-	// container uid; a deployment that runs minecraft as another id overrides it.
+	// RunAsUser / RunAsGroup / FSGroup are the Pod's runtime identity. They default
+	// to ROOT (0:0) for the same reason the operator's forwarding-init runs as
+	// root: the world volume is written by the game image's own UID (root for the
+	// images we ship), and Paper saves mode-0600 files a non-root writer/reader
+	// cannot replace (a uid-1000 restore cannot overwrite level.dat). DAC_OVERRIDE
+	// on the container covers images whose UID is neither root nor ours; FSGroup
+	// is omitted when zero.
 	RunAsUser  int64
 	RunAsGroup int64
 	FSGroup    int64
@@ -112,7 +114,6 @@ const (
 	defaultDeadline       = 30 * time.Minute
 	defaultCPULimit       = "1"
 	defaultMemLimit       = "1Gi"
-	defaultRunAsID        = int64(1000)
 	defaultTTL            = 10 * time.Minute
 )
 
@@ -143,15 +144,6 @@ func (c Config) withDefaults() Config {
 	if c.MemLimit == "" {
 		c.MemLimit = defaultMemLimit
 	}
-	if c.RunAsUser == 0 {
-		c.RunAsUser = defaultRunAsID
-	}
-	if c.RunAsGroup == 0 {
-		c.RunAsGroup = defaultRunAsID
-	}
-	if c.FSGroup == 0 {
-		c.FSGroup = defaultRunAsID
-	}
 	if c.TTLAfterFinished <= 0 {
 		c.TTLAfterFinished = defaultTTL
 	}