termivar_scanner/
realtime.rs

1//! Recorded event and subscription data models.
2//!
3//! ## Runtime scope
4//!
5//! - **Build:** opt-in via `platform-models`.
6//! - **Execution:** no repository runtime caller (not on the default scan path).
7//! - **Default `termivar scan`:** no.
8//! - **Support:** experimental in-memory models.
9//!
10//! `EventStream` is an in-process event journal. This module has no WebSocket
11//! listener, connection manager, subscriber delivery, or network broadcast.
12
13use dashmap::DashMap;
14use serde::{Deserialize, Deserializer, Serialize, Serializer};
15use std::sync::Arc;
16
17fn validate_bounded_f32(value: f32, minimum: f32, maximum: f32) -> bool {
18    value.is_finite() && (minimum..=maximum).contains(&value)
19}
20
21fn serialize_bounded_f32<S>(
22    value: &f32,
23    minimum: f32,
24    maximum: f32,
25    field: &'static str,
26    serializer: S,
27) -> Result<S::Ok, S::Error>
28where
29    S: Serializer,
30{
31    if !validate_bounded_f32(*value, minimum, maximum) {
32        return Err(serde::ser::Error::custom(format_args!(
33            "{field} must be finite and in {minimum}..={maximum}"
34        )));
35    }
36    serializer.serialize_f32(*value)
37}
38
39fn deserialize_bounded_f32<'de, D>(
40    deserializer: D,
41    minimum: f32,
42    maximum: f32,
43    field: &'static str,
44) -> Result<f32, D::Error>
45where
46    D: Deserializer<'de>,
47{
48    let value = f32::deserialize(deserializer)?;
49    if !validate_bounded_f32(value, minimum, maximum) {
50        return Err(serde::de::Error::custom(format_args!(
51            "{field} must be finite and in {minimum}..={maximum}"
52        )));
53    }
54    Ok(value)
55}
56
57mod percentage {
58    use super::*;
59
60    pub fn serialize<S>(value: &f32, serializer: S) -> Result<S::Ok, S::Error>
61    where
62        S: Serializer,
63    {
64        serialize_bounded_f32(value, 0.0, 100.0, "percentage", serializer)
65    }
66
67    pub fn deserialize<'de, D>(deserializer: D) -> Result<f32, D::Error>
68    where
69        D: Deserializer<'de>,
70    {
71        deserialize_bounded_f32(deserializer, 0.0, 100.0, "percentage")
72    }
73}
74
75mod normalized_score {
76    use super::*;
77
78    pub fn serialize<S>(value: &f32, serializer: S) -> Result<S::Ok, S::Error>
79    where
80        S: Serializer,
81    {
82        serialize_bounded_f32(value, 0.0, 1.0, "normalized score", serializer)
83    }
84
85    pub fn deserialize<'de, D>(deserializer: D) -> Result<f32, D::Error>
86    where
87        D: Deserializer<'de>,
88    {
89        deserialize_bounded_f32(deserializer, 0.0, 1.0, "normalized score")
90    }
91}
92
93mod optional_percentage {
94    use super::*;
95
96    pub fn serialize<S>(value: &Option<f32>, serializer: S) -> Result<S::Ok, S::Error>
97    where
98        S: Serializer,
99    {
100        if let Some(value) = value {
101            if !validate_bounded_f32(*value, 0.0, 100.0) {
102                return Err(serde::ser::Error::custom(
103                    "percentage must be finite and in 0..=100",
104                ));
105            }
106        }
107        value.serialize(serializer)
108    }
109
110    pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<f32>, D::Error>
111    where
112        D: Deserializer<'de>,
113    {
114        let value = Option::<f32>::deserialize(deserializer)?;
115        if let Some(value) = value {
116            if !validate_bounded_f32(value, 0.0, 100.0) {
117                return Err(serde::de::Error::custom(
118                    "percentage must be finite and in 0..=100",
119                ));
120            }
121        }
122        Ok(value)
123    }
124}
125
126/// Caller-supplied scan event record.
127#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
128#[serde(tag = "type", rename_all = "lowercase")]
129pub enum RealtimeEvent {
130    ScanStarted {
131        scan_id: String,
132        target: String,
133        timestamp: u64,
134    },
135    PhaseStarted {
136        scan_id: String,
137        phase: u8,
138        phase_name: String,
139        timestamp: u64,
140    },
141    PhaseProgress {
142        scan_id: String,
143        phase: u8,
144        #[serde(with = "percentage")]
145        progress: f32,
146        timestamp: u64,
147    },
148    FindingDiscovered {
149        scan_id: String,
150        phase: u8,
151        severity: String,
152        description: String,
153        timestamp: u64,
154    },
155    PhaseCompleted {
156        scan_id: String,
157        phase: u8,
158        findings_count: u64,
159        duration_ms: u64,
160        timestamp: u64,
161    },
162    ScanCompleted {
163        scan_id: String,
164        total_findings: u64,
165        #[serde(with = "normalized_score")]
166        risk_score: f32,
167        duration_ms: u64,
168        timestamp: u64,
169    },
170    Error {
171        scan_id: String,
172        error_message: String,
173        timestamp: u64,
174    },
175    Metrics {
176        scan_id: String,
177        requests: u64,
178        responses: u64,
179        findings: u64,
180        errors: u64,
181        #[serde(with = "optional_percentage")]
182        response_request_ratio_percent: Option<f32>,
183        timestamp: u64,
184    },
185}
186
187#[derive(Debug, Clone, Copy, PartialEq, Eq)]
188pub enum RealtimeEventValidationError {
189    EmptyScanId,
190    InvalidPhaseProgress,
191    InvalidRiskScore,
192    InvalidResponseRequestRatioPercent,
193}
194
195impl RealtimeEvent {
196    pub fn timestamp(&self) -> u64 {
197        match self {
198            Self::ScanStarted { timestamp, .. }
199            | Self::PhaseStarted { timestamp, .. }
200            | Self::PhaseProgress { timestamp, .. }
201            | Self::FindingDiscovered { timestamp, .. }
202            | Self::PhaseCompleted { timestamp, .. }
203            | Self::ScanCompleted { timestamp, .. }
204            | Self::Error { timestamp, .. }
205            | Self::Metrics { timestamp, .. } => *timestamp,
206        }
207    }
208
209    pub fn scan_id(&self) -> &str {
210        match self {
211            Self::ScanStarted { scan_id, .. }
212            | Self::PhaseStarted { scan_id, .. }
213            | Self::PhaseProgress { scan_id, .. }
214            | Self::FindingDiscovered { scan_id, .. }
215            | Self::PhaseCompleted { scan_id, .. }
216            | Self::ScanCompleted { scan_id, .. }
217            | Self::Error { scan_id, .. }
218            | Self::Metrics { scan_id, .. } => scan_id,
219        }
220    }
221
222    pub fn validate(&self) -> Result<(), RealtimeEventValidationError> {
223        if self.scan_id().is_empty() {
224            return Err(RealtimeEventValidationError::EmptyScanId);
225        }
226        match self {
227            Self::PhaseProgress { progress, .. }
228                if !validate_bounded_f32(*progress, 0.0, 100.0) =>
229            {
230                Err(RealtimeEventValidationError::InvalidPhaseProgress)
231            },
232            Self::ScanCompleted { risk_score, .. }
233                if !validate_bounded_f32(*risk_score, 0.0, 1.0) =>
234            {
235                Err(RealtimeEventValidationError::InvalidRiskScore)
236            },
237            Self::Metrics {
238                response_request_ratio_percent: Some(ratio),
239                ..
240            } if !validate_bounded_f32(*ratio, 0.0, 100.0) => {
241                Err(RealtimeEventValidationError::InvalidResponseRequestRatioPercent)
242            },
243            _ => Ok(()),
244        }
245    }
246}
247
248/// Cloneable, in-memory journal keyed by scan ID.
249#[derive(Debug, Clone, Default)]
250pub struct EventStream {
251    events: Arc<DashMap<String, Vec<RealtimeEvent>>>,
252}
253
254impl EventStream {
255    pub fn new() -> Self {
256        Self::default()
257    }
258
259    /// Records a valid event in memory. It does not deliver the event to a client.
260    ///
261    /// Invalid float ranges and empty scan IDs are rejected without mutating the journal.
262    pub fn record(&self, event: RealtimeEvent) -> Result<(), RealtimeEventValidationError> {
263        event.validate()?;
264        self.events
265            .entry(event.scan_id().to_string())
266            .or_default()
267            .push(event);
268        Ok(())
269    }
270
271    pub fn events(&self, scan_id: &str) -> Vec<RealtimeEvent> {
272        self.events
273            .get(scan_id)
274            .map(|events| events.clone())
275            .unwrap_or_default()
276    }
277
278    pub fn events_since(&self, scan_id: &str, since_exclusive: u64) -> Vec<RealtimeEvent> {
279        self.events
280            .get(scan_id)
281            .map(|events| {
282                events
283                    .iter()
284                    .filter(|event| event.timestamp() > since_exclusive)
285                    .cloned()
286                    .collect()
287            })
288            .unwrap_or_default()
289    }
290
291    pub fn recent_events(&self, scan_id: &str, limit: usize) -> Vec<RealtimeEvent> {
292        self.events
293            .get(scan_id)
294            .map(|events| {
295                let start = events.len().saturating_sub(limit);
296                events[start..].to_vec()
297            })
298            .unwrap_or_default()
299    }
300
301    /// Removes an in-memory journal and returns the number of removed events.
302    pub fn clear_events(&self, scan_id: &str) -> usize {
303        self.events
304            .remove(scan_id)
305            .map(|(_, events)| events.len())
306            .unwrap_or_default()
307    }
308
309    pub fn event_count(&self, scan_id: &str) -> usize {
310        self.events
311            .get(scan_id)
312            .map(|events| events.len())
313            .unwrap_or_default()
314    }
315}
316
317/// Caller-owned record of subscription state; no network connection is implied.
318#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
319pub struct Subscription {
320    pub subscriber_id: String,
321    pub scan_id: String,
322    pub subscribed_at: u64,
323    pub ended_at: Option<u64>,
324}
325
326impl Subscription {
327    pub fn new(subscriber_id: String, scan_id: String, subscribed_at: u64) -> Self {
328        Self {
329            subscriber_id,
330            scan_id,
331            subscribed_at,
332            ended_at: None,
333        }
334    }
335
336    pub fn is_active(&self) -> bool {
337        self.ended_at.is_none()
338    }
339
340    /// Records the end of this subscription.
341    ///
342    /// Returns `false` for a repeated end or a timestamp before subscription.
343    pub fn end(&mut self, ended_at: u64) -> bool {
344        if self.ended_at.is_some() || ended_at < self.subscribed_at {
345            return false;
346        }
347        self.ended_at = Some(ended_at);
348        true
349    }
350}
351
352#[cfg(test)]
353mod tests {
354    use super::*;
355
356    fn event(timestamp: u64) -> RealtimeEvent {
357        RealtimeEvent::PhaseProgress {
358            scan_id: "scan-1".to_string(),
359            phase: 1,
360            progress: 50.0,
361            timestamp,
362        }
363    }
364
365    #[test]
366    fn scan_started_exposes_identity_and_validates() {
367        let event = RealtimeEvent::ScanStarted {
368            scan_id: "scan-1".into(),
369            target: "https://example.test".into(),
370            timestamp: 42,
371        };
372
373        assert_eq!(event.scan_id(), "scan-1");
374        assert_eq!(event.timestamp(), 42);
375        assert_eq!(event.validate(), Ok(()));
376    }
377
378    #[test]
379    fn journal_records_without_claiming_delivery() {
380        let stream = EventStream::new();
381        stream.record(event(10)).unwrap();
382        stream.record(event(20)).unwrap();
383
384        assert_eq!(stream.event_count("scan-1"), 2);
385        assert_eq!(stream.events_since("scan-1", 10), vec![event(20)]);
386        assert!(stream.events("unknown").is_empty());
387    }
388
389    #[test]
390    fn recent_limit_zero_is_empty() {
391        let stream = EventStream::new();
392        stream.record(event(10)).unwrap();
393        assert!(stream.recent_events("scan-1", 0).is_empty());
394    }
395
396    #[test]
397    fn clear_reports_what_was_actually_removed() {
398        let stream = EventStream::new();
399        stream.record(event(10)).unwrap();
400
401        assert_eq!(stream.clear_events("scan-1"), 1);
402        assert_eq!(stream.clear_events("scan-1"), 0);
403    }
404
405    #[test]
406    fn subscription_end_is_truthful() {
407        let mut subscription = Subscription::new("observer".into(), "scan-1".into(), 100);
408        assert!(subscription.is_active());
409        assert!(!subscription.end(99));
410        assert!(subscription.is_active());
411        assert!(subscription.end(110));
412        assert!(!subscription.is_active());
413        assert!(!subscription.end(120));
414        assert_eq!(subscription.ended_at, Some(110));
415    }
416
417    #[test]
418    fn record_rejects_invalid_events_without_mutating_the_journal() {
419        let stream = EventStream::new();
420        let invalid = RealtimeEvent::PhaseProgress {
421            scan_id: "scan-1".into(),
422            phase: 1,
423            progress: f32::NAN,
424            timestamp: 10,
425        };
426
427        assert_eq!(
428            stream.record(invalid),
429            Err(RealtimeEventValidationError::InvalidPhaseProgress)
430        );
431        assert!(stream.events("scan-1").is_empty());
432    }
433
434    #[test]
435    fn valid_events_round_trip_and_wire_counts_are_u64() {
436        let event = RealtimeEvent::ScanCompleted {
437            scan_id: "scan-1".into(),
438            total_findings: u64::MAX,
439            risk_score: 0.75,
440            duration_ms: 25,
441            timestamp: 100,
442        };
443        let json = serde_json::to_string(&event).unwrap();
444        assert!(json.contains(&u64::MAX.to_string()));
445        assert_eq!(serde_json::from_str::<RealtimeEvent>(&json).unwrap(), event);
446    }
447
448    #[test]
449    fn serialization_and_deserialization_reject_invalid_floats() {
450        let invalid_progress = RealtimeEvent::PhaseProgress {
451            scan_id: "scan-1".into(),
452            phase: 1,
453            progress: -0.1,
454            timestamp: 1,
455        };
456        assert!(serde_json::to_string(&invalid_progress).is_err());
457
458        let invalid = RealtimeEvent::ScanCompleted {
459            scan_id: "scan-1".into(),
460            total_findings: 0,
461            risk_score: f32::INFINITY,
462            duration_ms: 25,
463            timestamp: 100,
464        };
465        assert!(serde_json::to_string(&invalid).is_err());
466
467        let invalid_score = r#"{"type":"scancompleted","scan_id":"scan-1","total_findings":0,"risk_score":1.1,"duration_ms":25,"timestamp":100}"#;
468        assert!(serde_json::from_str::<RealtimeEvent>(invalid_score).is_err());
469
470        let out_of_range = r#"{"type":"phaseprogress","scan_id":"scan-1","phase":1,"progress":100.1,"timestamp":1}"#;
471        assert!(serde_json::from_str::<RealtimeEvent>(out_of_range).is_err());
472
473        let invalid_ratio = RealtimeEvent::Metrics {
474            scan_id: "scan-1".into(),
475            requests: 1,
476            responses: 1,
477            findings: 0,
478            errors: 0,
479            response_request_ratio_percent: Some(-0.1),
480            timestamp: 1,
481        };
482        assert!(serde_json::to_string(&invalid_ratio).is_err());
483
484        let out_of_range_ratio = r#"{"type":"metrics","scan_id":"scan-1","requests":1,"responses":1,"findings":0,"errors":0,"response_request_ratio_percent":101.0,"timestamp":1}"#;
485        assert!(serde_json::from_str::<RealtimeEvent>(out_of_range_ratio).is_err());
486    }
487}
termivar_scanner/realtime.rs

termivar_scanner/
realtime.rs
