Communication subprotocol implementation

Cargo.toml

@@ -4,6 +4,7 @@ resolver = "3"
 
 [workspace.dependencies]
 axum = "0.8.8"
+axum-extra = { version = "0.12.5", features = ["typed-header"] }
 base64 = "0.22.1"
 bon = "3.8.0"
 chrono = { version = "0.4.42", features = ["serde"] }

s2energy-connection/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2024"
 
 [dependencies]
 axum.workspace = true
+axum-extra.workspace = true
 base64.workspace = true
 hmac.workspace = true
 rand.workspace = true

s2energy-connection/examples/communication-client.rs

@@ -0,0 +1,73 @@
+use std::{convert::Infallible, path::PathBuf, sync::Arc};
+
+use rustls::pki_types::{CertificateDer, pem::PemObject};
+use s2energy_connection::{
+    AccessToken, MessageVersion, S2NodeId,
+    communication::{Client, ClientConfig, ClientPairing, NodeConfig},
+};
+
+struct MemoryPairing {
+    communication_url: String,
+    tokens: Vec<AccessToken>,
+    server: S2NodeId,
+    client: S2NodeId,
+}
+
+impl ClientPairing for &mut MemoryPairing {
+    type Error = Infallible;
+
+    fn client_id(&self) -> S2NodeId {
+        self.client.clone()
+    }
+
+    fn server_id(&self) -> S2NodeId {
+        self.server.clone()
+    }
+
+    fn access_tokens(&self) -> impl AsRef<[AccessToken]> {
+        &self.tokens
+    }
+
+    fn communication_url(&self) -> impl AsRef<str> {
+        &self.communication_url
+    }
+
+    async fn set_access_tokens(&mut self, tokens: Vec<AccessToken>) -> Result<(), Self::Error> {
+        self.tokens = tokens;
+        Ok(())
+    }
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+    let client = Client::new(
+        ClientConfig {
+            additional_certificates: vec![
+                CertificateDer::from_pem_file(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("testdata").join("root.pem")).unwrap(),
+            ],
+            endpoint_description: None,
+        },
+        Arc::new(NodeConfig::builder(vec![MessageVersion("v1".into())]).build()),
+    );
+
+    let mut pairing = MemoryPairing {
+        communication_url: "https://localhost:8005/".into(),
+        tokens: vec![AccessToken("0123456789ABCDEF".into())],
+        server: S2NodeId("12".into()),
+        client: S2NodeId("34".into()),
+    };
+
+    let connection_info = client.connect(&mut pairing).await.unwrap();
+
+    println!(
+        "Url: {}, token: {}",
+        connection_info.communication_url, connection_info.communication_token.0
+    );
+
+    let connection_info = client.connect(&mut pairing).await.unwrap();
+
+    println!(
+        "Url: {}, token: {}",
+        connection_info.communication_url, connection_info.communication_token.0
+    );
+}

s2energy-connection/examples/communication-server.rs

