Merge pull request #806 from nymtech/bugfix/network-explorer-api-geoip

Bug fix: Network Explorer: Add freegeoip API key and split out tasks for country distributions

Cargo.lock

@@ -1599,6 +1599,7 @@ dependencies = [
  "schemars",
  "serde",
  "serde_json",
+ "thiserror",
  "tokio",
  "validator-client",
 ]

explorer-api/Cargo.toml

@@ -20,6 +20,7 @@ okapi = { version = "0.6.0-alpha-1", features = ["derive_json_schema"] }
 rocket_okapi = "0.7.0-alpha-1"
 log = "0.4.0"
 pretty_env_logger = "0.4.0"
+thiserror = "1.0.29"
 
 mixnet-contract = { path = "../common/mixnet-contract" }
 network-defaults = { path = "../common/network-defaults" }

explorer-api/README.md

@@ -1,17 +1,24 @@
 Network Explorer API
 ====================
 
-An API that can: 
+An API that provides data for the [Network Explorer](../explorer).
 
-* calculate how many nodes are in which country, by checking the IPs of all nodes against an external service
-* serve "hello world" via HTTP
+Features:
 
+  - geolocates mixnodes using https://freegeoip.app/
+  - calculates how many nodes are in each country
+  - proxies mixnode API requests to add HTTPS
+  
+## Running
 
-TODO:
+Supply the environment variable `GEO_IP_SERVICE_API_KEY` with a key from https://freegeoip.app/.
+
+Run as a service and reverse proxy with `nginx` to add `https` with Lets Encrypt.
+
+# TODO / Known Issues
+
+## TODO
 
 * record the number of mixnodes on a given date and write to a file for later retrieval
-* store the nodes per country state in a variable
-* grab mixnode description info via reqwest and serve it (avoid mixed-content errors)
-* serve it all over http
 * dependency injection
-* tests
+* tests

explorer-api/src/country_statistics/distribution.rs

@@ -0,0 +1,63 @@
+use log::info;
+
+use crate::country_statistics::country_nodes_distribution::CountryNodesDistribution;
+
+use crate::state::ExplorerApiStateContext;
+
+pub(crate) struct CountryStatisticsDistributionTask {
+    state: ExplorerApiStateContext,
+}
+
+impl CountryStatisticsDistributionTask {
+    pub(crate) fn new(state: ExplorerApiStateContext) -> Self {
+        CountryStatisticsDistributionTask { state }
+    }
+
+    pub(crate) fn start(mut self) {
+        info!("Spawning mix node country distribution task runner...");
+        tokio::spawn(async move {
+            let mut interval_timer = tokio::time::interval(std::time::Duration::from_secs(60 * 15)); // every 15 mins
+            loop {
+                // wait for the next interval tick
+                interval_timer.tick().await;
+                self.calculate_nodes_per_country().await;
+            }
+        });
+    }
+
+    /// Retrieves the current list of mixnodes from the validators and calculates how many nodes are in each country
+    async fn calculate_nodes_per_country(&mut self) {
+        let cache = self.state.inner.mix_nodes.get_location_cache().await;
+
+        let three_letter_iso_country_codes: Vec<String> = cache
+            .values()
+            .flat_map(|i| {
+                i.location
+                    .as_ref()
+                    .map(|j| j.three_letter_iso_country_code.clone())
+            })
+            .collect();
+
+        let mut distribution = CountryNodesDistribution::new();
+
+        info!("Calculating country distribution from located mixnodes...");
+        for three_letter_iso_country_code in three_letter_iso_country_codes {
+            *(distribution.entry(three_letter_iso_country_code)).or_insert(0) += 1;
+        }
+
+        // replace the shared distribution to be the new distribution
+        self.state
+            .inner
+            .country_node_distribution
+            .set_all(distribution)
+            .await;
+
+        info!(
+            "Mixnode country distribution done: {:?}",
+            self.state.inner.country_node_distribution.get_all().await
+        );
+
+        // keep state on disk, so that when this process dies it can start up again and users get some data
+        self.state.write_to_file().await;
+    }
+}

explorer-api/src/country_statistics/geolocate.rs

