SMPP

use smpp.qweruse smpp_bind, smpp_parts from smpp      // Quillo 0.6: only these

serde_json = { version = "1", features = ["preserve_order"] }rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] }webpki-roots = "1"

// A restaurant's booking texts through an SMS gateway over SMPP: a confirmation sent with a// delivery receipt asked for, a long reminder that goes in parts, a question whose answer comes// back from the phone, the receipts and replies drained by an `every` tick, and a wrong password.////   SMPP_HOST=smsc.example.net:2775 SMPP_SYSTEM_ID=quillo SMPP_PASSWORD=secret quillo run smpp.qwe//   then open http://localhost:8080/send, /long, /ask and /receipts//// Needs an SMSC (your SMS provider's SMPP account, or a simulator) at SMPP_HOST. SMPP_TLS=1 and// SMPP_CA for SMPP over TLS.use "smpp.qwer"let HOST = env("SMPP_HOST", "127.0.0.1:2775")fn link() = smpp_bind("sms", HOST, {    system_id: env("SMPP_SYSTEM_ID", "quillo"),    password: env("SMPP_PASSWORD", ""),    mode: "transceiver",    enquire_link: 30,    buffer: 1000,    tls: env("SMPP_TLS", "") == "1",    ca: env("SMPP_CA", ""),    timeout_ms: 5000,})// Receipts and replies, taken off the session twice a second.let receipts = []let replies = []fn drain() {    let got = smpp_receive("sms", 100, 0) ?? []    for m in got {        if m.kind == "receipt" { receipts = receipts + [{ id: m.receipt.id, stat: m.receipt.stat, delivered: m.receipt.delivered }] }        else { replies = replies + [{ from: m.from, text: m.text, encoding: m.encoding }] }    }}every(500, drain)route GET /send {    link()?    smpp_submit("sms", "QuilloCafe", "+447700900123", "Your table for 2 is booked for 19:30 tonight. Reply STOP to opt out.", { receipt: true })?}route GET /long {    link()?    let text = "Reminder: your table at Quillo Café is booked for 19:30 tonight, for 4 people. A deposit of €20 was taken {ref BK-2207}. " + "If you are running late, we keep the table for 15 minutes; after that it may be given to guests waiting at the bar. " + "To change the booking, reply with the new time or call us. Thank you!"    let sent = smpp_submit("sms", "QuilloCafe", "+447700900123", text, {})?    { parts: sent.parts, encoding: sent.encoding, ids: len(sent.message_ids), characters: len(text) }}// A question: the answer comes back as a deliver_sm, in UCS-2 here since it has an emoji.route GET /ask {    link()?    let before = len(replies)    smpp_submit("sms", "+447700900001", "+447700900123", "Can you come at 10?", { receipt: false })?    let waited = 0    while len(replies) == before && waited < 30 {        sleep(100)        waited += 1    }    len(replies) > before ? replies[len(replies) - 1] : { reply: null }}route GET /receipts {    let waited = 0    while len(receipts) == 0 && waited < 30 {        sleep(100)        waited += 1    }    { count: len(receipts), first: len(receipts) > 0 ? receipts[0] : null, status: smpp_status("sms")? }}route GET /wrong-password = { error: smpp_bind("intruder", HOST, { system_id: "quillo", password: "guess", timeout_ms: 3000 }).error }// No network: how texts are split and encoded.route GET /parts = {    plain: smpp_parts("See you at 19:30!", {})?,    euro: smpp_parts("Deposit: €20 [paid]", {}).units,    emoji: smpp_parts("See you 👍", {})?,    packed: smpp_parts("hello", { packed: true }).parts,    long: smpp_parts(join(for i in 0..40 => "word", " "), {}).count,    not_gsm: smpp_parts("Привет", { encoding: "gsm" }).error,}

