termivar_scanner/plugin/
context.rs

1use serde::Serialize;
2use std::{
3    fmt,
4    panic::AssertUnwindSafe,
5    sync::{Arc, Mutex, MutexGuard},
6    time::Duration,
7};
8use termivar_core::{ConfidenceScore, EntityId, Evidence, EvidenceSource};
9use tokio_util::sync::CancellationToken;
10use url::Url;
11
12use super::{
13    limits::{
14        ensure_input_budget, invalid_config, validate_identifier, PluginBudget,
15        MAX_PLUGIN_CASE_ID_BYTES, MAX_PLUGIN_URL_BYTES,
16    },
17    recorder::{
18        evidence_value_bytes, redact_value, sanitize_error_safely, PluginObservation,
19        PluginRedactionPolicy, SecretRedactionPolicy,
20    },
21    transport::{
22        origin_string, validate_authorized_origin, validate_scoped_url, PluginHttpMethod,
23        PluginHttpRequest, PluginHttpResponse, PluginRequestBroker,
24    },
25    PluginError,
26};
27
28pub struct PluginExecutionRequest {
29    subject: EntityId,
30    authorized_origin: Url,
31    case_id: String,
32    input: Vec<u8>,
33    budget: PluginBudget,
34    cancellation: CancellationToken,
35    broker: Arc<dyn PluginRequestBroker>,
36    redaction: Arc<dyn PluginRedactionPolicy>,
37    reliability: ConfidenceScore,
38}
39
40impl PluginExecutionRequest {
41    /// Creates a request with finite defaults, empty input, no confidence, and
42    /// the default secret redaction policy.
43    pub fn new(
44        subject: EntityId,
45        authorized_origin: Url,
46        case_id: impl Into<String>,
47        broker: Arc<dyn PluginRequestBroker>,
48    ) -> Result<Self, PluginError> {
49        validate_authorized_origin(&authorized_origin)?;
50        let case_id = case_id.into();
51        validate_identifier(&case_id, "plugin case id", MAX_PLUGIN_CASE_ID_BYTES)?;
52        Ok(Self {
53            subject,
54            authorized_origin,
55            case_id,
56            input: Vec::new(),
57            budget: PluginBudget::default(),
58            cancellation: CancellationToken::new(),
59            broker,
60            redaction: Arc::new(SecretRedactionPolicy::default()),
61            reliability: ConfidenceScore::NONE,
62        })
63    }
64
65    /// Sets opaque bounded invocation input.
66    pub fn with_input(mut self, input: Vec<u8>) -> Result<Self, PluginError> {
67        ensure_input_budget(&input, &self.budget)?;
68        self.input = input;
69        Ok(self)
70    }
71
72    /// Replaces the immutable budget snapshot.
73    pub fn with_budget(mut self, budget: PluginBudget) -> Result<Self, PluginError> {
74        ensure_input_budget(&self.input, &budget)?;
75        self.budget = budget;
76        Ok(self)
77    }
78
79    /// Narrows response capture to a host execution allowance.
80    ///
81    /// This operation can only reduce both the per-response and cumulative
82    /// ceilings already selected by the request provider.
83    pub fn restrict_response_body_bytes(mut self, maximum: u64) -> Self {
84        self.budget.max_response_body_bytes = self.budget.max_response_body_bytes.min(maximum);
85        self.budget.max_cumulative_body_bytes = self.budget.max_cumulative_body_bytes.min(maximum);
86        self
87    }
88
89    /// Uses a host-owned cancellation token; the invocation receives a child.
90    pub fn with_cancellation(mut self, cancellation: CancellationToken) -> Self {
91        self.cancellation = cancellation;
92        self
93    }
94
95    /// Replaces the host redaction policy.
96    pub fn with_redaction(mut self, redaction: Arc<dyn PluginRedactionPolicy>) -> Self {
97        self.redaction = redaction;
98        self
99    }
100
101    /// Sets host-assessed source reliability without granting claim authority.
102    pub fn with_reliability(mut self, reliability: ConfidenceScore) -> Self {
103        self.reliability = reliability;
104        self
105    }
106
107    /// Authorized evidence subject selected by the host.
108    pub fn subject(&self) -> &EntityId {
109        &self.subject
110    }
111
112    /// Exact authorized HTTP(S) origin selected by the host.
113    pub fn authorized_origin(&self) -> &Url {
114        &self.authorized_origin
115    }
116
117    /// Host verification/correlation identity.
118    pub fn case_id(&self) -> &str {
119        &self.case_id
120    }
121
122    /// Opaque bounded input bytes.
123    pub fn input(&self) -> &[u8] {
124        &self.input
125    }
126}
127
128impl fmt::Debug for PluginExecutionRequest {
129    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
130        formatter
131            .debug_struct("PluginExecutionRequest")
132            .field("subject", &"[redacted]")
133            .field("authorized_origin", &origin_string(&self.authorized_origin))
134            .field("case_id", &"[redacted]")
135            .field("input_bytes", &self.input.len())
136            .field("budget", &self.budget)
137            .field("reliability", &self.reliability)
138            .finish_non_exhaustive()
139    }
140}
141
142/// Usage receipt for one completed plugin invocation.
143#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
144pub struct PluginUsage {
145    requests: u64,
146    response_body_bytes: u64,
147    observations: u64,
148    observation_bytes: u64,
149}
150
151impl PluginUsage {
152    /// Broker dispatch attempts charged to the invocation.
153    pub const fn requests(self) -> u64 {
154        self.requests
155    }
156
157    /// Delivered response bytes charged to the invocation.
158    pub const fn response_body_bytes(self) -> u64 {
159        self.response_body_bytes
160    }
161
162    /// Evidence observations retained at successful completion.
163    pub const fn observations(self) -> u64 {
164        self.observations
165    }
166
167    /// Bounded observation-value representation bytes charged by the host.
168    pub const fn observation_bytes(self) -> u64 {
169        self.observation_bytes
170    }
171}
172
173/// Successful execution receipt. Failures are returned as [`PluginError`].
174#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
175pub struct PluginExecutionResult {
176    pub(super) plugin_id: String,
177    pub(super) observations: Vec<Evidence>,
178    pub(super) usage: PluginUsage,
179    pub(super) elapsed_ms: u64,
180}
181
182impl PluginExecutionResult {
183    /// Registered plugin identity.
184    pub fn plugin_id(&self) -> &str {
185        &self.plugin_id
186    }
187
188    /// Host-normalized evidence observations.
189    pub fn observations(&self) -> &[Evidence] {
190        &self.observations
191    }
192
193    /// Consumes the receipt and returns normalized observations.
194    pub fn into_observations(self) -> Vec<Evidence> {
195        self.observations
196    }
197
198    /// Bounded usage receipt.
199    pub const fn usage(&self) -> PluginUsage {
200        self.usage
201    }
202
203    /// Host-observed elapsed milliseconds.
204    pub const fn elapsed_ms(&self) -> u64 {
205        self.elapsed_ms
206    }
207}
208
209struct PluginContextState {
210    sealed: bool,
211    failure: Option<PluginError>,
212    requests: u64,
213    response_body_bytes: u64,
214    reserved_response_body_bytes: u64,
215    observation_bytes: u64,
216    observations: Vec<Evidence>,
217}
218
219/// Borrowed capability boundary for one plugin invocation.
220///
221/// The type intentionally does not implement `Clone`: request and recorder
222/// authority remain structurally tied to the invocation future.
223pub struct PluginContext {
224    plugin_id: String,
225    subject: EntityId,
226    authorized_origin: Url,
227    case_id: String,
228    input: Vec<u8>,
229    budget: PluginBudget,
230    pub(super) cancellation: CancellationToken,
231    broker: Arc<dyn PluginRequestBroker>,
232    pub(super) redaction: Arc<dyn PluginRedactionPolicy>,
233    reliability: ConfidenceScore,
234    pub(super) deadline: tokio::time::Instant,
235    state: Mutex<PluginContextState>,
236}
237
238struct PluginRequestReservation<'a> {
239    context: &'a PluginContext,
240    capture_limit: u64,
241    cancellation: CancellationToken,
242    active: bool,
243}
244
245impl PluginRequestReservation<'_> {
246    fn commit(mut self, delivered_body_bytes: u64) -> Result<(), PluginError> {
247        {
248            let mut state = self.context.lock_state()?;
249            ensure_state_active(&state)?;
250            let Some(reserved) = state
251                .reserved_response_body_bytes
252                .checked_sub(self.capture_limit)
253            else {
254                state.failure = Some(PluginError::HostStateUnavailable);
255                return Err(PluginError::HostStateUnavailable);
256            };
257            if self.context.cancellation.is_cancelled() {
258                return Err(PluginError::Cancelled);
259            }
260            if tokio::time::Instant::now() >= self.context.deadline {
261                state.failure = Some(PluginError::WallTimeExceeded);
262                return Err(PluginError::WallTimeExceeded);
263            }
264            let Some(cumulative) = state.response_body_bytes.checked_add(delivered_body_bytes)
265            else {
266                let error = PluginError::CumulativeBodyBudgetExceeded;
267                state.failure = Some(error.clone());
268                return Err(error);
269            };
270            if cumulative > self.context.budget.max_cumulative_body_bytes {
271                let error = PluginError::CumulativeBodyBudgetExceeded;
272                state.failure = Some(error.clone());
273                return Err(error);
274            }
275            state.reserved_response_body_bytes = reserved;
276            state.response_body_bytes = cumulative;
277        }
278        self.active = false;
279        self.cancellation.cancel();
280        Ok(())
281    }
282}
283
284impl Drop for PluginRequestReservation<'_> {
285    fn drop(&mut self) {
286        self.cancellation.cancel();
287        if !self.active {
288            return;
289        }
290        if let Ok(mut state) = self.context.state.lock() {
291            match state
292                .reserved_response_body_bytes
293                .checked_sub(self.capture_limit)
294            {
295                Some(reserved) => state.reserved_response_body_bytes = reserved,
296                None => state.failure = Some(PluginError::HostStateUnavailable),
297            }
298            if !state.sealed && state.failure.is_none() {
299                state.failure = Some(PluginError::RequestAbandoned);
300            }
301        }
302    }
303}
304
305impl PluginContext {
306    pub(super) fn from_request(
307        plugin_id: String,
308        request: PluginExecutionRequest,
309    ) -> Result<Self, PluginError> {
310        ensure_input_budget(&request.input, &request.budget)?;
311        let now = tokio::time::Instant::now();
312        let deadline = now
313            .checked_add(request.budget.max_wall_time())
314            .ok_or_else(|| invalid_config("plugin wall budget exceeds runtime clock range"))?;
315        Ok(Self {
316            plugin_id,
317            subject: request.subject,
318            authorized_origin: request.authorized_origin,
319            case_id: request.case_id,
320            input: request.input,
321            budget: request.budget,
322            cancellation: request.cancellation.child_token(),
323            broker: request.broker,
324            redaction: request.redaction,
325            reliability: request.reliability,
326            deadline,
327            state: Mutex::new(PluginContextState {
328                sealed: false,
329                failure: None,
330                requests: 0,
331                response_body_bytes: 0,
332                reserved_response_body_bytes: 0,
333                observation_bytes: 0,
334                observations: Vec::new(),
335            }),
336        })
337    }
338
339    /// Authorized evidence subject.
340    pub fn subject(&self) -> &EntityId {
341        &self.subject
342    }
343
344    /// Exact authorized HTTP(S) origin.
345    pub fn authorized_origin(&self) -> &Url {
346        &self.authorized_origin
347    }
348
349    /// Host verification/correlation identity.
350    pub fn case_id(&self) -> &str {
351        &self.case_id
352    }
353
354    /// Opaque host input bounded before plugin code is polled.
355    pub fn input(&self) -> &[u8] {
356        &self.input
357    }
358
359    /// Immutable resource budget.
360    pub const fn budget(&self) -> &PluginBudget {
361        &self.budget
362    }
363
364    /// Returns whether the host has cancelled this invocation.
365    pub fn is_cancelled(&self) -> bool {
366        self.cancellation.is_cancelled()
367    }
368
369    /// Waits until the host cancels this invocation.
370    pub async fn cancelled(&self) {
371        self.cancellation.cancelled().await;
372    }
373
374    /// Records one observation; host provenance and redaction are mandatory.
375    pub fn record(&self, observation: PluginObservation) -> Result<(), PluginError> {
376        self.ensure_active()?;
377        let raw_bytes = evidence_value_bytes(&observation.value);
378        let redacted_value = match std::panic::catch_unwind(AssertUnwindSafe(|| {
379            redact_value(self.redaction.as_ref(), observation.value)
380        })) {
381            Ok(value) => value,
382            Err(_) => return self.fail(PluginError::HostCallbackPanicked),
383        };
384        let redacted_bytes = evidence_value_bytes(&redacted_value);
385        if raw_bytes > self.budget.max_observation_bytes
386            || redacted_bytes > self.budget.max_observation_bytes
387        {
388            return self.fail(PluginError::ObservationBytesBudgetExceeded);
389        }
390        let charged_bytes = raw_bytes.max(redacted_bytes);
391
392        let source = EvidenceSource::new(self.plugin_id.clone(), observation.method)
393            .and_then(|source| source.with_correlation_id(self.case_id.clone()))
394            .map_err(|_| invalid_config("plugin observation provenance is invalid"))?;
395        let evidence = Evidence::new(
396            self.subject.clone(),
397            observation.kind,
398            observation.predicate,
399            redacted_value,
400            source,
401            self.reliability,
402        );
403
404        let mut state = self.lock_state()?;
405        ensure_state_active(&state)?;
406        if state.observations.len() as u64 >= self.budget.max_observations {
407            let error = PluginError::ObservationBudgetExceeded;
408            state.failure = Some(error.clone());
409            return Err(error);
410        }
411        let Some(next_bytes) = state.observation_bytes.checked_add(charged_bytes) else {
412            let error = PluginError::ObservationBytesBudgetExceeded;
413            state.failure = Some(error.clone());
414            return Err(error);
415        };
416        if next_bytes > self.budget.max_observation_bytes {
417            let error = PluginError::ObservationBytesBudgetExceeded;
418            state.failure = Some(error.clone());
419            return Err(error);
420        }
421        state.observation_bytes = next_bytes;
422        state.observations.push(evidence);
423        Ok(())
424    }
425
426    /// Dispatches one bodyless request through the host-owned bounded broker.
427    pub async fn request(
428        &self,
429        method: PluginHttpMethod,
430        url: Url,
431    ) -> Result<PluginHttpResponse, PluginError> {
432        if validate_scoped_url(&self.authorized_origin, &url).is_err() {
433            return self.fail(PluginError::ScopeViolation);
434        }
435        if url.as_str().len() > MAX_PLUGIN_URL_BYTES {
436            return self.fail(PluginError::ScopeViolation);
437        }
438
439        let capture_limit = {
440            let mut state = self.lock_state()?;
441            ensure_state_active(&state)?;
442            if self.cancellation.is_cancelled() {
443                return Err(PluginError::Cancelled);
444            }
445            if tokio::time::Instant::now() >= self.deadline {
446                state.failure = Some(PluginError::WallTimeExceeded);
447                return Err(PluginError::WallTimeExceeded);
448            }
449            if state.requests >= self.budget.max_requests {
450                state.failure = Some(PluginError::RequestBudgetExceeded);
451                return Err(PluginError::RequestBudgetExceeded);
452            }
453            let Some(committed_and_reserved) = state
454                .response_body_bytes
455                .checked_add(state.reserved_response_body_bytes)
456            else {
457                state.failure = Some(PluginError::CumulativeBodyBudgetExceeded);
458                return Err(PluginError::CumulativeBodyBudgetExceeded);
459            };
460            let Some(remaining_cumulative) = self
461                .budget
462                .max_cumulative_body_bytes
463                .checked_sub(committed_and_reserved)
464            else {
465                state.failure = Some(PluginError::CumulativeBodyBudgetExceeded);
466                return Err(PluginError::CumulativeBodyBudgetExceeded);
467            };
468            if remaining_cumulative == 0 {
469                state.failure = Some(PluginError::CumulativeBodyBudgetExceeded);
470                return Err(PluginError::CumulativeBodyBudgetExceeded);
471            }
472            if self.budget.max_response_body_bytes == 0 {
473                let error = PluginError::ResponseBodyBudgetUnavailable;
474                state.failure = Some(error.clone());
475                return Err(error);
476            }
477            let capture_limit = self
478                .budget
479                .max_response_body_bytes
480                .min(remaining_cumulative);
481            state.requests += 1;
482            let Some(reserved) = state
483                .reserved_response_body_bytes
484                .checked_add(capture_limit)
485            else {
486                state.failure = Some(PluginError::CumulativeBodyBudgetExceeded);
487                return Err(PluginError::CumulativeBodyBudgetExceeded);
488            };
489            state.reserved_response_body_bytes = reserved;
490            capture_limit
491        };
492
493        let request_cancellation = self.cancellation.child_token();
494        let reservation = PluginRequestReservation {
495            context: self,
496            capture_limit,
497            cancellation: request_cancellation.clone(),
498            active: true,
499        };
500        let request_timeout = self.budget.request_timeout();
501        if request_timeout.is_zero() {
502            return self.fail(PluginError::RequestTimeout);
503        }
504        let remaining = self
505            .deadline
506            .checked_duration_since(tokio::time::Instant::now())
507            .unwrap_or(Duration::ZERO);
508        if remaining.is_zero() {
509            return self.fail(PluginError::WallTimeExceeded);
510        }
511        let timeout = request_timeout.min(remaining);
512        let broker_request = PluginHttpRequest {
513            method,
514            url,
515            max_response_body_bytes: capture_limit,
516            cancellation: request_cancellation.clone(),
517        };
518        let broker = self.broker.execute(broker_request);
519        tokio::pin!(broker);
520        let sleep = tokio::time::sleep(timeout);
521        tokio::pin!(sleep);
522
523        let response = tokio::select! {
524            biased;
525            () = self.cancellation.cancelled() => {
526                request_cancellation.cancel();
527                return self.fail(PluginError::Cancelled);
528            }
529            () = &mut sleep => {
530                request_cancellation.cancel();
531                let error = if tokio::time::Instant::now() >= self.deadline {
532                    PluginError::WallTimeExceeded
533                } else {
534                    PluginError::RequestTimeout
535                };
536                return self.fail(error);
537            }
538            result = &mut broker => match result {
539                Ok(response) => response,
540                Err(error) => {
541                    let error = sanitize_error_safely(self.redaction.as_ref(), error)
542                        .unwrap_or(PluginError::HostCallbackPanicked);
543                    return self.fail(error);
544                },
545            },
546        };
547
548        if validate_scoped_url(&self.authorized_origin, response.final_url()).is_err()
549            || response.final_url().as_str().len() > MAX_PLUGIN_URL_BYTES
550        {
551            return self.fail(PluginError::ScopeViolation);
552        }
553        if response.delivered_body_bytes > capture_limit {
554            return self.fail(PluginError::ResponseBodyBudgetExceeded {
555                actual: response.delivered_body_bytes,
556                maximum: capture_limit,
557            });
558        }
559
560        reservation.commit(response.delivered_body_bytes)?;
561        Ok(response)
562    }
563
564    pub(super) fn ensure_active(&self) -> Result<(), PluginError> {
565        if self.cancellation.is_cancelled() {
566            return Err(PluginError::Cancelled);
567        }
568        if tokio::time::Instant::now() >= self.deadline {
569            return self.fail(PluginError::WallTimeExceeded);
570        }
571        let state = self.lock_state()?;
572        ensure_state_active(&state)
573    }
574
575    fn fail<T>(&self, error: PluginError) -> Result<T, PluginError> {
576        if let Ok(mut state) = self.state.lock() {
577            if state.failure.is_none() {
578                state.failure = Some(error.clone());
579            }
580        }
581        Err(error)
582    }
583
584    pub(super) fn discard(&self) {
585        self.cancellation.cancel();
586        if let Ok(mut state) = self.state.lock() {
587            state.sealed = true;
588            state.observations.clear();
589        }
590    }
591
592    pub(super) fn finish(&self) -> Result<(Vec<Evidence>, PluginUsage), PluginError> {
593        let mut state = self.lock_state()?;
594        ensure_state_active(&state)?;
595        if self.cancellation.is_cancelled() {
596            return Err(PluginError::Cancelled);
597        }
598        if tokio::time::Instant::now() >= self.deadline {
599            return Err(PluginError::WallTimeExceeded);
600        }
601        if state.reserved_response_body_bytes != 0 {
602            return Err(PluginError::RequestAbandoned);
603        }
604        state.sealed = true;
605        let observations = std::mem::take(&mut state.observations);
606        let usage = PluginUsage {
607            requests: state.requests,
608            response_body_bytes: state.response_body_bytes,
609            observations: observations.len() as u64,
610            observation_bytes: state.observation_bytes,
611        };
612        Ok((observations, usage))
613    }
614
615    fn lock_state(&self) -> Result<MutexGuard<'_, PluginContextState>, PluginError> {
616        self.state
617            .lock()
618            .map_err(|_| PluginError::HostStateUnavailable)
619    }
620}
621
622impl fmt::Debug for PluginContext {
623    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
624        let usage = self.state.lock().ok().map(|state| PluginUsage {
625            requests: state.requests,
626            response_body_bytes: state.response_body_bytes,
627            observations: state.observations.len() as u64,
628            observation_bytes: state.observation_bytes,
629        });
630        formatter
631            .debug_struct("PluginContext")
632            .field("plugin_id", &self.plugin_id)
633            .field("subject", &"[redacted]")
634            .field("authorized_origin", &origin_string(&self.authorized_origin))
635            .field("case_id", &"[redacted]")
636            .field("input_bytes", &self.input.len())
637            .field("budget", &self.budget)
638            .field("usage", &usage)
639            .finish_non_exhaustive()
640    }
641}
642
643fn ensure_state_active(state: &PluginContextState) -> Result<(), PluginError> {
644    if state.sealed {
645        return Err(PluginError::ContextSealed);
646    }
647    if let Some(error) = &state.failure {
648        return Err(error.clone());
649    }
650    Ok(())
651}
termivar_scanner/plugin/context.rs

termivar_scanner/plugin/
context.rs
