codetics-ocpp-service 0.1.0

// SPDX-License-Identifier: Apache-2.0
// Copyright 2026 Kevin Vavelin, Codetics (https://codetics.fr)

//! WebSocket opening handshake for charging stations (OCPP-J §3).
//!
//! The connection URL is `<endpoint>/<identity>`: the station identity is the
//! last path segment, percent-decoded. Any endpoint prefix is accepted, and a
//! version in the URL is ignored: only `Sec-WebSocket-Protocol` decides.

use axum::body::Body;
use axum::extract::{Extension, Request, State};
use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, header};
use axum::response::{IntoResponse, Response};
use hyper_util::rt::TokioIo;
use ocpp_core::OcppVersion;
use ocpp_core::version::negotiate;
use ocpp_ws::{DeflateConfig, handshake};

use super::security::{self, Denied};
use super::session::{self, SessionParams};
use crate::auth;
use crate::db;
use crate::server::ConnMeta;
use crate::state::App;

/// Max identity length (OCPP 2.0.1 Part 4 §3.1.1).
const MAX_IDENTITY_LEN: usize = 48;

fn reject(status: StatusCode, msg: &'static str) -> Response {
    (status, msg).into_response()
}

fn headers_all(h: &HeaderMap, name: header::HeaderName) -> impl Iterator<Item = &str> {
    h.get_all(name).iter().filter_map(|v| v.to_str().ok())
}

pub fn percent_decode(s: &str) -> Option<String> {
    let b = s.as_bytes();
    let mut out = Vec::with_capacity(b.len());
    let mut i = 0;
    while i < b.len() {
        if b[i] == b'%' {
            let hex = std::str::from_utf8(b.get(i + 1..i + 3)?).ok()?;
            out.push(u8::from_str_radix(hex, 16).ok()?);
            i += 3;
        } else {
            out.push(b[i]);
            i += 1;
        }
    }
    String::from_utf8(out).ok()
}

