iota_proxy/
peers.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{
6    collections::HashMap,
7    sync::{Arc, RwLock},
8    time::Duration,
9};
10
11use anyhow::Result;
12use bcs;
13use fastcrypto::{ed25519::Ed25519PublicKey, traits::ToFromBytes};
14use iota_sdk::{IotaClient, IotaClientBuilder, rpc_types::IotaObjectDataOptions};
15use iota_sdk_types::ObjectId;
16use iota_tls::Allower;
17use iota_types::{
18    dynamic_field::Field,
19    iota_system_state::{
20        iota_system_state_inner_v1::ValidatorV1,
21        iota_system_state_summary::{IotaSystemStateSummary, IotaValidatorSummary},
22    },
23};
24use itertools::Itertools;
25use tracing::{debug, error, info};
26
27/// AllowedPeers is a mapping of public key to AllowedPeer data
28pub type AllowedPeers = Arc<RwLock<HashMap<Ed25519PublicKey, AllowedPeer>>>;
29
30#[derive(Hash, PartialEq, Eq, Debug, Clone)]
31pub struct AllowedPeer {
32    pub name: String,
33    pub public_key: Ed25519PublicKey,
34}
35
36/// IotaNodeProvider queries the iota blockchain and keeps a record of known
37/// validators based on the response from iota_getValidators.  The node name,
38/// public key and other info is extracted from the chain and stored in this
39/// data structure.  We pass this struct to the tls verifier and it depends on
40/// the state contained within. Handlers also use this data in an Extractor
41/// extension to check incoming clients on the http api against known keys.
42#[derive(Debug, Clone)]
43pub struct IotaNodeProvider {
44    active_validator_nodes: AllowedPeers,
45    pending_validator_nodes: AllowedPeers,
46    static_nodes: AllowedPeers,
47    rpc_url: String,
48    rpc_poll_interval: Duration,
49}
50
51impl Allower for IotaNodeProvider {
52    fn allowed(&self, key: &Ed25519PublicKey) -> bool {
53        self.static_nodes.read().unwrap().contains_key(key)
54            || self
55                .active_validator_nodes
56                .read()
57                .unwrap()
58                .contains_key(key)
59            || self
60                .pending_validator_nodes
61                .read()
62                .unwrap()
63                .contains_key(key)
64    }
65}
66
67impl IotaNodeProvider {
68    pub fn new(
69        rpc_url: String,
70        rpc_poll_interval: Duration,
71        static_peers: Vec<AllowedPeer>,
72    ) -> Self {
73        // build our hashmap with the static pub keys. we only do this one time at
74        // binary startup.
75        let static_nodes: HashMap<Ed25519PublicKey, AllowedPeer> = static_peers
76            .into_iter()
77            .map(|v| (v.public_key.clone(), v))
78            .collect();
79        let static_nodes = Arc::new(RwLock::new(static_nodes));
80        let active_validator_nodes = Arc::new(RwLock::new(HashMap::new()));
81        let pending_validator_nodes = Arc::new(RwLock::new(HashMap::new()));
82        Self {
83            active_validator_nodes,
84            pending_validator_nodes,
85            static_nodes,
86            rpc_url,
87            rpc_poll_interval,
88        }
89    }
90
91    /// get is used to retrieve peer info in our handlers
92    pub fn get(&self, key: &Ed25519PublicKey) -> Option<AllowedPeer> {
93        debug!("look for {:?}", key);
94        // check static nodes first
95        if let Some(v) = self.static_nodes.read().unwrap().get(key) {
96            return Some(AllowedPeer {
97                name: v.name.to_owned(),
98                public_key: v.public_key.to_owned(),
99            });
100        }
101        // check active validators
102        if let Some(v) = self.active_validator_nodes.read().unwrap().get(key) {
103            return Some(AllowedPeer {
104                name: v.name.to_owned(),
105                public_key: v.public_key.to_owned(),
106            });
107        }
108        // check pending validators
109        if let Some(v) = self.pending_validator_nodes.read().unwrap().get(key) {
110            return Some(AllowedPeer {
111                name: v.name.to_owned(),
112                public_key: v.public_key.to_owned(),
113            });
114        }
115        None
116    }
117
118    /// Get a mutable reference to the allowed validator map
119    pub fn get_mut(&mut self) -> &mut AllowedPeers {
120        &mut self.active_validator_nodes
121    }
122
123    /// Here we allow all active validators to be added to the allow list.
124    fn update_active_validator_set(&self, summary: &IotaSystemStateSummary) {
125        let active_validator_summaries = summary
126            .iter_active_validators()
127            .cloned()
128            .collect::<Vec<IotaValidatorSummary>>();
129
130        // Here we allow all active validators to be added to the allow list to make it
131        // more flexible.
132        let active_validators = extract_validators_from_summaries(&active_validator_summaries);
133        let mut allow = self.active_validator_nodes.write().unwrap();
134        allow.clear();
135        allow.extend(active_validators);
136        info!(
137            "{} iota validators managed to make it on the allow list",
138            allow.len()
139        );
140    }
141
142    fn update_pending_validator_set(
143        &self,
144        pending_validators: Vec<ValidatorV1>,
145        protocol_version: Option<u64>,
146    ) {
147        let summaries = pending_validators
148            .into_iter()
149            .map(|v| v.into_iota_validator_summary(protocol_version))
150            .collect_vec();
151        let validators = extract_validators_from_summaries(&summaries);
152        let mut allow = self.pending_validator_nodes.write().unwrap();
153        allow.clear();
154        allow.extend(validators);
155        info!(
156            "{} iota pending validators managed to make it on the allow list",
157            allow.len()
158        );
159    }
160
161    async fn get_pending_validators(
162        iota_client: &IotaClient,
163        pending_active_validators_id: ObjectId,
164    ) -> Result<Vec<ValidatorV1>> {
165        let pending_validators_ids = iota_client
166            .read_api()
167            .get_dynamic_fields(pending_active_validators_id, None, None)
168            .await?
169            .data
170            .into_iter()
171            .map(|dyi| dyi.object_id)
172            .collect::<Vec<_>>();
173
174        let responses = iota_client
175            .read_api()
176            .multi_get_object_with_options(
177                pending_validators_ids,
178                IotaObjectDataOptions::default().with_bcs(),
179            )
180            .await?;
181
182        responses
183            .into_iter()
184            .map(|resp| {
185                let object_id = resp.object_id()?;
186                let bcs = resp.move_object_bcs().ok_or_else(|| {
187                    anyhow::anyhow!(
188                        "Object {object_id} does not exist or does not return bcs bytes",
189                    )
190                })?;
191                let field = bcs::from_bytes::<Field<u64, ValidatorV1>>(bcs).map_err(|e| {
192                anyhow::anyhow!(
193                    "Can't convert bcs bytes of object {object_id} to Field<u64, ValidatorV1>: {e}",
194                )
195            })?;
196
197                Ok(field.value)
198            })
199            .collect()
200    }
201
202    /// poll_peer_list will act as a refresh interval for our cache
203    pub fn poll_peer_list(&self) {
204        info!("Started polling for peers using rpc: {}", self.rpc_url);
205
206        let rpc_poll_interval = self.rpc_poll_interval;
207        let cloned_self = self.clone();
208        tokio::spawn(async move {
209            let mut interval = tokio::time::interval(rpc_poll_interval);
210            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
211
212            loop {
213                interval.tick().await;
214
215                match IotaClientBuilder::default()
216                    .build(&cloned_self.rpc_url)
217                    .await
218                {
219                    Ok(client) => {
220                        match client.governance_api().get_latest_iota_system_state().await {
221                            Ok(system_state) => {
222                                cloned_self.update_active_validator_set(&system_state);
223                                info!("Successfully updated active validators");
224
225                                let pending_active_validators_id = match &system_state {
226                                    IotaSystemStateSummary::V1(system_state) => {
227                                        system_state.pending_active_validators_id
228                                    }
229                                    IotaSystemStateSummary::V2(system_state) => {
230                                        system_state.pending_active_validators_id
231                                    }
232                                    _ => unimplemented!(
233                                        "a new IotaSystemStateSummary enum variant was added and needs to be handled"
234                                    ),
235                                };
236
237                                match Self::get_pending_validators(
238                                    &client,
239                                    pending_active_validators_id,
240                                )
241                                .await
242                                {
243                                    Ok(pending_validators) => {
244                                        cloned_self.update_pending_validator_set(
245                                            pending_validators,
246                                            Some(system_state.protocol_version().as_u64()),
247                                        );
248                                        info!("Successfully updated pending validators");
249                                    }
250                                    Err(e) => {
251                                        error!("Failed to get pending validators: {:?}", e);
252                                    }
253                                }
254                            }
255                            Err(e) => {
256                                error!("Failed to get latest iota system state: {:?}", e);
257                            }
258                        }
259                    }
260                    Err(e) => {
261                        error!("Failed to create IotaClient: {:?}", e);
262                    }
263                }
264            }
265        });
266    }
267}
268
269/// extract_validators_from_summaries will get the network pubkey bytes from a
270/// IotaValidatorSummary type. This type comes from a full node rpc result. The
271/// key here, if extracted successfully, will ultimately be stored in the allow
272/// list and let us communicate with those actual peers via tls.
273fn extract_validators_from_summaries(
274    validator_summaries: &[IotaValidatorSummary],
275) -> impl Iterator<Item = (Ed25519PublicKey, AllowedPeer)> + use<'_> {
276    validator_summaries.iter().filter_map(|vm| {
277        match Ed25519PublicKey::from_bytes(&vm.network_pubkey_bytes) {
278            Ok(public_key) => {
279                debug!(
280                    "adding public key {:?} for iota validator {:?}",
281                    public_key, vm.name
282                );
283                Some((
284                    public_key.clone(),
285                    AllowedPeer {
286                        name: vm.name.to_owned(),
287                        public_key,
288                    },
289                )) // scoped to filter_map
290            }
291            Err(error) => {
292                error!(
293                    "unable to decode public key for name: {:?} iota_address: {:?} error: {error}",
294                    vm.name, vm.iota_address
295                );
296                None // scoped to filter_map
297            }
298        }
299    })
300}
301
302#[cfg(test)]
303mod tests {
304    use iota_types::iota_system_state::iota_system_state_summary::IotaValidatorSummary;
305    use multiaddr::Multiaddr;
306
307    use super::*;
308    use crate::admin::{CertKeyPair, generate_self_cert};
309    #[test]
310    fn extract_validators_from_summary() {
311        let CertKeyPair(_, client_pub_key) = generate_self_cert("iota".into());
312        let p2p_address: Multiaddr = "/ip4/127.0.0.1/tcp/10000"
313            .parse()
314            .expect("expected a multiaddr value");
315        let summaries = vec![IotaValidatorSummary {
316            network_pubkey_bytes: Vec::from(client_pub_key.as_bytes()),
317            p2p_address: format!("{p2p_address}"),
318            primary_address: "empty".into(),
319            ..Default::default()
320        }];
321        let peers = extract_validators_from_summaries(&summaries);
322        assert_eq!(peers.count(), 1, "peers should have been a length of 1");
323    }
324}
iota_proxy/peers.rs

iota_proxy/
peers.rs
