dicom_ul/association/
mod.rs

1//! DICOM association module
2//!
3//! This module contains utilities for establishing associations
4//! between DICOM nodes via TCP/IP.
5//!
6//! As an association requester, often as a service class user (SCU),
7//! a new association can be started
8//! via the [`ClientAssociationOptions`] type.
9//! The minimum required properties are the accepted abstract syntaxes
10//! and the TCP socket address to the target node.
11//!
12//! As an association acceptor,
13//! usually taking the role of a service class provider (SCP),
14//! a newly created [TCP stream][1] can be passed to
15//! a previously prepared [`ServerAssociationOptions`].
16//!
17//!
18//! [1]: std::net::TcpStream
19pub mod client;
20pub mod server;
21#[cfg(test)]
22mod tests;
23
24mod uid;
25
26pub(crate) mod pdata;
27
28use std::{
29    backtrace::Backtrace,
30    io::{BufRead, BufReader, Cursor, Read},
31    time::Duration,
32};
33
34use bytes::{Buf, BytesMut};
35#[cfg(feature = "async")]
36pub use client::AsyncClientAssociation;
37pub use client::{ClientAssociation, ClientAssociationOptions};
38#[cfg(feature = "async")]
39pub use pdata::non_blocking::AsyncPDataWriter;
40pub use pdata::{PDataReader, PDataWriter};
41#[cfg(feature = "async")]
42pub use server::AsyncServerAssociation;
43pub use server::{ServerAssociation, ServerAssociationOptions};
44use snafu::{ensure, ResultExt, Snafu};
45
46use crate::{
47    pdu::{self, AssociationRJ, PresentationContextNegotiated, ReadPduSnafu, UserVariableItem},
48    write_pdu, Pdu,
49};
50
51type Result<T, E = Error> = std::result::Result<T, E>;
52
53#[derive(Debug, Snafu)]
54#[non_exhaustive]
55pub enum Error {
56    /// missing abstract syntax to begin negotiation
57    MissingAbstractSyntax { backtrace: Backtrace },
58
59    /// could not convert to sockeDUt address
60    ToAddress {
61        source: std::io::Error,
62        backtrace: Backtrace,
63    },
64
65    /// could not connect to server
66    Connect {
67        source: std::io::Error,
68        backtrace: Backtrace,
69    },
70
71    /// Could not set tcp read timeout
72    SetReadTimeout {
73        source: std::io::Error,
74        backtrace: Backtrace,
75    },
76
77    /// Could not set tcp write timeout
78    SetWriteTimeout {
79        source: std::io::Error,
80        backtrace: Backtrace,
81    },
82
83    /// failed to send association request
84    #[snafu(display("failed to send pdu: {}", source))]
85    SendPdu {
86        #[snafu(backtrace)]
87        source: crate::pdu::WriteError,
88    },
89
90    /// failed to receive association response
91    #[snafu(display("failed to receive pdu: {}", source))]
92    ReceivePdu {
93        #[snafu(backtrace)]
94        source: crate::pdu::ReadError,
95    },
96
97    #[snafu(display("unexpected response from peer `{:?}`", pdu))]
98    #[non_exhaustive]
99    UnexpectedPdu {
100        /// the PDU obtained from the server
101        pdu: Box<Pdu>,
102    },
103
104    #[snafu(display("unknown response from peer `{:?}`", pdu))]
105    #[non_exhaustive]
106    UnknownPdu {
107        /// the PDU obtained from the server, of variant Unknown
108        pdu: Box<Pdu>,
109    },
110
111    #[snafu(display("protocol version mismatch: expected {}, got {}", expected, got))]
112    ProtocolVersionMismatch {
113        expected: u16,
114        got: u16,
115        backtrace: Backtrace,
116    },
117
118    // Association rejected by the server
119    #[snafu(display("association rejected {}", association_rj.source))]
120    Rejected {
121        association_rj: AssociationRJ,
122        backtrace: Backtrace,
123    },
124
125    /// association aborted
126    Aborted { backtrace: Backtrace },
127
128    /// no presentation contexts accepted by the server
129    NoAcceptedPresentationContexts { backtrace: Backtrace },
130
131    /// failed to send PDU message on wire
132    #[non_exhaustive]
133    WireSend {
134        source: std::io::Error,
135        backtrace: Backtrace,
136    },
137
138    /// failed to read PDU message from wire
139    #[non_exhaustive]
140    WireRead {
141        source: std::io::Error,
142        backtrace: Backtrace,
143    },
144
145    /// Operation timed out
146    #[non_exhaustive]
147    Timeout {
148        source: std::io::Error,
149        backtrace: Backtrace,
150    },
151
152    #[snafu(display("failed close connection: {}", source))]
153    Close {
154        source: std::io::Error,
155        backtrace: Backtrace,
156    },
157
158    #[snafu(display(
159        "PDU is too large ({} bytes) to be sent to the remote application entity",
160        length
161    ))]
162    #[non_exhaustive]
163    SendTooLongPdu { length: usize, backtrace: Backtrace },
164
165    #[snafu(display("Connection closed by peer"))]
166    ConnectionClosed,
167
168    /// TLS configuration is missing
169    #[cfg(feature = "sync-tls")]
170    #[snafu(display("TLS configuration is required but not provided"))]
171    TlsConfigMissing { backtrace: Backtrace },
172
173    /// Invalid server name for TLS
174    #[cfg(feature = "sync-tls")]
175    #[snafu(display("Invalid server name for TLS connection"))]
176    InvalidServerName {
177        source: rustls::pki_types::InvalidDnsNameError,
178        backtrace: Backtrace,
179    },
180
181    /// Failed to establish TLS connection
182    #[cfg(feature = "sync-tls")]
183    #[snafu(display("Failed to establish TLS connection: {:?}", source))]
184    TlsConnection {
185        source: rustls::Error,
186        backtrace: Backtrace,
187    },
188}
189/// Struct to hold negotiated options after association is accepted
190pub(crate) struct NegotiatedOptions {
191    /// Maximum PDU length the peer can handle
192    peer_max_pdu_length: u32,
193    /// User variables accepted by the peer
194    user_variables: Vec<UserVariableItem>,
195    /// Presentation contexts accepted by the peer
196    presentation_contexts: Vec<PresentationContextNegotiated>,
197    /// The peer's AE title
198    peer_ae_title: String,
199}
200
201/// Socket configuration for associations
202#[derive(Debug, Clone, Copy, Default)]
203pub(crate) struct SocketOptions {
204    /// Timeout for individual read operations
205    read_timeout: Option<Duration>,
206    /// Timeout for individual send operations
207    write_timeout: Option<Duration>,
208    /// Timeout for connection establishment
209    connection_timeout: Option<Duration>,
210}
211
212/// Trait to close underlying socket
213pub trait CloseSocket {
214    fn close(&mut self) -> std::io::Result<()>;
215}
216
217impl CloseSocket for std::net::TcpStream {
218    fn close(&mut self) -> std::io::Result<()> {
219        self.shutdown(std::net::Shutdown::Both)
220    }
221}
222
223#[cfg(feature = "sync-tls")]
224impl CloseSocket for rustls::StreamOwned<rustls::ClientConnection, std::net::TcpStream> {
225    fn close(&mut self) -> std::io::Result<()> {
226        self.get_mut().shutdown(std::net::Shutdown::Both)
227    }
228}
229
230#[cfg(feature = "sync-tls")]
231impl CloseSocket for rustls::StreamOwned<rustls::ServerConnection, std::net::TcpStream> {
232    fn close(&mut self) -> std::io::Result<()> {
233        self.get_mut().shutdown(std::net::Shutdown::Both)
234    }
235}
236
237/// Trait that represents common properties of an association
238pub trait Association {
239    /// Obtain the remote DICOM node's application entity title.
240    fn peer_ae_title(&self) -> &str;
241
242    /// Retrieve the maximum PDU length
243    /// admitted by the association acceptor.
244    fn acceptor_max_pdu_length(&self) -> u32;
245
246    /// Retrieve the maximum PDU length
247    /// admitted by the association requestor.
248    fn requestor_max_pdu_length(&self) -> u32;
249
250    /// Retrieve the maximum PDU length
251    /// that this application entity is expecting to receive.
252    /// That's the same as acceptor_max_pdu_length() for
253    /// server objects, and as requestor_max_pdu_length()
254    /// for client objects.
255    ///
256    /// The current implementation is not required to fail
257    /// and/or abort the association
258    /// if a larger PDU is received.
259    fn local_max_pdu_length(&self) -> u32;
260
261    /// Retrieve the maximum PDU length
262    /// admitted by the peer.
263    /// That's the same as requestor_max_pdu_length() for
264    /// server objects, and as acceptor_max_pdu_length()
265    /// for client objects.
266    fn peer_max_pdu_length(&self) -> u32;
267
268    /// Obtain a view of the negotiated presentation contexts.
269    fn presentation_contexts(&self) -> &[PresentationContextNegotiated];
270
271    /// Retrieve the user variables that were taken from the server.
272    ///
273    /// It usually contains the maximum PDU length,
274    /// the implementation class UID, and the implementation version name.
275    fn user_variables(&self) -> &[UserVariableItem];
276}
277
278mod private {
279    use crate::{
280        pdu::{AbortRQServiceProviderReason, AbortRQSource},
281        Pdu,
282    };
283    use snafu::ResultExt;
284
285    /// Private trait which exposes "unsafe" methods that should not be called by the user
286    ///
287    /// `close` and `release` _should_ take ownership, and in the public interface, they
288    /// do. However, in order to implement `Drop` we need to expose a version of these
289    /// methods that don't take ownership.
290    ///
291    /// `send` and `receive` implementations are needed in order to provide
292    /// the implementation for `release`
293    pub trait SyncAssociationSealed<S: std::io::Read + std::io::Write + super::CloseSocket> {
294        fn close(&mut self) -> std::io::Result<()>;
295        fn send(&mut self, pdu: &Pdu) -> super::Result<()>;
296        fn receive(&mut self) -> super::Result<Pdu>;
297        fn release(&mut self) -> super::Result<()> {
298            let pdu = Pdu::ReleaseRQ;
299            self.send(&pdu)?;
300            let pdu = self.receive()?;
301
302            match pdu {
303                Pdu::ReleaseRP => {}
304                pdu @ Pdu::AbortRQ { .. }
305                | pdu @ Pdu::AssociationAC { .. }
306                | pdu @ Pdu::AssociationRJ { .. }
307                | pdu @ Pdu::AssociationRQ { .. }
308                | pdu @ Pdu::PData { .. }
309                | pdu @ Pdu::ReleaseRQ => return super::UnexpectedPduSnafu { pdu }.fail(),
310                pdu @ Pdu::Unknown { .. } => return super::UnknownPduSnafu { pdu }.fail(),
311            }
312            self.close().context(super::CloseSnafu)?;
313            Ok(())
314        }
315
316        fn abort(&mut self) -> super::Result<()>
317        where
318            Self: Sized,
319        {
320            let pdu = Pdu::AbortRQ {
321                source: AbortRQSource::ServiceProvider(
322                    AbortRQServiceProviderReason::ReasonNotSpecified,
323                ),
324            };
325            let out = self.send(&pdu);
326            let _ = self.close();
327            out
328        }
329    }
330
331    /// Private trait which exposes "unsafe" methods that should not be called by the user
332    ///
333    /// `close` and `release` _should_ take ownership, and in the public interface, they
334    /// do. However, in order to implement `Drop` we need to expose a version of these
335    /// methods that don't take ownership.
336    ///
337    /// `send` and `receive` implementations are needed in order to provide
338    /// the implementation for `release`
339    #[cfg(feature = "async")]
340    pub trait AsyncAssociationSealed<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin> {
341        fn close(&mut self) -> impl std::future::Future<Output = std::io::Result<()>> + Send
342        where
343            Self: Send;
344        fn send(
345            &mut self,
346            pdu: &Pdu,
347        ) -> impl std::future::Future<Output = super::Result<()>> + Send
348        where
349            Self: Send;
350        fn receive(&mut self) -> impl std::future::Future<Output = super::Result<Pdu>> + Send
351        where
352            Self: Send;
353        fn release(&mut self) -> impl std::future::Future<Output = super::Result<()>> + Send
354        where
355            Self: Send,
356        {
357            async move {
358                let pdu = Pdu::ReleaseRQ;
359                self.send(&pdu).await?;
360                let pdu = self.receive().await?;
361
362                match pdu {
363                    Pdu::ReleaseRP => {}
364                    pdu @ Pdu::AbortRQ { .. }
365                    | pdu @ Pdu::AssociationAC { .. }
366                    | pdu @ Pdu::AssociationRJ { .. }
367                    | pdu @ Pdu::AssociationRQ { .. }
368                    | pdu @ Pdu::PData { .. }
369                    | pdu @ Pdu::ReleaseRQ => return super::UnexpectedPduSnafu { pdu }.fail(),
370                    pdu @ Pdu::Unknown { .. } => return super::UnknownPduSnafu { pdu }.fail(),
371                }
372                self.close().await.context(super::CloseSnafu)?;
373                Ok(())
374            }
375        }
376
377        fn abort(&mut self) -> impl std::future::Future<Output = super::Result<()>> + Send
378        where
379            Self: Sized + Send,
380        {
381            let pdu = Pdu::AbortRQ {
382                source: AbortRQSource::ServiceProvider(
383                    AbortRQServiceProviderReason::ReasonNotSpecified,
384                ),
385            };
386            async move {
387                let out = self.send(&pdu).await;
388                let _ = self.close().await;
389                out
390            }
391        }
392    }
393}
394
395/// Trait that represents methods that can be made on a synchronous association.
396pub trait SyncAssociation<S: std::io::Read + std::io::Write + CloseSocket>:
397    private::SyncAssociationSealed<S> + Association
398{
399    /// Obtain access to the inner stream
400    /// connected to the association acceptor.
401    ///
402    /// This can be used to send the PDU in semantic fragments of the message,
403    /// thus using less memory.
404    ///
405    /// **Note:** reading and writing should be done with care
406    /// to avoid inconsistencies in the association state.
407    /// Do not call `send` and `receive` while not in a PDU boundary.
408    fn inner_stream(&mut self) -> &mut S;
409
410    /// Obtain mutable access to the inner stream and read buffer
411    fn get_mut(&mut self) -> (&mut S, &mut BytesMut);
412
413    /// Send a PDU message to the other intervenient.
414    fn send(&mut self, pdu: &Pdu) -> Result<()> {
415        private::SyncAssociationSealed::send(self, pdu)
416    }
417
418    /// Read a PDU message from the other intervenient.
419    fn receive(&mut self) -> Result<Pdu> {
420        private::SyncAssociationSealed::receive(self)
421    }
422
423    /// Send a provider initiated abort message
424    /// and shut down the TCP connection,
425    /// terminating the association.
426    fn abort(mut self) -> Result<()>
427    where
428        Self: Sized,
429    {
430        private::SyncAssociationSealed::abort(&mut self)
431    }
432
433    /// Iniate a graceful release of the association.
434    ///
435    /// A DIMSE A-RELEASE transaction is initiated by this application entity,
436    /// and the underlying socket is closed once settled.
437    ///
438    /// Note that as of version 0.9.1,
439    /// implementers of this trait no longer call this method on [`Drop`],
440    /// so remember to call `release` explicitly
441    /// at the end of all DIMSE transactions.
442    fn release(mut self) -> Result<()>
443    where
444        Self: Sized,
445    {
446        private::SyncAssociationSealed::release(&mut self)
447    }
448
449    /// Prepare a P-Data writer for sending
450    /// one or more data item PDUs.
451    ///
452    /// Returns a writer which automatically
453    /// splits the inner data into separate PDUs if necessary.
454    fn send_pdata(&mut self, presentation_context_id: u8) -> PDataWriter<&mut S> {
455        let max_pdu_length = self.peer_max_pdu_length();
456        PDataWriter::new(self.inner_stream(), presentation_context_id, max_pdu_length)
457    }
458
459    /// Prepare a P-Data reader for receiving
460    /// one or more data item PDUs.
461    ///
462    /// Returns a reader which automatically
463    /// receives more data PDUs once the bytes collected are consumed.
464    fn receive_pdata(&mut self) -> PDataReader<'_, &mut S> {
465        let max_pdu_length = self.local_max_pdu_length();
466        let (socket, read_buffer) = self.get_mut();
467        PDataReader::new(socket, max_pdu_length, read_buffer)
468    }
469}
470
471#[cfg(feature = "async")]
472/// Trait that represents methods that can be made on an asynchronous association.
473pub trait AsyncAssociation<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin>:
474    private::AsyncAssociationSealed<S> + Association
475{
476    /// Obtain access to the inner stream
477    /// connected to the association acceptor.
478    ///
479    /// This can be used to send the PDU in semantic fragments of the message,
480    /// thus using less memory.
481    ///
482    /// **Note:** reading and writing should be done with care
483    /// to avoid inconsistencies in the association state.
484    /// Do not call `send` and `receive` while not in a PDU boundary.
485    fn inner_stream(&mut self) -> &mut S;
486
487    /// Obtain mutable access to the inner stream and read buffer
488    fn get_mut(&mut self) -> (&mut S, &mut BytesMut);
489
490    /// Send a PDU message to the other intervenient.
491    fn send(&mut self, pdu: &Pdu) -> impl std::future::Future<Output = Result<()>> + Send
492    where
493        Self: Send,
494    {
495        async move { private::AsyncAssociationSealed::send(self, pdu).await }
496    }
497
498    /// Read a PDU message from the other intervenient.
499    fn receive(&mut self) -> impl std::future::Future<Output = Result<Pdu>> + Send
500    where
501        Self: Send,
502    {
503        async move { private::AsyncAssociationSealed::receive(self).await }
504    }
505
506    /// Send a provider initiated abort message
507    /// and shut down the TCP connection,
508    /// terminating the association.
509    fn abort(mut self) -> impl std::future::Future<Output = Result<()>> + Send
510    where
511        Self: Sized + Send,
512    {
513        async move { private::AsyncAssociationSealed::abort(&mut self).await }
514    }
515
516    /// Iniate a graceful release of the association.
517    ///
518    /// A DIMSE A-RELEASE transaction is initiated by this application entity,
519    /// and the underlying socket is closed once settled.
520    ///
521    /// Note that implementers of this trait
522    /// do not try to release the association on [`Drop`],
523    /// so remember to call `release` explicitly
524    /// at the end of all DIMSE transactions.
525    fn release(mut self) -> impl std::future::Future<Output = Result<()>> + Send
526    where
527        Self: Sized + Send,
528    {
529        async move { private::AsyncAssociationSealed::release(&mut self).await }
530    }
531
532    /// Prepare a P-Data writer for sending
533    /// one or more data item PDUs.
534    ///
535    /// Returns a writer which automatically
536    /// splits the inner data into separate PDUs if necessary.
537    fn send_pdata(&mut self, presentation_context_id: u8) -> AsyncPDataWriter<&mut S> {
538        let max_pdu_length = self.peer_max_pdu_length();
539        AsyncPDataWriter::new(self.inner_stream(), presentation_context_id, max_pdu_length)
540    }
541
542    /// Prepare a P-Data reader for receiving
543    /// one or more data item PDUs.
544    ///
545    /// Returns a reader which automatically
546    /// receives more data PDUs once the bytes collected are consumed.
547    fn receive_pdata(&mut self) -> PDataReader<'_, &mut S> {
548        let max_pdu_length = self.local_max_pdu_length();
549        let (socket, read_buffer) = self.get_mut();
550        PDataReader::new(socket, max_pdu_length, read_buffer)
551    }
552}
553
554// Helper function to perform an operation with timeout
555#[cfg(feature = "async")]
556async fn timeout<T>(
557    timeout: Option<Duration>,
558    block: impl std::future::Future<Output = Result<T>>,
559) -> Result<T> {
560    if let Some(timeout) = timeout {
561        tokio::time::timeout(timeout, block)
562            .await
563            .map_err(|_| std::io::Error::from(std::io::ErrorKind::TimedOut))
564            .context(crate::association::TimeoutSnafu)?
565    } else {
566        block.await
567    }
568}
569
570/// Encode a PDU into the provided buffer
571pub(crate) fn encode_pdu(buffer: &mut Vec<u8>, pdu: &Pdu, peer_max_pdu_length: u32) -> Result<()> {
572    write_pdu(buffer, pdu).context(SendPduSnafu)?;
573    if buffer.len() > peer_max_pdu_length as usize {
574        return SendTooLongPduSnafu {
575            length: buffer.len(),
576        }
577        .fail();
578    }
579    Ok(())
580}
581
582/// Helper function to get a PDU from a reader.
583///
584/// Chunks of data are read into `read_buffer`,
585/// which should be passed in subsequent calls
586/// to receive more PDUs from the same stream.
587pub fn read_pdu_from_wire<R>(
588    reader: &mut R,
589    read_buffer: &mut BytesMut,
590    max_pdu_length: u32,
591    strict: bool,
592) -> Result<Pdu>
593where
594    R: Read,
595{
596    let mut reader = BufReader::new(reader);
597    let msg = loop {
598        let mut buf = Cursor::new(&read_buffer[..]);
599        // try to read a PDU according to what's in the buffer
600        match pdu::read_pdu(&mut buf, max_pdu_length, strict).context(ReceivePduSnafu)? {
601            Some(pdu) => {
602                read_buffer.advance(buf.position() as usize);
603                break pdu;
604            }
605            None => {
606                // Reset position
607                buf.set_position(0)
608            }
609        }
610        // Use BufReader to get similar behavior to AsyncRead read_buf
611        let recv = reader
612            .fill_buf()
613            .context(ReadPduSnafu)
614            .context(ReceivePduSnafu)?;
615        let bytes_read = recv.len();
616        read_buffer.extend_from_slice(recv);
617        reader.consume(bytes_read);
618        ensure!(bytes_read != 0, ConnectionClosedSnafu);
619    };
620    Ok(msg)
621}
622
623/// Helper function to get a PDU from an async reader.
624///
625/// Chunks of data are read into `read_buffer`,
626/// which should be passed in subsequent calls
627/// to receive more PDUs from the same stream.
628#[cfg(feature = "async")]
629pub async fn read_pdu_from_wire_async<R: tokio::io::AsyncRead + Unpin>(
630    reader: &mut R,
631    read_buffer: &mut BytesMut,
632    max_pdu_length: u32,
633    strict: bool,
634) -> Result<Pdu> {
635    use tokio::io::AsyncReadExt;
636    // receive response
637
638    let msg = loop {
639        let mut buf = Cursor::new(&read_buffer[..]);
640        match pdu::read_pdu(&mut buf, max_pdu_length, strict).context(ReceivePduSnafu)? {
641            Some(pdu) => {
642                read_buffer.advance(buf.position() as usize);
643                break pdu;
644            }
645            None => {
646                // Reset position
647                buf.set_position(0)
648            }
649        }
650        let recv = reader
651            .read_buf(read_buffer)
652            .await
653            .context(ReadPduSnafu)
654            .context(ReceivePduSnafu)?;
655        ensure!(recv > 0, ConnectionClosedSnafu);
656    };
657    Ok(msg)
658}
dicom_ul/association/mod.rs

dicom_ul/association/
mod.rs
