confium_coordinator/coordinator/
net.rs

1//! TCP network protocol for coordinator ↔ signer communication.
2//!
3//! Wire format: 4-byte big-endian length prefix + JSON payload.
4//!
5//! Messages flow in both directions:
6//! - Signer → Coordinator: Register, Commitment, Share
7//! - Coordinator → Signer: Registered, SessionPending, CommitmentsReady, Signature
8//! - Client → Coordinator: CreateSession, GetStatus
9//! - Coordinator → Client: SessionCreated, Signature, Error
10
11use crate::coordinator::session::SignerId;
12use serde::{Deserialize, Serialize};
13use std::io;
14
15/// Protocol message exchanged over TCP between coordinator, signers, and clients.
16#[derive(Debug, Clone, Serialize, Deserialize)]
17#[serde(tag = "type")]
18pub enum ProtocolMessage {
19    /// Signer registers with coordinator.
20    Register {
21        /// Signer identity.
22        signer_id: SignerId,
23        /// Quorum this signer belongs to.
24        quorum_id: String,
25    },
26    /// Coordinator acknowledges registration.
27    Registered {
28        /// Signer identity.
29        signer_id: SignerId,
30    },
31    /// Client requests session creation.
32    CreateSession {
33        /// Quorum ID.
34        quorum_id: String,
35        /// Signing scheme.
36        scheme: String,
37        /// Message to sign.
38        message: Vec<u8>,
39        /// Threshold T.
40        threshold: u32,
41        /// Total parties N.
42        num_parties: u32,
43    },
44    /// Coordinator confirms session created.
45    SessionCreated {
46        /// Session ID.
47        session_id: String,
48    },
49    /// Coordinator notifies signers of pending session.
50    SessionPending {
51        /// Session ID.
52        session_id: String,
53        /// Message to sign.
54        message: Vec<u8>,
55        /// Threshold.
56        threshold: u32,
57    },
58    /// Signer submits commitment.
59    Commitment {
60        /// Session ID.
61        session_id: String,
62        /// Signer identity.
63        signer_id: SignerId,
64        /// Commitment bytes.
65        bytes: Vec<u8>,
66        /// Identity signature.
67        signature: Vec<u8>,
68    },
69    /// Coordinator notifies that commitments are collected.
70    CommitmentsReady {
71        /// Session ID.
72        session_id: String,
73    },
74    /// Signer submits share.
75    Share {
76        /// Session ID.
77        session_id: String,
78        /// Signer identity.
79        signer_id: SignerId,
80        /// Share bytes.
81        bytes: Vec<u8>,
82        /// Identity signature.
83        signature: Vec<u8>,
84    },
85    /// Coordinator returns aggregated signature.
86    Signature {
87        /// Session ID.
88        session_id: String,
89        /// Signature bytes.
90        bytes: Vec<u8>,
91        /// Algorithm.
92        algorithm: String,
93        /// Contributing signers.
94        contributing_signers: Vec<SignerId>,
95    },
96    /// Acknowledgement (commitment or share accepted, no further action needed).
97    Ack {
98        /// Session ID being acknowledged.
99        session_id: String,
100    },
101    /// Error response.
102    Error {
103        /// Error message.
104        message: String,
105    },
106    /// Status query.
107    GetStatus {
108        /// Session ID (optional).
109        session_id: Option<String>,
110    },
111    /// Status response.
112    Status {
113        /// Session ID.
114        session_id: String,
115        /// Current state.
116        state: String,
117    },
118    /// Liveness/readiness probe.
119    HealthCheck,
120    /// Health status response.
121    HealthStatus {
122        /// Server is running.
123        alive: bool,
124        /// Coordinator can accept sessions.
125        ready: bool,
126        /// Active session count.
127        session_count: usize,
128        /// Server uptime in seconds.
129        uptime_seconds: u64,
130    },
131    /// Prometheus metrics query.
132    MetricsQuery,
133    /// Prometheus metrics response (text exposition format).
134    MetricsResponse {
135        /// Prometheus text format metrics.
136        text: String,
137    },
138}
139
140/// Send a protocol message over any writable stream.
141pub fn send_message<S: std::io::Write + ?Sized>(
142    stream: &mut S,
143    msg: &ProtocolMessage,
144) -> io::Result<()> {
145    let json = serde_json::to_vec(msg)?;
146    let len = json.len() as u32;
147    stream.write_all(&len.to_be_bytes())?;
148    stream.write_all(&json)?;
149    stream.flush()?;
150    Ok(())
151}
152
153/// Receive a protocol message from any readable stream.
154pub fn recv_message<S: std::io::Read + ?Sized>(stream: &mut S) -> io::Result<ProtocolMessage> {
155    let mut len_buf = [0u8; 4];
156    stream.read_exact(&mut len_buf)?;
157    let len = u32::from_be_bytes(len_buf) as usize;
158
159    if len > 16 * 1024 * 1024 {
160        return Err(io::Error::new(
161            io::ErrorKind::InvalidData,
162            format!("message too large: {} bytes", len),
163        ));
164    }
165
166    let mut json_buf = vec![0u8; len];
167    stream.read_exact(&mut json_buf)?;
168
169    let msg: ProtocolMessage = serde_json::from_slice(&json_buf)?;
170    Ok(msg)
171}
172
173#[cfg(test)]
174mod tests {
175    use super::*;
176
177    #[test]
178    fn protocol_message_serializes() {
179        let msg = ProtocolMessage::Register {
180            signer_id: "alice".into(),
181            quorum_id: "test".into(),
182        };
183        let json = serde_json::to_vec(&msg).unwrap();
184        assert!(json.len() > 10);
185        let recovered: ProtocolMessage = serde_json::from_slice(&json).unwrap();
186        match recovered {
187            ProtocolMessage::Register {
188                signer_id,
189                quorum_id,
190            } => {
191                assert_eq!(signer_id, "alice");
192                assert_eq!(quorum_id, "test");
193            }
194            _ => panic!("wrong variant"),
195        }
196    }
197
198    #[test]
199    fn all_variants_round_trip() {
200        let messages = vec![
201            ProtocolMessage::Register {
202                signer_id: "a".into(),
203                quorum_id: "q".into(),
204            },
205            ProtocolMessage::Registered {
206                signer_id: "a".into(),
207            },
208            ProtocolMessage::CreateSession {
209                quorum_id: "q".into(),
210                scheme: "FROST-P256".into(),
211                message: vec![1, 2, 3],
212                threshold: 3,
213                num_parties: 5,
214            },
215            ProtocolMessage::SessionCreated {
216                session_id: "s1".into(),
217            },
218            ProtocolMessage::SessionPending {
219                session_id: "s1".into(),
220                message: vec![1, 2, 3],
221                threshold: 3,
222            },
223            ProtocolMessage::Commitment {
224                session_id: "s1".into(),
225                signer_id: "a".into(),
226                bytes: vec![4, 5],
227                signature: vec![6, 7],
228            },
229            ProtocolMessage::Signature {
230                session_id: "s1".into(),
231                bytes: vec![8, 9],
232                algorithm: "FROST-P256".into(),
233                contributing_signers: vec!["a".into(), "b".into()],
234            },
235            ProtocolMessage::Error {
236                message: "test".into(),
237            },
238        ];
239        for msg in &messages {
240            let json = serde_json::to_vec(msg).unwrap();
241            let recovered: ProtocolMessage = serde_json::from_slice(&json).unwrap();
242            let json2 = serde_json::to_vec(&recovered).unwrap();
243            assert_eq!(json, json2, "round-trip must preserve bytes");
244        }
245    }
246}
confium_coordinator/coordinator/net.rs

confium_coordinator/coordinator/
net.rs
