netget/server/dicom/
mod.rs

1//! DICOM DIMSE service provider. Rust owns the upper layer (association negotiation, PDU and
2//! PDV framing, release, abort), the DIMSE command sets, the dataset codec, C-ECHO and C-FIND
3//! matching; the handler decides associations, stored instances and query records.
4pub mod actions;
5pub mod dataset;
6pub mod pdu;
7
8use crate::llm::action_helper::call_llm;
9use crate::llm::actions::protocol_trait::ActionResult;
10use crate::logging::emit::Log;
11use crate::protocol::{Event, SpawnContext};
12use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
13use crate::server::connection::ConnectionId;
14use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
15use anyhow::{bail, Result};
16use dataset::{EXPLICIT_LE, IMPLICIT_LE};
17use pdu::{Pdu, C_CANCEL_RQ, C_ECHO_RQ, C_FIND_RQ, C_STORE_RQ, NO_DATASET};
18use serde_json::{json, Map, Value};
19use std::collections::HashMap;
20use std::{net::SocketAddr, sync::Arc, time::Duration};
21use tokio::io::AsyncWriteExt;
22
23pub const DEFAULT_AE_TITLE: &str = "NETGET";
24pub const IDLE_TIMEOUT: Duration = Duration::from_secs(60);
25/// The ARTIM timer: how long a new connection may take to send A-ASSOCIATE-RQ.
26const ARTIM: Duration = Duration::from_secs(30);
27
28struct Shared {
29    ctx: SpawnContext,
30    ae_title: String,
31    idle: Duration,
32}
33
34pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
35    let p = ctx.startup_params.as_ref();
36    let ae_title = p
37        .map(|p| p.get_optional_string("ae_title"))
38        .transpose()?
39        .flatten()
40        .unwrap_or_else(|| DEFAULT_AE_TITLE.to_owned());
41    anyhow::ensure!(
42        !ae_title.trim().is_empty()
43            && ae_title.len() <= 16
44            && ae_title
45                .bytes()
46                .all(|b| (0x20..0x7f).contains(&b) && b != b'\\'),
47        "ae_title is 1 to 16 printable characters"
48    );
49    let idle = p
50        .map(|p| p.get_optional_u64("idle_timeout_secs"))
51        .transpose()?
52        .flatten()
53        .unwrap_or(IDLE_TIMEOUT.as_secs());
54    anyhow::ensure!(
55        (1..=3600).contains(&idle),
56        "idle_timeout_secs must be 1..=3600"
57    );
58    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
59    let local = listener.local_addr()?;
60    Log::new(Some(&ctx.status_tx)).info(format!("DICOM SCP {ae_title} on {local}"));
61    let shared = Arc::new(Shared {
62        ctx: ctx.clone(),
63        ae_title: ae_title.trim().to_owned(),
64        idle: Duration::from_secs(idle),
65    });
66    let server_id = ctx.server_id;
67    let accept = tokio::spawn(async move {
68        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
69        loop {
70            // A-ASSOCIATE-RJ (rejected-transient, local limit exceeded) for a refused connection.
71            let refusal = pdu::encode(
72                &Pdu::AssociateRj {
73                    result: 2,
74                    source: 3,
75                    reason: 2,
76                },
77                "",
78                "",
79            );
80            let (stream, peer, permit) = match accept_bounded(
81                &listener,
82                &limiter,
83                &refusal,
84                "DICOM",
85                Some(&shared.ctx.status_tx),
86            )
87            .await
88            {
89                Ok(v) => v,
90                Err(_) => break,
91            };
92            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
93            let now = crate::utils::clock::Instant::now();
94            shared
95                .ctx
96                .state
97                .add_connection_to_server(
98                    server_id,
99                    ConnectionState {
100                        id,
101                        remote_addr: peer,
102                        local_addr: local,
103                        bytes_sent: 0,
104                        bytes_received: 0,
105                        packets_sent: 0,
106                        packets_received: 0,
107                        last_activity: now,
108                        status: ConnectionStatus::Active,
109                        status_changed_at: now,
110                        protocol_info: ProtocolConnectionInfo::empty(),
111                    },
112                )
113                .await;
114            let child = shared.clone();
115            shared
116                .ctx
117                .state
118                .spawn_server_task(server_id, async move {
119                    let _permit = permit;
120                    if let Err(e) = association(&child, id, stream).await {
121                        Log::new(Some(&child.ctx.status_tx))
122                            .debug(format!("DICOM connection {id}: {e}"));
123                    }
124                    child
125                        .ctx
126                        .state
127                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
128                        .await;
129                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
130                })
131                .await;
132        }
133    });
134    ctx.state.register_server_task(server_id, accept).await;
135    Ok(local)
136}
137
138fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
139    let summary = format!("DICOM connection {id} operation={operation} decision={decision}");
140    let log = Log::new(Some(&ctx.status_tx));
141    if decision.starts_with("fail_closed_") || decision == "model_silent" {
142        log.error(summary);
143    } else {
144        log.info(summary);
145    }
146}
147
148fn service(abstract_syntax: &str) -> Option<&'static str> {
149    match abstract_syntax {
150        pdu::VERIFICATION => Some("verification"),
151        pdu::PATIENT_ROOT_FIND => Some("patient_root_find"),
152        pdu::STUDY_ROOT_FIND => Some("study_root_find"),
153        s if s.starts_with(pdu::STORAGE_PREFIX) => Some("storage"),
154        _ => None,
155    }
156}
157
158async fn ask(
159    shared: &Shared,
160    id: ConnectionId,
161    event: Event,
162    operation: &str,
163) -> Result<Value, &'static str> {
164    let ctx = &shared.ctx;
165    let result = match call_llm(
166        &ctx.llm_client,
167        &ctx.state,
168        ctx.server_id,
169        Some(id),
170        &event,
171        &actions::DicomProtocol,
172    )
173    .await
174    {
175        Ok(r) => r,
176        Err(_) => {
177            outcome(ctx, id, operation, "fail_closed_llm_error");
178            return Err("fail_closed_llm_error");
179        }
180    };
181    if !result.failures.is_empty() {
182        outcome(ctx, id, operation, "fail_closed_invalid_reply");
183        return Err("fail_closed_invalid_reply");
184    }
185    let mut answers = Vec::new();
186    let mut pending = result.protocol_results;
187    while let Some(r) = pending.pop() {
188        match r {
189            ActionResult::Custom { data, .. } => answers.push(data),
190            ActionResult::Multiple(items) => pending.extend(items),
191            _ => {}
192        }
193    }
194    match answers.len() {
195        1 => Ok(answers.remove(0)),
196        0 => {
197            outcome(ctx, id, operation, "model_silent");
198            Err("model_silent")
199        }
200        _ => {
201            outcome(ctx, id, operation, "fail_closed_invalid_reply");
202            Err("fail_closed_invalid_reply")
203        }
204    }
205}
206
207async fn send<W: tokio::io::AsyncWrite + Unpin>(
208    shared: &Shared,
209    id: ConnectionId,
210    w: &mut W,
211    bytes: &[u8],
212) -> Result<()> {
213    w.write_all(bytes).await?;
214    shared
215        .ctx
216        .state
217        .update_connection_stats(
218            shared.ctx.server_id,
219            id,
220            None,
221            Some(bytes.len() as u64),
222            None,
223            Some(1),
224        )
225        .await;
226    Ok(())
227}
228
229async fn association(
230    shared: &Shared,
231    id: ConnectionId,
232    stream: tokio::net::TcpStream,
233) -> Result<()> {
234    let ctx = &shared.ctx;
235    let (mut r, mut w) = tokio::io::split(stream);
236    let first = match tokio::time::timeout(ARTIM, pdu::read(&mut r)).await {
237        Ok(Ok(Some(p))) => p,
238        Ok(Err(e)) => {
239            send(
240                shared,
241                id,
242                &mut w,
243                &pdu::encode(
244                    &Pdu::Abort {
245                        source: 2,
246                        reason: 2,
247                    },
248                    "",
249                    "",
250                ),
251            )
252            .await
253            .ok();
254            return Err(e);
255        }
256        _ => return Ok(()),
257    };
258    let Pdu::AssociateRq(rq) = first else {
259        send(
260            shared,
261            id,
262            &mut w,
263            &pdu::encode(
264                &Pdu::Abort {
265                    source: 2,
266                    reason: 2,
267                },
268                "",
269                "",
270            ),
271        )
272        .await?;
273        bail!("expected A-ASSOCIATE-RQ");
274    };
275    let reject = |reason: u8| {
276        pdu::encode(
277            &Pdu::AssociateRj {
278                result: 1,
279                source: 1,
280                reason,
281            },
282            "",
283            "",
284        )
285    };
286    if rq.called != shared.ae_title {
287        outcome(ctx, id, "associate", "protocol_refusal");
288        send(shared, id, &mut w, &reject(7)).await?;
289        return Ok(());
290    }
291    let contexts: Vec<Value> = rq.contexts.iter().map(|c| json!({"id": c.id, "abstract_syntax": c.abstract_syntax, "service": service(&c.abstract_syntax), "transfer_syntaxes": c.transfer_syntaxes})).collect();
292    let event = Event::new(
293        &actions::ASSOCIATE_EVENT,
294        json!({"calling_ae": rq.calling, "called_ae": rq.called, "contexts": contexts, "implementation": rq.implementation}),
295    );
296    match ask(shared, id, event, "associate").await {
297        Ok(a) if a["type"] == "dicom_accept" => outcome(ctx, id, "associate", "model_answer"),
298        Ok(a) if a["type"] == "dicom_reject" => {
299            outcome(ctx, id, "associate", "model_reject");
300            let reason = match a["reason"].as_str() {
301                Some("called_ae_not_recognized") => 7,
302                Some("no_reason") => 1,
303                _ => 3,
304            };
305            send(shared, id, &mut w, &reject(reason)).await?;
306            return Ok(());
307        }
308        _ => {
309            // Rejected-transient: the service is temporarily unable to decide.
310            send(
311                shared,
312                id,
313                &mut w,
314                &pdu::encode(
315                    &Pdu::AssociateRj {
316                        result: 2,
317                        source: 2,
318                        reason: 1,
319                    },
320                    "",
321                    "",
322                ),
323            )
324            .await?;
325            return Ok(());
326        }
327    }
328    let mut negotiated: HashMap<u8, (String, String)> = HashMap::new();
329    let results: Vec<(u8, u8, String)> = rq
330        .contexts
331        .iter()
332        .map(|c| {
333            if service(&c.abstract_syntax).is_none() {
334                return (c.id, 3, IMPLICIT_LE.to_owned());
335            }
336            let ts = [EXPLICIT_LE, IMPLICIT_LE]
337                .into_iter()
338                .find(|t| c.transfer_syntaxes.iter().any(|x| x == t));
339            match ts {
340                Some(t) => {
341                    negotiated.insert(c.id, (c.abstract_syntax.clone(), t.to_owned()));
342                    (c.id, 0, t.to_owned())
343                }
344                None => (c.id, 4, IMPLICIT_LE.to_owned()),
345            }
346        })
347        .collect();
348    send(
349        shared,
350        id,
351        &mut w,
352        &pdu::encode(
353            &Pdu::AssociateAc(results, pdu::MAX_PDU),
354            &rq.called,
355            &rq.calling,
356        ),
357    )
358    .await?;
359    let peer_max = rq.max_pdu;
360    let mut assembly = pdu::Assembly::default();
361    loop {
362        let next = match tokio::time::timeout(shared.idle, pdu::read(&mut r)).await {
363            Err(_) => {
364                send(
365                    shared,
366                    id,
367                    &mut w,
368                    &pdu::encode(
369                        &Pdu::Abort {
370                            source: 2,
371                            reason: 0,
372                        },
373                        "",
374                        "",
375                    ),
376                )
377                .await
378                .ok();
379                return Ok(());
380            }
381            Ok(Ok(Some(p))) => p,
382            Ok(Ok(None)) => return Ok(()),
383            Ok(Err(e)) => {
384                send(
385                    shared,
386                    id,
387                    &mut w,
388                    &pdu::encode(
389                        &Pdu::Abort {
390                            source: 2,
391                            reason: 2,
392                        },
393                        "",
394                        "",
395                    ),
396                )
397                .await
398                .ok();
399                return Err(e);
400            }
401        };
402        ctx.state
403            .update_connection_stats(ctx.server_id, id, Some(1), None, Some(1), None)
404            .await;
405        match next {
406            Pdu::Data(pdvs) => {
407                if pdvs.iter().any(|(c, _, _)| !negotiated.contains_key(c)) {
408                    send(
409                        shared,
410                        id,
411                        &mut w,
412                        &pdu::encode(
413                            &Pdu::Abort {
414                                source: 2,
415                                reason: 6,
416                            },
417                            "",
418                            "",
419                        ),
420                    )
421                    .await?;
422                    bail!("P-DATA on a context that was not accepted");
423                }
424                let message = match assembly.feed(pdvs) {
425                    Ok(Some(m)) => m,
426                    Ok(None) => continue,
427                    Err(e) => {
428                        send(
429                            shared,
430                            id,
431                            &mut w,
432                            &pdu::encode(
433                                &Pdu::Abort {
434                                    source: 2,
435                                    reason: 6,
436                                },
437                                "",
438                                "",
439                            ),
440                        )
441                        .await?;
442                        return Err(e);
443                    }
444                };
445                let (abs, ts) = negotiated[&message.context].clone();
446                for out in dimse(shared, id, &rq.calling, &abs, &ts, message, peer_max).await {
447                    send(shared, id, &mut w, &out).await?;
448                }
449            }
450            Pdu::ReleaseRq => {
451                send(shared, id, &mut w, &pdu::encode(&Pdu::ReleaseRp, "", "")).await?;
452                let _ = w.shutdown().await;
453                return Ok(());
454            }
455            Pdu::Abort { .. } => return Ok(()),
456            _ => {
457                send(
458                    shared,
459                    id,
460                    &mut w,
461                    &pdu::encode(
462                        &Pdu::Abort {
463                            source: 2,
464                            reason: 2,
465                        },
466                        "",
467                        "",
468                    ),
469                )
470                .await?;
471                bail!("unexpected PDU during the association");
472            }
473        }
474    }
475}
476
477fn response(
478    request_field: u16,
479    message_id: u16,
480    sop_class: &str,
481    status: u16,
482    instance: Option<&str>,
483    has_dataset: bool,
484    comment: Option<&str>,
485) -> Vec<u8> {
486    let mut f = vec![
487        (0x0000_0002u32, pdu::ui(sop_class)),
488        (0x0000_0100, pdu::us(request_field | 0x8000)),
489        (0x0000_0120, pdu::us(message_id)),
490        (
491            0x0000_0800,
492            pdu::us(if has_dataset { 0x0000 } else { NO_DATASET }),
493        ),
494        (0x0000_0900, pdu::us(status)),
495    ];
496    if let Some(i) = instance {
497        f.push((0x0000_1000, pdu::ui(i)));
498    }
499    if let Some(c) = comment {
500        f.push((0x0000_0902, json!({"vr": "LO", "Value": [c]})));
501    }
502    pdu::command(&f)
503}
504
505/// Answer one DIMSE message: the PDUs to send back.
506async fn dimse(
507    shared: &Shared,
508    id: ConnectionId,
509    calling: &str,
510    abstract_syntax: &str,
511    ts: &str,
512    m: pdu::Message,
513    peer_max: u32,
514) -> Vec<Vec<u8>> {
515    let ctx = &shared.ctx;
516    let field = pdu::field_u16(&m.command, 0x0000_0100).unwrap_or(0);
517    let msg_id = pdu::field_u16(&m.command, 0x0000_0110).unwrap_or(0);
518    let sop_class =
519        pdu::field_str(&m.command, 0x0000_0002).unwrap_or_else(|| abstract_syntax.to_owned());
520    let reply = |cmd: Vec<u8>| pdu::data_pdus(m.context, true, &cmd, peer_max);
521    match field {
522        C_ECHO_RQ => {
523            outcome(ctx, id, "c-echo", "protocol_answer");
524            reply(response(
525                C_ECHO_RQ, msg_id, &sop_class, 0x0000, None, false, None,
526            ))
527        }
528        C_STORE_RQ => {
529            let instance = pdu::field_str(&m.command, 0x0000_1000).unwrap_or_default();
530            let bytes = m.dataset.unwrap_or_default();
531            let decoded = dataset::decode(&bytes, ts);
532            let ds = match decoded {
533                Ok(d) => d,
534                Err(_) => {
535                    return reply(response(
536                        C_STORE_RQ,
537                        msg_id,
538                        &sop_class,
539                        0xC000,
540                        Some(&instance),
541                        false,
542                        Some("dataset could not be decoded"),
543                    ))
544                }
545            };
546            if dataset::text(&ds, 0x0008_0018).is_some_and(|i| i != instance)
547                || dataset::text(&ds, 0x0008_0016).is_some_and(|c| c != sop_class)
548            {
549                outcome(ctx, id, "c-store", "protocol_refusal");
550                return reply(response(
551                    C_STORE_RQ,
552                    msg_id,
553                    &sop_class,
554                    0xA900,
555                    Some(&instance),
556                    false,
557                    Some("SOP UIDs differ from the command"),
558                ));
559            }
560            let event = Event::new(
561                &actions::STORE_EVENT,
562                json!({"calling_ae": calling, "sop_class_uid": sop_class, "sop_instance_uid": instance, "dataset": ds, "bytes": bytes.len()}),
563            );
564            let (status, comment) = match ask(shared, id, event, "c-store").await {
565                Ok(a) if a["type"] == "dicom_store_status" => {
566                    let name = a["status"].as_str().unwrap_or_default();
567                    let code = actions::STORE_STATUS
568                        .iter()
569                        .find(|(n, _)| *n == name)
570                        .map(|(_, c)| *c)
571                        .unwrap_or(0xC000);
572                    outcome(
573                        ctx,
574                        id,
575                        "c-store",
576                        if code == 0 || code >> 12 == 0xB {
577                            "model_answer"
578                        } else {
579                            "model_reject"
580                        },
581                    );
582                    (code, a["comment"].as_str().map(str::to_owned))
583                }
584                Ok(_) => {
585                    outcome(ctx, id, "c-store", "fail_closed_invalid_reply");
586                    (0xA700, None)
587                }
588                Err(_) => (
589                    0xA700,
590                    Some("the service cannot store right now".to_owned()),
591                ),
592            };
593            reply(response(
594                C_STORE_RQ,
595                msg_id,
596                &sop_class,
597                status,
598                Some(&instance),
599                false,
600                comment.as_deref(),
601            ))
602        }
603        C_FIND_RQ => {
604            let model = if abstract_syntax == pdu::PATIENT_ROOT_FIND {
605                "patient_root"
606            } else {
607                "study_root"
608            };
609            let identifier = match m.dataset.as_deref().map(|b| dataset::decode(b, ts)) {
610                Some(Ok(d)) => d,
611                _ => {
612                    return reply(response(
613                        C_FIND_RQ,
614                        msg_id,
615                        &sop_class,
616                        0xC000,
617                        None,
618                        false,
619                        Some("identifier could not be decoded"),
620                    ))
621                }
622            };
623            let Some(level) = dataset::text(&identifier, 0x0008_0052)
624                .filter(|l| matches!(l.as_str(), "PATIENT" | "STUDY" | "SERIES" | "IMAGE"))
625            else {
626                outcome(ctx, id, "c-find", "protocol_refusal");
627                return reply(response(
628                    C_FIND_RQ,
629                    msg_id,
630                    &sop_class,
631                    0xA900,
632                    None,
633                    false,
634                    Some("QueryRetrieveLevel is missing or invalid"),
635                ));
636            };
637            let event = Event::new(
638                &actions::FIND_EVENT,
639                json!({"calling_ae": calling, "model": model, "level": level, "identifier": identifier}),
640            );
641            let matches = match ask(shared, id, event, "c-find").await {
642                Ok(a) if a["type"] == "dicom_find_matches" => {
643                    a["matches"].as_array().cloned().unwrap_or_default()
644                }
645                Ok(a) if a["type"] == "dicom_find_failed" => {
646                    outcome(ctx, id, "c-find", "model_reject");
647                    let code = if a["status"] == "out_of_resources" {
648                        0xA700
649                    } else {
650                        0xC000
651                    };
652                    return reply(response(
653                        C_FIND_RQ, msg_id, &sop_class, code, None, false, None,
654                    ));
655                }
656                _ => {
657                    return reply(response(
658                        C_FIND_RQ,
659                        msg_id,
660                        &sop_class,
661                        0xC000,
662                        None,
663                        false,
664                        Some("the service cannot query right now"),
665                    ))
666                }
667            };
668            let mut out = Vec::new();
669            for rec in matches.iter().filter_map(Value::as_object) {
670                if !dataset::matches(&identifier, rec) {
671                    continue;
672                }
673                let projected: Map<String, Value> = dataset::project(&identifier, rec);
674                let Ok(encoded) = dataset::encode(&projected, ts) else {
675                    outcome(ctx, id, "c-find", "fail_closed_invalid_reply");
676                    return reply(response(
677                        C_FIND_RQ,
678                        msg_id,
679                        &sop_class,
680                        0xC000,
681                        None,
682                        false,
683                        Some("a record could not be encoded"),
684                    ));
685                };
686                out.extend(reply(response(
687                    C_FIND_RQ, msg_id, &sop_class, 0xFF00, None, true, None,
688                )));
689                out.extend(pdu::data_pdus(m.context, false, &encoded, peer_max));
690            }
691            outcome(ctx, id, "c-find", "model_answer");
692            out.extend(reply(response(
693                C_FIND_RQ, msg_id, &sop_class, 0x0000, None, false, None,
694            )));
695            out
696        }
697        C_CANCEL_RQ => vec![],
698        other => {
699            outcome(ctx, id, "dimse", "protocol_refusal");
700            reply(response(
701                other, msg_id, &sop_class, 0x0211, None, false, None,
702            ))
703        }
704    }
705}
netget/server/dicom/mod.rs

netget/server/dicom/
mod.rs
