Struct CryptoConfig

Source

pub struct CryptoConfig {
    pub key_size: KeySize,
    pub refresh_packets: u32,
    pub preannounce_packets: u32,
    /* private fields */
}

Expand description

Clones share one zeroising passphrase allocation; Debug is redacted.

Fields§

§key_size: KeySize

AES key length for the session.

§refresh_packets: u32

Original packets per key, at most 2^25. Retransmissions do not count.

§preannounce_packets: u32

Announce the next key this many packets before switching. Must be below half the refresh period. Sender fails closed if a replaced slot still has retained unacknowledged ciphertext.

Implementations§

Source §

impl CryptoConfig

Source

pub fn new(passphrase: Vec<u8>, key_size: KeySize) -> Result<Self, CryptoError>

Takes ownership of 10..=79 non-NUL passphrase bytes. No plaintext fallback is permitted for a session configured with this value.

Examples found in repository ?

examples/receive.rs (line 30)

13fn main() -> Result<(), Box<dyn Error>> {
14    let args: Vec<_> = std::env::args().skip(1).collect();
15    let (Some(bind), Some(path), Some(expected)) = (args.first(), args.get(1), args.get(2)) else {
16        return Err("receive <bind> <output> <expected bytes> [latency ms]".into());
17    };
18    let bind: SocketAddr = bind.parse()?;
19    let expected: usize = expected.parse()?;
20    let mut config = Config::default();
21    if let Some(ms) = args.get(3) {
22        config.latency = Duration::from_millis(ms.parse()?);
23    }
24    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
25        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
26            Ok("256") => KeySize::Aes256,
27            Ok("128") | Err(_) => KeySize::Aes128,
28            _ => return Err("key bits must be 128 or 256".into()),
29        };
30        config.crypto = Some(CryptoConfig::new(secret.into_bytes(), size)?);
31        config.require_encryption = true;
32    }
33    let socket = UdpSocket::bind(bind)?;
34    socket.set_read_timeout(Some(Duration::from_millis(5)))?;
35    let mut output = File::create(path)?;
36    let start = Instant::now();
37    let mut listener = Listener::new(config, start);
38    let mut packet = [0; 2048];
39    let mut bytes = 0;
40    let mut connection = None;
41    println!("listening {}", socket.local_addr()?);
42    std::io::stdout().flush()?;
43    while start.elapsed() < Duration::from_secs(20) {
44        let now = Instant::now();
45        listener.handle_timeout(now);
46        while let Some(event) = listener.poll_event() {
47            match event {
48                Event::Connect { id, stream_id, .. } => {
49                    println!("stream_id={}", String::from_utf8_lossy(&stream_id));
50                    if listener.accept(id, now) {
51                        connection = Some(id);
52                    }
53                }
54                Event::Data { payload, .. } => {
55                    output.write_all(&payload)?;
56                    bytes += payload.len();
57                }
58                Event::Closed { .. } => {}
59            }
60        }
61        if bytes >= expected {
62            let stats = connection.and_then(|id| listener.connection_stats(id));
63            println!("received_bytes={bytes} stats={stats:?}");
64            if let Some(id) = connection {
65                listener.close(id, now);
66            }
67            while let Some((peer, b)) = listener.poll_transmit() {
68                socket.send_to(&b, peer)?;
69            }
70            return Ok(());
71        }
72        while let Some((peer, b)) = listener.poll_transmit() {
73            socket.send_to(&b, peer)?;
74        }
75        match socket.recv_from(&mut packet) {
76            Ok((n, peer)) => {
77                listener.handle_datagram(Instant::now(), peer, packet.get(..n).unwrap_or_default())
78            }
79            Err(e)
80                if matches!(
81                    e.kind(),
82                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
83                ) => {}
84            Err(e) => return Err(e.into()),
85        }
86    }
87    Err(format!(
88        "receiver timeout bytes={bytes} rejected={}",
89        listener.stats().rejected
90    )
91    .into())
92}

More examples

Hide additional examples

examples/send.rs (line 37)

