gatehouse_runtime_protocol/lifecycle/journal/storage/slot/
recovery.rs

1//! Explicit trusted Core operator policy. Not human authentication or auto recovery.
2use super::*;
3use sha2::{Digest, Sha256};
4
5pub struct OperatorRecovery {
6    slot: ServiceSlot,
7    journal: Journal,
8    authorization: String,
9}
10impl OperatorRecovery {
11    /// Caller must originate in an explicit operator command, never normal start.
12    /// Exact fence and lease invocation are required; authority is never inferred.
13    pub fn open_mapped(
14        base: &Path,
15        project: &str,
16        resource: &str,
17        fence: u64,
18        lease_invocation: &str,
19        authorization: &str,
20    ) -> Result<Self> {
21        if authorization.len() != 32
22            || !authorization
23                .bytes()
24                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
25        {
26            return Err(Invalid(
27                "operator authorization must be fresh 128-bit identity",
28            ));
29        }
30        let _parent = Journal::directory(base)?;
31        let root = ServiceSlot::mapped_root(base, project, resource)?;
32        let directory = Journal::directory(&root)?;
33        let bytes =
34            Journal::read(&directory, "service.json")?.ok_or(Invalid("missing service slot"))?;
35        let record: Record = serde_json::from_value(canonical::parse_canonical(&bytes)?)
36            .map_err(|_| Invalid("invalid operator slot record"))?;
37        record.validate()?;
38        if record.released {
39            return Err(Invalid("released slot does not require operator recovery"));
40        }
41        if record.fence != fence
42            || record.resolve.binding.lease_invocation != lease_invocation
43            || record.resolve.binding.project_id != project
44            || record.resolve.binding.resource_id != resource
45        {
46            return Err(Invalid(
47                "operator authorization does not match current resource fence",
48            ));
49        }
50        let slot = ServiceSlot {
51            root,
52            directory,
53            record,
54            bytes,
55            reserved: true,
56        };
57        slot.check()?;
58        let journal = Journal::open_reserved(
59            &slot.root.join(slot.record.journal_name()),
60            &slot.record.resolve,
61            &slot.record.plan,
62        )?;
63        // An ambiguous Create without an exact ID cannot be discovered/adopted.
64        journal.target()?;
65        let name = format!("operator-{authorization}.json");
66        if Journal::read(&slot.directory, &name)?.is_some() {
67            return Err(Invalid("operator authorization replay"));
68        }
69        let intent = canonical::encode(&serde_json::json!({
70            "schema":"gatehouse-lifecycle-operator/v1", "action":"cleanup-known-id",
71            "authorization":authorization, "fence":fence, "lease_invocation":lease_invocation,
72            "slot_sha256":format!("{:x}",Sha256::digest(&slot.bytes)),
73            "target":journal.target()?
74        }))?;
75        Journal::atomic(&slot.directory, &name, &intent)?;
76        Ok(Self {
77            slot,
78            journal,
79            authorization: authorization.into(),
80        })
81    }
82    pub fn resolve_request(&self) -> &Request {
83        self.slot.resolve_request()
84    }
85    pub fn target(&self) -> Result<Target> {
86        self.journal.target()
87    }
88    pub fn phase(&self) -> Phase {
89        self.journal.phase()
90    }
91    pub fn check(&self) -> Result<()> {
92        self.slot.check()
93    }
94    pub fn begin(&mut self, operation: Operation, invocation: &str) -> Result<Dispatch<'_>> {
95        self.slot.check()?;
96        if !matches!(
97            operation,
98            Operation::Status { .. } | Operation::Stop { .. } | Operation::Remove { .. }
99        ) {
100            return Err(Invalid(
101                "operator recovery forbids start/create/health/resolution",
102            ));
103        }
104        let mut request = self.journal.record.create.clone();
105        request.operation = operation;
106        request.binding.generation = self
107            .journal
108            .revision()
109            .checked_add(1)
110            .ok_or(Invalid("operator generation exhausted"))?;
111        request.binding.invocation = invocation.into();
112        request.binding.operation_key = request.operation_key()?;
113        self.journal.begin(&request, Purpose::OperatorCleanup)
114    }
115    pub fn finish(self) -> Result<()> {
116        self.finish_with(|| Ok(()))
117    }
118    fn finish_with(mut self, after_proof: impl FnOnce() -> Result<()>) -> Result<()> {
119        self.slot.check()?;
120        self.journal.finish_release()?;
121        let evidence = canonical::encode(&serde_json::json!({
122            "schema":"gatehouse-lifecycle-operator-proof/v1", "authorization":self.authorization,
123            "fence":self.slot.fence(), "slot_sha256":format!("{:x}",Sha256::digest(&self.slot.bytes)),
124            "journal_sha256":format!("{:x}",Sha256::digest(&self.journal.bytes)),
125            "target":self.journal.target()?
126        }))?;
127        Journal::atomic(
128            &self.slot.directory,
129            &format!("operator-{}.proof.json", self.authorization),
130            &evidence,
131        )?;
132        after_proof()?;
133        self.slot.finish(self.journal)
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use super::*;
140    use crate::lifecycle::{
141        journal::storage::tests::{absent, object, reply},
142        wire::ContainerState,
143    };
144
145    #[test]
146    fn persisted_operator_proof_never_substitutes_for_slot_release() {
147        let (d, resolve, plan) = super::super::tests::fixture();
148        let mut slot =
149            ServiceSlot::acquire_mapped(d.path(), resolve.clone(), plan.clone(), vec![]).unwrap();
150        let root = slot.root.clone();
151        let mut reservation = slot.reserve().unwrap();
152        let image = reply(
153            &resolve,
154            Outcome::Image {
155                image_id: format!("sha256:{}", "e".repeat(64)),
156                repo_digests: vec![],
157            },
158        );
159        reservation
160            .dispatch()
161            .unwrap()
162            .observe(Some(&image))
163            .unwrap();
164        let mut journal = reservation.into_journal(&"3".repeat(32)).unwrap();
165        let token = journal.dispatch().unwrap();
166        let create = token.request().clone();
167        let created = reply(
168            &create,
169            Outcome::Created {
170                object: object(&create, ContainerState::Created),
171            },
172        );
173        token.observe(Some(&created)).unwrap();
174        drop(journal);
175        drop(slot);
176        let open = |auth: &str| {
177            OperatorRecovery::open_mapped(
178                d.path(),
179                &resolve.binding.project_id,
180                &resolve.binding.resource_id,
181                1,
182                &resolve.binding.lease_invocation,
183                auth,
184            )
185        };
186        let first = "9".repeat(32);
187        let mut recovery = open(&first).unwrap();
188        let target = recovery.target().unwrap();
189        let token = recovery
190            .begin(Operation::Status { target }, &"4".repeat(32))
191            .unwrap();
192        let status = token.request().clone();
193        token.observe(Some(&absent(&status))).unwrap();
194        let before = fs::read(root.join("service.json")).unwrap();
195        let failure = recovery.finish_with(|| {
196            assert!(root.join(format!("operator-{first}.proof.json")).exists());
197            Err(Invalid("injected failure after proof before slot release"))
198        });
199        assert!(failure
200            .unwrap_err()
201            .to_string()
202            .contains("injected failure"));
203        assert_eq!(fs::read(root.join("service.json")).unwrap(), before);
204        let journal_root = root.join(format!("journal-{}", resolve.binding.lease_invocation));
205        assert!(journal_root.join("tombstone.json").exists());
206        assert!(!journal_root.join("active").exists());
207        assert!(ServiceSlot::acquire_mapped(d.path(), resolve.clone(), plan, vec![]).is_err());
208        assert!(open(&first).err().unwrap().to_string().contains("replay"));
209        let second = "8".repeat(32);
210        let recovery = open(&second).unwrap();
211        assert_eq!(recovery.phase(), Phase::VerifiedAbsent);
212        // No dispatch: durable absence and the exact tombstone finish the slot.
213        recovery.finish().unwrap();
214        let record: Record =
215            serde_json::from_slice(&fs::read(root.join("service.json")).unwrap()).unwrap();
216        assert!(record.released);
217        assert!(root.join(format!("operator-{second}.proof.json")).exists());
218        assert!(open(&"7".repeat(32))
219            .err()
220            .unwrap()
221            .to_string()
222            .contains("released slot"));
223    }
224}
gatehouse_runtime_protocol/lifecycle/journal/storage/slot/recovery.rs

gatehouse_runtime_protocol/lifecycle/journal/storage/slot/
recovery.rs
