confium_coordinator/coordinator/
client.rs

1//! TCP signer client — connects to coordinator, participates in signing sessions.
2//!
3//! Each signer:
4//! 1. Connects to coordinator via TCP
5//! 2. Registers with signer_id + quorum_id
6//! 3. Creates a session (or receives notification of pending session)
7//! 4. Submits commitment (round 1)
8//! 5. Submits share (round 2)
9//! 6. Receives aggregated signature (or error)
10//!
11//! Usage in e2e tests:
12//! ```ignore
13//! use confium_tc::coordinator::client::SignerClient;
14//!
15//! let mut client = SignerClient::connect("127.0.0.1:18432").unwrap();
16//! client.register("director-1", "biml-root").unwrap();
17//! client.submit_commitment("session-1", "director-1", &[0u8; 32]).unwrap();
18//! client.submit_share("session-1", "director-1", &[0u8; 32]).unwrap();
19//! ```
20
21use std::io;
22use std::net::TcpStream;
23
24use crate::coordinator::net::ProtocolMessage;
25use crate::coordinator::net::recv_message;
26use crate::coordinator::net::send_message;
27use crate::coordinator::net_server::SessionIo;
28
29/// TCP signer client.
30pub struct SignerClient {
31    stream: Box<dyn SessionIo>,
32}
33
34/// Polling read deadline over a session — see the SessionIo docs for
35/// why SO_RCVTIMEO is avoided. When the underlying session cannot be
36/// switched to non-blocking this degrades to the legacy socket
37/// timeout hook (unchanged behavior for exotic transports).
38struct Bounded<'a> {
39    session: &'a mut Box<dyn crate::coordinator::net_server::SessionIo>,
40    deadline: std::time::Instant,
41    nonblocking: bool,
42}
43
44impl<'a> Bounded<'a> {
45    fn new(
46        session: &'a mut Box<dyn crate::coordinator::net_server::SessionIo>,
47        timeout: std::time::Duration,
48    ) -> Self {
49        let nonblocking = session.set_nonblocking(true);
50        if !nonblocking {
51            let _ = session.set_read_timeout(Some(timeout));
52        }
53        Self {
54            session,
55            deadline: std::time::Instant::now() + timeout,
56            nonblocking,
57        }
58    }
59
60    fn expired(&self) -> bool {
61        std::time::Instant::now() >= self.deadline
62    }
63}
64
65impl std::io::Read for Bounded<'_> {
66    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
67        if !self.nonblocking {
68            return self.session.read(buf);
69        }
70        loop {
71            match self.session.read(buf) {
72                Ok(n) => return Ok(n),
73                Err(ref e)
74                    if e.kind() == io::ErrorKind::WouldBlock
75                        || e.kind() == io::ErrorKind::TimedOut =>
76                {
77                    if self.expired() {
78                        return Err(io::Error::new(
79                            io::ErrorKind::TimedOut,
80                            "coordinator response deadline exceeded",
81                        ));
82                    }
83                    std::thread::sleep(std::time::Duration::from_millis(2));
84                }
85                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
86                Err(e) => return Err(e),
87            }
88        }
89    }
90}
91
92impl Drop for Bounded<'_> {
93    fn drop(&mut self) {
94        if self.nonblocking {
95            self.session.set_nonblocking(false);
96        } else {
97            self.session.set_read_timeout(None);
98        }
99    }
100}
101
102impl SignerClient {
103    /// Connect to coordinator at `addr` (e.g., "127.0.0.1:18432").
104    pub fn connect(addr: &str) -> io::Result<Self> {
105        let stream = TcpStream::connect(addr)?;
106        Ok(Self {
107            stream: Box::new(stream),
108        })
109    }
110
111    /// Connect over any registry transport URL — plain
112    /// (`tcp://host:port`) or encrypted (`noise://host:port`, with
113    /// optional `key=`/`pinned=` parameters). Link the transport
114    /// crate (e.g. `confium-net-noise`) into the binary to enable
115    /// its scheme.
116    pub fn connect_url(url: &str) -> io::Result<Self> {
117        let transport = confium_net::connect(url)
118            .map_err(|e| io::Error::other(format!("connect {url}: {e}")))?;
119        Ok(Self {
120            stream: Box::new(confium_net::io::TransportIo::new(transport)),
121        })
122    }
123
124    /// Register this signer with the coordinator.
125    pub fn register(&mut self, signer_id: &str, quorum_id: &str) -> io::Result<()> {
126        send_message(
127            &mut self.stream,
128            &ProtocolMessage::Register {
129                signer_id: signer_id.into(),
130                quorum_id: quorum_id.into(),
131            },
132        )?;
133        let resp = recv_message(&mut self.stream)?;
134        match resp {
135            ProtocolMessage::Registered { signer_id: sid } if sid == signer_id => Ok(()),
136            _ => Err(io::Error::new(
137                io::ErrorKind::InvalidData,
138                "unexpected register response",
139            )),
140        }
141    }
142
143    /// Create a new signing session on the coordinator.
144    pub fn create_session(
145        &mut self,
146        quorum_id: &str,
147        scheme: &str,
148        message: &[u8],
149        threshold: u32,
150        num_parties: u32,
151    ) -> io::Result<String> {
152        send_message(
153            &mut self.stream,
154            &ProtocolMessage::CreateSession {
155                quorum_id: quorum_id.into(),
156                scheme: scheme.into(),
157                message: message.to_vec(),
158                threshold,
159                num_parties,
160            },
161        )?;
162        let resp = recv_message(&mut self.stream)?;
163        match resp {
164            ProtocolMessage::SessionCreated { session_id } => Ok(session_id),
165            ProtocolMessage::Error { message } => {
166                Err(io::Error::other(format!("coordinator error: {message}")))
167            }
168            _ => Err(io::Error::new(
169                io::ErrorKind::InvalidData,
170                "unexpected response",
171            )),
172        }
173    }
174
175    /// Submit a commitment for a session.
176    pub fn submit_commitment(
177        &mut self,
178        session_id: &str,
179        signer_id: &str,
180        commitment_bytes: &[u8],
181    ) -> io::Result<()> {
182        send_message(
183            &mut self.stream,
184            &ProtocolMessage::Commitment {
185                session_id: session_id.into(),
186                signer_id: signer_id.into(),
187                bytes: commitment_bytes.to_vec(),
188                signature: vec![0u8; 64],
189            },
190        )?;
191        // Wait for Ack or Error, under a polled deadline — NOT
192        // SO_RCVTIMEO: under MRI Ruby on windows-gnu the first recv on
193        // a timeout'd socket fails WSAENOTSOCK (audit ledger).
194        let mut bounded = Bounded::new(&mut self.stream, std::time::Duration::from_secs(5));
195        match recv_message(&mut bounded) {
196            Ok(ProtocolMessage::Ack { .. }) => Ok(()),
197            Ok(ProtocolMessage::Error { message }) => {
198                Err(io::Error::other(format!("coordinator error: {message}")))
199            }
200            Ok(_) => Ok(()), // tolerate unexpected but non-error responses
201            Err(e) => Err(e),
202        }
203    }
204
205    /// Submit a share for a session. Returns the aggregated signature if
206    /// this was the T-th share (threshold met).
207    pub fn submit_share(
208        &mut self,
209        session_id: &str,
210        signer_id: &str,
211        share_bytes: &[u8],
212    ) -> io::Result<Option<Vec<u8>>> {
213        send_message(
214            &mut self.stream,
215            &ProtocolMessage::Share {
216                session_id: session_id.into(),
217                signer_id: signer_id.into(),
218                bytes: share_bytes.to_vec(),
219                signature: vec![0u8; 64],
220            },
221        )?;
222
223        // Short read deadline — if the coordinator doesn't respond
224        // (threshold not met), the client errors instead of blocking
225        // forever. Polling, not SO_RCVTIMEO (see above).
226        let mut bounded = Bounded::new(&mut self.stream, std::time::Duration::from_secs(5));
227
228        match recv_message(&mut bounded) {
229            Ok(ProtocolMessage::Signature { bytes, .. }) => Ok(Some(bytes)),
230            Ok(ProtocolMessage::Ack { .. }) => Ok(None),
231            Ok(ProtocolMessage::Error { message }) => {
232                Err(io::Error::other(format!("coordinator error: {message}")))
233            }
234            Ok(_) => Ok(None),
235            Err(ref e)
236                if e.kind() == io::ErrorKind::WouldBlock || e.kind() == io::ErrorKind::TimedOut =>
237            {
238                Ok(None)
239            }
240            Err(e) => Err(e),
241        }
242    }
243
244    /// Query session status.
245    pub fn get_status(&mut self, session_id: &str) -> io::Result<String> {
246        send_message(
247            &mut self.stream,
248            &ProtocolMessage::GetStatus {
249                session_id: Some(session_id.into()),
250            },
251        )?;
252        let resp = recv_message(&mut self.stream)?;
253        match resp {
254            ProtocolMessage::Status { state, .. } => Ok(state),
255            _ => Err(io::Error::new(
256                io::ErrorKind::InvalidData,
257                "unexpected status response",
258            )),
259        }
260    }
261
262    /// Mutable access to the underlying session stream (TCP directly
263    /// or a registry transport). Used by the signer daemon for
264    /// low-level protocol message handling.
265    pub fn stream_mut(&mut self) -> &mut Box<dyn SessionIo> {
266        &mut self.stream
267    }
268
269    /// Receive the next protocol message on the session stream.
270    pub fn recv(&mut self) -> io::Result<ProtocolMessage> {
271        recv_message(self.stream.as_mut())
272    }
273
274    /// Send a protocol message on the session stream.
275    pub fn send(&mut self, msg: &ProtocolMessage) -> io::Result<()> {
276        send_message(self.stream.as_mut(), msg)
277    }
278}
confium_coordinator/coordinator/client.rs

confium_coordinator/coordinator/
client.rs
