lean_ctx_protocol/
engine_egress.rs

1// SPDX-License-Identifier: Apache-2.0
2//! Egress admission for an operator-owned Engine process (`engine egress-admit`).
3//!
4//! A host that forwards model requests itself (for example an organization's
5//! inference gateway) hands the request to the local Engine, which applies the
6//! same admission the lean-ctx proxy applies before a request leaves: secrets
7//! and personal data are masked, withheld content is never forwarded. The
8//! answer carries the admitted body, the request's most sensitive
9//! classification and the decision receipt. What the host does with the
10//! classification (destination rules) is the host's policy, not the Engine's.
11
12use std::fmt::Write as _;
13
14use serde::{Deserialize, Serialize};
15use serde_json::Value;
16use sha2::{Digest, Sha256};
17
18use crate::Sha256Digest;
19use crate::common::ValidationError;
20use crate::context_gateway::{
21    ClassificationV1, ContextDecisionReceiptV1, DeliveryOutcomeV1, EgressTransportFailureV1,
22    EgressTransportObservationV1, EgressTransportObserverV1, EgressTransportResultV1,
23    EgressTransportV1,
24};
25
26pub const ENGINE_EGRESS_SCHEMA_VERSION: u32 = 1;
27/// Opt-in preparation response that links an admission receipt for later reports.
28pub const ENGINE_EGRESS_PREPARE_SCHEMA_VERSION: u32 = 2;
29/// Maximum per-admission transport attempts persisted or host-reported.
30pub const MAX_EGRESS_ATTEMPTS: u32 = 32;
31/// Upper bound for one request document (matches the proxy's body bound).
32pub const MAX_ENGINE_EGRESS_REQUEST_BYTES: usize = 32 * 1024 * 1024;
33/// Host reports contain only bounded status metadata, never request content.
34pub const MAX_ENGINE_EGRESS_REPORT_BYTES: usize = 4096;
35
36#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
37#[serde(deny_unknown_fields)]
38pub struct EngineEgressAdmissionRequestV1 {
39    pub schema_version: u32,
40    /// Provider label as the request's wire shape names it (`anthropic`,
41    /// `openai`, ...). Informational: it is recorded in the receipt.
42    pub provider: String,
43    /// Base URL the host will send the request to. Locality (local vs.
44    /// remote) is derived from it; anything not provably local is remote.
45    pub upstream_base: String,
46    /// The model request document exactly as the host would forward it.
47    pub body: Value,
48}
49
50impl EngineEgressAdmissionRequestV1 {
51    pub fn validate(&self) -> Result<(), ValidationError> {
52        if self.schema_version != ENGINE_EGRESS_SCHEMA_VERSION {
53            return Err(ValidationError::new("unsupported egress schema version"));
54        }
55        if self.provider.trim().is_empty() || self.provider.len() > 64 {
56            return Err(ValidationError::new("egress provider label is invalid"));
57        }
58        if self.upstream_base.trim().is_empty() || self.upstream_base.len() > 2048 {
59            return Err(ValidationError::new("egress upstream base is invalid"));
60        }
61        if !self.body.is_object() {
62            return Err(ValidationError::new("egress body must be a JSON object"));
63        }
64        Ok(())
65    }
66}
67
68#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
69#[serde(rename_all = "snake_case")]
70pub enum EgressDispositionV1 {
71    /// Forward the request unchanged.
72    Forward,
73    /// Forward `body` instead of the original (values were masked).
74    Rewritten,
75    /// Do not forward anything.
76    Refused,
77}
78
79#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
80#[serde(deny_unknown_fields)]
81pub struct EngineEgressAdmissionResponseV1 {
82    pub schema_version: u32,
83    pub disposition: EgressDispositionV1,
84    /// The body to forward; absent when the request is refused.
85    #[serde(default, skip_serializing_if = "Option::is_none")]
86    pub body: Option<Value>,
87    /// Content-free refusal reason (reason codes only).
88    #[serde(default, skip_serializing_if = "Option::is_none")]
89    pub refusal: Option<String>,
90    /// Most sensitive classification found in the request. Absent when the
91    /// gateway inspected nothing (disabled, or an opaque payload): a host must
92    /// then treat the request as unclassified, never as public.
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub classification: Option<ClassificationV1>,
95    /// The decision receipt; absent when the gateway is disabled.
96    #[serde(default, skip_serializing_if = "Option::is_none")]
97    pub receipt: Option<ContextDecisionReceiptV1>,
98}
99
100impl EngineEgressAdmissionResponseV1 {
101    pub fn validate(&self) -> Result<(), ValidationError> {
102        if self.schema_version != ENGINE_EGRESS_SCHEMA_VERSION {
103            return Err(ValidationError::new("unsupported egress schema version"));
104        }
105        let consistent = match self.disposition {
106            EgressDispositionV1::Refused => self.body.is_none() && self.refusal.is_some(),
107            EgressDispositionV1::Forward | EgressDispositionV1::Rewritten => {
108                self.body.is_some() && self.refusal.is_none()
109            }
110        };
111        if !consistent {
112            return Err(ValidationError::new(
113                "egress body and refusal must match the disposition",
114            ));
115        }
116        if let Some(receipt) = &self.receipt {
117            receipt.validate()?;
118            let outcome_matches = match self.disposition {
119                EgressDispositionV1::Forward | EgressDispositionV1::Rewritten => {
120                    receipt.outcome == DeliveryOutcomeV1::Delivered
121                }
122                EgressDispositionV1::Refused => receipt.outcome != DeliveryOutcomeV1::Delivered,
123            };
124            if !outcome_matches {
125                return Err(ValidationError::new(
126                    "egress disposition does not match its decision receipt outcome",
127                ));
128            }
129        }
130        Ok(())
131    }
132}
133
134/// Explicit opt-in wire contract for hosts that need to report transport
135/// observations. The legacy `egress-admit` response remains V1 and unchanged.
136#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
137#[serde(deny_unknown_fields)]
138pub struct EngineEgressPrepareResponseV2 {
139    pub schema_version: u32,
140    pub decision: EngineEgressAdmissionResponseV1,
141    /// Exact UTF-8 JSON bytes to send. Hosts must send this string verbatim,
142    /// not reserialize `decision.body` (which could change the admitted bytes).
143    #[serde(default, skip_serializing_if = "Option::is_none")]
144    pub prepared_body: Option<String>,
145    /// Content-addressed preparation evidence; this is not a send claim.
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub admission_receipt_digest: Option<Sha256Digest>,
148}
149
150impl EngineEgressPrepareResponseV2 {
151    pub fn validate(&self) -> Result<(), ValidationError> {
152        if self.schema_version != ENGINE_EGRESS_PREPARE_SCHEMA_VERSION {
153            return Err(ValidationError::new(
154                "unsupported egress preparation schema version",
155            ));
156        }
157        self.decision.validate()?;
158        match self.decision.disposition {
159            EgressDispositionV1::Refused => {
160                if self.prepared_body.is_some() {
161                    return Err(ValidationError::new(
162                        "a refused preparation cannot carry prepared bytes",
163                    ));
164                }
165            }
166            EgressDispositionV1::Forward | EgressDispositionV1::Rewritten => {
167                let prepared = self.prepared_body.as_ref().ok_or_else(|| {
168                    ValidationError::new("a dispatchable preparation must carry exact bytes")
169                })?;
170                let parsed: Value = serde_json::from_str(prepared)
171                    .map_err(|_| ValidationError::new("prepared bytes are not JSON"))?;
172                if Some(&parsed) != self.decision.body.as_ref() {
173                    return Err(ValidationError::new(
174                        "prepared bytes do not match the admitted decision body",
175                    ));
176                }
177                let digest = digest_bytes(prepared.as_bytes())?;
178                if let Some(receipt) = &self.decision.receipt
179                    && receipt.final_context.as_ref() != Some(&digest)
180                {
181                    return Err(ValidationError::new(
182                        "prepared bytes do not match the decision receipt digest",
183                    ));
184                }
185            }
186        }
187        if self.admission_receipt_digest.is_some() && self.decision.receipt.is_none() {
188            return Err(ValidationError::new(
189                "egress admission digest requires its decision receipt",
190            ));
191        }
192        Ok(())
193    }
194}
195
196fn digest_bytes(bytes: &[u8]) -> Result<Sha256Digest, ValidationError> {
197    let mut digest = String::from("sha256:");
198    for byte in Sha256::digest(bytes) {
199        write!(&mut digest, "{byte:02x}")
200            .map_err(|_| ValidationError::new("egress digest encoding failed"))?;
201    }
202    Sha256Digest::new(digest)
203}
204
205/// Content-free transport facts reported by a host after it attempted to send
206/// an exact request that the Engine previously recorded as Prepared.
207#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
208#[serde(deny_unknown_fields)]
209pub struct EngineEgressHostReportRequestV1 {
210    pub schema_version: u32,
211    pub admission_receipt_digest: Sha256Digest,
212    pub attempt: u32,
213    pub transport: EgressTransportV1,
214    pub result: EgressTransportResultV1,
215    #[serde(default, skip_serializing_if = "Option::is_none")]
216    pub http_status: Option<u16>,
217    #[serde(default, skip_serializing_if = "Option::is_none")]
218    pub failure: Option<EgressTransportFailureV1>,
219}
220
221impl EngineEgressHostReportRequestV1 {
222    pub fn validate(&self) -> Result<(), ValidationError> {
223        if self.schema_version != ENGINE_EGRESS_SCHEMA_VERSION {
224            return Err(ValidationError::new("unsupported egress schema version"));
225        }
226        if self.attempt > MAX_EGRESS_ATTEMPTS {
227            return Err(ValidationError::new(
228                "egress attempt exceeds the supported limit",
229            ));
230        }
231        let report = EgressTransportObservationV1 {
232            schema_version: self.schema_version,
233            admission_receipt_digest: self.admission_receipt_digest.clone(),
234            attempt: self.attempt,
235            transport: self.transport,
236            observer: EgressTransportObserverV1::HostReported,
237            result: self.result,
238            http_status: self.http_status,
239            failure: self.failure,
240        };
241        report.validate()
242    }
243}
244
245/// The stored host report. The Engine sets `HostReported`; clients cannot
246/// submit an observer value in the request.
247#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
248#[serde(deny_unknown_fields)]
249pub struct EngineEgressHostReportResponseV1 {
250    pub schema_version: u32,
251    pub observation_digest: Sha256Digest,
252    pub observation: EgressTransportObservationV1,
253}
254
255impl EngineEgressHostReportResponseV1 {
256    pub fn validate(&self) -> Result<(), ValidationError> {
257        if self.schema_version != ENGINE_EGRESS_SCHEMA_VERSION {
258            return Err(ValidationError::new("unsupported egress schema version"));
259        }
260        self.observation.validate()?;
261        if self.observation.observer != EgressTransportObserverV1::HostReported
262            || self.observation.canonical_digest()? != self.observation_digest
263        {
264            return Err(ValidationError::new(
265                "host report response does not match its stored observation",
266            ));
267        }
268        Ok(())
269    }
270}
271
272#[cfg(test)]
273mod tests {
274    use super::*;
275
276    fn decision_receipt(outcome: DeliveryOutcomeV1) -> ContextDecisionReceiptV1 {
277        let final_context = (outcome == DeliveryOutcomeV1::Delivered)
278            .then(|| Sha256Digest::new(format!("sha256:{}", "a".repeat(64))).unwrap());
279        serde_json::from_value(serde_json::json!({
280            "schema_version": 1,
281            "receipt_id": "egress-test",
282            "mode": "developer",
283            "principal": {"kind": "unknown"},
284            "destination": {"provider": "openai", "locality": "remote"},
285            "sources": {"inspected": 0, "permitted": 0, "selected": 0, "blocked": 0},
286            "decisions": [],
287            "security": {
288                "redactions": 0, "blocked_objects": 0, "quarantined_objects": 0,
289                "injection_signals": 0, "incomplete_coverage": 0
290            },
291            "tokens": {"original": 0, "delivered": 0},
292            "final_context": final_context,
293            "outcome": outcome,
294            "duration_us": 0
295        }))
296        .unwrap()
297    }
298
299    fn decision(
300        disposition: EgressDispositionV1,
301        receipt: Option<ContextDecisionReceiptV1>,
302    ) -> EngineEgressAdmissionResponseV1 {
303        let refused = disposition == EgressDispositionV1::Refused;
304        EngineEgressAdmissionResponseV1 {
305            schema_version: ENGINE_EGRESS_SCHEMA_VERSION,
306            disposition,
307            body: (!refused).then(|| serde_json::json!({"model": "m"})),
308            refusal: refused.then(|| "request withheld".to_owned()),
309            classification: Some(ClassificationV1::Internal),
310            receipt,
311        }
312    }
313
314    #[test]
315    fn decision_receipt_outcome_must_match_egress_disposition() {
316        for disposition in [EgressDispositionV1::Forward, EgressDispositionV1::Rewritten] {
317            assert!(
318                decision(
319                    disposition,
320                    Some(decision_receipt(DeliveryOutcomeV1::Delivered))
321                )
322                .validate()
323                .is_ok()
324            );
325            for outcome in [DeliveryOutcomeV1::Withheld, DeliveryOutcomeV1::Failed] {
326                assert!(
327                    decision(disposition, Some(decision_receipt(outcome)))
328                        .validate()
329                        .is_err(),
330                    "{disposition:?} with {outcome:?} receipt"
331                );
332            }
333        }
334        for outcome in [DeliveryOutcomeV1::Withheld, DeliveryOutcomeV1::Failed] {
335            assert!(
336                decision(
337                    EgressDispositionV1::Refused,
338                    Some(decision_receipt(outcome))
339                )
340                .validate()
341                .is_ok()
342            );
343        }
344        assert!(
345            decision(
346                EgressDispositionV1::Refused,
347                Some(decision_receipt(DeliveryOutcomeV1::Delivered))
348            )
349            .validate()
350            .is_err()
351        );
352        // Disabled or best-effort operation may omit a durable decision receipt.
353        assert!(
354            decision(EgressDispositionV1::Forward, None)
355                .validate()
356                .is_ok()
357        );
358    }
359
360    #[test]
361    fn a_refusal_never_carries_a_body_and_a_forward_always_does() {
362        let mut response = EngineEgressAdmissionResponseV1 {
363            schema_version: ENGINE_EGRESS_SCHEMA_VERSION,
364            disposition: EgressDispositionV1::Refused,
365            body: None,
366            refusal: Some("request withheld by the context gateway (x)".into()),
367            classification: Some(ClassificationV1::Restricted),
368            receipt: None,
369        };
370        assert!(response.validate().is_ok());
371        response.body = Some(serde_json::json!({"model": "m"}));
372        assert!(response.validate().is_err(), "refused with a body");
373        response.disposition = EgressDispositionV1::Forward;
374        response.refusal = None;
375        assert!(response.validate().is_ok());
376        response.body = None;
377        assert!(response.validate().is_err(), "forward without a body");
378    }
379
380    #[test]
381    fn preparation_link_uses_a_separate_strict_versioned_response() {
382        let digest = Sha256Digest::new(format!("sha256:{}", "a".repeat(64))).unwrap();
383        let decision = EngineEgressAdmissionResponseV1 {
384            schema_version: ENGINE_EGRESS_SCHEMA_VERSION,
385            disposition: EgressDispositionV1::Forward,
386            body: Some(serde_json::json!({"model": "m"})),
387            refusal: None,
388            classification: Some(ClassificationV1::Internal),
389            receipt: None,
390        };
391        let response = EngineEgressPrepareResponseV2 {
392            schema_version: ENGINE_EGRESS_PREPARE_SCHEMA_VERSION,
393            decision,
394            prepared_body: Some(r#"{ "model" : "m" }"#.into()),
395            admission_receipt_digest: None,
396        };
397        assert!(response.validate().is_ok());
398        let legacy = serde_json::to_value(&response.decision).unwrap();
399        assert!(legacy.get("admission_receipt_digest").is_none());
400        let wire = serde_json::to_value(&response).unwrap();
401        assert_eq!(wire["prepared_body"], r#"{ "model" : "m" }"#);
402        let round_trip: EngineEgressPrepareResponseV2 =
403            serde_json::from_value(wire.clone()).unwrap();
404        assert_eq!(
405            round_trip.prepared_body.as_deref(),
406            Some(r#"{ "model" : "m" }"#),
407            "hosts send these exact UTF-8 bytes instead of reserializing the body"
408        );
409        let reserialized =
410            serde_json::to_string(round_trip.decision.body.as_ref().unwrap()).unwrap();
411        assert_ne!(
412            round_trip.prepared_body.as_deref(),
413            Some(reserialized.as_str())
414        );
415        let mut invalid = response.clone();
416        invalid.schema_version = ENGINE_EGRESS_SCHEMA_VERSION;
417        assert!(invalid.validate().is_err());
418        invalid.schema_version = ENGINE_EGRESS_PREPARE_SCHEMA_VERSION;
419        invalid.admission_receipt_digest = Some(digest);
420        assert!(
421            invalid.validate().is_err(),
422            "digest needs a decision receipt"
423        );
424        let mut value = wire;
425        value["unknown"] = serde_json::json!(true);
426        assert!(serde_json::from_value::<EngineEgressPrepareResponseV2>(value).is_err());
427    }
428
429    #[test]
430    fn requests_need_an_object_body_and_the_current_schema() {
431        let request = EngineEgressAdmissionRequestV1 {
432            schema_version: ENGINE_EGRESS_SCHEMA_VERSION,
433            provider: "openai".into(),
434            upstream_base: "https://api.openai.com".into(),
435            body: serde_json::json!({"model": "m", "messages": []}),
436        };
437        assert!(request.validate().is_ok());
438        let mut array = request.clone();
439        array.body = serde_json::json!([]);
440        assert!(array.validate().is_err());
441        let mut future = request;
442        future.schema_version = 2;
443        assert!(future.validate().is_err());
444    }
445
446    #[test]
447    fn host_report_request_cannot_choose_its_observer() {
448        let digest = Sha256Digest::new(format!("sha256:{}", "a".repeat(64))).unwrap();
449        let request = EngineEgressHostReportRequestV1 {
450            schema_version: ENGINE_EGRESS_SCHEMA_VERSION,
451            admission_receipt_digest: digest,
452            attempt: 1,
453            transport: EgressTransportV1::Http,
454            result: EgressTransportResultV1::ResponseObserved,
455            http_status: Some(503),
456            failure: None,
457        };
458        assert!(request.validate().is_ok());
459        let mut value = serde_json::to_value(&request).unwrap();
460        value["observer"] = serde_json::json!("engine_observed");
461        assert!(serde_json::from_value::<EngineEgressHostReportRequestV1>(value).is_err());
462
463        let mut over_limit = request;
464        over_limit.attempt = MAX_EGRESS_ATTEMPTS + 1;
465        assert!(over_limit.validate().is_err());
466    }
467
468    #[test]
469    fn host_report_response_is_bound_and_cannot_claim_engine_observation() {
470        let digest = Sha256Digest::new(format!("sha256:{}", "a".repeat(64))).unwrap();
471        let observation = EgressTransportObservationV1 {
472            schema_version: ENGINE_EGRESS_SCHEMA_VERSION,
473            admission_receipt_digest: digest,
474            attempt: 1,
475            transport: EgressTransportV1::Http,
476            observer: EgressTransportObserverV1::HostReported,
477            result: EgressTransportResultV1::ResponseObserved,
478            http_status: Some(200),
479            failure: None,
480        };
481        let response = EngineEgressHostReportResponseV1 {
482            schema_version: ENGINE_EGRESS_SCHEMA_VERSION,
483            observation_digest: observation.canonical_digest().unwrap(),
484            observation,
485        };
486        assert!(response.validate().is_ok());
487        let mut spoofed = response;
488        spoofed.observation.observer = EgressTransportObserverV1::EngineObserved;
489        assert!(spoofed.validate().is_err());
490    }
491}
lean_ctx_protocol/engine_egress.rs

lean_ctx_protocol/
engine_egress.rs
