Retain verified restored units and validate original installer identity bindings

core/archipelago/src/container/prod_orchestrator.rs

@@ -3283,6 +3283,16 @@ impl ProdContainerOrchestrator {
     }
 
     async fn prepare_for_start(&self, manifest: &AppManifest) -> Result<()> {
+        if super::supervised_update::installed_unit(
+            &self.data_dir,
+            &compute_container_name(manifest),
+        )?
+        .is_some()
+        {
+            // Already-reviewed managed configuration is authoritative. Applying
+            // today's catalog files/mount preparation could mutate restored data.
+            return Ok(());
+        }
         self.run_pre_start_hooks(&manifest.app.id).await?;
         self.ensure_bind_mount_sockets(manifest).await?;
         self.ensure_bind_mount_dirs(manifest).await?;
@@ -3585,6 +3595,17 @@ impl ProdContainerOrchestrator {
     }
 
     async fn ensure_resolved_source_available(&self, lm: &LoadedManifest) -> Result<()> {
+        if let Some((body, _)) = super::supervised_update::installed_unit(
+            &self.data_dir,
+            &compute_container_name(&lm.manifest),
+        )? {
+            let image = body
+                .lines()
+                .find_map(|line| line.trim().strip_prefix("Image="))
+                .context("Saved managed image missing")?;
+            anyhow::ensure!(self.runtime.image_exists(image).await?, "Saved managed image is unavailable; refusing to pull or recreate from a changed catalog");
+            return Ok(());
+        }
         let resolved = lm.manifest.app.container.resolve().ok_or_else(|| {
             anyhow::anyhow!(
                 "manifest for {} has invalid container source (neither image nor build)",

core/archipelago/src/container/supervised_runtime.rs

@@ -434,6 +434,58 @@ impl<B: DrainBarrier> Supervisor for SystemdSupervisor<B> {
                     == target.name,
             "Original unit or reviewed immutable target changed before update"
         );
+        if plan
+            .prepared
+            .manifest
+            .app
+            .container
+            .media_registration_identity
+        {
+            let identity =
+                crate::identity::NodeIdentity::load_existing(&self.data_dir.join("identity"))
+                    .await?;
+            let pin = super::registration_pin::load_existing(
+                &self.data_dir,
+                &plan.prepared.manifest.app.id,
+                &identity,
+            )?;
+            let mut expected = plan.prepared.manifest.clone();
+            super::registration_pin::apply_environment(&mut expected, &pin)?;
+            for entry in expected
+                .app
+                .environment
+                .iter()
+                .filter(|entry| entry.starts_with("ARCHIPELAGO_REGISTRATION_"))
+            {
+                let key = entry
+                    .split_once('=')
+                    .context("Malformed registration binding")?
+                    .0;
+                let in_manifest: Vec<_> = plan
+                    .prepared
+                    .manifest
+                    .app
+                    .environment
+                    .iter()
+                    .filter(|value| value.split_once('=').is_some_and(|(name, _)| name == key))
+                    .collect();
+                let in_unit: Vec<_> = plan
+                    .prepared
+                    .body
+                    .lines()
+                    .filter_map(|line| line.trim().strip_prefix("Environment="))
+                    .map(|value| value.trim_matches('"'))
+                    .filter(|value| value.split_once('=').is_some_and(|(name, _)| name == key))
+                    .collect();
+                anyhow::ensure!(
+                    in_manifest.len() == 1
+                        && in_manifest[0] == entry
+                        && in_unit.len() == 1
+                        && in_unit[0] == entry,
+                    "Reviewed registration identity does not match the existing installer pin"
+                );
+            }
+        }
         Ok(plan.prepared.clone())
     }
     async fn target_hooks(

core/archipelago/src/container/supervised_update.rs

@@ -532,8 +532,8 @@ fn installed_path(data: &Path, name: &str) -> Result<PathBuf> {
 }
 fn publish_installed(guard: &Guard, record: &Journal) -> Result<()> {
     anyhow::ensure!(
-        record.phase == Phase::Committed,
-        "Only committed units may be published"
+        matches!(record.phase, Phase::Committed | Phase::Restored),
+        "Only verified terminal units may be published"
     );
     let dir = guard.directory().join("installed-units");
     match std::fs::DirBuilder::new().mode(0o700).create(&dir) {
@@ -550,7 +550,11 @@ fn publish_installed(guard: &Guard, record: &Journal) -> Result<()> {
             schema: 1,
             operation: record.id.clone(),
             name: member.original.name.clone(),
-            body: member.target_body.clone(),
+            body: if record.phase == Phase::Restored {
+                member.pinned_original_body.clone()
+            } else {
+                member.target_body.clone()
+            },
             mode: member.original.file_mode,
         };
         let temporary = dir.join(format!(".{}.tmp", uuid::Uuid::new_v4()));
@@ -904,9 +908,13 @@ async fn restore(guard: &Guard, record: &mut Journal, supervisor: &impl Supervis
     }
     record.phase = Phase::Restored;
     save(guard, record)?;
+    publish_installed(guard, record)?;
     supervisor
         .release_barrier(&record.id, Completion::Restored)
         .await?;
+    for member in &record.members {
+        guard.release_hold(&member.original.name, &record.id)?;
+    }
     record.cleanup_done = true;
     save(guard, record)
 }
@@ -931,9 +939,13 @@ pub(crate) async fn recover(guard: &Guard, supervisor: &impl Supervisor) -> Resu
                 }
             }
             Phase::Restored => {
+                publish_installed(guard, &record)?;
                 supervisor
                     .release_barrier(&record.id, Completion::Restored)
                     .await?;
+                for member in &record.members {
+                    guard.release_hold(&member.original.name, &record.id)?;
+                }
             } // Never release a newer owner.
             _ => restore(guard, &mut record, supervisor).await?,
         }
@@ -1300,7 +1312,11 @@ mod tests {
         assert_eq!(restored.image, "e".repeat(64));
         assert_eq!(restored.config_sha256, runtime.original.config_sha256);
         assert_ne!(restored.id, format!("{:064x}", 1));
-        assert!(super::super::update_transaction::is_held(root.path(), "movie").unwrap());
+        assert!(!super::super::update_transaction::is_held(root.path(), "movie").unwrap());
+        assert_eq!(
+            installed_unit(root.path(), "movie").unwrap().unwrap().0,
+            *runtime.body.lock().unwrap()
+        );
         assert_eq!(records(&guard).unwrap()[0].phase, Phase::Restored);
     }
     #[tokio::test]