pub async fn upgrade(
    State(app): State<App>,
    Extension(meta): Extension<ConnMeta>,
    mut req: Request,
) -> Response {
    let peer = meta.peer;
    let h = req.headers();
    if req.method() != Method::GET {
        return reject(
            StatusCode::METHOD_NOT_ALLOWED,
            "WebSocket upgrade requires GET",
        );
    }
    let is_upgrade = headers_all(h, header::CONNECTION)
        .any(|v| handshake::header_has_token(v, "upgrade"))
        && headers_all(h, header::UPGRADE).any(|v| handshake::header_has_token(v, "websocket"));
    if !is_upgrade {
        return reject(
            StatusCode::BAD_REQUEST,
            "expected a WebSocket upgrade request",
        );
    }
    if h.get(header::SEC_WEBSOCKET_VERSION)
        .and_then(|v| v.to_str().ok())
        != Some("13")
    {
        let mut r = reject(
            StatusCode::UPGRADE_REQUIRED,
            "unsupported WebSocket version",
        );
        r.headers_mut().insert(
            header::SEC_WEBSOCKET_VERSION,
            HeaderValue::from_static("13"),
        );
        return r;
    }
    let Some(key) = h
        .get(header::SEC_WEBSOCKET_KEY)
        .and_then(|v| v.to_str().ok())
        .filter(|k| handshake::valid_client_key(k))
    else {
        return reject(StatusCode::BAD_REQUEST, "invalid Sec-WebSocket-Key");
    };
    let accept = handshake::accept_key(key);

    let identity = req
        .uri()
        .path()
        .trim_end_matches('/')
        .rsplit('/')
        .next()
        .and_then(percent_decode)
        .unwrap_or_default();
    if identity.is_empty() || identity.chars().count() > MAX_IDENTITY_LEN || identity.contains(':')
    {
        return reject(StatusCode::NOT_FOUND, "unknown charging station");
    }

    // OCPP-J §3.2: unknown station → 404.
    let station = match db::stations::get_by_identity(&app.db, &identity).await {
        Ok(Some(s)) => s,
        Ok(None) => {
            tracing::info!(%identity, %peer, "connection from unregistered charging station rejected");
            return reject(StatusCode::NOT_FOUND, "unknown charging station");
        }
        Err(e) => {
            tracing::error!(error = %e, "station lookup failed");
            return reject(StatusCode::SERVICE_UNAVAILABLE, "temporarily unavailable");
        }
    };

    // HTTP Basic authentication (Security Profiles 1/2): username must equal the identity.
    let creds = h
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(auth::parse_basic);
    let basic = match (&station.auth_password_hash, creds) {
        (_, None) => None,
        (Some(phc), Some((user, password))) => {
            Some(user == identity && app.auth.verify(&identity, &password, phc).await)
        }
        (None, Some(_)) => Some(false),
    };
    let decision = security::check(&security::Inputs {
        profile: station.security_profile,
        tls: meta.tls.as_ref(),
        basic,
        has_password: station.auth_password_hash.is_some(),
        require_password: app.cfg.require_password,
        identity: &identity,
        serial_number: station.serial_number.as_deref(),
        cso_name: app.cfg.cso_name.as_deref(),
    });
    if let Err(denied) = decision {
        tracing::warn!(%identity, %peer, reason = ?denied, "charging station connection refused");
        return match denied {
            Denied::BadCredentials => {
                let mut r = reject(StatusCode::UNAUTHORIZED, "authentication required");
                r.headers_mut().insert(
                    header::WWW_AUTHENTICATE,
                    HeaderValue::from_static("Basic realm=\"OCPP\", charset=\"UTF-8\""),
                );
                r
            }
            Denied::InvalidCertificate(reason) => {
                // A00.FR.429: recommended to log a security event in the CSMS.
                let event = serde_json::json!({ "type": "InvalidChargingStationCertificate", "reason": reason, "peer": peer.to_string() });
                if let Err(e) =
                    db::events::insert(&app.db, station.id, "SecurityEvent", &event).await
                {
                    tracing::warn!(error = %e, "failed to store security event");
                }
                reject(
                    StatusCode::FORBIDDEN,
                    "invalid charging station certificate",
                )
            }
            Denied::ProfileMismatch(msg) => reject(StatusCode::FORBIDDEN, msg),
        };
    }

    // Version: client preference order, restricted to service and per-station settings.
    let acceptable: Vec<OcppVersion> = app
        .cfg
        .versions
        .iter()
        .copied()
        .filter(|v| {
            station.allowed_versions.as_ref().is_none_or(|allowed| {
                allowed
                    .iter()
                    .any(|a| a.parse::<OcppVersion>().ok() == Some(*v))
            })
        })
        .collect();
    let negotiated = negotiate(headers_all(h, header::SEC_WEBSOCKET_PROTOCOL), &acceptable)
        .map(|(v, token)| (v, token.to_owned()));
    if negotiated.is_none() {
        let offered: Vec<&str> = headers_all(h, header::SEC_WEBSOCKET_PROTOCOL).collect();
        tracing::warn!(%identity, ?offered, "no acceptable OCPP subprotocol offered");
    }

    let deflate_cfg = DeflateConfig {
        enabled: app.cfg.ws_deflate,
        force_client_no_context_takeover: app.cfg.ws_force_client_no_context_takeover,
        ..Default::default()
    };
    let deflate = if negotiated.is_some() {
        ocpp_ws::deflate::negotiate_server(
            headers_all(h, header::SEC_WEBSOCKET_EXTENSIONS),
            &deflate_cfg,
        )
    } else {
        None
    };

    let forwarded = h
        .get("x-forwarded-for")
        .and_then(|v| v.to_str().ok())
        .map(|v| v.split(',').next().unwrap_or("").trim().to_owned());
    let remote_addr = Some(forwarded.unwrap_or_else(|| peer.to_string()));

    let on_upgrade = hyper::upgrade::on(&mut req);
    let mut response = Response::builder()
        .status(StatusCode::SWITCHING_PROTOCOLS)
        .header(header::UPGRADE, "websocket")
        .header(header::CONNECTION, "Upgrade")
        .header(header::SEC_WEBSOCKET_ACCEPT, accept);
    if let Some((_, token)) = &negotiated {
        // Echo the token exactly as offered.
        response = response.header(header::SEC_WEBSOCKET_PROTOCOL, token.as_str());
    }
    if let Some((_, ext)) = &deflate {
        response = response.header(header::SEC_WEBSOCKET_EXTENSIONS, ext.as_str());
    }
    let response = match response.body(Body::empty()) {
        Ok(r) => r,
        Err(_) => return reject(StatusCode::BAD_REQUEST, "invalid header value"),
    };

    let params = SessionParams {
        app: app.clone(),
        station,
        version: negotiated.map(|(v, _)| v),
        deflate: deflate.map(|(p, _)| (p, deflate_cfg)),
        remote_addr,
    };
    tokio::spawn(async move {
        match on_upgrade.await {
            Ok(upgraded) => session::run(params, TokioIo::new(upgraded)).await,
            Err(e) => tracing::warn!(error = %e, "WebSocket upgrade failed"),
        }
    });
    response
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn percent_decoding() {
        assert_eq!(percent_decode("RDAM%7C123").unwrap(), "RDAM|123");
        assert_eq!(percent_decode("RDAM%20123").unwrap(), "RDAM 123");
        assert!(percent_decode("bad%2").is_none());
    }
}