//! SMPP 3.4 for sending and receiving SMS through a carrier's or provider's SMSC: a bind as transmitter, receiver or transceiver kept alive with enquire_link (and bound again after a dropped link), submit_sm with GSM 7-bit or UCS-2 chosen for the text and long texts split with concatenation headers, delivery receipts asked for, and incoming messages and receipts buffered, put back together and read.//! version: 1.0.0//!//! [dependencies]//! serde_json = { version = "1", features = ["preserve_order"] }//! rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] }//! webpki-roots = "1"use serde_json::{json, Value};use std::collections::{HashMap, VecDeque};use std::io::{Read, Write};use std::net::{SocketAddr, TcpStream, ToSocketAddrs};use std::sync::atomic::{AtomicU64, Ordering};use std::sync::{Arc, Condvar, Mutex};use std::time::{Duration, Instant};// ---- Options, TCP with timeouts, and TLS by rustls (ring) ----fn o_str(o: &Value, k: &str) -> Option<String> {    match o.get(k) {        Some(Value::String(s)) => Some(s.clone()),        Some(Value::Number(n)) => Some(n.to_string()),        Some(Value::Bool(b)) => Some(b.to_string()),        _ => None,    }}fn o_i64(o: &Value, k: &str, d: i64) -> i64 {    match o.get(k) {        Some(Value::Number(n)) => n.as_i64().or(n.as_f64().map(|f| f as i64)).unwrap_or(d),        Some(Value::String(s)) => s.trim().parse().unwrap_or(d),        _ => d,    }}fn o_bool(o: &Value, k: &str, d: bool) -> bool {    match o.get(k) {        Some(Value::Bool(b)) => *b,        Some(Value::String(s)) => matches!(s.to_ascii_lowercase().as_str(), "true" | "yes" | "1" | "on"),        Some(Value::Number(n)) => n.as_i64() != Some(0),        _ => d,    }}/// `timeout_ms` from the options (10 s when not given).fn timeout_of(o: &Value) -> Duration {    Duration::from_millis(o_i64(o, "timeout_ms", 10_000).clamp(100, 600_000) as u64)}/// `host`, `host:port`, `[v6]:port` or a bare IPv6 address, with the protocol's port when none is given.fn split_host(host: &str, default_port: u16) -> Result<(String, u16), String> {    let h = host.trim();    if h.is_empty() {        return Err("no host given: write host or host:port".into());    }    let (name, port) = if let Some(rest) = h.strip_prefix('[') {        let end = rest.find(']').ok_or(format!("{}: a [ without its ]", host))?;        let after = &rest[end + 1..];        (rest[..end].to_string(), after.strip_prefix(':').map(|p| p.to_string()))    } else if h.matches(':').count() == 1 {        let (a, b) = h.split_once(':').unwrap();        (a.to_string(), Some(b.to_string()))    } else {        (h.to_string(), None)    };    let port = match port {        Some(p) => p.parse::<u16>().map_err(|_| format!("{}: the port is not a number from 1 to 65535", host))?,        None => default_port,    };    Ok((name, port))}/// Connects with a time limit to each address the name has, and keeps the limit on reads and writes.fn connect(name: &str, port: u16, t: Duration) -> Result<TcpStream, String> {    let addrs: Vec<SocketAddr> = (name, port).to_socket_addrs().map_err(|e| format!("{}: {}", name, e))?.collect();    if addrs.is_empty() {        return Err(format!("{} has no address", name));    }    let mut last = String::new();    for a in addrs {        match TcpStream::connect_timeout(&a, t) {            Ok(s) => {                let _ = s.set_read_timeout(Some(t));                let _ = s.set_write_timeout(Some(t));                let _ = s.set_nodelay(true);                return Ok(s);            }            Err(e) => last = format!("cannot reach {}: {}", a, e),        }    }    Err(last)}/// Takes any certificate: for `insecure: true`, on a test bench only.#[derive(Debug)]struct AcceptAny(Arc<rustls::crypto::CryptoProvider>);impl rustls::client::danger::ServerCertVerifier for AcceptAny {    fn verify_server_cert(&self, _: &rustls::pki_types::CertificateDer<'_>, _: &[rustls::pki_types::CertificateDer<'_>], _: &rustls::pki_types::ServerName<'_>, _: &[u8], _: rustls::pki_types::UnixTime) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {        Ok(rustls::client::danger::ServerCertVerified::assertion())    }    fn verify_tls12_signature(&self, m: &[u8], c: &rustls::pki_types::CertificateDer<'_>, d: &rustls::DigitallySignedStruct) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {        rustls::crypto::verify_tls12_signature(m, c, d, &self.0.signature_verification_algorithms)    }    fn verify_tls13_signature(&self, m: &[u8], c: &rustls::pki_types::CertificateDer<'_>, d: &rustls::DigitallySignedStruct) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {        rustls::crypto::verify_tls13_signature(m, c, d, &self.0.signature_verification_algorithms)    }    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {        self.0.signature_verification_algorithms.supported_schemes()    }}fn b64_decode(text: &str) -> Vec<u8> {    let mut out = vec![];    let (mut acc, mut bits) = (0u32, 0);    for c in text.bytes() {        let v = match c {            b'A'..=b'Z' => c - b'A',            b'a'..=b'z' => c - b'a' + 26,            b'0'..=b'9' => c - b'0' + 52,            b'+' | b'-' => 62,            b'/' | b'_' => 63,            _ => continue,        } as u32;        acc = acc << 6 | v;        bits += 6;        if bits >= 8 {            bits -= 8;            out.push((acc >> bits) as u8);        }    }    out}/// The certificates in PEM text.fn pem_certs(text: &str) -> Vec<Vec<u8>> {    let mut out = vec![];    let mut rest = text;    while let Some(a) = rest.find("-----BEGIN CERTIFICATE-----") {        let body = &rest[a + 27..];        let Some(b) = body.find("-----END CERTIFICATE-----") else { break };        out.push(b64_decode(&body[..b]));        rest = &body[b + 25..];    }    out}/// A rustls client set up from the options: Mozilla's roots, plus `ca` (PEM text, or a PEM file's path); `insecure: true` takes any certificate.fn tls_config(o: &Value) -> Result<Arc<rustls::ClientConfig>, String> {    let provider = Arc::new(rustls::crypto::ring::default_provider());    let builder = rustls::ClientConfig::builder_with_provider(provider.clone()).with_safe_default_protocol_versions().map_err(|e| e.to_string())?;    let config = if o_bool(o, "insecure", false) {        builder.dangerous().with_custom_certificate_verifier(Arc::new(AcceptAny(provider))).with_no_client_auth()    } else {        let mut roots = rustls::RootCertStore { roots: webpki_roots::TLS_SERVER_ROOTS.to_vec() };        if let Some(ca) = o_str(o, "ca").filter(|s| !s.trim().is_empty()) {            let text = if ca.contains("-----BEGIN") { ca.clone() } else { std::fs::read_to_string(&ca).map_err(|e| format!("cannot read the ca file {}: {}", ca, e))? };            let certs = pem_certs(&text);            if certs.is_empty() {                return Err("`ca` holds no PEM certificate (-----BEGIN CERTIFICATE-----)".into());            }            for c in certs {                roots.add(rustls::pki_types::CertificateDer::from(c)).map_err(|e| format!("a `ca` certificate cannot be used: {}", e))?;            }        }        builder.with_root_certificates(roots).with_no_client_auth()    };    Ok(Arc::new(config))}enum Stream {    Plain(TcpStream),    Tls(Box<rustls::StreamOwned<rustls::ClientConnection, TcpStream>>),}impl Stream {    fn tcp(&self) -> &TcpStream {        match self {            Stream::Plain(s) => s,            Stream::Tls(s) => &s.sock,        }    }}impl Read for Stream {    fn read(&mut self, b: &mut [u8]) -> std::io::Result<usize> {        match self {            Stream::Plain(s) => s.read(b),            Stream::Tls(s) => match s.read(b) {                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => Ok(0),                r => r,            },        }    }}impl Write for Stream {    fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {        match self {            Stream::Plain(s) => s.write(b),            Stream::Tls(s) => s.write(b),        }    }    fn flush(&mut self) -> std::io::Result<()> {        match self {            Stream::Plain(s) => s.flush(),            Stream::Tls(s) => s.flush(),        }    }}/// TLS over a connected socket, the handshake done now so a bad certificate is said at once. `server_name` in the options overrides the name checked.fn tls_wrap(mut tcp: TcpStream, name: &str, o: &Value) -> Result<Stream, String> {    let config = tls_config(o)?;    let sni = o_str(o, "server_name").unwrap_or_else(|| name.to_string());    let server = rustls::pki_types::ServerName::try_from(sni.clone()).map_err(|_| format!("{} is not a name TLS can check", sni))?;    let mut conn = rustls::ClientConnection::new(config, server).map_err(|e| e.to_string())?;    while conn.is_handshaking() {        conn.complete_io(&mut tcp).map_err(|e| format!("TLS with {} failed: {} (a private CA? pass its PEM as `ca`)", name, e))?;    }    Ok(Stream::Tls(Box::new(rustls::StreamOwned::new(conn, tcp))))}// ---- Text: GSM 03.38, UCS-2, Latin-1 ----const GSM: &str = "@£$¥èéùìòÇ\nØø\rÅåΔ_ΦΓΛΩΠΨΣΘΞ\u{1b}ÆæßÉ !\"#¤%&'()*+,-./0123456789:;<=>?¡ABCDEFGHIJKLMNOPQRSTUVWXYZÄÖÑܧ¿abcdefghijklmnopqrstuvwxyzäöñüà";const GSM_EXT: &[(char, u8)] = &[('\u{0c}', 0x0A), ('^', 0x14), ('{', 0x28), ('}', 0x29), ('\\', 0x2F), ('[', 0x3C), ('~', 0x3D), (']', 0x3E), ('|', 0x40), ('€', 0x65)];/// Each character as its septets (two for the extension table), or None when one is not in GSM 03.38.fn gsm_units(text: &str) -> Option<Vec<Vec<u8>>> {    let table: Vec<char> = GSM.chars().collect();    text.chars()        .map(|c| {            if let Some(i) = table.iter().position(|&t| t == c && c != '\u{1b}') {                Some(vec![i as u8])            } else {                GSM_EXT.iter().find(|e| e.0 == c).map(|e| vec![0x1b, e.1])            }        })        .collect()}fn gsm_decode(b: &[u8]) -> String {    let table: Vec<char> = GSM.chars().collect();    let mut out = String::new();    let mut esc = false;    for &x in b {        let x = x & 0x7f;        if esc {            esc = false;            out.push(GSM_EXT.iter().find(|e| e.1 == x).map_or(' ', |e| e.0));        } else if x == 0x1b {            esc = true;        } else {            out.push(table[x as usize]);        }    }    out}/// Septets packed eight to seven octets, after `fill` bits of padding (to line up after a UDH).fn pack7(septets: &[u8], fill: usize) -> Vec<u8> {    let bits = fill + septets.len() * 7;    let mut out = vec![0u8; (bits + 7) / 8];    for (i, s) in septets.iter().enumerate() {        let at = fill + i * 7;        let v = (*s as u16) << (at % 8);        out[at / 8] |= v as u8;        if at / 8 + 1 < out.len() {            out[at / 8 + 1] |= (v >> 8) as u8;        }    }    out}fn unpack7(b: &[u8], fill: usize) -> Vec<u8> {    let n = (b.len() * 8).saturating_sub(fill) / 7;    (0..n)        .map(|i| {            let at = fill + i * 7;            let lo = b[at / 8] as u16;            let hi = *b.get(at / 8 + 1).unwrap_or(&0) as u16;            (((hi << 8 | lo) >> (at % 8)) & 0x7f) as u8        })        .collect()}struct Encoded {    name: &'static str,    data_coding: u8,    parts: Vec<Vec<u8>>,    units: usize,}static REF: AtomicU64 = AtomicU64::new(1);/// The text split into short messages: each with a concatenation UDH when there are several.fn encode_text(text: &str, opts: &Value) -> Result<Encoded, String> {    let want = o_str(opts, "encoding").unwrap_or_else(|| "auto".into()).to_ascii_lowercase();    let packed = o_bool(opts, "packed", false);    let gsm = gsm_units(text);    let (name, dc, units): (&'static str, u8, Vec<Vec<u8>>) = match (want.as_str(), gsm) {        ("auto", Some(u)) | ("gsm", Some(u)) => ("gsm", 0, u),        ("gsm", None) => {            let bad: String = text.chars().filter(|c| gsm_units(&c.to_string()).is_none()).take(5).collect();            return Err(format!("{:?} is not in the GSM alphabet: use encoding \"ucs2\" (or \"auto\")", bad));        }        ("auto", None) | ("ucs2", _) => ("ucs2", 8, text.chars().map(|c| {            let mut b = [0u16; 2];            c.encode_utf16(&mut b).iter().flat_map(|u| u.to_be_bytes()).collect()        }).collect()),        ("latin1", _) => {            let u: Option<Vec<Vec<u8>>> = text.chars().map(|c| if (c as u32) < 256 { Some(vec![c as u32 as u8]) } else { None }).collect();            ("latin1", 3, u.ok_or("the text has characters outside Latin-1: use \"ucs2\"")?)        }        ("ascii", _) => {            if !text.is_ascii() {                return Err("the text is not ASCII: use \"auto\"".into());            }            ("ascii", 1, text.bytes().map(|b| vec![b]).collect())        }        (x, _) => return Err(format!("encoding {}: auto, gsm, ucs2, latin1 or ascii", x)),    };    // A message's room: 160 septets or 140 octets; 153 septets or 134 octets a part after the UDH.    let (single, part) = match name {        "gsm" => (160, 153),        _ => (140, 134),    };    let size = |u: &Vec<u8>| if name == "gsm" { u.len() } else { u.len() };    let total: usize = units.iter().map(size).sum();    let long_mode = o_str(opts, "long").unwrap_or_else(|| "udh".into());    let body = |chunk: &[Vec<u8>], fill: usize| -> Vec<u8> {        let flat: Vec<u8> = chunk.iter().flatten().cloned().collect();        if name == "gsm" && packed { pack7(&flat, fill) } else { flat }    };    if total <= single || long_mode == "payload" {        return Ok(Encoded { name, data_coding: dc, parts: vec![body(&units, 0)], units: total });    }    let mut chunks: Vec<Vec<Vec<u8>>> = vec![vec![]];    let mut used = 0;    for u in units {        // A character is never split between parts (an escape and its septet, a surrogate pair).        if used + size(&u) > part {            chunks.push(vec![]);            used = 0;        }        used += size(&u);        chunks.last_mut().unwrap().push(u);    }    if chunks.len() > 255 {        return Err(format!("{} parts: a message has at most 255", chunks.len()));    }    let r = (REF.fetch_add(1, Ordering::Relaxed) % 255 + 1) as u8;    let n = chunks.len() as u8;    let parts = chunks        .iter()        .enumerate()        .map(|(i, c)| {            let mut p = vec![0x05, 0x00, 0x03, r, n, i as u8 + 1];            // Six octets of UDH are 48 bits: one fill bit lines the septets up again.            p.extend(body(c, 1));            p        })        .collect();    Ok(Encoded { name, data_coding: dc, parts, units: total })}fn decode_text(b: &[u8], dc: u8, packed: bool, fill: usize) -> (String, &'static str) {    match dc & 0x0f {        _ if dc & 0xf0 == 0xf0 => {            if dc & 0x04 != 0 { (hex(b), "binary") } else if packed { (gsm_decode(&unpack7(b, fill)), "gsm") } else { (gsm_decode(b), "gsm") }        }        0 => {            if packed { (gsm_decode(&unpack7(b, fill)), "gsm") } else { (gsm_decode(b), "gsm") }        }        1 => (String::from_utf8_lossy(b).to_string(), "ascii"),        3 => (b.iter().map(|&c| c as char).collect(), "latin1"),        8 => {            let u: Vec<u16> = b.chunks(2).map(|p| u16::from_be_bytes([p[0], *p.get(1).unwrap_or(&0)])).collect();            (String::from_utf16_lossy(&u), "ucs2")        }        _ => (hex(b), "binary"),    }}fn hex(b: &[u8]) -> String {    b.iter().map(|x| format!("{:02X}", x)).collect()}// ---- PDUs ----const BIND_RX: u32 = 0x01;const BIND_TX: u32 = 0x02;const SUBMIT: u32 = 0x04;const DELIVER: u32 = 0x05;const UNBIND: u32 = 0x06;const BIND_TRX: u32 = 0x09;const ENQUIRE: u32 = 0x15;const RESP: u32 = 0x8000_0000;fn status_text(s: u32) -> String {    let t = match s {        0x00 => "ok",        0x01 => "message length is invalid",        0x02 => "command length is invalid",        0x03 => "invalid command ID",        0x04 => "incorrect bind status for this command",        0x05 => "already bound",        0x06 => "invalid priority flag",        0x07 => "invalid registered delivery flag",        0x08 => "system error at the SMSC",        0x0A => "invalid source address",        0x0B => "invalid destination address",        0x0C => "message ID is invalid",        0x0D => "bind failed",        0x0E => "invalid password",        0x0F => "invalid system ID",        0x14 => "message queue full",        0x15 => "invalid service type",        0x33 => "invalid number of destinations",        0x45 => "submit failed",        0x58 => "throttled: too many messages a second",        0x61 => "invalid scheduled delivery time",        0x62 => "invalid validity period",        0x63 => "predefined message not found",        0x64 => "ESME receiver temporary error",        0x65 => "ESME receiver permanent error",        0x66 => "ESME receiver reject message error",        0xFF => "unknown error",        _ => "",    };    if t.is_empty() { format!("SMPP error 0x{:08X}", s) } else { format!("{} (0x{:08X})", t, s) }}fn cstr(out: &mut Vec<u8>, s: &str, max: usize) -> Result<(), String> {    if s.len() >= max {        return Err(format!("{:?} is longer than the {} characters SMPP allows there", s, max - 1));    }    out.extend_from_slice(s.as_bytes());    out.push(0);    Ok(())}fn pdu(id: u32, status: u32, seq: u32, body: &[u8]) -> Vec<u8> {    let mut p = Vec::with_capacity(16 + body.len());    p.extend(((16 + body.len()) as u32).to_be_bytes());    p.extend(id.to_be_bytes());    p.extend(status.to_be_bytes());    p.extend(seq.to_be_bytes());    p.extend_from_slice(body);    p}struct Reader<'a> {    b: &'a [u8],    at: usize,}impl<'a> Reader<'a> {    fn u8(&mut self) -> u8 {        let v = *self.b.get(self.at).unwrap_or(&0);        self.at += 1;        v    }    fn cstr(&mut self) -> String {        let rest = self.b.get(self.at..).unwrap_or(&[]);        let end = rest.iter().position(|&c| c == 0).unwrap_or(rest.len());        let s = String::from_utf8_lossy(&rest[..end]).to_string();        self.at += end + 1;        s    }    fn bytes(&mut self, n: usize) -> &'a [u8] {        let end = (self.at + n).min(self.b.len());        let s = &self.b[self.at.min(end)..end];        self.at = end;        s    }}fn addr_of(a: &str) -> (u8, u8, String) {    let t = a.trim();    let digits = t.trim_start_matches('+');    if digits.is_empty() || !digits.chars().all(|c| c.is_ascii_digit()) {        (5, 0, t.to_string()) // alphanumeric    } else if !t.starts_with('+') && digits.len() <= 6 {        (3, 0, digits.to_string()) // a short code    } else {        (1, 1, digits.to_string()) // international, E.164    }}fn read_receipt(text: &str) -> Value {    let keys = ["id:", "sub:", "dlvrd:", "submit date:", "done date:", "stat:", "err:", "text:"];    let mut found: Vec<(usize, &str)> = keys.iter().filter_map(|k| text.to_ascii_lowercase().find(k).map(|i| (i, *k))).collect();    found.sort();    let mut m = serde_json::Map::new();    for (j, (i, k)) in found.iter().enumerate() {        let end = found.get(j + 1).map_or(text.len(), |n| n.0);        let v = text[i + k.len()..end].trim().to_string();        let key = k.trim_end_matches(':').replace(' ', "_");        m.insert(key, json!(v));    }    let stat = m.get("stat").and_then(|v| v.as_str()).unwrap_or("").to_ascii_uppercase();    m.insert("delivered".into(), json!(stat == "DELIVRD"));    let meaning = match stat.as_str() {        "DELIVRD" => "delivered",        "EXPIRED" => "expired before it could be delivered",        "DELETED" => "deleted",        "UNDELIV" => "undeliverable",        "ACCEPTD" => "accepted",        "UNKNOWN" => "unknown",        "REJECTD" => "rejected",        "ENROUTE" => "on its way",        _ => "",    };    m.insert("meaning".into(), json!(meaning));    Value::Object(m)}/// A deliver_sm read: a message from a phone, or a delivery receipt; concatenated parts come with `part`.fn read_deliver(body: &[u8]) -> Value {    let mut r = Reader { b: body, at: 0 };    let service = r.cstr();    let (_, _) = (r.u8(), r.u8());    let from = r.cstr();    let (_, _) = (r.u8(), r.u8());    let to = r.cstr();    let esm = r.u8();    let _pid = r.u8();    let _prio = r.u8();    let _sched = r.cstr();    let _valid = r.cstr();    let _reg = r.u8();    let _rep = r.u8();    let dc = r.u8();    let _def = r.u8();    let len = r.u8() as usize;    let mut sm = r.bytes(len).to_vec();    let mut tlv: HashMap<u16, Vec<u8>> = HashMap::new();    while r.at + 4 <= body.len() {        let tag = u16::from_be_bytes([r.u8(), r.u8()]);        let l = u16::from_be_bytes([r.u8(), r.u8()]) as usize;        tlv.insert(tag, r.bytes(l).to_vec());    }    if let Some(p) = tlv.get(&0x0424) {        sm = p.clone();    }    let mut part = Value::Null;    let mut udh_len = 0;    if esm & 0x40 != 0 && !sm.is_empty() {        udh_len = sm[0] as usize + 1;        let udh = sm[1..udh_len.min(sm.len())].to_vec();        let mut i = 0;        while i + 1 < udh.len() {            let (iei, l) = (udh[i], udh[i + 1] as usize);            let d = &udh[(i + 2).min(udh.len())..(i + 2 + l).min(udh.len())];            if iei == 0 && d.len() == 3 {                part = json!({ "ref": d[0], "count": d[1], "index": d[2] });            } else if iei == 8 && d.len() == 4 {                part = json!({ "ref": (d[0] as u16) << 8 | d[1] as u16, "count": d[2], "index": d[3] });            }            i += 2 + l;        }        sm = sm[udh_len.min(sm.len())..].to_vec();    }    let fill = if udh_len > 0 { (7 - (udh_len * 8) % 7) % 7 } else { 0 };    let receipt = esm & 0x3c == 0x04 || esm & 0x3c == 0x08;    let (text, enc) = if receipt { (String::from_utf8_lossy(&sm).to_string(), "ascii") } else { decode_text(&sm, dc, false, fill) };    let mut m = json!({        "kind": if receipt { "receipt" } else { "message" },        "from": from, "to": to, "text": text, "encoding": enc, "data_coding": dc, "service_type": service, "part": part,        "at": std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_or(0, |d| d.as_millis() as i64),    });    if receipt {        let mut rc = read_receipt(&text);        if let Some(id) = tlv.get(&0x001E) {            rc["id"] = json!(String::from_utf8_lossy(id).trim_end_matches('\0'));        }        if let Some(st) = tlv.get(&0x0427).and_then(|v| v.first()) {            let s = ["", "ENROUTE", "DELIVERED", "EXPIRED", "DELETED", "UNDELIVERABLE", "ACCEPTED", "UNKNOWN", "REJECTED"].get(*st as usize).copied().unwrap_or("");            rc["message_state"] = json!(s);            if *st == 2 {                rc["delivered"] = json!(true);            }        }        // SMSCs write the ID in the receipt as they like: the submit_sm_resp's hex ID, or the same number in decimal.        let id = rc["id"].as_str().unwrap_or("").to_string();        let mut forms = vec![id.clone()];        if !id.is_empty() && id.chars().all(|c| c.is_ascii_digit()) {            if let Ok(n) = id.parse::<u64>() {                forms.push(format!("{:x}", n));                forms.push(format!("{:X}", n));            }            if let Ok(n) = u64::from_str_radix(&id, 16) {                forms.push(n.to_string());            }        } else if let Ok(n) = u64::from_str_radix(&id, 16) {            forms.push(n.to_string());        }        rc["id_forms"] = json!(forms);        m["receipt"] = rc;    }    m}// ---- Sessions ----struct Shared {    state: String,    seq: u32,    outbox: VecDeque<Vec<u8>>,    answers: HashMap<u32, (u32, Vec<u8>)>,    waiting: std::collections::HashSet<u32>,    queue: VecDeque<Value>,    parts: HashMap<String, Vec<Value>>,    refused: u64,    received: u64,    sent: u64,    last_error: Option<String>,    stop: bool,    smsc: String,}struct Session {    cfg: Value,    mode: u32,    capacity: usize,    shared: Mutex<Shared>,    ready: Condvar,}static SESSIONS: Mutex<Option<HashMap<String, Arc<Session>>>> = Mutex::new(None);fn session(name: &str) -> Result<Arc<Session>, String> {    let g = SESSIONS.lock().unwrap_or_else(|e| e.into_inner());    g.as_ref().and_then(|m| m.get(name).cloned()).ok_or(format!("no SMPP session {}: call smpp_bind first", name))}fn lock(s: &Session) -> std::sync::MutexGuard<'_, Shared> {    s.shared.lock().unwrap_or_else(|e| e.into_inner())}fn next_seq(sh: &mut Shared) -> u32 {    sh.seq = if sh.seq >= 0x7fff_ffff { 1 } else { sh.seq + 1 };    sh.seq}fn read_pdu(st: &mut Stream, buf: &mut Vec<u8>) -> Result<Option<(u32, u32, u32, Vec<u8>)>, String> {    if buf.len() >= 16 {        let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize;        if !(16..=1 << 20).contains(&len) {            return Err(format!("a PDU of {} bytes: this is not SMPP", len));        }        if buf.len() >= len {            let p: Vec<u8> = buf.drain(..len).collect();            let g = |i: usize| u32::from_be_bytes([p[i], p[i + 1], p[i + 2], p[i + 3]]);            return Ok(Some((g(4), g(8), g(12), p[16..].to_vec())));        }    }    let mut b = [0u8; 8192];    match st.read(&mut b) {        Ok(0) => Err("the SMSC closed the connection".into()),        Ok(n) => {            buf.extend_from_slice(&b[..n]);            Ok(None)        }        Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => Ok(None),        Err(e) => Err(e.to_string()),    }}fn bind_now(s: &Session) -> Result<(Stream, Vec<u8>, String), String> {    let host = o_str(&s.cfg, "host").unwrap_or_default();    let (name, port) = split_host(&host, 2775)?;    let t = timeout_of(&s.cfg);    let tcp = connect(&name, port, t)?;    let mut st = if o_bool(&s.cfg, "tls", false) { tls_wrap(tcp, &name, &s.cfg)? } else { Stream::Plain(tcp) };    let mut body = vec![];    cstr(&mut body, &o_str(&s.cfg, "system_id").unwrap_or_default(), 16)?;    cstr(&mut body, &o_str(&s.cfg, "password").unwrap_or_default(), 9)?;    cstr(&mut body, &o_str(&s.cfg, "system_type").unwrap_or_default(), 13)?;    body.push(0x34);    body.push(o_i64(&s.cfg, "addr_ton", 0) as u8);    body.push(o_i64(&s.cfg, "addr_npi", 0) as u8);    cstr(&mut body, &o_str(&s.cfg, "address_range").unwrap_or_default(), 41)?;    let seq = next_seq(&mut lock(s));    st.write_all(&pdu(s.mode, 0, seq, &body)).map_err(|e| e.to_string())?;    let _ = st.tcp().set_read_timeout(Some(Duration::from_millis(100)));    let until = Instant::now() + t;    let mut buf = vec![];    while Instant::now() < until {        if let Some((id, status, rseq, b)) = read_pdu(&mut st, &mut buf)? {            if id == s.mode | RESP && rseq == seq {                if status != 0 {                    return Err(format!("the SMSC refused the bind: {}", status_text(status)));                }                let smsc = Reader { b: &b, at: 0 }.cstr();                return Ok((st, buf, smsc));            }            if id == RESP {                return Err(format!("the SMSC answered the bind with generic_nack: {}", status_text(status)));            }        }    }    Err(format!("{} did not answer the bind in time (timeout_ms)", host))}fn deliver(s: &Session, sh: &mut Shared, m: Value) -> u32 {    // Parts of a long message wait for each other and come out as one.    if let (Some(r), Some(count)) = (m["part"]["ref"].as_u64(), m["part"]["count"].as_u64()) {        let key = format!("{}|{}|{}", m["from"].as_str().unwrap_or(""), r, count);        let got = sh.parts.entry(key.clone()).or_default();        got.push(m);        if (got.len() as u64) < count {            return 0;        }        let mut all = sh.parts.remove(&key).unwrap_or_default();        all.sort_by_key(|p| p["part"]["index"].as_u64().unwrap_or(0));        let text: String = all.iter().map(|p| p["text"].as_str().unwrap_or("")).collect();        let mut whole = all[0].clone();        whole["text"] = json!(text);        whole["parts"] = json!(count);        whole["part"] = Value::Null;        sh.queue.push_back(whole);    } else {        let mut m = m;        m["parts"] = json!(1);        sh.queue.push_back(m);    }    sh.received += 1;    s.ready.notify_all();    0}fn run(s: Arc<Session>, mut st: Stream, mut buf: Vec<u8>) {    let ka = Duration::from_secs(o_i64(&s.cfg, "enquire_link", 30).clamp(1, 3600) as u64);    let mut last_in = Instant::now();    let mut last_out = Instant::now();    let mut enquiry: Option<Instant> = None;    let _ = st.tcp().set_read_timeout(Some(Duration::from_millis(20)));    let mut backoff = 1;    loop {        let (out, stop) = {            let mut sh = lock(&s);            (sh.outbox.drain(..).collect::<Vec<_>>(), sh.stop)        };        let mut broken: Option<String> = None;        for p in out {            if let Err(e) = st.write_all(&p).and_then(|_| st.flush()) {                broken = Some(format!("cannot write: {}", e));                break;            }            last_out = Instant::now();        }        if broken.is_none() {            loop {                match read_pdu(&mut st, &mut buf) {                    Ok(Some((id, status, seq, body))) => {                        last_in = Instant::now();                        enquiry = None;                        let mut sh = lock(&s);                        match id {                            ENQUIRE => sh.outbox.push_back(pdu(ENQUIRE | RESP, 0, seq, &[])),                            DELIVER => {                                let m = read_deliver(&body);                                // A full buffer answers "queue full", so the SMSC delivers it again later.                                let code = if sh.queue.len() >= s.capacity {                                    sh.refused += 1;                                    0x14                                } else {                                    deliver(&s, &mut sh, m)                                };                                sh.outbox.push_back(pdu(DELIVER | RESP, code, seq, &[0]));                            }                            UNBIND => {                                sh.outbox.push_back(pdu(UNBIND | RESP, 0, seq, &[]));                                let out: Vec<_> = sh.outbox.drain(..).collect();                                sh.state = "unbound".into();                                sh.last_error = Some("the SMSC unbound the session".into());                                drop(sh);                                for p in out {                                    let _ = st.write_all(&p);                                }                                s.ready.notify_all();                                return;                            }                            x if x & RESP != 0 => {                                if x == UNBIND | RESP {                                    sh.state = "unbound".into();                                    s.ready.notify_all();                                    return;                                }                                if sh.waiting.remove(&seq) {                                    sh.answers.insert(seq, (status, body));                                    s.ready.notify_all();                                }                            }                            _ => sh.outbox.push_back(pdu(RESP, 0x03, seq, &[])),                        }                    }                    Ok(None) => break,                    Err(e) => {                        broken = Some(e);                        break;                    }                }            }        }        if stop {            let mut sh = lock(&s);            if sh.state == "bound" {                let seq = next_seq(&mut sh);                sh.state = "unbinding".into();                drop(sh);                let _ = st.write_all(&pdu(UNBIND, 0, seq, &[]));                last_in = Instant::now();                continue;            }            if sh.state == "unbinding" && last_in.elapsed() < timeout_of(&s.cfg) {                drop(sh);                continue;            }            sh.state = "unbound".into();            s.ready.notify_all();            return;        }        if broken.is_none() && (last_in.elapsed() >= ka || last_out.elapsed() >= ka) {            match enquiry {                None => {                    let mut sh = lock(&s);                    let seq = next_seq(&mut sh);                    sh.outbox.push_back(pdu(ENQUIRE, 0, seq, &[]));                    enquiry = Some(Instant::now());                    last_out = Instant::now();                }                Some(at) if at.elapsed() >= timeout_of(&s.cfg) => broken = Some("no answer to enquire_link: the link is dead".into()),                _ => {}            }        }        if let Some(e) = broken {            {                let mut sh = lock(&s);                sh.state = "reconnecting".into();                sh.last_error = Some(e);                sh.waiting.clear();                s.ready.notify_all();            }            loop {                std::thread::sleep(Duration::from_secs(backoff));                if lock(&s).stop {                    lock(&s).state = "unbound".into();                    return;                }                match bind_now(&s) {                    Ok((n, b, smsc)) => {                        st = n;                        buf = b;                        let _ = st.tcp().set_read_timeout(Some(Duration::from_millis(20)));                        let mut sh = lock(&s);                        sh.state = "bound".into();                        sh.smsc = smsc;                        backoff = 1;                        last_in = Instant::now();                        enquiry = None;                        break;                    }                    Err(e) => {                        lock(&s).last_error = Some(e);                        backoff = (backoff * 2).min(30);                    }                }            }        }    }}/// Binds session `name` to the SMSC at `host` (host:port, 2775 by default). Options: `system_id`, `password`, `system_type`, `mode` ("transceiver", "transmitter" or "receiver"), `enquire_link` (30 s keep-alive), `buffer` (1000 messages and receipts kept for smpp_receive; past it the SMSC is told "queue full" and delivers again later), `tls`, `ca`, `insecure`, `timeout_ms`, `addr_ton`, `addr_npi`, `address_range`. A dropped link binds again by itself. Calling it again for a bound session gives its status.pub fn smpp_bind(name: &str, host: &str, opts: serde_json::Value) -> Result<Value, String> {    if let Ok(old) = session(name) {        let state = lock(&old).state.clone();        if state == "bound" || state == "reconnecting" {            return smpp_status(name);        }    }    let mode = match o_str(&opts, "mode").unwrap_or_else(|| "transceiver".into()).as_str() {        "transceiver" | "trx" => BIND_TRX,        "transmitter" | "tx" => BIND_TX,        "receiver" | "rx" => BIND_RX,        x => return Err(format!("mode {}: transceiver, transmitter or receiver", x)),    };    let mut cfg = opts.clone();    cfg["host"] = json!(host);    let s = Arc::new(Session {        cfg,        mode,        capacity: o_i64(&opts, "buffer", 1000).clamp(1, 1_000_000) as usize,        shared: Mutex::new(Shared { state: "binding".into(), seq: 0, outbox: VecDeque::new(), answers: HashMap::new(), waiting: Default::default(), queue: VecDeque::new(), parts: HashMap::new(), refused: 0, received: 0, sent: 0, last_error: None, stop: false, smsc: String::new() }),        ready: Condvar::new(),    });    let (st, buf, smsc) = bind_now(&s)?;    {        let mut sh = lock(&s);        sh.state = "bound".into();        sh.smsc = smsc;    }    SESSIONS.lock().unwrap_or_else(|e| e.into_inner()).get_or_insert_with(HashMap::new).insert(name.to_string(), s.clone());    let sc = s.clone();    std::thread::spawn(move || run(sc, st, buf));    smpp_status(name)}/// How `text` would go: `{ encoding, data_coding, units (septets for GSM, octets otherwise), count, parts: [hex of each short message] }`, GSM 7-bit when every character is in its alphabet, else UCS-2, split with a concatenation header past 160 (GSM) or 70 (UCS-2) characters. Options as smpp_submit: `encoding`, `packed`.pub fn smpp_parts(text: &str, opts: serde_json::Value) -> Result<Value, String> {    let e = encode_text(text, &opts)?;    Ok(json!({ "encoding": e.name, "data_coding": e.data_coding, "units": e.units, "count": e.parts.len(), "parts": e.parts.iter().map(|p| hex(p)).collect::<Vec<_>>() }))}/// Sends an SMS on session `name` with submit_sm: `from` (a number, a short code or a name of up to 11 letters), `to`, `text`. Long texts go in parts that phones put together. Options: `receipt` (true: ask for a delivery receipt), `encoding` ("auto", "gsm", "ucs2", "latin1"), `packed` (GSM septets packed, for SMSCs that want it), `long` ("udh" or "payload": one message_payload TLV), `service_type`, `retries` (3, when the SMSC says it is throttled). Gives `{ message_ids, parts, encoding }`.pub fn smpp_submit(name: &str, from: &str, to: &str, text: &str, opts: serde_json::Value) -> Result<Value, String> {    let s = session(name)?;    if s.mode == BIND_RX {        return Err(format!("session {} is bound as a receiver: bind a transmitter or transceiver to send", name));    }    let e = encode_text(text, &opts)?;    let (sto, snp, src) = addr_of(from);    let (dto, dnp, dst) = addr_of(to);    if dto == 5 {        return Err(format!("{} is not a phone number", to));    }    if sto == 5 && src.chars().count() > 11 {        return Err(format!("a sender name has at most 11 characters: {}", from));    }    let receipt = o_bool(&opts, "receipt", true);    let payload = o_str(&opts, "long").as_deref() == Some("payload");    let mut bodies = vec![];    for p in &e.parts {        let mut b = vec![];        cstr(&mut b, &o_str(&opts, "service_type").unwrap_or_default(), 6)?;        b.extend([sto, snp]);        cstr(&mut b, &src, 21)?;        b.extend([dto, dnp]);        cstr(&mut b, &dst, 21)?;        b.push(if e.parts.len() > 1 { 0x40 } else { 0 });        b.extend([0, 0]);        b.extend([0, 0]); // schedule and validity: empty C-strings        b.push(if receipt { 1 } else { 0 });        b.push(0);        b.push(e.data_coding);        b.push(0);        if payload && p.len() > 254 {            b.push(0);            b.extend(0x0424u16.to_be_bytes());            b.extend((p.len() as u16).to_be_bytes());            b.extend(p);        } else {            if p.len() > 254 {                return Err("a part over 254 octets: use long \"payload\" or \"udh\"".into());            }            b.push(p.len() as u8);            b.extend(p);        }        bodies.push(b);    }    let t = timeout_of(&s.cfg);    let retries = o_i64(&opts, "retries", 3).clamp(0, 20);    let mut ids = vec![];    for body in bodies {        let mut attempt = 0;        loop {            let seq = {                let mut sh = lock(&s);                if sh.state != "bound" {                    return Err(format!("session {} is {}: {}", name, sh.state, sh.last_error.clone().unwrap_or_default()));                }                let seq = next_seq(&mut sh);                sh.waiting.insert(seq);                sh.outbox.push_back(pdu(SUBMIT, 0, seq, &body));                seq            };            let until = Instant::now() + t;            let mut sh = lock(&s);            while !sh.answers.contains_key(&seq) {                let left = until.saturating_duration_since(Instant::now());                if left.is_zero() || !sh.waiting.contains(&seq) {                    break;                }                sh = s.ready.wait_timeout(sh, left.min(Duration::from_millis(50))).unwrap_or_else(|e| e.into_inner()).0;            }            let Some((status, rb)) = sh.answers.remove(&seq) else {                sh.waiting.remove(&seq);                return Err(format!("no submit_sm_resp in time (timeout_ms); {} part(s) went before", ids.len()));            };            drop(sh);            if status == 0x58 && attempt < retries {                attempt += 1;                std::thread::sleep(Duration::from_millis(500 * attempt as u64));                continue;            }            if status != 0 {                return Err(format!("the SMSC refused part {} of {}: {}", ids.len() + 1, e.parts.len(), status_text(status)));            }            ids.push(Reader { b: &rb, at: 0 }.cstr());            lock(&s).sent += 1;            break;        }    }    Ok(json!({ "message_ids": ids, "parts": e.parts.len(), "encoding": e.name, "data_coding": e.data_coding }))}/// Up to `max` messages and delivery receipts that came on session `name`, waiting up to `wait_ms` for the first: `{ kind ("message" or "receipt"), from, to, text, encoding, parts, receipt: { id, id_forms, stat, err, delivered, meaning, submit_date, done_date } }`. Long messages come whole; `id_forms` holds the ID in hex and decimal, as SMSCs write it either way, to match it with smpp_submit's message_ids.pub fn smpp_receive(name: &str, max: i64, wait_ms: i64) -> Result<Vec<Value>, String> {    let s = session(name)?;    let until = Instant::now() + Duration::from_millis(wait_ms.clamp(0, 600_000) as u64);    let mut sh = lock(&s);    while sh.queue.is_empty() {        let left = until.saturating_duration_since(Instant::now());        if left.is_zero() {            break;        }        sh = s.ready.wait_timeout(sh, left).unwrap_or_else(|e| e.into_inner()).0;    }    let n = (max.max(1) as usize).min(sh.queue.len());    Ok(sh.queue.drain(..n).collect())}/// How session `name` is: `{ state ("bound", "reconnecting", "unbound"), smsc, sent, received, waiting, refused, last_error }`.pub fn smpp_status(name: &str) -> Result<Value, String> {    let s = session(name)?;    let sh = lock(&s);    let mode = match s.mode {        BIND_TX => "transmitter",        BIND_RX => "receiver",        _ => "transceiver",    };    Ok(json!({ "name": name, "state": sh.state, "mode": mode, "smsc": sh.smsc, "sent": sh.sent, "received": sh.received, "waiting": sh.queue.len(), "refused": sh.refused, "last_error": sh.last_error }))}/// Unbinds session `name` and closes it, waiting for the SMSC's unbind_resp up to the session's timeout.pub fn smpp_unbind(name: &str) -> Result<Value, String> {    let s = session(name)?;    lock(&s).stop = true;    let until = Instant::now() + timeout_of(&s.cfg) + Duration::from_millis(500);    let mut sh = lock(&s);    while sh.state != "unbound" && Instant::now() < until {        sh = s.ready.wait_timeout(sh, Duration::from_millis(50)).unwrap_or_else(|e| e.into_inner()).0;    }    drop(sh);    let out = smpp_status(name)?;    SESSIONS.lock().unwrap_or_else(|e| e.into_inner()).get_or_insert_with(HashMap::new).remove(name);    Ok(out)}