updating stored allow list on file changes

Cargo.lock

@@ -3658,6 +3658,7 @@ dependencies = [
 name = "nym-network-requester"
 version = "1.1.13"
 dependencies = [
+ "async-file-watcher",
  "async-trait",
  "clap 4.1.11",
  "client-core",

common/async-file-watcher/src/lib.rs

@@ -27,7 +27,7 @@ pub struct AsyncFileWatcher {
 }
 
 impl AsyncFileWatcher {
-    pub async fn new_file_changes_watcher<P: AsRef<Path>>(
+    pub fn new_file_changes_watcher<P: AsRef<Path>>(
         path: P,
         event_sender: FileWatcherEventSender,
     ) -> notify::Result<Self> {

service-providers/network-requester/Cargo.toml

@@ -32,6 +32,7 @@ tokio-tungstenite = "0.17.2"
 url = { workspace = true }
 
 # internal
+async-file-watcher = { path = "../../common/async-file-watcher" }
 client-core = { path = "../../common/client-core" }
 nym-config = { path = "../../common/config" }
 nym-crypto = { path = "../../common/crypto" }

service-providers/network-requester/src/allowed_hosts/filter.rs

@@ -4,6 +4,7 @@
 use super::HostsStore;
 use crate::allowed_hosts::group::HostsGroup;
 use crate::allowed_hosts::standard_list::StandardList;
+use crate::allowed_hosts::stored_allowed_hosts::StoredAllowedHosts;
 use std::net::{IpAddr, SocketAddr};
 
 enum RequestHost {
@@ -26,7 +27,7 @@ enum RequestHost {
 /// domains as allowed. That list is loaded once at startup from Mozilla's canonical
 /// publicsuffix list.
 pub(crate) struct OutboundRequestFilter {
-    pub(super) allowed_hosts: HostsStore,
+    pub(super) allowed_hosts: StoredAllowedHosts,
     pub(super) standard_list: StandardList,
     root_domain_list: publicsuffix::List,
     unknown_hosts: HostsStore,
@@ -41,7 +42,7 @@ impl OutboundRequestFilter {
     /// Automatcially fetches the latest standard allowed list from the Nym website, so that all
     /// requesters are able to support the same minimal functionality out of the box.
     pub(crate) fn new(
-        allowed_hosts: HostsStore,
+        allowed_hosts: StoredAllowedHosts,
         standard_list: StandardList,
         unknown_hosts: HostsStore,
     ) -> OutboundRequestFilter {
@@ -58,8 +59,9 @@ impl OutboundRequestFilter {
         }
     }
 
-    fn check_allowed_hosts(&self, host: &RequestHost) -> bool {
-        self.check_group(&self.allowed_hosts.data, host)
+    async fn check_allowed_hosts(&self, host: &RequestHost) -> bool {
+        let guard = self.allowed_hosts.get().await;
+        self.check_group(&guard.data, host)
     }
 
     async fn check_standard_list(&self, host: &RequestHost) -> bool {
@@ -103,7 +105,7 @@ impl OutboundRequestFilter {
 
     async fn check_request_host(&mut self, request_host: &RequestHost) -> bool {
         // first check our own allow list
-        let local_allowed = self.check_allowed_hosts(&request_host);
+        let local_allowed = self.check_allowed_hosts(&request_host).await;
 
         // if it's locally allowed, no point in checking the standard list
         if local_allowed {

service-providers/network-requester/src/allowed_hosts/hosts.rs

@@ -31,6 +31,12 @@ impl HostsStore {
         }
     }
 
+    pub(crate) fn try_reload(&mut self) -> io::Result<()> {
+        let hosts = Self::load_from_storefile(&self.storefile)?;
+        self.data = HostsGroup::new(hosts);
+        Ok(())
+    }
+
     pub(crate) fn contains_domain(&self, host: &str) -> bool {
         self.data.contains_domain(host)
     }

service-providers/network-requester/src/allowed_hosts/mod.rs

@@ -1,8 +1,12 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
 mod filter;
 mod group;
 mod host;
 mod hosts;
 pub(crate) mod standard_list;
+pub(crate) mod stored_allowed_hosts;
 
 pub(crate) use filter::OutboundRequestFilter;
 pub(crate) use hosts::HostsStore;

service-providers/network-requester/src/allowed_hosts/stored_allowed_hosts.rs

@@ -0,0 +1,120 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::allowed_hosts::HostsStore;
+use async_file_watcher::{AsyncFileWatcher, FileWatcherEventReceiver};
+use futures::channel::mpsc;
+use futures::StreamExt;
+use nym_task::TaskClient;
+use std::io;
+use std::path::Path;
+use std::sync::Arc;
+use tokio::sync::{RwLock, RwLockReadGuard};
+
+#[derive(Debug, Clone)]
+pub(crate) struct StoredAllowedHosts {
+    inner: Arc<RwLock<HostsStore>>,
+}
+
+impl StoredAllowedHosts {
+    pub(crate) fn new<P: AsRef<Path>>(path: P) -> Self {
+        let allowed_hosts =
+            HostsStore::new(HostsStore::default_base_dir(), path.as_ref().to_path_buf());
+
+        StoredAllowedHosts {
+            inner: Arc::new(RwLock::new(allowed_hosts)),
+        }
+    }
+
+    pub(crate) async fn reload(&self) -> io::Result<()> {
+        log::debug!("reloading stored allowed hosts");
+        self.inner.write().await.try_reload()
+    }
+
+    pub(crate) async fn get(&self) -> RwLockReadGuard<'_, HostsStore> {
+        self.inner.read().await
+    }
+}
+
+pub(crate) struct StoredAllowedHostsReloader {
+    stored_hosts: StoredAllowedHosts,
+    events_receiver: FileWatcherEventReceiver,
+
+    // Listens to shutdown commands from higher up
+    shutdown_listener: TaskClient,
+}
+
+impl StoredAllowedHostsReloader {
+    pub(crate) fn new(
+        stored_hosts: StoredAllowedHosts,
+        events_receiver: FileWatcherEventReceiver,
+        shutdown_listener: TaskClient,
+    ) -> Self {
+        StoredAllowedHostsReloader {
+            events_receiver,
+            stored_hosts,
+            shutdown_listener,
+        }
+    }
+
+    pub(crate) async fn run(&mut self) {
+        while !self.shutdown_listener.is_shutdown() {
+            tokio::select! {
+                biased;
+                _ = self.shutdown_listener.recv() => {
+                    log::trace!("StoredAllowedHostsReloader: Received shutdown");
+                }
+                event = self.events_receiver.next() => {
+                    let Some(event) = event else {
+                        log::trace!("StoredAllowedHostsReloader: sender channel has terminated");
+                        break
+                    };
+                    log::debug!("the file has changed - {event:?}");
+                    log::debug!("reloading stored hosts");
+                    if let Err(err) = self.stored_hosts.reload().await {
+                        log::error!("failed to reload stored hosts: {err}")
+                    }
+                }
+            }
+        }
+
+        log::debug!("StoredAllowedHostsReloader: Exiting");
+    }
+
+    pub(crate) fn start(mut self) {
+        tokio::spawn(async move { self.run().await });
+    }
+}
+
+async fn run_watcher(mut watcher: AsyncFileWatcher, mut shutdown: TaskClient) {
+    tokio::select! {
+        biased;
+        _ = shutdown.recv() => {
+            log::trace!("AsyncFileWatcher: Received shutdown");
+        }
+        res = watcher.watch() => {
+            log::trace!("AsyncFileWatcher: finished with {res:?}");
+        }
+    }
+    log::debug!("AsyncFileWatcher: Exiting");
+}
+
+fn start_watcher(watcher: AsyncFileWatcher, shutdown: TaskClient) {
+    tokio::spawn(async move { run_watcher(watcher, shutdown).await });
+}
+
+pub(crate) async fn start_allowed_list_reloader(
+    stored_list: StoredAllowedHosts,
+    shutdown_listener: TaskClient,
+) {
+    let (events_sender, events_receiver) = mpsc::unbounded();
+    let file = stored_list.get().await.storefile.clone();
+
+    let watcher = AsyncFileWatcher::new_file_changes_watcher(file, events_sender)
+        .expect("failed to create file watcher");
+    let reloader =
+        StoredAllowedHostsReloader::new(stored_list, events_receiver, shutdown_listener.clone());
+
+    start_watcher(watcher, shutdown_listener);
+    reloader.start()
+}

service-providers/network-requester/src/core.rs

@@ -3,6 +3,7 @@
 
 use crate::allowed_hosts;
 use crate::allowed_hosts::standard_list::StandardListUpdater;
+use crate::allowed_hosts::stored_allowed_hosts::{start_allowed_list_reloader, StoredAllowedHosts};
 use crate::allowed_hosts::{OutboundRequestFilter, StandardList};
 use crate::config::Config;
 use crate::error::NetworkRequesterError;
@@ -52,6 +53,7 @@ pub struct NRServiceProviderBuilder {
     enable_statistics: bool,
     stats_provider_addr: Option<Recipient>,
     standard_list: StandardList,
+    allowed_hosts: StoredAllowedHosts,
 }
 
 struct NRServiceProvider {
@@ -158,10 +160,7 @@ impl NRServiceProviderBuilder {
     ) -> NRServiceProviderBuilder {
         let standard_list = StandardList::new();
 
-        let allowed_hosts = allowed_hosts::HostsStore::new(
-            allowed_hosts::HostsStore::default_base_dir(),
-            PathBuf::from("allowed.list"),
-        );
+        let allowed_hosts = StoredAllowedHosts::new("allowed.list");
 
         let unknown_hosts = allowed_hosts::HostsStore::new(
             allowed_hosts::HostsStore::default_base_dir(),
@@ -169,7 +168,7 @@ impl NRServiceProviderBuilder {
         );
 
         let outbound_request_filter =
-            OutboundRequestFilter::new(allowed_hosts, standard_list.clone(), unknown_hosts);
+            OutboundRequestFilter::new(allowed_hosts.clone(), standard_list.clone(), unknown_hosts);
 
         NRServiceProviderBuilder {
             config,
@@ -178,6 +177,7 @@ impl NRServiceProviderBuilder {
             enable_statistics,
             stats_provider_addr,
             standard_list,
+            allowed_hosts,
         }
     }
 
@@ -240,6 +240,9 @@ impl NRServiceProviderBuilder {
         )
         .start();
 
+        // start the allowed.list watcher and updater
+        start_allowed_list_reloader(self.allowed_hosts, shutdown.subscribe()).await;
+
         let service_provider = NRServiceProvider {
             outbound_request_filter: self.outbound_request_filter,
             open_proxy: self.open_proxy,