netget/server/ocpp/
mod.rs

1//! OCPP-J central system (CSMS). Rust negotiates the subprotocol, frames every RPC message,
2//! correlates ids and enforces one outstanding CALL per direction; the handler decides each
3//! answer and any CSMS-initiated call.
4pub mod actions;
5pub mod frame;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, SpawnContext};
11use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
12use crate::server::connection::ConnectionId;
13use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
14use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
15use anyhow::{bail, ensure, Context, Result};
16use frame::{Frame, Version};
17use futures::{SinkExt, StreamExt};
18use serde_json::{json, Value};
19use std::net::SocketAddr;
20use std::sync::Arc;
21use std::time::Duration;
22use tokio::sync::{mpsc, Mutex};
23use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
24use tokio_tungstenite::tungstenite::protocol::WebSocketConfig;
25use tokio_tungstenite::tungstenite::Message;
26
27pub const CALL_TIMEOUT: Duration = Duration::from_secs(30);
28pub const DEFAULT_VERSIONS: [&str; 2] = ["1.6", "2.0.1"];
29pub const IDLE_TIMEOUT: Duration = Duration::from_secs(3600);
30const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30);
31
32struct Shared {
33    ctx: SpawnContext,
34    versions: Vec<Version>,
35    call_timeout: Duration,
36}
37
38type Sink = Arc<
39    Mutex<
40        futures::stream::SplitSink<
41            tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
42            Message,
43        >,
44    >,
45>;
46
47/// The CSMS-initiated CALL awaiting an answer on one connection.
48#[derive(Default)]
49struct Pending {
50    call: Option<(String, String, tokio::time::Instant)>,
51    next: u64,
52}
53
54pub fn ws_config() -> WebSocketConfig {
55    let mut c = WebSocketConfig::default();
56    c.max_message_size = Some(frame::MAX_MESSAGE_BYTES);
57    c.max_frame_size = Some(frame::MAX_MESSAGE_BYTES);
58    c
59}
60
61pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
62    let p = ctx.startup_params.as_ref();
63    let versions = match p
64        .map(|p| p.get_optional_array("ocpp_versions"))
65        .transpose()?
66        .flatten()
67    {
68        None => DEFAULT_VERSIONS
69            .iter()
70            .map(|v| Version::from_label(v).expect("known"))
71            .collect(),
72        Some(list) => {
73            let v: Vec<Version> = list
74                .iter()
75                .map(|v| {
76                    v.as_str()
77                        .and_then(Version::from_label)
78                        .context("ocpp_versions entries must be \"1.6\" or \"2.0.1\"")
79                })
80                .collect::<Result<_>>()?;
81            ensure!(!v.is_empty(), "ocpp_versions must not be empty");
82            v
83        }
84    };
85    let call_timeout = p
86        .map(|p| p.get_optional_u64("call_timeout_secs"))
87        .transpose()?
88        .flatten()
89        .unwrap_or(CALL_TIMEOUT.as_secs());
90    ensure!(
91        (1..=300).contains(&call_timeout),
92        "call_timeout_secs must be 1..=300"
93    );
94    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
95    let local = listener.local_addr()?;
96    Log::new(Some(&ctx.status_tx)).info(format!(
97        "OCPP CSMS listening on ws://{local}/<charge point id> ({})",
98        versions
99            .iter()
100            .map(|v| v.subprotocol())
101            .collect::<Vec<_>>()
102            .join(", ")
103    ));
104    let shared = Arc::new(Shared {
105        ctx: ctx.clone(),
106        versions,
107        call_timeout: Duration::from_secs(call_timeout),
108    });
109    let server_id = ctx.server_id;
110    let accept = tokio::spawn(async move {
111        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
112        loop {
113            let (socket, peer, permit) = match accept_bounded(&listener, &limiter, b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", "OCPP", Some(&shared.ctx.status_tx)).await {
114                Ok(v) => v,
115                Err(_) => break,
116            };
117            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
118            let now = crate::utils::clock::Instant::now();
119            shared
120                .ctx
121                .state
122                .add_connection_to_server(
123                    server_id,
124                    ConnectionState {
125                        id,
126                        remote_addr: peer,
127                        local_addr: local,
128                        bytes_sent: 0,
129                        bytes_received: 0,
130                        packets_sent: 0,
131                        packets_received: 0,
132                        last_activity: now,
133                        status: ConnectionStatus::Active,
134                        status_changed_at: now,
135                        protocol_info: ProtocolConnectionInfo::empty(),
136                    },
137                )
138                .await;
139            let child = shared.clone();
140            shared
141                .ctx
142                .state
143                .spawn_server_task(server_id, async move {
144                    let _permit = permit;
145                    if let Err(e) = connection(&child, id, socket).await {
146                        Log::new(Some(&child.ctx.status_tx))
147                            .debug(format!("OCPP connection {id} ended: {e}"));
148                    }
149                    child
150                        .ctx
151                        .state
152                        .remove_peer_handle(server_id, id.as_u32())
153                        .await;
154                    child
155                        .ctx
156                        .state
157                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
158                        .await;
159                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
160                })
161                .await;
162        }
163    });
164    ctx.state.register_server_task(server_id, accept).await;
165    Ok(local)
166}
167
168fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
169    let summary = format!("OCPP connection {id} action={operation} decision={decision}");
170    let log = Log::new(Some(&ctx.status_tx));
171    if decision.starts_with("fail_closed_") || decision == "model_silent" {
172        log.error(summary);
173    } else {
174        log.info(summary);
175    }
176}
177
178async fn send(shared: &Shared, id: ConnectionId, sink: &Sink, text: String) -> Result<usize> {
179    let n = text.len();
180    tokio::time::timeout(
181        Duration::from_secs(10),
182        sink.lock().await.send(Message::Text(text)),
183    )
184    .await
185    .context("OCPP write deadline")??;
186    shared
187        .ctx
188        .state
189        .update_connection_stats(
190            shared.ctx.server_id,
191            id,
192            None,
193            Some(n as u64),
194            None,
195            Some(1),
196        )
197        .await;
198    Ok(n)
199}
200
201async fn decide(shared: &Shared, id: ConnectionId, event: Event, accept: &[&str]) -> Result<Value> {
202    let ctx = &shared.ctx;
203    let result = match call_llm(
204        &ctx.llm_client,
205        &ctx.state,
206        ctx.server_id,
207        Some(id),
208        &event,
209        &actions::OcppProtocol,
210    )
211    .await
212    {
213        Ok(r) => r,
214        Err(e) => {
215            outcome(ctx, id, event.id(), "fail_closed_llm_error");
216            return Err(e);
217        }
218    };
219    if !result.failures.is_empty() {
220        outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
221        bail!("OCPP handler supplied an invalid action");
222    }
223    let mut found = None;
224    let mut pending = result.protocol_results;
225    while let Some(r) = pending.pop() {
226        match r {
227            ActionResult::Custom { name, mut data } if accept.contains(&name.as_str()) => {
228                data["type"] = json!(name);
229                if found.replace(data).is_some() {
230                    outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
231                    bail!("OCPP handler supplied more than one answer");
232                }
233            }
234            ActionResult::CloseConnection if accept.contains(&"disconnect") => {
235                found = Some(json!({"type": "disconnect"}));
236            }
237            ActionResult::Multiple(items) => pending.extend(items),
238            _ => {}
239        }
240    }
241    found
242        .context("OCPP handler did not answer")
243        .inspect_err(|_| outcome(ctx, id, event.id(), "model_silent"))
244}
245
246async fn connection(
247    shared: &Arc<Shared>,
248    id: ConnectionId,
249    socket: tokio::net::TcpStream,
250) -> Result<()> {
251    let ctx = &shared.ctx;
252    let chosen: Arc<std::sync::Mutex<Option<(Version, String)>>> = Arc::default();
253    let slot = chosen.clone();
254    let offered = shared.versions.clone();
255    let callback = move |request: &Request,
256                         mut response: Response|
257          -> std::result::Result<Response, ErrorResponse> {
258        let refuse = |code: u16, why: &str| {
259            let mut r = ErrorResponse::new(Some(why.to_owned()));
260            *r.status_mut() = tokio_tungstenite::tungstenite::http::StatusCode::from_u16(code)
261                .unwrap_or(tokio_tungstenite::tungstenite::http::StatusCode::BAD_REQUEST);
262            r
263        };
264        let cp = frame::charge_point_id(request.uri().path())
265            .map_err(|e| refuse(404, &e.to_string()))?;
266        let asked: Vec<&str> = request
267            .headers()
268            .get_all("Sec-WebSocket-Protocol")
269            .iter()
270            .filter_map(|v| v.to_str().ok())
271            .flat_map(|v| v.split(','))
272            .map(str::trim)
273            .collect();
274        // The server's preference order wins among what the charge point offered.
275        let Some(version) = offered
276            .iter()
277            .copied()
278            .find(|v| asked.contains(&v.subprotocol()))
279        else {
280            return Err(refuse(
281                400,
282                "no supported OCPP subprotocol offered (ocpp1.6, ocpp2.0.1)",
283            ));
284        };
285        response.headers_mut().insert(
286            "Sec-WebSocket-Protocol",
287            version.subprotocol().parse().expect("static header"),
288        );
289        *slot.lock().expect("handshake slot") = Some((version, cp));
290        Ok(response)
291    };
292    let ws = tokio::time::timeout(
293        HANDSHAKE_TIMEOUT,
294        tokio_tungstenite::accept_hdr_async_with_config(socket, callback, Some(ws_config())),
295    )
296    .await
297    .context("OCPP handshake deadline")??;
298    let (version, cp) = chosen
299        .lock()
300        .expect("handshake slot")
301        .clone()
302        .context("handshake completed without a negotiated version")?;
303    ctx.state
304        .with_server_mut(ctx.server_id, |s| {
305            if let Some(c) = s.connections.get_mut(&id) {
306                c.protocol_info = ProtocolConnectionInfo::new(
307                    json!({"charge_point_id": cp, "ocpp_version": version.label()}),
308                );
309            }
310        })
311        .await;
312    Log::new(Some(&ctx.status_tx)).info(format!(
313        "OCPP charge point {cp} connected (connection {id}, OCPP {})",
314        version.label()
315    ));
316    let (sink, mut stream) = ws.split();
317    let sink: Sink = Arc::new(Mutex::new(sink));
318    let pending: Arc<Mutex<Pending>> = Arc::default();
319    let peer_rx =
320        crate::server::peer_support::register_peer_channel(&ctx.state, ctx.server_id, id.as_u32())
321            .await;
322    let commands = ctx
323        .state
324        .spawn_server_task(
325            ctx.server_id,
326            peer_commands(
327                shared.clone(),
328                id,
329                version,
330                peer_rx,
331                sink.clone(),
332                pending.clone(),
333            ),
334        )
335        .await;
336    let result = async {
337        loop {
338            // A pending CSMS call that outlived its timeout is forgotten.
339            {
340                let mut p = pending.lock().await;
341                if p.call.as_ref().is_some_and(|(_, _, at)| at.elapsed() > shared.call_timeout) {
342                    let (mid, action, _) = p.call.take().expect("checked");
343                    Log::new(Some(&ctx.status_tx)).warn(format!("OCPP {cp}: {action} ({mid}) unanswered after {}s", shared.call_timeout.as_secs()));
344                }
345            }
346            let message = match tokio::time::timeout(IDLE_TIMEOUT, stream.next()).await {
347                Err(_) => bail!("charge point idle for {}s", IDLE_TIMEOUT.as_secs()),
348                Ok(None) => return Ok(()),
349                Ok(Some(m)) => m?,
350            };
351            let text = match message {
352                Message::Text(t) => t,
353                Message::Close(_) => return Ok(()),
354                Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => continue,
355                Message::Binary(_) => bail!("OCPP-J is text only; binary frame received"),
356            };
357            ctx.state.update_connection_stats(ctx.server_id, id, Some(text.len() as u64), None, Some(1), None).await;
358            match frame::parse(&text) {
359                Err((Some(mid), why)) => {
360                    outcome(ctx, id, "frame", "protocol_refusal");
361                    send(shared, id, &sink, frame::error_frame(&mid, version.formation_code(), &why)).await?;
362                }
363                Err((None, why)) => {
364                    outcome(ctx, id, "frame", "protocol_refusal");
365                    send(shared, id, &sink, frame::error_frame("-1", version.formation_code(), &why)).await?;
366                }
367                Ok(Frame::Call { id: mid, action, payload }) => {
368                    let reply = answer_call(shared, id, version, &cp, &mid, &action, payload).await;
369                    send(shared, id, &sink, reply).await?;
370                }
371                Ok(Frame::Result { id: mid, payload }) => {
372                    if let Some(action) = take_pending(&pending, &mid).await {
373                        let data = json!({"charge_point_id": cp, "action": action, "message_id": mid, "payload": payload});
374                        if let Some(next) = react(shared, id, data).await {
375                            if next["type"] == "disconnect" {
376                                return Ok(());
377                            }
378                            start_call(shared, id, version, &sink, &pending, &next).await?;
379                        }
380                    }
381                }
382                Ok(Frame::Error { id: mid, code, description, details }) => {
383                    if let Some(action) = take_pending(&pending, &mid).await {
384                        let data = json!({"charge_point_id": cp, "action": action, "message_id": mid, "error": {"code": code, "description": description, "details": details}});
385                        if let Some(next) = react(shared, id, data).await {
386                            if next["type"] == "disconnect" {
387                                return Ok(());
388                            }
389                            start_call(shared, id, version, &sink, &pending, &next).await?;
390                        }
391                    }
392                }
393            }
394        }
395    }
396    .await;
397    commands.abort();
398    let _ = sink.lock().await.close().await;
399    result
400}
401
402async fn take_pending(pending: &Mutex<Pending>, mid: &str) -> Option<String> {
403    let mut p = pending.lock().await;
404    match &p.call {
405        Some((id, _, _)) if id == mid => p.call.take().map(|(_, action, _)| action),
406        // An answer to nothing pending is ignored (OCPP-J §4.1.4).
407        _ => None,
408    }
409}
410
411async fn answer_call(
412    shared: &Shared,
413    id: ConnectionId,
414    version: Version,
415    cp: &str,
416    mid: &str,
417    action: &str,
418    payload: Value,
419) -> String {
420    let ctx = &shared.ctx;
421    if let Err(e) = frame::check_request(version, action, &payload) {
422        outcome(ctx, id, action, "protocol_refusal");
423        return frame::error_frame(mid, version.occurrence_code(), &e.to_string());
424    }
425    let event = Event::new(
426        &actions::CALL_EVENT,
427        json!({"charge_point_id": cp, "ocpp_version": version.label(), "action": action, "message_id": mid, "payload": payload}),
428    );
429    let failed = || {
430        frame::error_frame(
431            mid,
432            "InternalError",
433            "the central system cannot answer right now",
434        )
435    };
436    match decide(shared, id, event, &["ocpp_call_result", "ocpp_call_error"]).await {
437        Err(_) => failed(),
438        Ok(v) if v["type"] == "ocpp_call_result" => {
439            match frame::check_response(version, action, &v["payload"]).and_then(|_| {
440                frame::encode(&Frame::Result {
441                    id: mid.into(),
442                    payload: v["payload"].clone(),
443                })
444            }) {
445                Ok(text) => {
446                    outcome(ctx, id, action, "model_answer");
447                    text
448                }
449                Err(_) => {
450                    outcome(ctx, id, action, "fail_closed_invalid_reply");
451                    failed()
452                }
453            }
454        }
455        Ok(v) => {
456            let code = v["code"].as_str().unwrap_or("");
457            match frame::validate_error_code(version, code) {
458                Ok(()) => {
459                    outcome(ctx, id, action, "model_reject");
460                    let details = if v["details"].is_object() {
461                        v["details"].clone()
462                    } else {
463                        json!({})
464                    };
465                    frame::encode(&Frame::Error {
466                        id: mid.into(),
467                        code: code.into(),
468                        description: v["description"].as_str().unwrap_or("").into(),
469                        details,
470                    })
471                    .unwrap_or_else(|_| failed())
472                }
473                Err(_) => {
474                    outcome(ctx, id, action, "fail_closed_invalid_reply");
475                    failed()
476                }
477            }
478        }
479    }
480}
481
482/// Let the handler react to a charge point's answer; returns a follow-up call or disconnect.
483async fn react(shared: &Shared, id: ConnectionId, data: Value) -> Option<Value> {
484    let label = data["action"].as_str().unwrap_or("").to_owned();
485    match decide(
486        shared,
487        id,
488        Event::new(&actions::CALL_RESPONSE_EVENT, data),
489        &["ocpp_send_call", "disconnect"],
490    )
491    .await
492    {
493        Ok(v) => {
494            outcome(&shared.ctx, id, &label, "model_answer");
495            Some(v)
496        }
497        // Silence after a response is fine: nothing more to send.
498        Err(_) => None,
499    }
500}
501
502async fn start_call(
503    shared: &Shared,
504    id: ConnectionId,
505    version: Version,
506    sink: &Sink,
507    pending: &Mutex<Pending>,
508    call: &Value,
509) -> Result<usize> {
510    actions::validate_answer(call)?;
511    let action = call["action"].as_str().unwrap_or_default().to_owned();
512    frame::check_request(version, &action, &call["payload"])?;
513    let mut p = pending.lock().await;
514    ensure!(
515        p.call.is_none(),
516        "a CSMS call is already outstanding on this connection; wait for its answer"
517    );
518    p.next += 1;
519    let mid = format!("csms-{}", p.next);
520    let text = frame::encode(&Frame::Call {
521        id: mid.clone(),
522        action: action.clone(),
523        payload: call["payload"].clone(),
524    })?;
525    p.call = Some((mid, action, tokio::time::Instant::now()));
526    drop(p);
527    send(shared, id, sink, text).await
528}
529
530async fn peer_commands(
531    shared: Arc<Shared>,
532    id: ConnectionId,
533    version: Version,
534    mut rx: mpsc::Receiver<ClientCommand>,
535    sink: Sink,
536    pending: Arc<Mutex<Pending>>,
537) {
538    while let Some(command) = rx.recv().await {
539        let outcome: Result<ClientSendOutcome> = match command.action["type"].as_str() {
540            Some("ocpp_send_call") => {
541                match start_call(&shared, id, version, &sink, &pending, &command.action).await {
542                    Ok(n) => Ok(ClientSendOutcome::Sent { bytes_sent: n }),
543                    Err(e) => Ok(ClientSendOutcome::Rejected {
544                        error: e.to_string(),
545                    }),
546                }
547            }
548            Some("disconnect") => {
549                let _ = sink.lock().await.close().await;
550                Ok(ClientSendOutcome::Disconnected)
551            }
552            _ => Ok(ClientSendOutcome::Rejected {
553                error: "OCPP peers accept ocpp_send_call or disconnect".into(),
554            }),
555        };
556        shared
557            .ctx
558            .state
559            .record_access_log(
560                crate::state::AccessLogOwner::Server(shared.ctx.server_id.as_u32()),
561                "OCPP",
562                Some(id.as_u32()),
563                "injected_action",
564                json!({"type": command.action["type"], "action": command.action["action"]}),
565                vec![outcome
566                    .as_ref()
567                    .map(|o| serde_json::to_value(o).unwrap_or(Value::Null))
568                    .unwrap_or(Value::Null)],
569            )
570            .await;
571        let _ = command.reply_tx.send(outcome);
572    }
573}
netget/server/ocpp/mod.rs

netget/server/ocpp/
mod.rs