@@ -0,0 +1,139 @@
+use log::{info, warn};
+use reqwest::Error as ReqwestError;
+use thiserror::Error;
+
+use crate::mix_nodes::{GeoLocation, Location};
+use crate::state::ExplorerApiStateContext;
+
+pub(crate) struct GeoLocateTask {
+    state: ExplorerApiStateContext,
+}
+
+impl GeoLocateTask {
+    pub(crate) fn new(state: ExplorerApiStateContext) -> Self {
+        GeoLocateTask { state }
+    }
+
+    pub(crate) fn start(mut self) {
+        info!("Spawning mix node locator task runner...");
+        tokio::spawn(async move {
+            let mut interval_timer = tokio::time::interval(std::time::Duration::from_millis(50));
+            loop {
+                // wait for the next interval tick
+                interval_timer.tick().await;
+                self.locate_mix_nodes().await;
+            }
+        });
+    }
+
+    async fn locate_mix_nodes(&mut self) {
+        let mixnode_bonds = self.state.inner.mix_nodes.get().await.value;
+
+        for (i, cache_item) in mixnode_bonds.values().enumerate() {
+            if self
+                .state
+                .inner
+                .mix_nodes
+                .is_location_valid(&cache_item.mix_node.identity_key)
+                .await
+            {
+                // when the cached location is valid, don't locate and continue to next mix node
+                continue;
+            }
+
+            // the mix node has not been located or is the cache time has expired
+            match locate(&cache_item.mix_node.host).await {
+                Ok(geo_location) => {
+                    let location = Location::new(geo_location);
+
+                    trace!(
+                        "{} mix nodes already located. Ip {} is located in {:#?}",
+                        i,
+                        cache_item.mix_node.host,
+                        location.three_letter_iso_country_code,
+                    );
+
+                    if i > 0 && (i % 100) == 0 {
+                        info!(
+                            "Located {} mixnodes...",
+                            i + 1,
+                        );
+                    }
+
+                    self.state
+                        .inner
+                        .mix_nodes
+                        .set_location(&cache_item.mix_node.identity_key, Some(location))
+                        .await;
+
+                    // one node has been located, so return out of the loop
+                    return;
+                }
+                Err(e) => match e {
+                    LocateError::ReqwestError(e) => warn!(
+                        "❌ Oh no! Location for {} failed {}",
+                        cache_item.mix_node.host, e
+                    ),
+                    LocateError::NotFound(e) => {
+                            warn!(
+                            "❌ Location for {} not found. Response body: {}",
+                            cache_item.mix_node.host, e
+                        );
+                        self.state
+                            .inner
+                            .mix_nodes
+                            .set_location(&cache_item.mix_node.identity_key, None)
+                            .await;
+                    },
+                    LocateError::RateLimited(e) => warn!(
+                        "❌ Oh no, we've been rate limited! Location for {} failed. Response body: {}",
+                        cache_item.mix_node.host, e
+                    ),
+                },
+            }
+        }
+
+        trace!("All mix nodes located");
+    }
+}
+
+#[derive(Debug, Error)]
+enum LocateError {
+    #[error("Oops, we have made too many requests and are being rate limited. Request body: {0}")]
+    RateLimited(String),
+
+    #[error("Geolocation not found. Request body: {0}")]
+    NotFound(String),
+
+    #[error(transparent)]
+    ReqwestError(#[from] ReqwestError),
+}
+
+async fn locate(ip: &str) -> Result<GeoLocation, LocateError> {
+    let api_key = ::std::env::var("GEO_IP_SERVICE_API_KEY")
+        .expect("Env var GEO_IP_SERVICE_API_KEY is not set");
+    let uri = format!("{}/{}?apikey={}", crate::GEO_IP_SERVICE, ip, api_key);
+    match reqwest::get(uri.clone()).await {
+        Ok(response) => {
+            if response.status() == 429 {
+                return Err(LocateError::RateLimited(
+                    response
+                        .text()
+                        .await
+                        .unwrap_or_else(|_| "(the response body is empty)".to_string()),
+                ));
+            }
+            if response.status() == 404 {
+                return Err(LocateError::NotFound(
+                    response
+                        .text()
+                        .await
+                        .unwrap_or_else(|_| "(the response body is empty)".to_string()),
+                ));
+            }
+            let location = response.json::<GeoLocation>().await?;
+            Ok(location)
+        }
+        Err(e) => Err(LocateError::ReqwestError(e)),
+    }
+}

explorer-api/src/country_statistics/mod.rs

@@ -1,96 +1,4 @@
-use log::{info, trace, warn};
-use reqwest::Error as ReqwestError;
-
-use crate::country_statistics::country_nodes_distribution::CountryNodesDistribution;
-use crate::mix_nodes::{GeoLocation, Location};
-use crate::state::ExplorerApiStateContext;
-
 pub mod country_nodes_distribution;
+pub mod distribution;
+pub mod geolocate;
 pub mod http;
-
-pub(crate) struct CountryStatistics {
-    state: ExplorerApiStateContext,
-}
-
-impl CountryStatistics {
-    pub(crate) fn new(state: ExplorerApiStateContext) -> Self {
-        CountryStatistics { state }
-    }
-
-    pub(crate) fn start(mut self) {
-        info!("Spawning task runner...");
-        tokio::spawn(async move {
-            let mut interval_timer = tokio::time::interval(std::time::Duration::from_secs(60 * 60));
-            loop {
-                // wait for the next interval tick
-                interval_timer.tick().await;
-
-                info!("Running task...");
-                self.calculate_nodes_per_country().await;
-                info!("Done");
-            }
-        });
-    }
-
-    /// Retrieves the current list of mixnodes from the validators and calculates how many nodes are in each country
-    async fn calculate_nodes_per_country(&mut self) {
-        // force the mixnode cache to invalidate
-        let mixnode_bonds = self.state.inner.mix_nodes.refresh_and_get().await.value;
-
-        let mut distribution = CountryNodesDistribution::new();
-
-        info!("Locating mixnodes...");
-        for (i, cache_item) in mixnode_bonds.values().enumerate() {
-            match locate(&cache_item.bond.mix_node.host).await {
-                Ok(geo_location) => {
-                    let location = Location::new(geo_location);
-
-                    *(distribution.entry(location.three_letter_iso_country_code.to_string()))
-                        .or_insert(0) += 1;
-
-                    trace!(
-                        "Ip {} is located in {:#?}",
-                        cache_item.bond.mix_node.host,
-                        location.three_letter_iso_country_code,
-                    );
-
-                    self.state
-                        .inner
-                        .mix_nodes
-                        .set_location(&cache_item.bond.mix_node.identity_key, location)
-                        .await;
-
-                    if (i % 100) == 0 {
-                        info!(
-                            "Located {} mixnodes in {} countries",
-                            i + 1,
-                            distribution.len()
-                        );
-                    }
-                }
-                Err(e) => warn!("❌ Oh no! Location failed {}", e),
-            }
-        }
-
-        // replace the shared distribution to be the new distribution
-        self.state
-            .inner
-            .country_node_distribution
-            .set_all(distribution)
-            .await;
-
-        info!(
-            "Locating mixnodes done: {:?}",
-            self.state.inner.country_node_distribution.get_all().await
-        );
-
-        // keep state on disk, so that when this process dies it can start up again and users get some data
-        self.state.write_to_file().await;
-    }
-}
-
-async fn locate(ip: &str) -> Result<GeoLocation, ReqwestError> {
-    let response = reqwest::get(format!("{}{}", crate::GEO_IP_SERVICE, ip)).await?;
-    let location = response.json::<GeoLocation>().await?;
-    Ok(location)
-}

explorer-api/src/main.rs

@@ -12,7 +12,7 @@ mod mix_nodes;
 mod ping;
 mod state;
 
-const GEO_IP_SERVICE: &str = "https://freegeoip.app/json/";
+const GEO_IP_SERVICE: &str = "https://api.freegeoip.app/json";
 
 #[tokio::main]
 async fn main() {
@@ -36,7 +36,12 @@ impl ExplorerApi {
         info!("Explorer API starting up...");
 
         // spawn concurrent tasks
-        country_statistics::CountryStatistics::new(self.state.clone()).start();
+        mix_nodes::tasks::MixNodesTasks::new(self.state.clone()).start();
+        country_statistics::distribution::CountryStatisticsDistributionTask::new(
+            self.state.clone(),
+        )
+        .start();
+        country_statistics::geolocate::GeoLocateTask::new(self.state.clone()).start();
         http::start(self.state.clone());
 
         // wait for user to press ctrl+C

explorer-api/src/mix_node/http.rs

@@ -34,27 +34,7 @@ pub(crate) struct PrettyMixNodeBondWithLocation {
 pub(crate) async fn list(
     state: &State<ExplorerApiStateContext>,
 ) -> Json<Vec<PrettyMixNodeBondWithLocation>> {
-    Json(
-        state
-            .inner
-            .mix_nodes
-            .get()
-            .await
-            .value
-            .values()
-            .map(|i| {
-                let mix_node = i.bond.clone();
-                PrettyMixNodeBondWithLocation {
-                    location: i.location.clone(),
-                    bond_amount: mix_node.bond_amount,
-                    total_delegation: mix_node.total_delegation,
-                    owner: mix_node.owner,
-                    layer: mix_node.layer,
-                    mix_node: mix_node.mix_node,
-                }
-            })
-            .collect::<Vec<PrettyMixNodeBondWithLocation>>(),
-    )
+    Json(state.inner.mix_nodes.get_mixnodes_with_location().await)
 }
 
 #[openapi(tag = "mix_node")]

explorer-api/src/mix_nodes/mod.rs

@@ -1,3 +1,4 @@
+pub(crate) mod tasks;
 mod utils;
 
 use std::collections::HashMap;
@@ -7,6 +8,7 @@ use std::time::{Duration, SystemTime};
 use rocket::tokio::sync::RwLock;
 use serde::{Deserialize, Serialize};
 
+use crate::mix_node::http::PrettyMixNodeBondWithLocation;
 use crate::mix_nodes::utils::map_2_letter_to_3_letter_country_code;
 use mixnet_contract::{Delegation, MixNodeBond, RawDelegationData, UnpackedDelegation};
 use network_defaults::{
@@ -14,8 +16,9 @@ use network_defaults::{
 };
 use validator_client::nymd::QueryNymdClient;
 
-pub(crate) type LocationCache = HashMap<String, Location>;
+pub(crate) type LocationCache = HashMap<String, LocationCacheItem>;
 
+#[allow(dead_code)]
 #[derive(Debug, Deserialize)]
 pub(crate) struct GeoLocation {
     pub(crate) ip: String,
@@ -31,6 +34,12 @@ pub(crate) struct GeoLocation {
     pub(crate) metro_code: u32,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub(crate) struct LocationCacheItem {
+    pub(crate) location: Option<Location>,
+    pub(crate) valid_until: SystemTime,
+}
+
 #[derive(Clone, Debug, JsonSchema, Serialize, Deserialize)]
 pub(crate) struct Location {
     pub(crate) two_letter_iso_country_code: String,
@@ -53,16 +62,19 @@ impl Location {
     }
 }
 
-#[derive(Clone, Debug)]
-pub(crate) struct MixNodeBondWithLocation {
-    pub(crate) location: Option<Location>,
-    pub(crate) bond: MixNodeBond,
+impl LocationCacheItem {
+    pub(crate) fn new_from_location(location: Option<Location>) -> Self {
+        LocationCacheItem {
+            location,
+            valid_until: SystemTime::now() + Duration::from_secs(60 * 60 * 24), // valid for 1 day
+        }
+    }
 }
 
 #[derive(Clone, Debug)]
 pub(crate) struct MixNodesResult {
     pub(crate) valid_until: SystemTime,
-    pub(crate) value: HashMap<String, MixNodeBondWithLocation>,
+    pub(crate) value: HashMap<String, MixNodeBond>,
     location_cache: LocationCache,
 }
 
@@ -92,23 +104,28 @@ impl ThreadsafeMixNodesResult {
         }
     }
 
+    pub(crate) async fn is_location_valid(&self, identity_key: &str) -> bool {
+        self.inner
+            .read()
+            .await
+            .location_cache
+            .get(identity_key)
+            .map(|cache_item| cache_item.valid_until > SystemTime::now())
+            .unwrap_or(false)
+    }
+
     pub(crate) async fn get_location_cache(&self) -> LocationCache {
         self.inner.read().await.location_cache.clone()
     }
 
-    pub(crate) async fn set_location(&self, identity_key: &str, location: Location) {
+    pub(crate) async fn set_location(&self, identity_key: &str, location: Option<Location>) {
         let mut guard = self.inner.write().await;
 
         // cache the location for this mix node so that it can be used when the mix node list is refreshed
-        guard
-            .location_cache
-            .insert(identity_key.to_string(), location.clone());
-
-        // add the location to the mix node
-        guard
-            .value
-            .entry(identity_key.to_string())
-            .and_modify(|item| item.location = Some(location));
+        guard.location_cache.insert(
+            identity_key.to_string(),
+            LocationCacheItem::new_from_location(location),
+        );
     }
 
     pub(crate) async fn get(&self) -> MixNodesResult {
@@ -124,25 +141,34 @@ impl ThreadsafeMixNodesResult {
         self.inner.read().await.clone()
     }
 
-    pub(crate) async fn refresh_and_get(&self) -> MixNodesResult {
-        self.refresh().await;
-        self.inner.read().await.clone()
+    pub(crate) async fn get_mixnodes_with_location(&self) -> Vec<PrettyMixNodeBondWithLocation> {
+        let guard = self.inner.read().await;
+        guard
+            .value
+            .values()
+            .map(|bond| {
+                let location = guard.location_cache.get(&bond.mix_node.identity_key);
+                let copy = bond.clone();
+                PrettyMixNodeBondWithLocation {
+                    location: location.and_then(|l| l.location.clone()),
+                    bond_amount: copy.bond_amount,
+                    total_delegation: copy.total_delegation,
+                    owner: copy.owner,
+                    layer: copy.layer,
+                    mix_node: copy.mix_node,
+                }
+            })
+            .collect()
     }
 
-    async fn refresh(&self) {
+    pub(crate) async fn refresh(&self) {
         // get mixnodes and cache the new value
         let value = retrieve_mixnodes().await;
         let location_cache = self.inner.read().await.location_cache.clone();
         *self.inner.write().await = MixNodesResult {
             value: value
                 .into_iter()
-                .map(|bond| {
-                    let location = location_cache.get(&bond.mix_node.identity_key).cloned(); // add the location, if we've located this mix node before
-                    (
-                        bond.mix_node.identity_key.to_string(),
-                        MixNodeBondWithLocation { location, bond },
-                    )
-                })
+                .map(|bond| (bond.mix_node.identity_key.to_string(), bond))
                 .collect(),
             valid_until: SystemTime::now() + Duration::from_secs(60 * 10), // valid for 10 minutes
             location_cache,

explorer-api/src/mix_nodes/tasks.rs

@@ -0,0 +1,26 @@
+use crate::state::ExplorerApiStateContext;
+
+pub(crate) struct MixNodesTasks {
+    state: ExplorerApiStateContext,
+}
+
+impl MixNodesTasks {
+    pub(crate) fn new(state: ExplorerApiStateContext) -> Self {
+        MixNodesTasks { state }
+    }
+
+    pub(crate) fn start(self) {
+        info!("Spawning mix nodes task runner...");
+        tokio::spawn(async move {
+            let mut interval_timer = tokio::time::interval(std::time::Duration::from_secs(60 * 60)); // every hour
+            loop {
+                // wait for the next interval tick
+                interval_timer.tick().await;
+
+                info!("Updating mix node cache...");
+                self.state.inner.mix_nodes.refresh().await;
+                info!("Done");
+            }
+        });
+    }
+}

explorer-api/src/state.rs

@@ -27,12 +27,7 @@ pub struct ExplorerApiState {
 
 impl ExplorerApiState {
     pub(crate) async fn get_mix_node(&self, pubkey: &str) -> Option<MixNodeBond> {
-        self.mix_nodes
-            .get()
-            .await
-            .value
-            .get(pubkey)
-            .map(|cache_item| cache_item.bond.clone())
+        self.mix_nodes.get().await.value.get(pubkey).cloned()
     }
 }
 
@@ -46,14 +41,12 @@ pub struct ExplorerApiStateOnDisk {
 #[derive(Clone)]
 pub(crate) struct ExplorerApiStateContext {
     pub(crate) inner: ExplorerApiState,
-    state_file: String,
 }
 
 impl ExplorerApiStateContext {
     pub(crate) fn new() -> Self {
         ExplorerApiStateContext {
             inner: ExplorerApiStateContext::read_from_file(),
-            state_file: std::env::var("API_STATE_FILE").unwrap_or_else(|_| STATE_FILE.to_string()),
         }
     }