netget/server/smpp/
mod.rs

1//! SMPP 3.4 SMSC. Rust owns PDU framing, the bind state machine, sequence numbers, response
2//! status codes, message ids, enquire_link, unbind, delivery receipts and every bound; the
3//! handler decides binds (unless credentials are configured) and the fate of each message.
4pub mod actions;
5pub mod wire;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, EventType, SpawnContext};
11use crate::server::connection::ConnectionId;
12use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
13use anyhow::{Context, Result};
14use serde_json::{json, Value};
15use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
16use std::sync::Arc;
17use std::{net::SocketAddr, time::Duration};
18use tokio::{
19    io::{AsyncWriteExt, WriteHalf},
20    net::{TcpListener, TcpStream},
21};
22use wire::Pdu;
23
24pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
25pub const MAX_IDLE_TIMEOUT_SECS: u64 = 86400;
26pub const DEFAULT_SYSTEM_ID: &str = "NETGET";
27
28#[derive(Clone)]
29struct Config {
30    system_id: String,
31    credentials: Option<(String, String)>,
32    idle: Duration,
33    next_message_id: Arc<AtomicU64>,
34}
35
36fn config(ctx: &SpawnContext) -> Result<Config> {
37    let params = ctx.startup_params.as_ref();
38    let get = |name: &str| -> Result<Option<String>> {
39        Ok(params
40            .map(|p| p.get_optional_string(name))
41            .transpose()?
42            .flatten())
43    };
44    let system_id = get("system_id")?.unwrap_or_else(|| DEFAULT_SYSTEM_ID.into());
45    anyhow::ensure!(
46        !system_id.is_empty() && system_id.len() <= 15,
47        "system_id must be 1 to 15 characters"
48    );
49    let credentials = match (get("esme_system_id")?, get("password")?) {
50        (Some(id), Some(pw)) => {
51            anyhow::ensure!(
52                id.len() <= 15 && pw.len() <= 8,
53                "esme_system_id is at most 15 characters, password at most 8"
54            );
55            Some((id, pw))
56        }
57        (None, None) => None,
58        _ => anyhow::bail!("esme_system_id and password are configured together"),
59    };
60    let secs = params
61        .map(|p| p.get_optional_u64("idle_timeout_secs"))
62        .transpose()?
63        .flatten()
64        .unwrap_or(IDLE_TIMEOUT.as_secs());
65    anyhow::ensure!(
66        (1..=MAX_IDLE_TIMEOUT_SECS).contains(&secs),
67        "idle_timeout_secs must be between 1 and {MAX_IDLE_TIMEOUT_SECS}"
68    );
69    Ok(Config {
70        system_id,
71        credentials,
72        idle: Duration::from_secs(secs),
73        next_message_id: Arc::new(AtomicU64::new(1)),
74    })
75}
76
77pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
78    let cfg = config(&ctx)?;
79    let listener = TcpListener::bind(ctx.legacy_listen_addr()).await?;
80    let addr = listener.local_addr()?;
81    Log::new(Some(&ctx.status_tx)).info(format!("SMPP SMSC listening on {addr}"));
82    let state = ctx.state.clone();
83    let server_id = ctx.server_id;
84    let accept = tokio::spawn(async move {
85        let limiter = crate::server::accept_bounded::ConnectionLimiter::new(
86            crate::server::accept_bounded::DEFAULT_MAX_CONNECTIONS,
87        );
88        loop {
89            let (socket, peer, permit) = match crate::server::accept_bounded::accept_bounded(
90                &listener,
91                &limiter,
92                b"",
93                "SMPP",
94                Some(&ctx.status_tx),
95            )
96            .await
97            {
98                Ok(v) => v,
99                Err(_) => break,
100            };
101            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
102            let now = crate::utils::clock::Instant::now();
103            ctx.state
104                .add_connection_to_server(
105                    server_id,
106                    ConnectionState {
107                        id,
108                        remote_addr: peer,
109                        local_addr: addr,
110                        bytes_sent: 0,
111                        bytes_received: 0,
112                        packets_sent: 0,
113                        packets_received: 0,
114                        last_activity: now,
115                        status: ConnectionStatus::Active,
116                        status_changed_at: now,
117                        protocol_info: ProtocolConnectionInfo::empty(),
118                    },
119                )
120                .await;
121            let child = ctx.clone();
122            let cfg = cfg.clone();
123            ctx.state
124                .spawn_server_task(server_id, async move {
125                    let _permit = permit;
126                    if let Err(e) = session(&child, id, socket, peer, &cfg).await {
127                        Log::new(Some(&child.status_tx))
128                            .warn(format!("SMPP connection {id} ended: {e}"));
129                    }
130                    child
131                        .state
132                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
133                        .await;
134                    let _ = child.status_tx.send("__UPDATE_UI__".into());
135                })
136                .await;
137        }
138    });
139    state.register_server_task(server_id, accept).await;
140    Ok(addr)
141}
142
143fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
144    let summary = format!("SMPP connection {id} operation={operation} decision={decision}");
145    let log = Log::new(Some(&ctx.status_tx));
146    if decision.starts_with("fail_closed_") || decision == "model_silent" {
147        log.error(summary);
148    } else {
149        log.info(summary);
150    }
151}
152
153/// The handler's one answer, or `Err(status)` for the failure path (already logged).
154async fn decide(
155    ctx: &SpawnContext,
156    id: ConnectionId,
157    event_type: &'static EventType,
158    data: Value,
159    failure_status: u32,
160) -> std::result::Result<(String, Value), u32> {
161    let op = event_type.id.clone();
162    let event = Event::new(event_type, data);
163    let result = match call_llm(
164        &ctx.llm_client,
165        &ctx.state,
166        ctx.server_id,
167        Some(id),
168        &event,
169        &actions::SmppProtocol,
170    )
171    .await
172    {
173        Ok(result) => result,
174        Err(error) => {
175            outcome(ctx, id, &op, "fail_closed_llm_error");
176            return Err(
177                if crate::llm::is_overload_error(&error) && failure_status == wire::ESME_RSYSERR {
178                    wire::ESME_RTHROTTLED
179                } else {
180                    failure_status
181                },
182            );
183        }
184    };
185    if !result.failures.is_empty() {
186        outcome(ctx, id, &op, "fail_closed_invalid_reply");
187        return Err(failure_status);
188    }
189    let mut found = Vec::new();
190    let mut pending = result.protocol_results;
191    while let Some(result) = pending.pop() {
192        match result {
193            ActionResult::Custom { name, data } if name.starts_with("smpp_") => {
194                found.push((name, data))
195            }
196            ActionResult::Multiple(items) => pending.extend(items),
197            _ => {}
198        }
199    }
200    match found.len() {
201        1 => Ok(found.remove(0)),
202        0 => {
203            outcome(ctx, id, &op, "model_silent");
204            Err(failure_status)
205        }
206        _ => {
207            outcome(ctx, id, &op, "fail_closed_invalid_reply");
208            Err(failure_status)
209        }
210    }
211}
212
213struct Session<'a> {
214    ctx: &'a SpawnContext,
215    id: ConnectionId,
216    writer: WriteHalf<TcpStream>,
217    next_sequence: AtomicU32,
218}
219
220impl Session<'_> {
221    async fn send(&mut self, pdu: Pdu) -> Result<()> {
222        let bytes = pdu.encode();
223        tokio::time::timeout(wire::IO_TIMEOUT, self.writer.write_all(&bytes))
224            .await
225            .context("SMPP write deadline")??;
226        self.ctx
227            .state
228            .update_connection_stats(
229                self.ctx.server_id,
230                self.id,
231                None,
232                Some(bytes.len() as u64),
233                None,
234                Some(1),
235            )
236            .await;
237        Ok(())
238    }
239
240    async fn respond(&mut self, req: &Pdu, status: u32, body: Vec<u8>) -> Result<()> {
241        self.send(Pdu::new(
242            req.command_id | wire::RESP,
243            status,
244            req.sequence,
245            body,
246        ))
247        .await
248    }
249
250    fn sequence(&self) -> u32 {
251        // SMPP sequence numbers run 1..=0x7FFFFFFF.
252        let n = self.next_sequence.fetch_add(1, Ordering::Relaxed);
253        (n % 0x7FFF_FFFF) + 1
254    }
255}
256
257fn mode_name(command_id: u32) -> &'static str {
258    match command_id {
259        wire::BIND_RECEIVER => "receiver",
260        wire::BIND_TRANSMITTER => "transmitter",
261        _ => "transceiver",
262    }
263}
264
265fn same_secret(a: &[u8], b: &[u8]) -> bool {
266    a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
267}
268
269async fn session(
270    ctx: &SpawnContext,
271    id: ConnectionId,
272    socket: TcpStream,
273    peer: SocketAddr,
274    cfg: &Config,
275) -> Result<()> {
276    let (mut reader, writer) = tokio::io::split(socket);
277    let mut s = Session {
278        ctx,
279        id,
280        writer,
281        next_sequence: AtomicU32::new(0),
282    };
283    let mut bound: Option<(&'static str, String)> = None;
284    loop {
285        let Some(pdu) = wire::read_pdu(&mut reader, cfg.idle).await? else {
286            return Ok(());
287        };
288        ctx.state
289            .update_connection_stats(
290                ctx.server_id,
291                id,
292                Some(pdu.body.len() as u64 + 16),
293                None,
294                Some(1),
295                None,
296            )
297            .await;
298        match pdu.command_id {
299            wire::BIND_RECEIVER | wire::BIND_TRANSMITTER | wire::BIND_TRANSCEIVER => {
300                if bound.is_some() {
301                    s.respond(&pdu, wire::ESME_RALYBND, Vec::new()).await?;
302                    continue;
303                }
304                let Ok(bind) = wire::parse_bind(&pdu.body) else {
305                    s.send(Pdu::new(
306                        wire::GENERIC_NACK,
307                        wire::ESME_RINVCMDLEN,
308                        pdu.sequence,
309                        Vec::new(),
310                    ))
311                    .await?;
312                    return Ok(());
313                };
314                let mode = mode_name(pdu.command_id);
315                let status = match &cfg.credentials {
316                    Some((user, pass)) => {
317                        if bind.system_id != *user {
318                            wire::ESME_RINVSYSID
319                        } else if !same_secret(bind.password.as_bytes(), pass.as_bytes()) {
320                            wire::ESME_RINVPASWD
321                        } else {
322                            wire::ESME_ROK
323                        }
324                    }
325                    None => match decide(
326                        ctx,
327                        id,
328                        &actions::BIND_EVENT,
329                        json!({
330                            "mode": mode,
331                            "system_id": bind.system_id,
332                            "password": bind.password,
333                            "system_type": bind.system_type,
334                            "interface_version": bind.interface_version,
335                            "remote_addr": peer.to_string(),
336                        }),
337                        wire::ESME_RBINDFAIL,
338                    )
339                    .await
340                    {
341                        Ok((name, _)) if name == "smpp_bind_accept" => {
342                            outcome(ctx, id, "smpp_bind", "model_answer");
343                            wire::ESME_ROK
344                        }
345                        Ok((name, data)) if name == "smpp_bind_reject" => {
346                            outcome(ctx, id, "smpp_bind", "model_reject");
347                            data["status"]
348                                .as_str()
349                                .and_then(wire::status_code)
350                                .unwrap_or(wire::ESME_RBINDFAIL)
351                        }
352                        Ok(_) => {
353                            outcome(ctx, id, "smpp_bind", "fail_closed_invalid_reply");
354                            wire::ESME_RBINDFAIL
355                        }
356                        Err(status) => status,
357                    },
358                };
359                let mut body = Vec::new();
360                if status == wire::ESME_ROK {
361                    wire::put_cstring(&mut body, &cfg.system_id, 16);
362                }
363                s.respond(&pdu, status, body).await?;
364                if status != wire::ESME_ROK {
365                    return Ok(());
366                }
367                Log::new(Some(&ctx.status_tx)).info(format!(
368                    "SMPP connection {id}: {} bound as {mode}",
369                    bind.system_id
370                ));
371                bound = Some((mode, bind.system_id));
372            }
373            wire::SUBMIT_SM => {
374                let Some((mode, system_id)) = bound.clone() else {
375                    s.respond(&pdu, wire::ESME_RINVBNDSTS, vec![0]).await?;
376                    continue;
377                };
378                if mode == "receiver" {
379                    s.respond(&pdu, wire::ESME_RINVBNDSTS, vec![0]).await?;
380                    continue;
381                }
382                let message = match wire::parse_message(&pdu.body) {
383                    Ok(m) => m,
384                    Err(_) => {
385                        s.respond(&pdu, wire::ESME_RINVMSGLEN, vec![0]).await?;
386                        continue;
387                    }
388                };
389                submit(&mut s, cfg, &pdu, &message, mode, &system_id, peer).await?;
390            }
391            wire::ENQUIRE_LINK => s.respond(&pdu, wire::ESME_ROK, Vec::new()).await?,
392            wire::UNBIND => {
393                s.respond(&pdu, wire::ESME_ROK, Vec::new()).await?;
394                return Ok(());
395            }
396            id if id & wire::RESP != 0 => {} // deliver_sm_resp and the like: nothing to do
397            _ => {
398                s.send(Pdu::new(
399                    wire::GENERIC_NACK,
400                    wire::ESME_RINVCMDID,
401                    pdu.sequence,
402                    Vec::new(),
403                ))
404                .await?;
405            }
406        }
407    }
408}
409
410async fn submit(
411    s: &mut Session<'_>,
412    cfg: &Config,
413    pdu: &Pdu,
414    message: &wire::Message,
415    mode: &str,
416    system_id: &str,
417    peer: SocketAddr,
418) -> Result<()> {
419    let (ctx, id) = (s.ctx, s.id);
420    let text = wire::decode_text(message.data_coding, &message.payload);
421    let mut data = json!({
422        "system_id": system_id,
423        "source_addr": message.source_addr,
424        "destination_addr": message.destination_addr,
425        "data_coding": message.data_coding,
426        "registered_delivery": message.registered_delivery & 0x03 != 0,
427        "esm_class": message.esm_class,
428        "remote_addr": peer.to_string(),
429    });
430    match &text {
431        Some(t) => data["text"] = json!(t),
432        None => {
433            data["data"] = json!(message
434                .payload
435                .iter()
436                .map(|b| format!("{b:02x}"))
437                .collect::<String>())
438        }
439    }
440    let answer = decide(ctx, id, &actions::SUBMIT_EVENT, data, wire::ESME_RSYSERR).await;
441    let accepted = match answer {
442        Ok((name, data)) if name == "smpp_accept" => {
443            outcome(ctx, id, "smpp_submit", "model_answer");
444            data
445        }
446        Ok((name, data)) if name == "smpp_reject" => {
447            outcome(ctx, id, "smpp_submit", "model_reject");
448            let status = data["status"]
449                .as_str()
450                .and_then(wire::status_code)
451                .unwrap_or(wire::ESME_RSUBMITFAIL);
452            return s.respond(pdu, status, vec![0]).await;
453        }
454        Ok(_) => {
455            outcome(ctx, id, "smpp_submit", "fail_closed_invalid_reply");
456            return s.respond(pdu, wire::ESME_RSYSERR, vec![0]).await;
457        }
458        Err(status) => return s.respond(pdu, status, vec![0]).await,
459    };
460    let message_id = accepted["message_id"]
461        .as_str()
462        .map(str::to_string)
463        .unwrap_or_else(|| {
464            format!(
465                "NG{:08X}",
466                cfg.next_message_id.fetch_add(1, Ordering::Relaxed)
467            )
468        });
469    let mut body = Vec::new();
470    wire::put_cstring(&mut body, &message_id, 65);
471    s.respond(pdu, wire::ESME_ROK, body).await?;
472    if mode == "transmitter" {
473        return Ok(());
474    }
475    let now = chrono::Utc::now().format("%y%m%d%H%M").to_string();
476    if let Some(stat) = accepted["receipt"].as_str() {
477        if message.registered_delivery & 0x03 != 0 {
478            let original = text.clone().unwrap_or_default();
479            let receipt = wire::receipt_text(&message_id, stat, &now, &now, &original);
480            let mut receipted = message_id.as_bytes().to_vec();
481            receipted.push(0);
482            let deliver = wire::Message {
483                source_ton: message.dest_ton,
484                source_npi: message.dest_npi,
485                source_addr: message.destination_addr.clone(),
486                dest_ton: message.source_ton,
487                dest_npi: message.source_npi,
488                destination_addr: message.source_addr.clone(),
489                esm_class: 0x04,
490                data_coding: 0,
491                payload: receipt.into_bytes(),
492                tlvs: vec![
493                    (wire::TLV_RECEIPTED_MESSAGE_ID, receipted),
494                    (wire::TLV_MESSAGE_STATE, vec![wire::message_state(stat)]),
495                ],
496                ..Default::default()
497            };
498            let seq = s.sequence();
499            s.send(Pdu::new(
500                wire::DELIVER_SM,
501                0,
502                seq,
503                wire::encode_message(&deliver),
504            ))
505            .await?;
506        }
507    }
508    if let Some(reply) = accepted["reply_text"].as_str() {
509        let (coding, payload) = wire::encode_text(reply);
510        let deliver = wire::Message {
511            source_ton: message.dest_ton,
512            source_npi: message.dest_npi,
513            source_addr: message.destination_addr.clone(),
514            dest_ton: message.source_ton,
515            dest_npi: message.source_npi,
516            destination_addr: message.source_addr.clone(),
517            data_coding: coding,
518            payload,
519            ..Default::default()
520        };
521        let seq = s.sequence();
522        s.send(Pdu::new(
523            wire::DELIVER_SM,
524            0,
525            seq,
526            wire::encode_message(&deliver),
527        ))
528        .await?;
529    }
530    Ok(())
531}
netget/server/smpp/mod.rs

netget/server/smpp/
mod.rs
