mkit_server/hooks/
inspection.rs

1//! Synchronous `PRE_RECEIVE` inspection over the existing signed hook channels.
2//!
3//! Calls carry metadata and optional R-193 raw-pack retrieval credentials only.
4//! Public object serving cannot supply unpublished content (SPEC-SERVER §6.4).
5
6use core::time::Duration;
7use std::collections::BTreeSet;
8use std::sync::Arc;
9
10use super::client::Rpc;
11use super::map;
12use super::proto::v1 as pb;
13use super::proto::v1::__buffa::oneof::inspect_response::Verdict;
14use super::{DEFAULT_TIMEOUT, HookChannel, HookClient};
15use crate::error::ServerError;
16use crate::op::Operation;
17
18/// A validated launch-profile verdict. Quarantine rejects the push at stage 5.
19/// Quarantine is a rejection under the launch amendment.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub enum InspectVerdict {
22    /// This inspector permits the batch.
23    Pass,
24    /// Reject this push with sanitized public policy text.
25    Reject(String),
26}
27
28/// One named synchronous inspector over a shared hook client.
29pub struct RemoteInspector<C> {
30    name: String,
31    client: Arc<HookClient<C>>,
32    timeout: Duration,
33}
34
35impl<C> Clone for RemoteInspector<C> {
36    fn clone(&self) -> Self {
37        Self {
38            name: self.name.clone(),
39            client: self.client.clone(),
40            timeout: self.timeout,
41        }
42    }
43}
44
45impl<C> core::fmt::Debug for RemoteInspector<C> {
46    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
47        f.debug_struct("RemoteInspector")
48            .field("name", &self.name)
49            .field("timeout", &self.timeout)
50            .finish_non_exhaustive()
51    }
52}
53
54impl<C> RemoteInspector<C> {
55    /// The inspector's configuration identity, with the default 5 s timeout.
56    #[must_use]
57    pub fn new(name: impl Into<String>, client: Arc<HookClient<C>>) -> Self {
58        Self {
59            name: name.into(),
60            client,
61            timeout: DEFAULT_TIMEOUT,
62        }
63    }
64
65    /// The stable configured identity used to derive logical inspection ids.
66    #[must_use]
67    pub fn id(&self) -> &str {
68        &self.name
69    }
70
71    /// Bound each remote attempt by this timeout.
72    #[must_use]
73    pub fn with_timeout(mut self, timeout: Duration) -> Self {
74        self.timeout = timeout;
75        self
76    }
77}
78
79impl<C: HookChannel> RemoteInspector<C> {
80    /// Inspect one complete assigned batch before apply. Every attempt signs
81    /// afresh; the caller retains `inspection_id` across logical retries.
82    ///
83    /// # Errors
84    /// Retryable unavailable for transport failures or invalid metadata/verdicts.
85    pub async fn inspect(
86        &self,
87        op: &Operation,
88        inspection_id: &str,
89        objects: &[pb::InspectObject],
90    ) -> Result<InspectVerdict, ServerError> {
91        self.inspect_request(op, inspection_id, objects, None).await
92    }
93
94    async fn inspect_request(
95        &self,
96        op: &Operation,
97        inspection_id: &str,
98        objects: &[pb::InspectObject],
99        retrieval: Option<pb::InspectRetrieval>,
100    ) -> Result<InspectVerdict, ServerError> {
101        if self.name.is_empty() || inspection_id.is_empty() || !objects.iter().all(valid_object) {
102            return Err(map::unavailable("inspection", "invalid request metadata"));
103        }
104        let request = pb::InspectRequest {
105            operation: map::authorize_request(op, self.client.server_audience()).operation,
106            objects: objects.to_vec(),
107            scanner_retrieval: retrieval.into(),
108            phase: Some(pb::InspectPhase::INSPECT_PHASE_PRE_RECEIVE.into()),
109            inspection_id: Some(inspection_id.to_owned()),
110            ..Default::default()
111        };
112        let answer: pb::InspectResponse = self
113            .client
114            .decide(Rpc::Inspect, &request, self.timeout)
115            .await
116            .map_err(|failure| map::unavailable("inspection", failure.0))?;
117        validate_answer(answer, objects)
118    }
119}
120
121impl<C: HookChannel> crate::pipeline::inspection::ContentInspector for RemoteInspector<C> {
122    fn id(&self) -> &str {
123        self.id()
124    }
125
126    fn retrieval_timeout(&self) -> Duration {
127        self.timeout
128    }
129
130    fn inspect_with_retrieval<'a>(
131        &'a self,
132        op: &'a Operation,
133        id: &'a str,
134        objects: &'a [pb::InspectObject],
135        retrieval: Option<pb::InspectRetrieval>,
136    ) -> crate::BoxFuture<'a, Result<InspectVerdict, ServerError>> {
137        Box::pin(self.inspect_request(op, id, objects, retrieval))
138    }
139
140    fn inspect<'a>(
141        &'a self,
142        op: &'a Operation,
143        inspection_id: &'a str,
144        objects: &'a [pb::InspectObject],
145    ) -> crate::BoxFuture<'a, Result<InspectVerdict, ServerError>> {
146        Box::pin(self.inspect(op, inspection_id, objects))
147    }
148}
149
150fn valid_object(object: &pb::InspectObject) -> bool {
151    object.id.as_ref().is_some_and(|id| id.len() == 32)
152        && object.size.is_some()
153        && object.kind.is_some_and(|kind| {
154            [
155                pb::InspectObjectKind::INSPECT_OBJECT_KIND_BLOB,
156                pb::InspectObjectKind::INSPECT_OBJECT_KIND_CHUNKED_FILE,
157                pb::InspectObjectKind::INSPECT_OBJECT_KIND_CHUNK,
158            ]
159            .into_iter()
160            .any(|value| kind == value)
161        })
162}
163
164fn policy_text(text: Option<String>) -> String {
165    text.filter(|text| !text.is_empty() && text.len() <= 512 && !text.chars().any(char::is_control))
166        .unwrap_or_else(|| "inspection rejected".to_owned())
167}
168
169fn validate_answer(
170    response: pb::InspectResponse,
171    objects: &[pb::InspectObject],
172) -> Result<InspectVerdict, ServerError> {
173    let bad = |reason| map::unavailable("inspection", reason);
174    let ids: BTreeSet<&[u8]> = objects.iter().filter_map(|o| o.id.as_deref()).collect();
175    if response
176        .flagged_objects
177        .iter()
178        .any(|id| id.len() != 32 || !ids.contains(id.as_slice()))
179    {
180        return Err(bad("flagged object outside batch"));
181    }
182    let verdict = response.verdict.ok_or_else(|| bad("absent verdict"))?;
183    if response.takedown_reason.is_some() && !matches!(verdict, Verdict::Reject(_)) {
184        return Err(bad("takedown reason on non-reject verdict"));
185    }
186    match verdict {
187        Verdict::Pass(_) if response.flagged_objects.is_empty() => Ok(InspectVerdict::Pass),
188        Verdict::Pass(_) => Err(bad("flagged objects on pass")),
189        // PRE_RECEIVE rejects ignore takedown_reason, including unregistered
190        // tokens: denying this push creates no global takedown (SPEC §6.4).
191        Verdict::Reject(reject) => Ok(InspectVerdict::Reject(policy_text(reject.message))),
192        Verdict::Quarantine(quarantine) => {
193            if quarantine.reason.as_ref().is_some_and(|r| r.len() > 512) {
194                return Err(bad("quarantine reason too long"));
195            }
196            Ok(InspectVerdict::Reject(policy_text(quarantine.reason)))
197        }
198        Verdict::Defer(_) => Err(bad("defer invalid at pre-receive")),
199    }
200}
201
202#[cfg(test)]
203mod tests {
204    use futures_executor::block_on;
205    use mkit_core::refs::RefWriteCondition;
206
207    use super::*;
208    use crate::error::Code;
209    use crate::hooks::tests::{MockChannel, Step, channel_of, client};
210    use crate::op::{OpKind, RefUpdate};
211    use crate::principal::Principal;
212    use crate::repo::{NamespaceKey, RepoId, RepoName};
213    use crate::rt::ManualSleep;
214
215    fn objects() -> Vec<pb::InspectObject> {
216        vec![pb::InspectObject {
217            id: Some(vec![0x22; 32]),
218            size: Some(123),
219            kind: Some(pb::InspectObjectKind::INSPECT_OBJECT_KIND_BLOB.into()),
220            ..Default::default()
221        }]
222    }
223
224    fn answer(json: &str) -> Result<InspectVerdict, ServerError> {
225        validate_answer(serde_json::from_str(json).unwrap(), &objects())
226    }
227
228    #[test]
229    fn launch_verdicts_validate_and_sanitize_policy_text() {
230        assert_eq!(answer(r#"{"pass":{}}"#).unwrap(), InspectVerdict::Pass);
231        assert_eq!(
232            answer(r#"{"reject":{"code":"unauthenticated","message":"policy"},"takedownReason":"anything"}"#).unwrap(),
233            InspectVerdict::Reject("policy".into())
234        );
235        assert_eq!(
236            answer(r#"{"quarantine":{"reason":"hold"}}"#).unwrap(),
237            InspectVerdict::Reject("hold".into())
238        );
239        assert_eq!(
240            answer(r#"{"reject":{"message":"secret\ntext"}}"#).unwrap(),
241            InspectVerdict::Reject("inspection rejected".into())
242        );
243        let flagged = "IiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiI=";
244        assert!(
245            answer(&format!(
246                r#"{{"reject":{{}},"flaggedObjects":["{flagged}"]}}"#
247            ))
248            .is_ok()
249        );
250        assert!(
251            answer(&format!(
252                r#"{{"quarantine":{{}},"flaggedObjects":["{flagged}"]}}"#
253            ))
254            .is_ok()
255        );
256    }
257
258    #[test]
259    fn invalid_verdicts_fail_closed() {
260        for json in [
261            "{}",
262            r#"{"defer":{"retryAfterMs":1}}"#,
263            r#"{"pass":{},"flaggedObjects":["IiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiIiI="]}"#,
264            r#"{"reject":{},"flaggedObjects":["AQ=="]}"#,
265            r#"{"reject":{},"flaggedObjects":["AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="]}"#,
266            r#"{"pass":{},"takedownReason":"policy"}"#,
267            r#"{"quarantine":{},"takedownReason":"policy"}"#,
268        ] {
269            assert_eq!(
270                answer(json).unwrap_err().code(),
271                Code::Unavailable,
272                "{json}"
273            );
274        }
275        let long = serde_json::json!({"quarantine": {"reason": "x".repeat(513)}});
276        assert_eq!(
277            answer(&long.to_string()).unwrap_err().code(),
278            Code::Unavailable
279        );
280    }
281
282    #[test]
283    fn signed_retries_keep_metadata_and_inspection_id_with_fresh_nonce() {
284        let client = client(
285            MockChannel::new(Step::json(r#"{"pass":{}}"#)),
286            ManualSleep::new(),
287        );
288        let remote = RemoteInspector::new("scanner", client.clone());
289        let inspector: &dyn crate::pipeline::inspection::ContentInspector = &remote;
290        assert_eq!(inspector.id(), "scanner");
291        assert_eq!(
292            inspector.phase(),
293            crate::pipeline::inspection::InspectorPhase::Sync
294        );
295        assert_eq!(
296            inspector.on_unavailable(),
297            crate::pipeline::inspection::OnUnavailable::FailClosed
298        );
299        let op = Operation::new(
300            RepoId {
301                namespace: NamespaceKey::deployment_default(),
302                name: RepoName::new("test").unwrap(),
303            },
304            Principal::Anonymous,
305            None,
306            OpKind::UpdateRef(RefUpdate {
307                name: "refs/heads/main".into(),
308                condition: RefWriteCondition::Missing,
309                new: Some([0x33; 32]),
310            }),
311        );
312        for _ in 0..2 {
313            assert_eq!(
314                block_on(inspector.inspect(&op, "inspection:stable", &objects())).unwrap(),
315                InspectVerdict::Pass
316            );
317        }
318        let seen = channel_of(&client).seen.lock().unwrap();
319        assert_eq!(seen[0].body, seen[1].body);
320        assert_eq!(seen[0].procedure, Rpc::Inspect.path());
321        let request: pb::InspectRequest = serde_json::from_slice(&seen[0].body).unwrap();
322        assert_eq!(request.inspection_id.as_deref(), Some("inspection:stable"));
323        assert_eq!(
324            request.phase,
325            Some(pb::InspectPhase::INSPECT_PHASE_PRE_RECEIVE.into())
326        );
327        assert_eq!(request.objects, objects());
328        let nonce = |i: usize| {
329            &seen[i]
330                .headers
331                .iter()
332                .find(|(name, _)| *name == "X-Mkit-Hook-Nonce")
333                .unwrap()
334                .1
335        };
336        assert_ne!(nonce(0), nonce(1));
337    }
338    #[test]
339    fn retrieval_descriptor_is_metadata_and_covered_by_hook_signature() {
340        use crate::pipeline::inspection::ContentInspector as _;
341        use crate::scanner_retrieval::{Assignment, PackGrant, RetrievalConfig};
342        use mkit_core::hash::to_hex;
343        let client = client(
344            MockChannel::new(Step::json(r#"{"pass":{}}"#)),
345            ManualSleep::new(),
346        );
347        let remote = RemoteInspector::new("scanner", client.clone());
348        let scanner = ed25519_dalek::SigningKey::from_bytes(&[0x65; 32]);
349        let config = RetrievalConfig::parse(
350            &format!("active current {}", to_hex(&[0x64; 32])),
351            &to_hex(&scanner.verifying_key().to_bytes()),
352        )
353        .unwrap();
354        let assignment = Assignment {
355            namespace: "default".into(),
356            repo_name: "test".into(),
357            repository: "test".into(),
358            ref_name: "refs/heads/main".into(),
359            signer: [0x66; 32],
360            packs: vec![PackGrant {
361                id: [0x67; 32],
362                length: 123,
363                tickets: vec![[0x68; 32]],
364            }],
365        };
366        let op = Operation::new(
367            RepoId {
368                namespace: NamespaceKey::deployment_default(),
369                name: RepoName::new("test").unwrap(),
370            },
371            Principal::Anonymous,
372            None,
373            OpKind::UpdateRef(RefUpdate {
374                name: "refs/heads/main".into(),
375                condition: RefWriteCondition::Missing,
376                new: Some([0x33; 32]),
377            }),
378        );
379        let descriptors: Vec<_> = (0..2)
380            .map(|_| {
381                config
382                    .mint(
383                        "https://vcs.example",
384                        "inspection:stable",
385                        &assignment,
386                        remote.retrieval_timeout(),
387                        100,
388                    )
389                    .unwrap()
390            })
391            .collect();
392        for descriptor in &descriptors {
393            assert_eq!(
394                block_on(remote.inspect_with_retrieval(
395                    &op,
396                    "inspection:stable",
397                    &objects(),
398                    Some(descriptor.clone()),
399                ))
400                .unwrap(),
401                InspectVerdict::Pass
402            );
403        }
404        let seen = channel_of(&client).seen.lock().unwrap();
405        for (call, descriptor) in seen.iter().zip(&descriptors) {
406            crate::hooks::tests::verify(call);
407            let request: pb::InspectRequest = serde_json::from_slice(&call.body).unwrap();
408            assert_eq!(request.scanner_retrieval.as_option(), Some(descriptor));
409            assert_eq!(request.objects, objects());
410            assert_eq!(request.inspection_id.as_deref(), Some("inspection:stable"));
411        }
412        assert_ne!(descriptors[0].capability, descriptors[1].capability);
413        assert_ne!(seen[0].body, seen[1].body);
414    }
415}
mkit_server/hooks/inspection.rs

mkit_server/hooks/
inspection.rs