14fn main() -> Result<(), Box<dyn Error>> {
15    let args: Vec<_> = std::env::args().skip(1).collect();
16    let (Some(destination), Some(stream_id), Some(path)) = (args.first(), args.get(1), args.get(2))
17    else {
18        return Err("usage: send <ip:port> <stream-id> <input.ts> [SRT-bytes/sec] [latency-ms]; LAB ONLY; optional SAPFLOW_SRT_PASSPHRASE env".into());
19    };
20    let destination: SocketAddr = destination.parse()?;
21    let mut config = CallerConfig {
22        stream_id: stream_id.clone(),
23        ..CallerConfig::default()
24    };
25    if let Some(bandwidth) = args.get(3) {
26        config.send.bandwidth = bandwidth.parse()?;
27    }
28    if let Some(latency) = args.get(4) {
29        config.latency = Duration::from_millis(latency.parse()?);
30    }
31    if let Ok(secret) = std::env::var("SAPFLOW_SRT_PASSPHRASE") {
32        let size = match std::env::var("SAPFLOW_SRT_KEY_BITS").as_deref() {
33            Ok("256") => KeySize::Aes256,
34            Ok("128") | Err(_) => KeySize::Aes128,
35            _ => return Err("key bits must be 128 or 256".into()),
36        };
37        let mut crypto = CryptoConfig::new(secret.into_bytes(), size)?;
38        if let Ok(refresh) = std::env::var("SAPFLOW_SRT_REFRESH") {
39            crypto.refresh_packets = refresh.parse()?;
40            crypto.preannounce_packets = crypto.refresh_packets / 4;
41        }
42        config.crypto = Some(crypto);
43    }
44    let encrypted = config.crypto.is_some();
45    let mut latency = config.latency;
46    let socket = UdpSocket::bind(if destination.is_ipv4() {
47        "0.0.0.0:0"
48    } else {
49        "[::]:0"
50    })?;
51    let mut caller = Caller::new(config, destination, Instant::now())?;
52    let mut file = File::open(path)?;
53    let mut input = [0u8; 1316];
54    let mut packet = [0u8; 2048];
55    let mut eof = false;
56    let mut drained_at = None;
57    let deadline = Instant::now() + Duration::from_secs(120);
58    eprintln!("LAB ONLY: encrypted={encrypted}; stream ID is visible on the network");
59    loop {
60        let now = Instant::now();
61        if now >= deadline {
62            caller.stop(now);
63            return Err("lab sender exceeded 120-second run budget".into());
64        }
65        caller.handle_timeout(now);
66        while let Some(event) = caller.poll_event() {
67            match event {
68                CallerEvent::Connected {
69                    latency: negotiated_latency,
70                    max_payload,
71                    ..
72                } => {
73                    latency = negotiated_latency;
74                    eprintln!(
75                        "connected latency_ms={} payload_limit={max_payload}",
76                        latency.as_millis()
77                    );
78                }
79                CallerEvent::Closed { reason } => {
80                    return Err(format!("SRT session closed: {reason:?}").into());
81                }
82            }
83        }
84        // Eight payloads at most are read ahead. Retained retransmissions
85        // remain accounted for, so slow receivers cannot drive file growth.
86        while caller.is_connected() && !eof && caller.stats().send.buffered_packets < 8 {
87            let len = file.read(&mut input)?;
88            if len == 0 {
89                eof = true;
90            } else {
91                caller.enqueue(now, input.get(..len).unwrap_or_default().to_vec())?;
92            }
93        }
94        while let Some((peer, bytes)) = caller.poll_transmit(Instant::now()) {
95            socket.send_to(&bytes, peer)?;
96        }
97        if eof && caller.stats().send.buffered_packets == 0 {
98            let drained = *drained_at.get_or_insert(now);
99            if now.saturating_duration_since(drained) >= latency + Duration::from_millis(250) {
100                caller.stop(now);
101                if let Some((peer, bytes)) = caller.poll_transmit(now) {
102                    socket.send_to(&bytes, peer)?;
103                }
104                let stats = caller.stats();
105                println!(
106                    "acknowledged_bytes={} acknowledged_packets={} retransmitted_packets={} rtt_us={} peak_buffered_bytes={}",
107                    stats.send.acknowledged_bytes,
108                    stats.send.acknowledged_packets,
109                    stats.send.retransmitted_packets,
110                    stats.send.rtt_us,
111                    stats.send.peak_buffered_bytes
112                );
113                return Ok(());
114            }
115        }
116        let wait = caller
117            .poll_timeout()
118            .map(|at| at.saturating_duration_since(Instant::now()))
119            .unwrap_or(Duration::from_millis(10))
120            .clamp(Duration::from_micros(100), Duration::from_millis(10));
121        socket.set_read_timeout(Some(wait))?;
122        match socket.recv_from(&mut packet) {
123            Ok((len, peer)) => {
124                caller.handle_datagram(Instant::now(), peer, packet.get(..len).unwrap_or_default())
125            }
126            Err(error)
127                if matches!(
128                    error.kind(),
129                    ErrorKind::TimedOut | ErrorKind::WouldBlock | ErrorKind::Interrupted
130                ) => {}
131            Err(error) => {
132                caller.stop(Instant::now());
133                return Err(error.into());
134            }
135        }
136    }
137}