sipx_transport/
policy.rs

1//! Bounded endpoint observation and the two narrow policy seams.
2
3use std::net::IpAddr;
4use std::sync::{Arc, Mutex};
5
6use sipx_sip::{Header, Message, Request};
7use tokio::sync::mpsc;
8
9use crate::counters::Meters;
10use crate::{ConnectionKey, Target, TransportKind};
11
12/// One exact IP network used by live source admission.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
14pub struct SourcePrefix {
15    network: IpAddr,
16    bits: u8,
17}
18
19impl SourcePrefix {
20    /// Build a prefix, rejecting a width outside its address family.
21    #[must_use]
22    pub fn new(network: IpAddr, bits: u8) -> Option<Self> {
23        let max = if network.is_ipv4() { 32 } else { 128 };
24        (bits <= max).then_some(Self { network, bits })
25    }
26
27    /// One address, expressed as its narrowest prefix.
28    #[must_use]
29    pub const fn address(address: IpAddr) -> Self {
30        let bits = if address.is_ipv4() { 32 } else { 128 };
31        Self {
32            network: address,
33            bits,
34        }
35    }
36
37    /// Whether `candidate` belongs to this network.
38    #[must_use]
39    pub fn contains(self, candidate: IpAddr) -> bool {
40        match (self.network, candidate) {
41            (IpAddr::V4(network), IpAddr::V4(candidate)) => prefix_matches(
42                u32::from(network).into(),
43                u32::from(candidate).into(),
44                self.bits,
45                32,
46            ),
47            (IpAddr::V6(network), IpAddr::V6(candidate)) => {
48                prefix_matches(u128::from(network), u128::from(candidate), self.bits, 128)
49            }
50            _ => false,
51        }
52    }
53}
54
55fn prefix_matches(network: u128, candidate: u128, bits: u8, width: u8) -> bool {
56    if bits == 0 {
57        return true;
58    }
59    let shift = u32::from(width.saturating_sub(bits));
60    (network >> shift) == (candidate >> shift)
61}
62
63#[derive(Debug, Clone)]
64struct AdmissionGeneration {
65    number: u64,
66    prefixes: Option<Arc<[SourcePrefix]>>,
67}
68
69/// Atomic publication point for source-admission generations.
70#[derive(Debug)]
71pub(crate) struct SourceAdmission {
72    current: Mutex<AdmissionGeneration>,
73    limit: usize,
74}
75
76impl Default for SourceAdmission {
77    fn default() -> Self {
78        Self::new(1024)
79    }
80}
81
82impl SourceAdmission {
83    pub(crate) fn new(limit: usize) -> Self {
84        Self {
85            current: Mutex::new(AdmissionGeneration {
86                number: 0,
87                prefixes: None,
88            }),
89            limit,
90        }
91    }
92    /// Admit an address and return the immutable generation that made the decision.
93    pub(crate) fn admit(&self, address: IpAddr) -> Option<u64> {
94        let current = self
95            .current
96            .lock()
97            .unwrap_or_else(std::sync::PoisonError::into_inner);
98        let allowed = current
99            .prefixes
100            .as_ref()
101            .is_none_or(|prefixes| prefixes.iter().any(|prefix| prefix.contains(address)));
102        allowed.then_some(current.number)
103    }
104
105    pub(crate) fn replace(&self, prefixes: Vec<SourcePrefix>) -> crate::Result<u64> {
106        if prefixes.len() > self.limit {
107            return Err(crate::Error::SourceAdmissionCapacity {
108                max: self.limit,
109                attempted: prefixes.len(),
110            });
111        }
112        Ok(self.publish(Some(prefixes.into())))
113    }
114
115    pub(crate) fn clear(&self) -> u64 {
116        self.publish(None)
117    }
118
119    fn publish(&self, prefixes: Option<Arc<[SourcePrefix]>>) -> u64 {
120        let mut current = self
121            .current
122            .lock()
123            .unwrap_or_else(std::sync::PoisonError::into_inner);
124        current.number = current.number.wrapping_add(1).max(1);
125        current.prefixes = prefixes;
126        current.number
127    }
128}
129
130/// What a pre-transaction request policy decided.
131#[derive(Debug)]
132pub enum RequestPolicyDecision {
133    /// Continue without adding fields.
134    Allow,
135    /// Refuse the send before a transaction exists.
136    Reject(String),
137    /// Continue after appending application-owned headers.
138    AddHeaders(Vec<Header>),
139}
140
141/// An immutable pre-transaction request policy.
142pub trait RequestPolicy: Send + Sync {
143    /// Decide one send. The request and target cannot be mutated or replaced.
144    fn decide(&self, request: &Request, target: &Target) -> RequestPolicyDecision;
145}
146
147/// Cloneable configured request policy.
148#[derive(Clone)]
149pub struct RequestPolicyRef(Arc<dyn RequestPolicy>);
150
151impl RequestPolicyRef {
152    /// Wrap a request policy for [`crate::Config`].
153    #[must_use]
154    pub fn new(policy: impl RequestPolicy + 'static) -> Self {
155        Self(Arc::new(policy))
156    }
157
158    pub(crate) fn decide(&self, request: &Request, target: &Target) -> RequestPolicyDecision {
159        self.0.decide(request, target)
160    }
161}
162
163impl std::fmt::Debug for RequestPolicyRef {
164    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
165        formatter.write_str("RequestPolicyRef(..)")
166    }
167}
168
169/// Which side of the endpoint boundary a message crossed.
170#[derive(Debug, Clone, Copy, PartialEq, Eq)]
171pub enum MessageDirection {
172    /// Parsed network input.
173    Inbound,
174    /// Finalized network output.
175    Outbound,
176}
177
178/// How the transaction layer classified an observed message.
179#[derive(Debug, Clone, Copy, PartialEq, Eq)]
180pub enum TransactionClass {
181    /// A new server transaction.
182    ServerCreated,
183    /// A message matched an existing transaction.
184    Matched,
185    /// No transaction matched.
186    Unmatched,
187    /// A new client transaction.
188    ClientCreated,
189    /// A request deliberately sent without a transaction.
190    Direct,
191}
192
193/// A parsed inbound or finalized outbound SIP message.
194#[derive(Debug, Clone)]
195pub struct MessageObservation {
196    /// Immutable message snapshot.
197    pub message: Message,
198    /// Endpoint address.
199    pub local: std::net::SocketAddr,
200    /// Remote address.
201    pub peer: std::net::SocketAddr,
202    /// Wire transport.
203    pub transport: TransportKind,
204    /// Boundary direction.
205    pub direction: MessageDirection,
206    /// Transaction-layer classification.
207    pub transaction: TransactionClass,
208}
209
210/// Stable identity of one pooled connection incarnation.
211#[derive(Debug, Clone, PartialEq, Eq, Hash)]
212pub struct ConnectionId {
213    /// Pool key, including verified identity and WebSocket resource when present.
214    pub key: ConnectionKey,
215    /// Pool incarnation; replacement receives another value.
216    pub generation: u64,
217    /// Source-admission generation retained by an inbound connection.
218    pub admission_generation: Option<u64>,
219}
220
221/// A connection lifecycle transition.
222#[derive(Debug, Clone, Copy, PartialEq, Eq)]
223pub enum ConnectionState {
224    /// An inbound socket was accepted into the pool.
225    Accepted,
226    /// An outbound or inbound stream became usable.
227    Opened,
228    /// A secure transport authenticated its peer.
229    Authenticated,
230    /// The connection joined the bounded pool.
231    Pooled,
232    /// Existing pooled transport carried another send.
233    Reused,
234    /// Opening, authentication or framing failed.
235    Failed,
236    /// The incarnation closed.
237    Closed,
238}
239
240/// One connection transition.
241#[derive(Debug, Clone)]
242pub struct ConnectionObservation {
243    /// Stable connection incarnation.
244    pub connection: ConnectionId,
245    /// Transition.
246    pub state: ConnectionState,
247}
248
249/// One item on the bounded endpoint observation stream.
250#[derive(Debug, Clone)]
251pub enum EndpointObservation {
252    /// A parsed or finalized SIP message.
253    Message(Box<MessageObservation>),
254    /// A connection lifecycle transition.
255    Connection(ConnectionObservation),
256}
257
258/// Shared non-blocking fan-in for every endpoint task.
259#[derive(Debug)]
260pub(crate) struct ObservationHub {
261    sink: Mutex<Option<mpsc::Sender<EndpointObservation>>>,
262    meters: Arc<Meters>,
263}
264
265impl ObservationHub {
266    pub(crate) fn new(meters: Arc<Meters>) -> Self {
267        Self {
268            sink: Mutex::new(None),
269            meters,
270        }
271    }
272
273    pub(crate) fn subscribe(&self, capacity: usize) -> mpsc::Receiver<EndpointObservation> {
274        let (sender, receiver) = mpsc::channel(capacity.max(1));
275        *self
276            .sink
277            .lock()
278            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(sender);
279        receiver
280    }
281
282    pub(crate) fn emit(&self, event: EndpointObservation) {
283        let mut sink = self
284            .sink
285            .lock()
286            .unwrap_or_else(std::sync::PoisonError::into_inner);
287        let Some(sender) = sink.as_ref() else {
288            return;
289        };
290        match sender.try_send(event) {
291            Ok(()) => {}
292            Err(mpsc::error::TrySendError::Full(_)) => self.meters.observation_drop(),
293            Err(mpsc::error::TrySendError::Closed(_)) => *sink = None,
294        }
295    }
296}
297
298pub(crate) fn connection_event(
299    key: ConnectionKey,
300    generation: u64,
301    admission_generation: Option<u64>,
302    state: ConnectionState,
303) -> EndpointObservation {
304    EndpointObservation::Connection(ConnectionObservation {
305        connection: ConnectionId {
306            key,
307            generation,
308            admission_generation,
309        },
310        state,
311    })
312}
313
314/// Resolve deliberate `Other` construction before deciding whether policy may append the field.
315pub(crate) fn policy_header(name: &sipx_sip::HeaderName) -> (sipx_sip::HeaderName, bool) {
316    use sipx_sip::HeaderName;
317    let semantic = HeaderName::parse(&bytes::Bytes::copy_from_slice(name.canonical()));
318    let allowed = matches!(
319        semantic,
320        HeaderName::AlertInfo
321            | HeaderName::CallInfo
322            | HeaderName::Organization
323            | HeaderName::Priority
324            | HeaderName::Subject
325            | HeaderName::UserAgent
326            | HeaderName::Other(_)
327    );
328    (semantic, allowed)
329}
330
331pub(crate) fn duplicate_policy_header(request: &Request, semantic: &sipx_sip::HeaderName) -> bool {
332    !matches!(semantic, sipx_sip::HeaderName::Other(_)) && request.headers.get(semantic).is_some()
333}
334
335#[cfg(test)]
336#[allow(
337    clippy::unwrap_used,
338    clippy::expect_used,
339    clippy::panic,
340    clippy::indexing_slicing
341)]
342mod tests {
343    use super::*;
344    use bytes::Bytes;
345    use sipx_sip::HeaderName;
346    use std::net::{Ipv4Addr, Ipv6Addr};
347
348    #[test]
349    fn prefixes_match_only_their_network_and_family() {
350        let v4 = SourcePrefix::new(Ipv4Addr::new(192, 0, 2, 0).into(), 24).unwrap();
351        assert!(v4.contains(Ipv4Addr::new(192, 0, 2, 42).into()));
352        assert!(!v4.contains(Ipv4Addr::new(192, 0, 3, 1).into()));
353        assert!(!v4.contains(Ipv6Addr::LOCALHOST.into()));
354    }
355
356    #[test]
357    fn replacement_publishes_complete_generations() {
358        let admission = SourceAdmission::default();
359        assert_eq!(admission.admit(IpAddr::V4(Ipv4Addr::LOCALHOST)), Some(0));
360        let one = admission
361            .replace(vec![SourcePrefix::address(IpAddr::V4(Ipv4Addr::LOCALHOST))])
362            .unwrap();
363        assert_eq!(admission.admit(IpAddr::V4(Ipv4Addr::LOCALHOST)), Some(one));
364        assert_eq!(admission.admit(IpAddr::V4(Ipv4Addr::UNSPECIFIED)), None);
365        let two = admission.clear();
366        assert!(two > one);
367        assert_eq!(
368            admission.admit(IpAddr::V4(Ipv4Addr::UNSPECIFIED)),
369            Some(two)
370        );
371    }
372
373    #[test]
374    fn oversized_replacement_preserves_the_old_generation() {
375        let admission = SourceAdmission::new(1);
376        let first = admission
377            .replace(vec![SourcePrefix::address(IpAddr::V4(Ipv4Addr::LOCALHOST))])
378            .unwrap();
379        let error = admission
380            .replace(vec![
381                SourcePrefix::address(IpAddr::V4(Ipv4Addr::LOCALHOST)),
382                SourcePrefix::address(IpAddr::V4(Ipv4Addr::UNSPECIFIED)),
383            ])
384            .unwrap_err();
385        assert!(matches!(
386            error,
387            crate::Error::SourceAdmissionCapacity { .. }
388        ));
389        assert_eq!(
390            admission.admit(IpAddr::V4(Ipv4Addr::LOCALHOST)),
391            Some(first)
392        );
393        assert_eq!(admission.admit(IpAddr::V4(Ipv4Addr::UNSPECIFIED)), None);
394    }
395
396    #[test]
397    fn request_policy_allows_only_application_fields_and_unknown_extensions() {
398        for name in [HeaderName::Subject, HeaderName::Organization] {
399            assert!(policy_header(&name).1);
400        }
401        assert!(policy_header(&HeaderName::Other(Bytes::from_static(b"X-Trace"))).1);
402        for name in [
403            HeaderName::Contact,
404            HeaderName::ContentType,
405            HeaderName::Event,
406        ] {
407            assert!(!policy_header(&name).1);
408        }
409        for raw in [b"vIa".as_slice(), b"v".as_slice()] {
410            let (semantic, allowed) =
411                policy_header(&HeaderName::Other(Bytes::copy_from_slice(raw)));
412            assert_eq!(semantic, HeaderName::Via);
413            assert!(!allowed);
414        }
415    }
416}
sipx_transport/policy.rs

sipx_transport/
policy.rs
