dicom_ul/association/
client.rs

1//! Association requester module
2//!
3//! The module provides an abstraction for a DICOM association
4//! in which this application entity is the one requesting the association.
5//! See [`ClientAssociationOptions`]
6//! for details and examples on how to create an association.
7use bytes::BytesMut;
8use std::{
9    borrow::Cow,
10    convert::TryInto,
11    net::{TcpStream, ToSocketAddrs},
12    time::Duration,
13};
14
15#[cfg(feature = "async")]
16use crate::association::AsyncAssociation;
17use crate::{
18    association::{
19        encode_pdu, private::SyncAssociationSealed, read_pdu_from_wire, Association,
20        NegotiatedOptions, SocketOptions, SyncAssociation,
21    },
22    pdu::{
23        write_pdu, AbortRQSource, AssociationAC, AssociationRQ, Pdu, PresentationContextNegotiated,
24        PresentationContextProposed, PresentationContextResultReason, UserIdentity,
25        UserIdentityType, UserVariableItem, DEFAULT_MAX_PDU, LARGE_PDU_SIZE, PDU_HEADER_SIZE,
26    },
27    AeAddr, IMPLEMENTATION_CLASS_UID, IMPLEMENTATION_VERSION_NAME,
28};
29use snafu::{ensure, ResultExt};
30
31use super::{uid::trim_uid, Result};
32
33// stray module from 0.9.0, remove in 0.10.0
34#[deprecated(since = "0.9.1")]
35pub mod non_blocking {}
36
37#[cfg(feature = "sync-tls")]
38pub type TlsStream = rustls::StreamOwned<rustls::ClientConnection, std::net::TcpStream>;
39#[cfg(feature = "async-tls")]
40pub type AsyncTlsStream = tokio_rustls::client::TlsStream<tokio::net::TcpStream>;
41
42pub use crate::association::CloseSocket;
43
44/// Helper function to establish a TCP client connection
45fn tcp_connection<T>(ae_address: &AeAddr<T>, opts: &SocketOptions) -> Result<TcpStream>
46where
47    T: ToSocketAddrs,
48{
49    // NOTE: TcpStream::connect_timeout needs a single SocketAddr, whereas TcpStream::connect can
50    // take multiple
51    let conn_result: Result<TcpStream> = if let Some(timeout) = opts.connection_timeout {
52        let addresses = ae_address
53            .to_socket_addrs()
54            .context(super::ToAddressSnafu)?;
55        let mut result = Result::Err(std::io::Error::from(std::io::ErrorKind::AddrNotAvailable));
56        for address in addresses {
57            result = TcpStream::connect_timeout(&address, timeout);
58            if result.is_ok() {
59                break;
60            }
61        }
62        result.context(super::ConnectSnafu)
63    } else {
64        TcpStream::connect(ae_address).context(super::ConnectSnafu)
65    };
66
67    let socket = conn_result?;
68    socket
69        .set_read_timeout(opts.read_timeout)
70        .context(super::SetReadTimeoutSnafu)?;
71    socket
72        .set_write_timeout(opts.write_timeout)
73        .context(super::SetWriteTimeoutSnafu)?;
74
75    Ok(socket)
76}
77
78/// Helper function to establish a TLS client connection
79#[cfg(feature = "sync-tls")]
80fn tls_connection<T>(
81    ae_address: &AeAddr<T>,
82    server_name: &str,
83    opts: &SocketOptions,
84    tls_config: std::sync::Arc<rustls::ClientConfig>,
85) -> Result<TlsStream>
86where
87    T: ToSocketAddrs,
88{
89    use std::convert::TryFrom;
90
91    let socket = tcp_connection(ae_address, opts)?;
92    let server_name = rustls::pki_types::ServerName::try_from(server_name.to_string())
93        .context(super::InvalidServerNameSnafu)?;
94
95    let conn = rustls::ClientConnection::new(tls_config.clone(), server_name)
96        .context(super::TlsConnectionSnafu)?;
97
98    Ok(rustls::StreamOwned::new(conn, socket))
99}
100
101/// A DICOM association builder for a client node.
102/// The final outcome is a [`ClientAssociation`].
103///
104/// This is the standard way of requesting and establishing
105/// an association with another DICOM node,
106/// that one usually taking the role of a service class provider (SCP).
107///
108/// You can create either a blocking or non-blocking client by calling either
109/// `establish` or `establish_async` respectively.
110///
111/// > **⚠️ Warning:** It is highly recommended to set `read_timeout` and `write_timeout` to a reasonable
112/// > value for the async client since there is _no_ default timeout on
113/// > [`TcpStream`]
114///
115/// ## Basic usage
116///
117/// ### Synchronous API
118///
119/// ```no_run
120/// # use dicom_ul::association::client::ClientAssociationOptions;
121/// # use std::time::Duration;
122/// # fn run() -> Result<(), Box<dyn std::error::Error>> {
123/// let association = ClientAssociationOptions::new()
124///    .with_presentation_context("1.2.840.10008.1.1", vec!["1.2.840.10008.1.2.1", "1.2.840.10008.1.2"])
125///    .read_timeout(Duration::from_secs(60))
126///    .write_timeout(Duration::from_secs(60))
127///    .establish("129.168.0.5:104")?;
128/// # Ok(())
129/// # }
130/// ```
131///
132/// ### Asynchronous API
133///
134/// Include the `async` feature in your `Cargo.toml`
135///
136/// ```no_run
137/// # use dicom_ul::association::client::ClientAssociationOptions;
138/// # use std::time::Duration;
139/// # #[cfg(feature = "async")]
140/// # #[tokio::main]
141/// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
142/// let association = ClientAssociationOptions::new()
143///    .with_presentation_context("1.2.840.10008.1.1", vec!["1.2.840.10008.1.2.1", "1.2.840.10008.1.2"])
144///    .read_timeout(Duration::from_secs(60))
145///    .write_timeout(Duration::from_secs(60))
146///    .establish_async("129.168.0.5:104")
147///    .await?;
148/// # Ok(())
149/// # }
150/// ```
151///
152/// ## TLS Support
153///
154/// Enabling one of the Cargo features `sync-tls` or `async-tls`
155/// unlocks the methods for configuring TLS.
156/// Call `tls_config` and `server_name`
157/// to establish the association over a secure transport connection.
158///
159/// ### TLS in synchronous API
160///
161/// Include the `sync-tls` feature in your `Cargo.toml`.
162///
163/// ### TLS in asynchronous API
164///
165/// Include the `async-tls` feature in your `Cargo.toml`.
166///
167/// ### Example
168///
169/// ```no_run
170/// # use std::time::Duration;
171/// # use std::sync::Arc;
172/// # #[cfg(feature = "sync-tls")]
173/// # fn run() -> Result<(), Box<dyn std::error::Error>> {
174/// use dicom_dictionary_std::uids;
175/// use dicom_ul::{ClientAssociation, ClientAssociationOptions};
176/// use rustls::{
177///     ClientConfig, RootCertStore,
178///     pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject},
179/// };
180/// // Loading certificates and keys for demonstration purposes
181/// let ca_cert = CertificateDer::from_pem_slice(std::fs::read("ssl/ca.crt")?.as_ref())
182///     .expect("Failed to load client cert");
183///
184/// // Server certificate -- signed by CA
185/// let server_cert = CertificateDer::from_pem_slice(std::fs::read("ssl/server.crt")?.as_ref())
186///     .expect("Failed to load server cert");
187///
188/// // Client cert and private key -- signed by CA
189/// let client_cert = CertificateDer::from_pem_slice(std::fs::read("ssl/client.crt")?.as_ref())
190///     .expect("Failed to load client cert");
191/// let client_private_key = PrivateKeyDer::from_pem_slice(std::fs::read("ssl/client.key")?.as_ref())
192///     .expect("Failed to load client private key");
193///
194/// // Create a root cert store for the client which includes the server certificate
195/// let mut certs = RootCertStore::empty();
196/// certs.add_parsable_certificates(vec![ca_cert.clone()]);
197///
198/// let config = ClientConfig::builder()
199///     .with_root_certificates(certs)
200///     .with_client_auth_cert(vec![client_cert, ca_cert], client_private_key)
201///     .expect("Failed to create client TLS config");
202///
203/// let association: ClientAssociation<_> = ClientAssociationOptions::new()
204///    .with_presentation_context(
205///         uids::VERIFICATION,
206///         vec![uids::EXPLICIT_VR_LITTLE_ENDIAN, uids::IMPLICIT_VR_LITTLE_ENDIAN]
207///    )
208///    .tls_config(config)
209///    .read_timeout(Duration::from_secs(60))
210///    .write_timeout(Duration::from_secs(60))
211///    .establish_with_tls("REMOTE_DCM@129.168.0.5:104")?;
212/// # Ok(())
213/// # }
214/// ```
215///
216/// For an association with the async API,
217/// call `establish_tls_async` or `establish_with_async_tls`
218/// instead of `establish_tls` or `establish_with_tls`.
219///
220/// ## Presentation contexts
221///
222/// At least one presentation context must be specified,
223/// using the method [`with_presentation_context`](Self::with_presentation_context)
224/// and supplying both an abstract syntax and list of transfer syntaxes.
225///
226/// A helper method [`with_abstract_syntax`](Self::with_abstract_syntax) will
227/// include by default the transfer syntaxes
228/// _Implicit VR Little Endian_ and _Explicit VR Little Endian_
229/// in the resulting presentation context.
230///
231/// ```no_run
232/// # use dicom_ul::association::client::ClientAssociationOptions;
233/// # fn run() -> Result<(), Box<dyn std::error::Error>> {
234/// let association = ClientAssociationOptions::new()
235///     .with_abstract_syntax("1.2.840.10008.1.1")
236///     .establish("129.168.0.5:104")?;
237/// # Ok(())
238/// # }
239/// ```
240#[derive(Debug, Clone)]
241pub struct ClientAssociationOptions<'a> {
242    /// the calling AE title
243    calling_ae_title: Cow<'a, str>,
244    /// the called AE title
245    called_ae_title: Option<Cow<'a, str>>,
246    /// the requested application context name
247    application_context_name: Cow<'a, str>,
248    /// the list of requested presentation contexts
249    presentation_contexts: Vec<(Cow<'a, str>, Vec<Cow<'a, str>>)>,
250    /// the expected protocol version
251    protocol_version: u16,
252    /// the maximum PDU length requested for receiving PDUs
253    max_pdu_length: u32,
254    /// whether to receive PDUs in strict mode
255    strict: bool,
256    /// User identity username
257    username: Option<Cow<'a, str>>,
258    /// User identity password
259    password: Option<Cow<'a, str>>,
260    /// User identity Kerberos service ticket
261    kerberos_service_ticket: Option<Cow<'a, str>>,
262    /// User identity SAML assertion
263    saml_assertion: Option<Cow<'a, str>>,
264    /// User identity JWT
265    jwt: Option<Cow<'a, str>>,
266    /// Socket options for TCP connections
267    socket_options: SocketOptions,
268    /// TLS configuration to use for the connection
269    #[cfg(feature = "sync-tls")]
270    tls_config: Option<std::sync::Arc<rustls::ClientConfig>>,
271    /// Server name for TLS
272    #[cfg(feature = "sync-tls")]
273    server_name: Option<String>,
274}
275
276impl Default for ClientAssociationOptions<'_> {
277    fn default() -> Self {
278        ClientAssociationOptions {
279            // the calling AE title
280            calling_ae_title: "THIS-SCU".into(),
281            // the called AE title
282            called_ae_title: None,
283            // the requested application context name
284            application_context_name: "1.2.840.10008.3.1.1.1".into(),
285            // the list of requested presentation contexts
286            presentation_contexts: Vec::new(),
287            protocol_version: 1,
288            max_pdu_length: DEFAULT_MAX_PDU,
289            strict: true,
290            username: None,
291            password: None,
292            kerberos_service_ticket: None,
293            saml_assertion: None,
294            jwt: None,
295            socket_options: SocketOptions {
296                read_timeout: None,
297                write_timeout: None,
298                connection_timeout: None,
299            },
300            #[cfg(feature = "sync-tls")]
301            tls_config: None,
302            #[cfg(feature = "sync-tls")]
303            server_name: None,
304        }
305    }
306}
307
308impl<'a> ClientAssociationOptions<'a> {
309    /// Create a new set of options for establishing an association.
310    pub fn new() -> Self {
311        Self::default()
312    }
313    /// Define the calling application entity title for the association,
314    /// which refers to this DICOM node.
315    ///
316    /// The default is `THIS-SCU`.
317    pub fn calling_ae_title<T>(mut self, calling_ae_title: T) -> Self
318    where
319        T: Into<Cow<'a, str>>,
320    {
321        self.calling_ae_title = calling_ae_title.into();
322        self
323    }
324
325    /// Define the called application entity title for the association,
326    /// which refers to the target DICOM node.
327    ///
328    /// The default is `ANY-SCP`.
329    /// Passing an empty string resets the AE title to the default
330    /// (or to the one passed via [`establish_with`](ClientAssociationOptions::establish_with)).
331    pub fn called_ae_title<T>(mut self, called_ae_title: T) -> Self
332    where
333        T: Into<Cow<'a, str>>,
334    {
335        let cae = called_ae_title.into();
336        if cae.is_empty() {
337            self.called_ae_title = None;
338        } else {
339            self.called_ae_title = Some(cae);
340        }
341        self
342    }
343
344    /// Include this presentation context
345    /// in the list of proposed presentation contexts.
346    pub fn with_presentation_context<T>(
347        mut self,
348        abstract_syntax_uid: T,
349        transfer_syntax_uids: Vec<T>,
350    ) -> Self
351    where
352        T: Into<Cow<'a, str>>,
353    {
354        let transfer_syntaxes: Vec<Cow<'a, str>> = transfer_syntax_uids
355            .into_iter()
356            .map(|t| trim_uid(t.into()))
357            .collect();
358        self.presentation_contexts
359            .push((trim_uid(abstract_syntax_uid.into()), transfer_syntaxes));
360        self
361    }
362
363    /// Helper to add this abstract syntax
364    /// with the default transfer syntaxes
365    /// to the list of proposed presentation contexts.
366    pub fn with_abstract_syntax<T>(self, abstract_syntax_uid: T) -> Self
367    where
368        T: Into<Cow<'a, str>>,
369    {
370        let default_transfer_syntaxes: Vec<Cow<'a, str>> =
371            vec!["1.2.840.10008.1.2.1".into(), "1.2.840.10008.1.2".into()];
372        self.with_presentation_context(abstract_syntax_uid.into(), default_transfer_syntaxes)
373    }
374
375    /// Override the maximum PDU length
376    /// that this application entity will admit.
377    pub fn max_pdu_length(mut self, value: u32) -> Self {
378        self.max_pdu_length = value;
379        self
380    }
381
382    /// Override strict mode:
383    /// whether receiving PDUs must not
384    /// surpass the negotiated maximum PDU length.
385    pub fn strict(mut self, strict: bool) -> Self {
386        self.strict = strict;
387        self
388    }
389
390    /// Sets the user identity username
391    pub fn username<T>(mut self, username: T) -> Self
392    where
393        T: Into<Cow<'a, str>>,
394    {
395        let username = username.into();
396        if username.is_empty() {
397            self.username = None;
398        } else {
399            self.username = Some(username);
400            self.saml_assertion = None;
401            self.jwt = None;
402            self.kerberos_service_ticket = None;
403        }
404        self
405    }
406
407    /// Sets the user identity password
408    pub fn password<T>(mut self, password: T) -> Self
409    where
410        T: Into<Cow<'a, str>>,
411    {
412        let password = password.into();
413        if password.is_empty() {
414            self.password = None;
415        } else {
416            self.password = Some(password);
417            self.saml_assertion = None;
418            self.jwt = None;
419            self.kerberos_service_ticket = None;
420        }
421        self
422    }
423
424    /// Sets the user identity username and password
425    pub fn username_password<T, U>(mut self, username: T, password: U) -> Self
426    where
427        T: Into<Cow<'a, str>>,
428        U: Into<Cow<'a, str>>,
429    {
430        let username = username.into();
431        let password = password.into();
432        if username.is_empty() {
433            self.username = None;
434            self.password = None;
435        } else {
436            self.username = Some(username);
437            self.password = Some(password);
438            self.saml_assertion = None;
439            self.jwt = None;
440            self.kerberos_service_ticket = None;
441        }
442        self
443    }
444
445    /// Sets the user identity Kerberos service ticket
446    pub fn kerberos_service_ticket<T>(mut self, kerberos_service_ticket: T) -> Self
447    where
448        T: Into<Cow<'a, str>>,
449    {
450        let kerberos_service_ticket = kerberos_service_ticket.into();
451        if kerberos_service_ticket.is_empty() {
452            self.kerberos_service_ticket = None;
453        } else {
454            self.kerberos_service_ticket = Some(kerberos_service_ticket);
455            self.username = None;
456            self.password = None;
457            self.saml_assertion = None;
458            self.jwt = None;
459        }
460        self
461    }
462
463    /// Sets the user identity SAML assertion
464    pub fn saml_assertion<T>(mut self, saml_assertion: T) -> Self
465    where
466        T: Into<Cow<'a, str>>,
467    {
468        let saml_assertion = saml_assertion.into();
469        if saml_assertion.is_empty() {
470            self.saml_assertion = None;
471        } else {
472            self.saml_assertion = Some(saml_assertion);
473            self.username = None;
474            self.password = None;
475            self.jwt = None;
476            self.kerberos_service_ticket = None;
477        }
478        self
479    }
480
481    /// Sets the user identity JWT
482    pub fn jwt<T>(mut self, jwt: T) -> Self
483    where
484        T: Into<Cow<'a, str>>,
485    {
486        let jwt = jwt.into();
487        if jwt.is_empty() {
488            self.jwt = None;
489        } else {
490            self.jwt = Some(jwt);
491            self.username = None;
492            self.password = None;
493            self.saml_assertion = None;
494            self.kerberos_service_ticket = None;
495        }
496        self
497    }
498
499    /// Set the TLS configuration to use for the connection
500    #[cfg(feature = "sync-tls")]
501    pub fn tls_config(mut self, config: impl Into<std::sync::Arc<rustls::ClientConfig>>) -> Self {
502        self.tls_config = Some(config.into());
503        self
504    }
505
506    /// Set the server name to use for the TLS connection
507    #[cfg(feature = "sync-tls")]
508    pub fn server_name(mut self, server_name: &str) -> Self {
509        self.server_name = Some(server_name.to_string());
510        self
511    }
512
513    /// Initiate simple TCP connection to the given address
514    /// and request a new DICOM association,
515    /// negotiating the presentation contexts in the process.
516    pub fn establish<A: ToSocketAddrs>(
517        self,
518        address: A,
519    ) -> Result<ClientAssociation<std::net::TcpStream>> {
520        let addr = AeAddr::new_socket_addr(address);
521        let socket = tcp_connection(&addr, &self.socket_options)?;
522        self.establish_impl(addr, socket)
523    }
524
525    /// Initiate simple TCP connection to the given address
526    /// and request a new DICOM association,
527    /// negotiating the presentation contexts in the process.
528    #[cfg(feature = "sync-tls")]
529    pub fn establish_tls<A: ToSocketAddrs>(
530        self,
531        address: A,
532    ) -> Result<ClientAssociation<TlsStream>> {
533        match (&self.tls_config, &self.server_name) {
534            (Some(tls_config), Some(server_name)) => {
535                let addr = AeAddr::new_socket_addr(address);
536                let socket =
537                    tls_connection(&addr, server_name, &self.socket_options, tls_config.clone())?;
538                self.establish_impl(addr, socket)
539            }
540            _ => super::TlsConfigMissingSnafu.fail()?,
541        }
542    }
543
544    /// Initiate the TCP connection to the given address
545    /// and request a new DICOM association,
546    /// negotiating the presentation contexts in the process.
547    ///
548    /// This method allows you to specify the called AE title
549    /// alongside with the socket address.
550    /// See [AeAddr](`crate::AeAddr`) for more details.
551    /// However, the AE title in this parameter
552    /// is overridden by any `called_ae_title` option
553    /// previously received.
554    ///
555    /// # Example
556    ///
557    /// ```no_run
558    /// # use dicom_ul::association::client::ClientAssociationOptions;
559    /// # fn run() -> Result<(), Box<dyn std::error::Error>> {
560    /// let association = ClientAssociationOptions::new()
561    ///     .with_abstract_syntax("1.2.840.10008.1.1")
562    ///     // called AE title in address
563    ///     .establish_with("MY-STORAGE@10.0.0.100:104")?;
564    /// # Ok(())
565    /// # }
566    /// ```
567    #[allow(unreachable_patterns)]
568    pub fn establish_with(self, ae_address: &str) -> Result<ClientAssociation<TcpStream>> {
569        match ae_address.try_into() {
570            Ok(ae_address) => {
571                let socket = tcp_connection(&ae_address, &self.socket_options)?;
572                self.establish_impl(ae_address, socket)
573            }
574            Err(_) => {
575                let addr = AeAddr::new_socket_addr(ae_address);
576                let socket = tcp_connection(&addr, &self.socket_options)?;
577                self.establish_impl(addr, socket)
578            }
579        }
580    }
581
582    /// Initiate TLS connection to the given address
583    /// and request a new DICOM association,
584    /// negotiating the presentation contexts in the process.
585    ///
586    /// This method allows you to specify the called AE title
587    /// alongside with the socket address.
588    /// See [AeAddr](`crate::AeAddr`) for more details.
589    /// However, the AE title in this parameter
590    /// is overridden by any `called_ae_title` option
591    /// previously received.
592    ///
593    /// # Example
594    ///
595    /// ```no_run
596    /// # use dicom_ul::association::client::ClientAssociationOptions;
597    /// # fn run() -> Result<(), Box<dyn std::error::Error>> {
598    /// let association = ClientAssociationOptions::new()
599    ///     .with_abstract_syntax("1.2.840.10008.1.1")
600    ///     // called AE title in address
601    ///     .establish_with("MY-STORAGE@10.0.0.100:104")?;
602    /// # Ok(())
603    /// # }
604    /// ```
605    #[allow(unreachable_patterns)]
606    #[cfg(feature = "sync-tls")]
607    pub fn establish_with_tls(self, ae_address: &str) -> Result<ClientAssociation<TlsStream>> {
608        match (&self.tls_config, &self.server_name) {
609            (Some(tls_config), Some(server_name)) => match ae_address.try_into() {
610                Ok(ae_address) => {
611                    let socket = tls_connection(
612                        &ae_address,
613                        server_name,
614                        &self.socket_options,
615                        tls_config.clone(),
616                    )?;
617                    self.establish_impl(ae_address, socket)
618                }
619                Err(_) => {
620                    let addr = AeAddr::new_socket_addr(ae_address);
621                    let socket = tls_connection(
622                        &addr,
623                        server_name,
624                        &self.socket_options,
625                        tls_config.clone(),
626                    )?;
627                    self.establish_impl(addr, socket)
628                }
629            },
630            _ => super::TlsConfigMissingSnafu.fail()?,
631        }
632    }
633
634    /// Set the read timeout for the underlying TCP socket
635    ///
636    /// This is used to set both the read and write timeout.
637    pub fn read_timeout(self, timeout: Duration) -> Self {
638        Self {
639            socket_options: SocketOptions {
640                read_timeout: Some(timeout),
641                write_timeout: self.socket_options.write_timeout,
642                connection_timeout: self.socket_options.connection_timeout,
643            },
644            ..self
645        }
646    }
647
648    /// Set the write timeout for the underlying TCP socket
649    pub fn write_timeout(self, timeout: Duration) -> Self {
650        Self {
651            socket_options: SocketOptions {
652                read_timeout: self.socket_options.read_timeout,
653                write_timeout: Some(timeout),
654                connection_timeout: self.socket_options.connection_timeout,
655            },
656            ..self
657        }
658    }
659
660    /// Set the connection timeout for the underlying TCP socket
661    pub fn connection_timeout(self, timeout: Duration) -> Self {
662        Self {
663            socket_options: SocketOptions {
664                read_timeout: self.socket_options.read_timeout,
665                write_timeout: self.socket_options.write_timeout,
666                connection_timeout: Some(timeout),
667            },
668            ..self
669        }
670    }
671
672    /// Construct the A-ASSOCIATE-RQ PDU given the options and the AE title.
673    fn create_a_associate_req(
674        &'a self,
675        ae_title: Option<&str>,
676    ) -> Result<(Vec<PresentationContextProposed>, Pdu)> {
677        let ClientAssociationOptions {
678            calling_ae_title,
679            called_ae_title,
680            application_context_name,
681            presentation_contexts,
682            protocol_version,
683            max_pdu_length,
684            username,
685            password,
686            kerberos_service_ticket,
687            saml_assertion,
688            jwt,
689            ..
690        } = self;
691        // fail if no presentation contexts were provided: they represent intent,
692        // should not be omitted by the user
693        ensure!(
694            !presentation_contexts.is_empty(),
695            crate::association::MissingAbstractSyntaxSnafu
696        );
697
698        // choose called AE title
699        let called_ae_title: &str = match (&called_ae_title, ae_title) {
700            (Some(aec), Some(aet)) => {
701                if aec != aet {
702                    tracing::warn!(
703                        "Option `called_ae_title` overrides the AE title from `{aet}` to `{aec}`"
704                    );
705                }
706                aec
707            }
708            (Some(aec), None) => aec,
709            (None, Some(aec)) => aec,
710            (None, None) => "ANY-SCP",
711        };
712
713        let presentation_contexts_proposed: Vec<_> = presentation_contexts
714            .iter()
715            .enumerate()
716            .map(|(i, presentation_context)| PresentationContextProposed {
717                id: (2 * i + 1) as u8,
718                abstract_syntax: presentation_context.0.to_string(),
719                transfer_syntaxes: presentation_context
720                    .1
721                    .iter()
722                    .map(|uid| uid.to_string())
723                    .collect(),
724            })
725            .collect();
726
727        let mut user_variables = vec![
728            UserVariableItem::MaxLength(*max_pdu_length),
729            UserVariableItem::ImplementationClassUID(IMPLEMENTATION_CLASS_UID.to_string()),
730            UserVariableItem::ImplementationVersionName(IMPLEMENTATION_VERSION_NAME.to_string()),
731        ];
732
733        if let Some(user_identity) = Self::determine_user_identity(
734            username.as_deref(),
735            password.as_deref(),
736            kerberos_service_ticket.as_deref(),
737            saml_assertion.as_deref(),
738            jwt.as_deref(),
739        ) {
740            user_variables.push(UserVariableItem::UserIdentityItem(user_identity));
741        }
742
743        Ok((
744            presentation_contexts_proposed.clone(),
745            Pdu::AssociationRQ(AssociationRQ {
746                protocol_version: *protocol_version,
747                calling_ae_title: calling_ae_title.to_string(),
748                called_ae_title: called_ae_title.to_string(),
749                application_context_name: application_context_name.to_string(),
750                presentation_contexts: presentation_contexts_proposed,
751                user_variables,
752            }),
753        ))
754    }
755
756    /// Process the A-ASSOCIATE-AC PDU received from the SCP.
757    ///
758    /// Returns the negotiated options for the association
759    fn process_a_association_resp(
760        &self,
761        msg: Pdu,
762        presentation_contexts_proposed: &[PresentationContextProposed],
763    ) -> Result<NegotiatedOptions> {
764        match msg {
765            Pdu::AssociationAC(AssociationAC {
766                protocol_version: protocol_version_scp,
767                application_context_name: _,
768                presentation_contexts: presentation_contexts_scp,
769                calling_ae_title: _,
770                called_ae_title,
771                user_variables,
772            }) => {
773                ensure!(
774                    self.protocol_version == protocol_version_scp,
775                    crate::association::ProtocolVersionMismatchSnafu {
776                        expected: self.protocol_version,
777                        got: protocol_version_scp,
778                    }
779                );
780
781                let acceptor_max_pdu_length = user_variables
782                    .iter()
783                    .find_map(|item| match item {
784                        UserVariableItem::MaxLength(len) => Some(*len),
785                        _ => None,
786                    })
787                    .unwrap_or(DEFAULT_MAX_PDU);
788
789                // treat 0 as practically unlimited
790                let acceptor_max_pdu_length = if acceptor_max_pdu_length == 0 {
791                    u32::MAX
792                } else {
793                    acceptor_max_pdu_length
794                };
795
796                let presentation_contexts: Vec<_> = presentation_contexts_scp
797                    .into_iter()
798                    .filter(|c| {
799                        c.reason == PresentationContextResultReason::Acceptance
800                            && presentation_contexts_proposed.iter().any(|p| p.id == c.id)
801                    })
802                    .map(|c| {
803                        let pcp = presentation_contexts_proposed
804                            .iter()
805                            .find(|pc| pc.id == c.id)
806                            .unwrap();
807                        PresentationContextNegotiated {
808                            id: c.id,
809                            reason: c.reason,
810                            transfer_syntax: c.transfer_syntax,
811                            abstract_syntax: pcp.abstract_syntax.clone(),
812                        }
813                    })
814                    .collect();
815                if presentation_contexts.is_empty() {
816                    return crate::association::NoAcceptedPresentationContextsSnafu.fail();
817                }
818                Ok(NegotiatedOptions {
819                    presentation_contexts,
820                    peer_max_pdu_length: acceptor_max_pdu_length,
821                    user_variables,
822                    peer_ae_title: called_ae_title,
823                })
824            }
825            Pdu::AssociationRJ(association_rj) => {
826                crate::association::RejectedSnafu { association_rj }.fail()
827            }
828            pdu @ Pdu::AbortRQ { .. }
829            | pdu @ Pdu::ReleaseRQ
830            | pdu @ Pdu::AssociationRQ { .. }
831            | pdu @ Pdu::PData { .. }
832            | pdu @ Pdu::ReleaseRP => crate::association::UnexpectedPduSnafu { pdu }.fail(),
833            pdu @ Pdu::Unknown { .. } => crate::association::UnknownPduSnafu { pdu }.fail(),
834        }
835    }
836
837    /// Establish the association with the given AE address.
838    fn establish_impl<T, S>(
839        self,
840        ae_address: AeAddr<T>,
841        mut socket: S,
842    ) -> Result<ClientAssociation<S>>
843    where
844        T: ToSocketAddrs,
845        S: CloseSocket + std::io::Read + std::io::Write,
846    {
847        let (pc_proposed, a_associate) = self.create_a_associate_req(ae_address.ae_title())?;
848        let mut buffer: Vec<u8> = Vec::with_capacity(self.max_pdu_length as usize);
849
850        write_pdu(&mut buffer, &a_associate).context(super::SendPduSnafu)?;
851        socket.write_all(&buffer).context(super::WireSendSnafu)?;
852        buffer.clear();
853
854        let mut buf = BytesMut::with_capacity(
855            (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
856        );
857        let resp = read_pdu_from_wire(&mut socket, &mut buf, self.max_pdu_length, self.strict)?;
858        let negotiated_options = self.process_a_association_resp(resp, &pc_proposed);
859        match negotiated_options {
860            Err(e) => {
861                // abort connection
862                let _ = write_pdu(
863                    &mut buffer,
864                    &Pdu::AbortRQ {
865                        source: AbortRQSource::ServiceUser,
866                    },
867                );
868                let _ = socket.write_all(&buffer);
869                buffer.clear();
870                Err(e)
871            }
872            Ok(NegotiatedOptions {
873                presentation_contexts,
874                peer_max_pdu_length,
875                user_variables,
876                peer_ae_title,
877            }) => {
878                Ok(ClientAssociation {
879                    presentation_contexts,
880                    requestor_max_pdu_length: self.max_pdu_length,
881                    acceptor_max_pdu_length: peer_max_pdu_length,
882                    socket,
883                    write_buffer: buffer,
884                    strict: self.strict,
885                    // Fixes #589, instead of creating a new buffer, we pass the existing buffer into the Association object.
886                    read_buffer: buf,
887                    read_timeout: self.socket_options.read_timeout,
888                    write_timeout: self.socket_options.write_timeout,
889                    user_variables,
890                    peer_ae_title,
891                })
892            }
893        }
894    }
895
896    fn determine_user_identity<T>(
897        username: Option<T>,
898        password: Option<T>,
899        kerberos_service_ticket: Option<T>,
900        saml_assertion: Option<T>,
901        jwt: Option<T>,
902    ) -> Option<UserIdentity>
903    where
904        T: Into<Cow<'a, str>>,
905    {
906        if let Some(username) = username {
907            if let Some(password) = password {
908                return Some(UserIdentity::new(
909                    false,
910                    UserIdentityType::UsernamePassword,
911                    username.into().as_bytes().to_vec(),
912                    password.into().as_bytes().to_vec(),
913                ));
914            } else {
915                return Some(UserIdentity::new(
916                    false,
917                    UserIdentityType::Username,
918                    username.into().as_bytes().to_vec(),
919                    vec![],
920                ));
921            }
922        }
923
924        if let Some(kerberos_service_ticket) = kerberos_service_ticket {
925            return Some(UserIdentity::new(
926                false,
927                UserIdentityType::KerberosServiceTicket,
928                kerberos_service_ticket.into().as_bytes().to_vec(),
929                vec![],
930            ));
931        }
932
933        if let Some(saml_assertion) = saml_assertion {
934            return Some(UserIdentity::new(
935                false,
936                UserIdentityType::SamlAssertion,
937                saml_assertion.into().as_bytes().to_vec(),
938                vec![],
939            ));
940        }
941
942        if let Some(jwt) = jwt {
943            return Some(UserIdentity::new(
944                false,
945                UserIdentityType::Jwt,
946                jwt.into().as_bytes().to_vec(),
947                vec![],
948            ));
949        }
950
951        None
952    }
953}
954
955/// A DICOM upper level association from the perspective
956/// of a requesting application entity.
957///
958/// The most common operations of an established association are
959/// [`send`](SyncAssociation::send)
960/// and [`receive`](SyncAssociation::receive).
961/// Sending large P-Data fragments may be easier through the P-Data sender
962/// abstraction (see [`send_pdata`](SyncAssociation::send_pdata)).
963///
964/// Call `release` at the end
965/// to perform a standard C-RELEASE message exchange
966/// and shut down the underlying TCP connection.
967/// Not calling this method will only close the socket
968/// without gracefully releasing the association.
969#[derive(Debug)]
970pub struct ClientAssociation<S> {
971    /// The presentation contexts accorded with the acceptor application entity,
972    /// without the rejected ones.
973    presentation_contexts: Vec<PresentationContextNegotiated>,
974    /// The maximum PDU length that this application entity is expecting to receive
975    requestor_max_pdu_length: u32,
976    /// The maximum PDU length that the remote application entity accepts
977    acceptor_max_pdu_length: u32,
978    /// The TCP stream to the other DICOM node
979    socket: S,
980    /// Buffer to write PDUs to the wire, prevents needing to allocate on every send
981    write_buffer: Vec<u8>,
982    /// whether to receive PDUs in strict mode
983    strict: bool,
984    /// Timeout for individual socket Reads
985    read_timeout: Option<Duration>,
986    /// Timeout for individual socket Writes.
987    write_timeout: Option<Duration>,
988    /// Buffer to assemble PDU before parsing
989    read_buffer: BytesMut,
990    /// User variables that were taken from the server
991    user_variables: Vec<UserVariableItem>,
992    /// The AE title of the peer
993    peer_ae_title: String,
994}
995
996impl<S> Association for ClientAssociation<S>
997where
998    S: CloseSocket + std::io::Read + std::io::Write,
999{
1000    fn peer_ae_title(&self) -> &str {
1001        &self.peer_ae_title
1002    }
1003
1004    /// Retrieve the maximum PDU length
1005    /// that the association acceptor is expecting to receive.
1006    fn acceptor_max_pdu_length(&self) -> u32 {
1007        self.acceptor_max_pdu_length
1008    }
1009
1010    /// Retrieve the maximum PDU length
1011    /// that the association requestor is expecting to receive.
1012    fn requestor_max_pdu_length(&self) -> u32 {
1013        self.requestor_max_pdu_length
1014    }
1015
1016    /// Retrieve the maximum PDU length that this application entity
1017    /// (the association requestor) is expecting to receive.
1018    fn local_max_pdu_length(&self) -> u32 {
1019        self.requestor_max_pdu_length
1020    }
1021
1022    /// Retrieve the maximum PDU length that the peer application entity
1023    /// (the association acceptor) is expecting to receive.
1024    fn peer_max_pdu_length(&self) -> u32 {
1025        self.acceptor_max_pdu_length
1026    }
1027
1028    fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
1029        &self.presentation_contexts
1030    }
1031
1032    fn user_variables(&self) -> &[UserVariableItem] {
1033        &self.user_variables
1034    }
1035}
1036
1037impl<S> ClientAssociation<S>
1038where
1039    S: CloseSocket + std::io::Read + std::io::Write,
1040{
1041    /// Retrieve read timeout for the association
1042    pub fn read_timeout(&self) -> Option<Duration> {
1043        self.read_timeout
1044    }
1045
1046    /// Retrieve write timeout for the association
1047    pub fn write_timeout(&self) -> Option<Duration> {
1048        self.write_timeout
1049    }
1050
1051    /// Retrieve the maximum PDU length
1052    /// that the association acceptor is expecting to receive.
1053    pub fn acceptor_max_pdu_length(&self) -> u32 {
1054        self.acceptor_max_pdu_length
1055    }
1056
1057    /// Retrieve the maximum PDU length
1058    /// that the association requestor is expecting to receive.
1059    pub fn requestor_max_pdu_length(&self) -> u32 {
1060        self.requestor_max_pdu_length
1061    }
1062
1063    /// Retrieve the user variables that were taken from the server.
1064    ///
1065    /// It usually contains the maximum PDU length,
1066    /// the implementation class UID, and the implementation version name.
1067    pub fn user_variables(&self) -> &[UserVariableItem] {
1068        &self.user_variables
1069    }
1070
1071    /// Retrieve the list of negotiated presentation contexts.
1072    pub fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
1073        &self.presentation_contexts
1074    }
1075}
1076
1077// compatibility filler, remove in 0.10.0
1078impl<S> ClientAssociation<S>
1079where
1080    S: CloseSocket + std::io::Read + std::io::Write,
1081{
1082    /// Send a PDU message to the other intervenient.
1083    pub fn send(&mut self, pdu: &Pdu) -> Result<()> {
1084        SyncAssociation::send(self, pdu)
1085    }
1086
1087    /// Read a PDU message from the other intervenient.
1088    pub fn receive(&mut self) -> Result<Pdu> {
1089        SyncAssociation::receive(self)
1090    }
1091
1092    /// Prepare a P-Data writer for sending
1093    /// one or more data item PDUs.
1094    ///
1095    /// Returns a writer which automatically
1096    /// splits the inner data into separate PDUs if necessary.
1097    pub fn send_pdata(
1098        &mut self,
1099        presentation_context_id: u8,
1100    ) -> crate::association::pdata::PDataWriter<&mut S> {
1101        SyncAssociation::send_pdata(self, presentation_context_id)
1102    }
1103
1104    /// Iniate a graceful release of the association.
1105    ///
1106    /// A DIMSE A-RELEASE transaction is initiated by this application entity,
1107    /// and the underlying socket is closed once settled.
1108    ///
1109    /// Note that as of version 0.9.1,
1110    /// `ClientAssociation` no longer calls this method on [`Drop`],
1111    /// so remember to call `release` explicitly
1112    /// at the end of all DIMSE transactions.
1113    pub fn release(self) -> Result<()> {
1114        SyncAssociation::release(self)
1115    }
1116
1117    /// Send a provider initiated abort message
1118    /// and shut down the TCP connection,
1119    /// terminating the association.
1120    pub fn abort(self) -> Result<()> {
1121        SyncAssociation::abort(self)
1122    }
1123
1124    /// Prepare a P-Data reader for receiving
1125    /// one or more data item PDUs.
1126    ///
1127    /// Returns a reader which automatically
1128    /// receives more data PDUs once the bytes collected are consumed.
1129    pub fn receive_pdata(&mut self) -> crate::association::pdata::PDataReader<'_, &mut S> {
1130        SyncAssociation::receive_pdata(self)
1131    }
1132
1133    /// Obtain access to the inner stream
1134    /// connected to the association acceptor.
1135    ///
1136    /// This can be used to send the PDU in semantic fragments of the message,
1137    /// thus using less memory.
1138    ///
1139    /// **Note:** reading and writing should be done with care
1140    /// to avoid inconsistencies in the association state.
1141    /// Do not call `send` and `receive` while not in a PDU boundary.
1142    pub fn inner_stream(&mut self) -> &mut S {
1143        SyncAssociation::inner_stream(self)
1144    }
1145}
1146
1147impl<S> SyncAssociationSealed<S> for ClientAssociation<S>
1148where
1149    S: CloseSocket + std::io::Read + std::io::Write,
1150{
1151    /// Send a PDU message to the other intervenient.
1152    fn send(&mut self, pdu: &Pdu) -> Result<()> {
1153        self.write_buffer.clear();
1154        encode_pdu(
1155            &mut self.write_buffer,
1156            pdu,
1157            self.acceptor_max_pdu_length + PDU_HEADER_SIZE,
1158        )?;
1159        self.socket
1160            .write_all(&self.write_buffer)
1161            .context(super::WireSendSnafu)
1162    }
1163
1164    /// Read a PDU message from the other intervenient.
1165    fn receive(&mut self) -> Result<Pdu> {
1166        read_pdu_from_wire(
1167            &mut self.socket,
1168            &mut self.read_buffer,
1169            self.requestor_max_pdu_length,
1170            self.strict,
1171        )
1172    }
1173
1174    fn close(&mut self) -> std::io::Result<()> {
1175        self.socket.close()
1176    }
1177}
1178
1179impl<S> SyncAssociation<S> for ClientAssociation<S>
1180where
1181    S: CloseSocket + std::io::Read + std::io::Write,
1182{
1183    fn inner_stream(&mut self) -> &mut S {
1184        &mut self.socket
1185    }
1186
1187    fn get_mut(&mut self) -> (&mut S, &mut BytesMut) {
1188        let Self {
1189            socket,
1190            read_buffer,
1191            ..
1192        } = self;
1193        (socket, read_buffer)
1194    }
1195}
1196
1197/// Trait with the behavior to synchronously release an association
1198#[deprecated(since = "0.9.1", note = "Call `SyncAssociation::release` instead")]
1199pub trait Release {
1200    #[deprecated(since = "0.9.1", note = "Call `SyncAssociation::release` instead")]
1201    fn release(&mut self) -> Result<()>;
1202}
1203
1204#[allow(deprecated)]
1205impl Release for ClientAssociation<std::net::TcpStream> {
1206    fn release(&mut self) -> Result<()> {
1207        SyncAssociationSealed::release(self)
1208    }
1209}
1210
1211#[cfg(feature = "async")]
1212/// Initiate simple TCP connection to the given address
1213pub(crate) async fn async_connection<T>(
1214    ae_address: &AeAddr<T>,
1215    opts: &SocketOptions,
1216) -> Result<tokio::net::TcpStream>
1217where
1218    T: tokio::net::ToSocketAddrs,
1219{
1220    super::timeout(opts.connection_timeout, async {
1221        tokio::net::TcpStream::connect(ae_address.socket_addr())
1222            .await
1223            .context(crate::association::ConnectSnafu)
1224    })
1225    .await
1226}
1227
1228/// Initiate TLS connection to the given address
1229#[cfg(feature = "async-tls")]
1230pub(crate) async fn async_tls_connection<T>(
1231    ae_address: &AeAddr<T>,
1232    server_name: &str,
1233    opts: &SocketOptions,
1234    tls_config: std::sync::Arc<rustls::ClientConfig>,
1235) -> Result<AsyncTlsStream>
1236where
1237    T: tokio::net::ToSocketAddrs,
1238{
1239    use rustls::pki_types::ServerName;
1240    use std::convert::TryFrom;
1241
1242    let tcp_stream = async_connection(ae_address, opts).await?;
1243    let connector = tokio_rustls::TlsConnector::from(tls_config);
1244    let domain = ServerName::try_from(server_name.to_string())
1245        .context(crate::association::InvalidServerNameSnafu)?;
1246    // NOTE: When tokio-rustls is updated to return a rustls::Error instead of std::io::Error,
1247    // switch to `crate::association::TlsConnectionSnafu` for context.
1248    let tls_stream = connector
1249        .connect(domain, tcp_stream)
1250        .await
1251        .context(crate::association::ConnectSnafu)?;
1252    Ok(tls_stream)
1253}
1254
1255/// A DICOM upper level association from the perspective
1256/// of a requesting application entity.
1257///
1258/// The most common operations of an established association are
1259/// [`send`](AsyncAssociation::release) and [`receive`](AsyncAssociation::release).
1260/// Sending large P-Data fragments may be easier through the P-Data sender
1261/// abstraction (see [`send_pdata`](AsyncAssociation::send_pdata)).
1262///
1263/// Call [`release`](AsyncAssociation::release) at the end
1264/// to perform a standard C-RELEASE message exchange
1265/// and shut down the underlying TCP connection.
1266/// Not calling this method will only close the socket
1267/// without gracefully releasing the association.
1268#[cfg(feature = "async")]
1269#[derive(Debug)]
1270pub struct AsyncClientAssociation<S> {
1271    /// The presentation contexts accorded with the acceptor application entity,
1272    /// without the rejected ones.
1273    presentation_contexts: Vec<PresentationContextNegotiated>,
1274    /// The maximum PDU length that this application entity is expecting to receive
1275    requestor_max_pdu_length: u32,
1276    /// The maximum PDU length that the remote application entity accepts
1277    acceptor_max_pdu_length: u32,
1278    /// The TCP stream to the other DICOM node
1279    socket: S,
1280    /// Buffer to assemble PDU before sending it on wire
1281    write_buffer: Vec<u8>,
1282    /// whether to receive PDUs in strict mode
1283    strict: bool,
1284    /// Timeout for individual socket Reads
1285    read_timeout: Option<Duration>,
1286    /// Timeout for individual socket Writes.
1287    write_timeout: Option<Duration>,
1288    /// Buffer to assemble PDU before parsing
1289    read_buffer: BytesMut,
1290    /// User variables that were taken from the server
1291    user_variables: Vec<UserVariableItem>,
1292    /// The AE title of the peer
1293    peer_ae_title: String,
1294}
1295
1296#[cfg(feature = "async")]
1297impl<'a> ClientAssociationOptions<'a> {
1298    async fn establish_impl_async<T, S>(
1299        self,
1300        ae_address: AeAddr<T>,
1301        mut socket: S,
1302    ) -> Result<AsyncClientAssociation<S>>
1303    where
1304        T: tokio::net::ToSocketAddrs,
1305        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
1306    {
1307        use tokio::io::AsyncWriteExt;
1308        let (pc_proposed, a_associate) = self.create_a_associate_req(ae_address.ae_title())?;
1309        let mut write_buffer: Vec<u8> = Vec::with_capacity(DEFAULT_MAX_PDU as usize);
1310
1311        // send request
1312        write_pdu(&mut write_buffer, &a_associate).context(crate::association::SendPduSnafu)?;
1313        super::timeout(self.socket_options.write_timeout, async {
1314            socket
1315                .write_all(&write_buffer)
1316                .await
1317                .context(crate::association::WireSendSnafu)?;
1318            Ok(())
1319        })
1320        .await?;
1321        write_buffer.clear();
1322
1323        // read buffer is prepared according to the requestor's max pdu length
1324        let mut read_buffer = BytesMut::with_capacity(
1325            (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
1326        );
1327        let resp = super::timeout(self.socket_options.read_timeout, async {
1328            super::read_pdu_from_wire_async(
1329                &mut socket,
1330                &mut read_buffer,
1331                self.max_pdu_length,
1332                self.strict,
1333            )
1334            .await
1335        })
1336        .await?;
1337        let negotiated_options = self.process_a_association_resp(resp, &pc_proposed);
1338        match negotiated_options {
1339            Err(e) => {
1340                // abort connection
1341                let _ = write_pdu(
1342                    &mut write_buffer,
1343                    &Pdu::AbortRQ {
1344                        source: AbortRQSource::ServiceUser,
1345                    },
1346                );
1347                socket
1348                    .write_all(&write_buffer)
1349                    .await
1350                    .context(crate::association::WireSendSnafu)?;
1351                write_buffer.clear();
1352                Err(e)
1353            }
1354            Ok(NegotiatedOptions {
1355                presentation_contexts,
1356                peer_max_pdu_length,
1357                user_variables,
1358                peer_ae_title,
1359            }) => {
1360                Ok(AsyncClientAssociation {
1361                    presentation_contexts,
1362                    requestor_max_pdu_length: self.max_pdu_length,
1363                    acceptor_max_pdu_length: peer_max_pdu_length,
1364                    socket,
1365                    write_buffer,
1366                    strict: self.strict,
1367                    // Fixes #589, instead of creating a new buffer, we pass the existing buffer into the Association object.
1368                    read_buffer,
1369                    read_timeout: self.socket_options.read_timeout,
1370                    write_timeout: self.socket_options.write_timeout,
1371                    user_variables,
1372                    peer_ae_title,
1373                })
1374            }
1375        }
1376    }
1377
1378    /// Initiate the TCP connection to the given address
1379    /// and request a new DICOM association,
1380    /// negotiating the presentation contexts in the process.
1381    pub async fn establish_async<A: tokio::net::ToSocketAddrs>(
1382        self,
1383        address: A,
1384    ) -> Result<AsyncClientAssociation<tokio::net::TcpStream>> {
1385        let addr = AeAddr::new_socket_addr(address);
1386        let socket = async_connection(&addr, &self.socket_options).await?;
1387        self.establish_impl_async(addr, socket).await
1388    }
1389
1390    /// Initiate the TCP connection to the given address
1391    /// and request a new DICOM association,
1392    /// negotiating the presentation contexts in the process.
1393    #[cfg(feature = "async-tls")]
1394    pub async fn establish_tls_async<A: tokio::net::ToSocketAddrs>(
1395        self,
1396        address: A,
1397    ) -> Result<AsyncClientAssociation<AsyncTlsStream>> {
1398        match (&self.tls_config, &self.server_name) {
1399            (Some(tls_config), Some(server_name)) => {
1400                let addr = AeAddr::new_socket_addr(address);
1401                let socket = async_tls_connection(
1402                    &addr,
1403                    server_name,
1404                    &self.socket_options,
1405                    tls_config.clone(),
1406                )
1407                .await?;
1408                self.establish_impl_async(addr, socket).await
1409            }
1410            _ => crate::association::TlsConfigMissingSnafu.fail()?,
1411        }
1412    }
1413
1414    /// Initiate async TCP connection to the given address
1415    /// and request a new DICOM association,
1416    /// negotiating the presentation contexts in the process.
1417    ///
1418    /// This method allows you to specify the called AE title
1419    /// alongside with the socket address.
1420    /// See [AeAddr](`crate::AeAddr`) for more details.
1421    /// However, the AE title in this parameter
1422    /// is overridden by any `called_ae_title` option
1423    /// previously received.
1424    ///
1425    /// # Example
1426    ///
1427    /// ```no_run
1428    /// # use dicom_ul::association::client::ClientAssociationOptions;
1429    /// # #[tokio::main]
1430    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
1431    /// let association = ClientAssociationOptions::new()
1432    ///     .with_abstract_syntax("1.2.840.10008.1.1")
1433    ///     // called AE title in address
1434    ///     .establish_with_async("MY-STORAGE@10.0.0.100:104")
1435    ///     .await?;
1436    /// # Ok(())
1437    /// # }
1438    /// ```
1439    #[allow(unreachable_patterns)]
1440    pub async fn establish_with_async(
1441        self,
1442        ae_address: &str,
1443    ) -> Result<AsyncClientAssociation<tokio::net::TcpStream>> {
1444        match ae_address.try_into() {
1445            Ok(ae_address) => {
1446                let socket = async_connection(&ae_address, &self.socket_options).await?;
1447                self.establish_impl_async(ae_address, socket).await
1448            }
1449            Err(_) => {
1450                let addr = AeAddr::new_socket_addr(ae_address);
1451                let socket = async_connection(&addr, &self.socket_options).await?;
1452                self.establish_impl_async(addr, socket).await
1453            }
1454        }
1455    }
1456
1457    /// Initiate async TLS connection to the given address
1458    /// and request a new DICOM association,
1459    /// negotiating the presentation contexts in the process.
1460    ///
1461    /// This method allows you to specify the called AE title
1462    /// alongside with the socket address.
1463    /// See [AeAddr](`crate::AeAddr`) for more details.
1464    /// However, the AE title in this parameter
1465    /// is overridden by any `called_ae_title` option
1466    /// previously received.
1467    ///
1468    /// # Example
1469    ///
1470    /// ```no_run
1471    /// # use dicom_ul::association::client::ClientAssociationOptions;
1472    /// # #[tokio::main]
1473    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
1474    /// let association = ClientAssociationOptions::new()
1475    ///     .with_abstract_syntax("1.2.840.10008.1.1")
1476    ///     // called AE title in address
1477    ///     .establish_with_async_tls("MY-STORAGE@10.0.0.100:104")
1478    ///     .await?;
1479    /// # Ok(())
1480    /// # }
1481    /// ```
1482    #[cfg(feature = "async-tls")]
1483    #[allow(unreachable_patterns)]
1484    pub async fn establish_with_async_tls(
1485        self,
1486        ae_address: &str,
1487    ) -> Result<AsyncClientAssociation<AsyncTlsStream>> {
1488        match (&self.tls_config, &self.server_name) {
1489            (Some(tls_config), Some(server_name)) => match ae_address.try_into() {
1490                Ok(ae_address) => {
1491                    let socket = async_tls_connection(
1492                        &ae_address,
1493                        server_name,
1494                        &self.socket_options,
1495                        tls_config.clone(),
1496                    )
1497                    .await?;
1498                    self.establish_impl_async(ae_address, socket).await
1499                }
1500                Err(_) => {
1501                    let addr = AeAddr::new_socket_addr(ae_address);
1502                    let socket = async_tls_connection(
1503                        &addr,
1504                        server_name,
1505                        &self.socket_options,
1506                        tls_config.clone(),
1507                    )
1508                    .await?;
1509                    self.establish_impl_async(addr, socket).await
1510                }
1511            },
1512            _ => crate::association::TlsConfigMissingSnafu.fail()?,
1513        }
1514    }
1515}
1516
1517#[cfg(feature = "async")]
1518impl<S> Association for AsyncClientAssociation<S> {
1519    fn peer_ae_title(&self) -> &str {
1520        &self.peer_ae_title
1521    }
1522
1523    /// Retrieve the maximum PDU length
1524    /// that the association acceptor is expecting to receive.
1525    fn acceptor_max_pdu_length(&self) -> u32 {
1526        self.acceptor_max_pdu_length
1527    }
1528
1529    /// Retrieve the maximum PDU length
1530    /// that the association requestor is expecting to receive.
1531    fn requestor_max_pdu_length(&self) -> u32 {
1532        self.requestor_max_pdu_length
1533    }
1534
1535    /// Retrieve the maximum PDU length that this application entity
1536    /// (the association requestor) is expecting to receive.
1537    fn local_max_pdu_length(&self) -> u32 {
1538        self.requestor_max_pdu_length
1539    }
1540
1541    /// Retrieve the maximum PDU length that the peer application entity
1542    /// (the association acceptor) is expecting to receive.
1543    fn peer_max_pdu_length(&self) -> u32 {
1544        self.acceptor_max_pdu_length
1545    }
1546
1547    fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
1548        &self.presentation_contexts
1549    }
1550
1551    fn user_variables(&self) -> &[UserVariableItem] {
1552        &self.user_variables
1553    }
1554}
1555
1556#[cfg(feature = "async")]
1557impl<S> AsyncClientAssociation<S> {
1558    /// Retrieve read timeout for the association
1559    pub fn read_timeout(&self) -> Option<Duration> {
1560        self.read_timeout
1561    }
1562
1563    /// Retrieve write timeout for the association
1564    pub fn write_timeout(&self) -> Option<Duration> {
1565        self.write_timeout
1566    }
1567
1568    /// Retrieve the maximum PDU length
1569    /// that the association acceptor is expecting to receive.
1570    pub fn acceptor_max_pdu_length(&self) -> u32 {
1571        self.acceptor_max_pdu_length
1572    }
1573
1574    /// Retrieve the maximum PDU length
1575    /// that the association requestor is expecting to receive.
1576    pub fn requestor_max_pdu_length(&self) -> u32 {
1577        self.requestor_max_pdu_length
1578    }
1579
1580    /// Retrieve the user variables that were taken from the server.
1581    ///
1582    /// It usually contains the maximum PDU length,
1583    /// the implementation class UID, and the implementation version name.
1584    pub fn user_variables(&self) -> &[UserVariableItem] {
1585        &self.user_variables
1586    }
1587
1588    /// Retrieve the list of negotiated presentation contexts.
1589    pub fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
1590        &self.presentation_contexts
1591    }
1592}
1593
1594// compatibility filler, remove in 0.10.0
1595#[cfg(feature = "async")]
1596impl<S> AsyncClientAssociation<S>
1597where
1598    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
1599{
1600    /// Obtain access to the inner stream
1601    /// connected to the association acceptor.
1602    ///
1603    /// This can be used to send the PDU in semantic fragments of the message,
1604    /// thus using less memory.
1605    ///
1606    /// **Note:** reading and writing should be done with care
1607    /// to avoid inconsistencies in the association state.
1608    /// Do not call `send` and `receive` while not in a PDU boundary.
1609    pub fn inner_stream(&mut self) -> &mut S {
1610        AsyncAssociation::inner_stream(self)
1611    }
1612
1613    /// Send a PDU message to the other intervenient.
1614    pub async fn send(&mut self, msg: &Pdu) -> Result<()> {
1615        AsyncAssociation::send(self, msg).await
1616    }
1617
1618    /// Read a PDU message from the other intervenient.
1619    pub async fn receive(&mut self) -> Result<Pdu> {
1620        AsyncAssociation::receive(self).await
1621    }
1622
1623    /// Iniate a graceful release of the association.
1624    ///
1625    /// A DIMSE A-RELEASE transaction is initiated by this application entity,
1626    /// and the underlying socket is closed once settled.
1627    ///
1628    /// Note that implementers of this trait
1629    /// do not try to release the association on [`Drop`],
1630    /// so remember to call `release` explicitly
1631    /// at the end of all DIMSE transactions.
1632    pub async fn release(self) -> Result<()> {
1633        AsyncAssociation::release(self).await
1634    }
1635
1636    /// Send a provider initiated abort message
1637    /// and shut down the TCP connection,
1638    /// terminating the association.
1639    pub async fn abort(self) -> Result<()> {
1640        AsyncAssociation::abort(self).await
1641    }
1642
1643    /// Prepare a P-Data writer for sending
1644    /// one or more data item PDUs.
1645    ///
1646    /// Returns a writer which automatically
1647    /// splits the inner data into separate PDUs if necessary.
1648    pub fn send_pdata(
1649        &mut self,
1650        presentation_context_id: u8,
1651    ) -> crate::association::pdata::non_blocking::AsyncPDataWriter<&mut S> {
1652        AsyncAssociation::send_pdata(self, presentation_context_id)
1653    }
1654
1655    /// Prepare a P-Data reader for receiving
1656    /// one or more data item PDUs.
1657    ///
1658    /// Returns a reader which automatically
1659    /// receives more data PDUs once the bytes collected are consumed.
1660    pub fn receive_pdata(&mut self) -> crate::association::pdata::PDataReader<'_, &mut S> {
1661        AsyncAssociation::receive_pdata(self)
1662    }
1663}
1664
1665#[cfg(feature = "async")]
1666impl<S> super::private::AsyncAssociationSealed<S> for AsyncClientAssociation<S>
1667where
1668    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
1669{
1670    async fn send(&mut self, msg: &Pdu) -> Result<()> {
1671        use tokio::io::AsyncWriteExt;
1672
1673        self.write_buffer.clear();
1674        encode_pdu(
1675            &mut self.write_buffer,
1676            msg,
1677            self.acceptor_max_pdu_length + PDU_HEADER_SIZE,
1678        )?;
1679        super::timeout(self.write_timeout, async {
1680            self.socket
1681                .write_all(&self.write_buffer)
1682                .await
1683                .context(crate::association::WireSendSnafu)
1684        })
1685        .await
1686    }
1687
1688    async fn receive(&mut self) -> Result<Pdu> {
1689        use crate::association::read_pdu_from_wire_async;
1690        super::timeout(self.read_timeout, async {
1691            read_pdu_from_wire_async(
1692                &mut self.socket,
1693                &mut self.read_buffer,
1694                self.requestor_max_pdu_length,
1695                self.strict,
1696            )
1697            .await
1698        })
1699        .await
1700    }
1701
1702    async fn close(&mut self) -> std::io::Result<()> {
1703        use tokio::io::AsyncWriteExt;
1704        self.socket.shutdown().await
1705    }
1706}
1707
1708#[cfg(feature = "async")]
1709impl<S> AsyncAssociation<S> for AsyncClientAssociation<S>
1710where
1711    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
1712{
1713    fn inner_stream(&mut self) -> &mut S {
1714        &mut self.socket
1715    }
1716
1717    fn get_mut(&mut self) -> (&mut S, &mut BytesMut) {
1718        let Self {
1719            socket,
1720            read_buffer,
1721            ..
1722        } = self;
1723        (socket, read_buffer)
1724    }
1725}
1726
1727#[cfg(test)]
1728mod tests {
1729    use super::*;
1730    #[cfg(feature = "async")]
1731    use crate::association::read_pdu_from_wire_async;
1732    use std::io::Write;
1733
1734    impl<'a> ClientAssociationOptions<'a> {
1735        pub(crate) fn establish_with_extra_pdus<T>(
1736            &self,
1737            ae_address: AeAddr<T>,
1738            extra_pdus: Vec<Pdu>,
1739        ) -> Result<ClientAssociation<std::net::TcpStream>>
1740        where
1741            T: ToSocketAddrs,
1742        {
1743            let (pc_proposed, a_associate) = self.create_a_associate_req(ae_address.ae_title())?;
1744            let mut socket = tcp_connection(&ae_address, &self.socket_options)?;
1745            let mut write_buffer: Vec<u8> = Vec::with_capacity(DEFAULT_MAX_PDU as usize);
1746            // send request
1747
1748            write_pdu(&mut write_buffer, &a_associate).context(crate::association::SendPduSnafu)?;
1749            for pdu in extra_pdus {
1750                write_pdu(&mut write_buffer, &pdu).context(crate::association::SendPduSnafu)?;
1751            }
1752            socket
1753                .write_all(&write_buffer)
1754                .context(crate::association::WireSendSnafu)?;
1755            write_buffer.clear();
1756
1757            let mut read_buffer = BytesMut::with_capacity(
1758                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
1759            );
1760            let resp = read_pdu_from_wire(
1761                &mut socket,
1762                &mut read_buffer,
1763                self.max_pdu_length,
1764                self.strict,
1765            )?;
1766            let NegotiatedOptions {
1767                presentation_contexts,
1768                peer_max_pdu_length,
1769                user_variables,
1770                peer_ae_title,
1771            } = self
1772                .process_a_association_resp(resp, &pc_proposed)
1773                .expect("Failed to process a associate response");
1774            Ok(ClientAssociation {
1775                presentation_contexts,
1776                requestor_max_pdu_length: self.max_pdu_length,
1777                acceptor_max_pdu_length: peer_max_pdu_length,
1778                socket,
1779                write_buffer,
1780                strict: self.strict,
1781                // Fixes #589, instead of creating a new buffer, we pass the existing buffer into the Association object.
1782                read_buffer,
1783                read_timeout: self.socket_options.read_timeout,
1784                write_timeout: self.socket_options.write_timeout,
1785                user_variables,
1786                peer_ae_title,
1787            })
1788        }
1789
1790        #[cfg(feature = "async")]
1791        pub(crate) async fn establish_with_extra_pdus_async<T>(
1792            &self,
1793            ae_address: AeAddr<T>,
1794            extra_pdus: Vec<Pdu>,
1795        ) -> Result<AsyncClientAssociation<tokio::net::TcpStream>>
1796        where
1797            T: tokio::net::ToSocketAddrs,
1798        {
1799            use tokio::io::AsyncWriteExt;
1800
1801            let (pc_proposed, a_associate) = self.create_a_associate_req(ae_address.ae_title())?;
1802            let mut socket = async_connection(&ae_address, &self.socket_options).await?;
1803            let mut buffer: Vec<u8> = Vec::with_capacity(DEFAULT_MAX_PDU as usize);
1804            // send request
1805
1806            write_pdu(&mut buffer, &a_associate).context(crate::association::SendPduSnafu)?;
1807            for pdu in extra_pdus {
1808                write_pdu(&mut buffer, &pdu).context(crate::association::SendPduSnafu)?;
1809            }
1810            socket
1811                .write_all(&buffer)
1812                .await
1813                .context(crate::association::WireSendSnafu)?;
1814            buffer.clear();
1815
1816            let mut buf = BytesMut::with_capacity(
1817                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
1818            );
1819            let resp =
1820                read_pdu_from_wire_async(&mut socket, &mut buf, self.max_pdu_length, self.strict)
1821                    .await?;
1822            let NegotiatedOptions {
1823                presentation_contexts,
1824                peer_max_pdu_length,
1825                user_variables,
1826                peer_ae_title,
1827            } = self
1828                .process_a_association_resp(resp, &pc_proposed)
1829                .expect("Failed to process a associate response");
1830            Ok(AsyncClientAssociation {
1831                presentation_contexts,
1832                requestor_max_pdu_length: self.max_pdu_length,
1833                acceptor_max_pdu_length: peer_max_pdu_length,
1834                socket,
1835                write_buffer: buffer,
1836                strict: self.strict,
1837                // Fixes #589, instead of creating a new buffer, we pass the existing buffer into the Association object.
1838                read_buffer: buf,
1839                read_timeout: self.socket_options.read_timeout,
1840                write_timeout: self.socket_options.write_timeout,
1841                user_variables,
1842                peer_ae_title,
1843            })
1844        }
1845
1846        // Broken implementation of server establish which reproduces behavior that #589 introduced
1847        pub fn broken_establish<T>(
1848            &self,
1849            ae_address: AeAddr<T>,
1850        ) -> Result<ClientAssociation<std::net::TcpStream>>
1851        where
1852            T: ToSocketAddrs,
1853        {
1854            let (pc_proposed, a_associate) = self.create_a_associate_req(ae_address.ae_title())?;
1855            let mut socket = tcp_connection(&ae_address, &self.socket_options)?;
1856            let mut buffer: Vec<u8> = Vec::with_capacity(DEFAULT_MAX_PDU as usize);
1857            // send request
1858            write_pdu(&mut buffer, &a_associate).context(crate::association::SendPduSnafu)?;
1859            socket
1860                .write_all(&buffer)
1861                .context(crate::association::WireSendSnafu)?;
1862            buffer.clear();
1863
1864            let mut buf = BytesMut::with_capacity(
1865                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
1866            );
1867            let resp = read_pdu_from_wire(&mut socket, &mut buf, self.max_pdu_length, self.strict)?;
1868            let NegotiatedOptions {
1869                presentation_contexts,
1870                peer_max_pdu_length,
1871                user_variables,
1872                peer_ae_title,
1873            } = self
1874                .process_a_association_resp(resp, &pc_proposed)
1875                .expect("Failed to process a associate response");
1876            Ok(ClientAssociation {
1877                presentation_contexts,
1878                requestor_max_pdu_length: self.max_pdu_length,
1879                acceptor_max_pdu_length: peer_max_pdu_length,
1880                socket,
1881                write_buffer: buffer,
1882                strict: self.strict,
1883                read_buffer: BytesMut::with_capacity(
1884                    (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
1885                ),
1886                read_timeout: self.socket_options.read_timeout,
1887                write_timeout: self.socket_options.write_timeout,
1888                user_variables,
1889                peer_ae_title,
1890            })
1891        }
1892
1893        #[cfg(feature = "async")]
1894        // Broken implementation of server establish which reproduces behavior that #589 introduced
1895        pub async fn broken_establish_async<T>(
1896            &self,
1897            ae_address: AeAddr<T>,
1898        ) -> Result<AsyncClientAssociation<tokio::net::TcpStream>>
1899        where
1900            T: tokio::net::ToSocketAddrs,
1901        {
1902            use tokio::io::AsyncWriteExt;
1903
1904            let (pc_proposed, a_associate) = self.create_a_associate_req(ae_address.ae_title())?;
1905            let mut socket = async_connection(&ae_address, &self.socket_options).await?;
1906            let mut buffer: Vec<u8> = Vec::with_capacity(DEFAULT_MAX_PDU as usize);
1907            // send request
1908            write_pdu(&mut buffer, &a_associate).context(crate::association::SendPduSnafu)?;
1909            socket
1910                .write_all(&buffer)
1911                .await
1912                .context(crate::association::WireSendSnafu)?;
1913            buffer.clear();
1914
1915            let mut buf = BytesMut::with_capacity(
1916                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
1917            );
1918            let resp =
1919                read_pdu_from_wire_async(&mut socket, &mut buf, self.max_pdu_length, self.strict)
1920                    .await?;
1921            let NegotiatedOptions {
1922                presentation_contexts,
1923                peer_max_pdu_length,
1924                user_variables,
1925                peer_ae_title,
1926            } = self
1927                .process_a_association_resp(resp, &pc_proposed)
1928                .expect("Failed to process a associate response");
1929            Ok(AsyncClientAssociation {
1930                presentation_contexts,
1931                requestor_max_pdu_length: self.max_pdu_length,
1932                acceptor_max_pdu_length: peer_max_pdu_length,
1933                socket,
1934                write_buffer: buffer,
1935                strict: self.strict,
1936                read_buffer: BytesMut::with_capacity(
1937                    (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
1938                ),
1939                read_timeout: self.socket_options.read_timeout,
1940                write_timeout: self.socket_options.write_timeout,
1941                user_variables,
1942                peer_ae_title,
1943            })
1944        }
1945    }
1946}
dicom_ul/association/client.rs

dicom_ul/association/
client.rs