@@ -0,0 +1,107 @@
+use std::{
+    convert::Infallible,
+    net::SocketAddr,
+    path::PathBuf,
+    sync::{Arc, Mutex},
+};
+
+use axum_server::tls_rustls::RustlsConfig;
+use s2energy_connection::{
+    AccessToken, MessageVersion, S2NodeId,
+    communication::{NodeConfig, PairingLookupResult, Server, ServerConfig, ServerPairing, ServerPairingStore},
+};
+
+struct MemoryPairingStoreInner {
+    token: AccessToken,
+    config: Arc<NodeConfig>,
+    server: S2NodeId,
+    client: S2NodeId,
+}
+
+#[derive(Clone)]
+struct MemoryPairingStore(Arc<Mutex<MemoryPairingStoreInner>>);
+
+impl MemoryPairingStore {
+    fn new() -> Self {
+        MemoryPairingStore(Arc::new(Mutex::new(MemoryPairingStoreInner {
+            token: AccessToken("0123456789ABCDEF".into()),
+            config: Arc::new(NodeConfig::builder(vec![MessageVersion("v1".into())]).build()),
+            server: S2NodeId("12".into()),
+            client: S2NodeId("34".into()),
+        })))
+    }
+}
+
+impl ServerPairingStore for MemoryPairingStore {
+    type Error = Infallible;
+
+    type Pairing<'a>
+        = MemoryPairingStore
+    where
+        Self: 'a;
+
+    async fn lookup(
+        &self,
+        request: s2energy_connection::communication::PairingLookup,
+    ) -> Result<s2energy_connection::communication::PairingLookupResult<Self::Pairing<'_>>, Self::Error> {
+        let this = self.0.lock().unwrap();
+        if this.client == request.client && this.server == request.server {
+            Ok(PairingLookupResult::Pairing(self.clone()))
+        } else {
+            Ok(PairingLookupResult::NeverPaired)
+        }
+    }
+}
+
+impl ServerPairing for MemoryPairingStore {
+    type Error = Infallible;
+
+    fn access_token(&self) -> impl AsRef<AccessToken> {
+        self.0.lock().unwrap().token.clone()
+    }
+
+    fn config(&self) -> impl AsRef<NodeConfig> {
+        self.0.lock().unwrap().config.clone()
+    }
+
+    async fn set_access_token(&mut self, token: AccessToken) -> Result<(), Self::Error> {
+        self.0.lock().unwrap().token = token;
+        Ok(())
+    }
+
+    async fn update_remote_node_description(&mut self, _node_description: s2energy_connection::S2NodeDescription) {
+        println!("Received updated node description");
+    }
+
+    async fn update_remote_endpoint_description(&mut self, _endpoint_description: s2energy_connection::S2EndpointDescription) {
+        println!("Received updated endpoint description");
+    }
+}
+
+#[tokio::main(flavor = "current_thread")]
+async fn main() {
+    let server = Server::new(
+        ServerConfig {
+            base_url: "localhost".into(),
+            endpoint_description: None,
+        },
+        MemoryPairingStore::new(),
+    );
+
+    let addr = SocketAddr::from(([127, 0, 0, 1], 8005));
+
+    let rustls_config = RustlsConfig::from_pem_file(
+        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+            .join("testdata")
+            .join("localhost.chain.pem"),
+        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("testdata").join("localhost.key"),
+    )
+    .await
+    .unwrap();
+
+    println!("listening on http://{}", addr);
+    axum_server::bind_rustls(addr, rustls_config)
+        .serve(server.get_router().into_make_service())
+        .await
+        .unwrap();
+}

s2energy-connection/examples/pairing-client.rs

@@ -1,7 +1,8 @@
 use std::sync::Arc;
 
-use s2energy_connection::pairing::{
-    Client, ClientConfig, Deployment, EndpointConfig, MessageVersion, PairingRemote, S2NodeDescription, S2NodeId, S2Role,
+use s2energy_connection::{
+    Deployment, MessageVersion, S2NodeDescription, S2NodeId, S2Role,
+    pairing::{Client, ClientConfig, EndpointConfig, PairingRemote},
 };
 
 const PAIRING_TOKEN: &[u8] = &[1, 2, 3];

s2energy-connection/examples/pairing-server.rs

@@ -2,8 +2,9 @@ use axum_server::tls_rustls::RustlsConfig;
 use rustls::pki_types::{CertificateDer, pem::PemObject};
 use std::{net::SocketAddr, path::PathBuf, sync::Arc};
 
-use s2energy_connection::pairing::{
-    EndpointConfig, MessageVersion, PairingToken, S2NodeDescription, S2NodeId, S2Role, Server, ServerConfig,
+use s2energy_connection::{
+    MessageVersion, S2NodeDescription, S2NodeId, S2Role,
+    pairing::{EndpointConfig, PairingToken, Server, ServerConfig},
 };
 
 #[allow(unused)]

s2energy-connection/src/common/mod.rs

@@ -0,0 +1,41 @@
+use axum::Json;
+
+pub(crate) mod wire;
+
+use reqwest::{StatusCode, Url};
+use wire::PairingVersion;
+
+use crate::common::wire::WirePairingVersion;
+
+pub(crate) const SUPPORTED_PAIRING_VERSIONS: &[PairingVersion] = &[PairingVersion::V1];
+
+pub(crate) async fn root() -> Json<&'static [PairingVersion]> {
+    Json(SUPPORTED_PAIRING_VERSIONS)
+}
+
+pub(crate) enum BaseError {
+    TransportFailed,
+    ProtocolError,
+    NoSupportedVersion,
+}
+
+pub(crate) async fn negotiate_version(client: &reqwest::Client, url: Url) -> Result<PairingVersion, BaseError> {
+    let response = client.get(url).send().await.map_err(|_| BaseError::TransportFailed)?;
+    let status = response.status();
+    if status != StatusCode::OK {
+        return Err(BaseError::ProtocolError);
+    }
+
+    let supported_versions = response
+        .json::<Vec<WirePairingVersion>>()
+        .await
+        .map_err(|_| BaseError::ProtocolError)?;
+
+    for version in supported_versions.into_iter().filter_map(|v| v.try_into().ok()) {
+        if SUPPORTED_PAIRING_VERSIONS.contains(&version) {
+            return Ok(version);
+        }
+    }
+
+    Err(BaseError::NoSupportedVersion)
+}