codetics_ocpp_ws/
client.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)
3
4//! Minimal WebSocket client, used by integration tests and the
5//! charging-station simulator. [`connect`] opens plain TCP; [`handshake`]
6//! runs over any established stream (e.g. a TLS stream).
7
8use std::io;
9
10use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf};
11use tokio::net::TcpStream;
12
13use crate::conn::{Role, WsConfig, WsReader, WsWriter};
14use crate::deflate::{self, DeflateConfig};
15use crate::{base64, handshake};
16
17pub struct ClientOptions<'a> {
18    /// `host:port`
19    pub addr: &'a str,
20    /// Request path, e.g. `/ocpp/CS001` (already percent-encoded).
21    pub path: &'a str,
22    pub subprotocols: &'a [&'a str],
23    /// Offer permessage-deflate.
24    pub deflate: bool,
25    /// Value for `Authorization: Basic …`, as (username, password bytes).
26    pub basic_auth: Option<(&'a str, &'a [u8])>,
27}
28
29pub struct ClientConnection<S = TcpStream> {
30    pub reader: WsReader<ReadHalf<S>>,
31    pub writer: WsWriter<WriteHalf<S>>,
32    /// Subprotocol selected by the server, if any.
33    pub subprotocol: Option<String>,
34    pub deflate: bool,
35}
36
37#[derive(Debug)]
38pub enum ConnectError {
39    Io(io::Error),
40    /// The server answered with a non-101 status.
41    Status(u16),
42    BadResponse(&'static str),
43}
44
45impl std::fmt::Display for ConnectError {
46    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47        match self {
48            ConnectError::Io(e) => write!(f, "{e}"),
49            ConnectError::Status(s) => write!(f, "handshake rejected with HTTP {s}"),
50            ConnectError::BadResponse(m) => write!(f, "bad handshake response: {m}"),
51        }
52    }
53}
54
55impl std::error::Error for ConnectError {}
56
57impl From<io::Error> for ConnectError {
58    fn from(e: io::Error) -> Self {
59        ConnectError::Io(e)
60    }
61}
62
63/// Opens a TCP connection to `opts.addr` and performs the WebSocket handshake.
64pub async fn connect(opts: &ClientOptions<'_>) -> Result<ClientConnection, ConnectError> {
65    let stream = TcpStream::connect(opts.addr).await?;
66    stream.set_nodelay(true)?;
67    handshake(stream, opts).await
68}
69
70/// Performs the WebSocket opening handshake over an established stream.
71pub async fn handshake<S>(
72    mut stream: S,
73    opts: &ClientOptions<'_>,
74) -> Result<ClientConnection<S>, ConnectError>
75where
76    S: AsyncRead + AsyncWrite + Unpin,
77{
78    let key = {
79        use std::hash::{BuildHasher, Hasher};
80        let a = std::collections::hash_map::RandomState::new()
81            .build_hasher()
82            .finish();
83        let b = std::collections::hash_map::RandomState::new()
84            .build_hasher()
85            .finish();
86        let mut k = a.to_le_bytes().to_vec();
87        k.extend_from_slice(&b.to_le_bytes());
88        base64::encode(&k)
89    };
90    let mut req = format!(
91        "GET {} HTTP/1.1\r\nHost: {}\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Key: {key}\r\nSec-WebSocket-Version: 13\r\n",
92        opts.path, opts.addr
93    );
94    if !opts.subprotocols.is_empty() {
95        req.push_str(&format!(
96            "Sec-WebSocket-Protocol: {}\r\n",
97            opts.subprotocols.join(", ")
98        ));
99    }
100    if opts.deflate {
101        req.push_str("Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits\r\n");
102    }
103    if let Some((user, pass)) = opts.basic_auth {
104        let mut creds = format!("{user}:").into_bytes();
105        creds.extend_from_slice(pass);
106        req.push_str(&format!(
107            "Authorization: Basic {}\r\n",
108            base64::encode(&creds)
109        ));
110    }
111    req.push_str("\r\n");
112    stream.write_all(req.as_bytes()).await?;
113
114    let mut buf = Vec::with_capacity(1024);
115    let header_end = loop {
116        if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
117            break i + 4;
118        }
119        if buf.len() > 16 * 1024 {
120            return Err(ConnectError::BadResponse("response headers too large"));
121        }
122        if stream.read_buf(&mut buf).await? == 0 {
123            return Err(ConnectError::BadResponse(
124                "connection closed during handshake",
125            ));
126        }
127    };
128    let head = std::str::from_utf8(&buf[..header_end])
129        .map_err(|_| ConnectError::BadResponse("non UTF-8 headers"))?;
130    let mut lines = head.split("\r\n");
131    let status: u16 = lines
132        .next()
133        .and_then(|l| l.split_whitespace().nth(1))
134        .and_then(|s| s.parse().ok())
135        .ok_or(ConnectError::BadResponse("invalid status line"))?;
136    if status != 101 {
137        return Err(ConnectError::Status(status));
138    }
139    let mut accept = None;
140    let mut subprotocol = None;
141    let mut extensions = None;
142    for line in lines {
143        if let Some((name, value)) = line.split_once(':') {
144            let value = value.trim().to_owned();
145            match name.trim().to_ascii_lowercase().as_str() {
146                "sec-websocket-accept" => accept = Some(value),
147                "sec-websocket-protocol" => subprotocol = Some(value),
148                "sec-websocket-extensions" => extensions = Some(value),
149                _ => {}
150            }
151        }
152    }
153    if accept.as_deref() != Some(handshake::accept_key(&key).as_str()) {
154        return Err(ConnectError::BadResponse("invalid Sec-WebSocket-Accept"));
155    }
156    let params = extensions
157        .as_deref()
158        .and_then(deflate::accept_client_response);
159    if params.is_some() && !opts.deflate {
160        return Err(ConnectError::BadResponse(
161            "server enabled an extension that was not offered",
162        ));
163    }
164    let leftover = buf[header_end..].to_vec();
165    let (r, w) = tokio::io::split(stream);
166    let cfg = DeflateConfig::default();
167    Ok(ClientConnection {
168        reader: WsReader::new(r, Role::Client, WsConfig::default(), params, leftover),
169        writer: WsWriter::new(w, Role::Client, params.map(|p| (p, cfg))),
170        subprotocol,
171        deflate: params.is_some(),
172    })
173}
codetics_ocpp_ws/client.rs

codetics_ocpp_ws/
client.rs
