first commit

.gitignore

@@ -0,0 +1,4 @@
+/target
+test-script.sh
+p2p-failover.config.yaml
+.p2p-trustkey

Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "p2p-failover"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+dateparser = "0.2.1"
+chrono = "0.4.39"
+serde = { version = "1.0", features = ["derive"] }
+serde_yaml = "0.9"
+reqwest = "0.12.12"
+tokio = { version = "=1.40.0", features = ["full"] }
+notify = "8.0.0"

README.md

@@ -0,0 +1,87 @@
+# P2P Failover
+
+A peer-to-peer active-passive failover system written in Rust that manages process execution across multiple nodes based on availability and priority.
+
+## Overview
+
+P2P Failover ensures that a process runs on the highest priority available node, with automatic failover if that node becomes unavailable.
+
+Key features:
+
+- Automatic process management based on node priority
+- Real-time node health monitoring
+- TCP-based peer communication
+- YAML configuration (apologies)
+
+## Installation
+
+1. Clone the repository
+2. Build the project:
+
+```bash
+cargo build --release
+```
+
+## Configuration
+
+Create a `p2p-failover.config.yaml` file in your project directory. Here's an example configuration:
+
+```yaml
+ddns:
+- name: pc
+  ddns: ''
+  ip: 127.0.0.1
+  port: 8080
+  preference: 1
+  priority: 100
+  last_updated: 2025-02-04 19:19:18 UTC
+- name: phone
+  ddns: ''
+  ip: 100.11.111.111
+  port: 8081
+  preference: 1
+  priority: 20
+  last_updated: 2025-01-09 16:45:00 UTC
+config_metadata:
+  name: pc
+  last_updated: 2025-01-11 10:00:00 UTC
+execution:
+  instructions: ./test-program.sh
+  last_updated: 2025-01-11 10:00:00 UTC
+```
+
+### Configuration Fields
+
+- `ddns`: List of nodes in the network
+  - `name`: Unique identifier for the node
+  - `ddns`: Domain name (optional)
+  - `ip`: IP address
+  - `port`: TCP port for node communication
+  - `preference`: Connection preference (0 for DDNS, 1 for IP)
+  - `priority`: Node priority (higher number = higher priority)
+  - `last_updated`: Timestamp of last update
+- `config_metadata`: Node-specific metadata
+  - `name`: Name of this node
+  - `last_updated`: Configuration timestamp
+- `execution`: Process execution settings
+  - `instructions`: Command to execute
+  - `last_updated`: Last modification timestamp
+
+## Environment Variables
+
+- `P2P_CONFIG_PATH`: Path to config file (default: `p2p-failover.config.yaml`)
+- `VERBOSE`: Enable verbose logging (1/true)
+- `DEBUG`: Enable debug logging (1/true)
+
+Note: When `DEBUG` is set to `1`, `VERBOSE` is automatically turned on.
+
+## How It Works
+
+1. Each node monitors the health of other nodes in the network through periodic heartbeats
+2. The node with the highest priority and availability runs the specified process
+3. If a higher priority node becomes available, the process gets killed and started on the other node
+4. If the active node fails, the next highest priority available node takes over
+
+## License
+
+MIT

example.p2p-failover.config.yaml

@@ -0,0 +1,21 @@
+ddns:
+- name: pc
+  ddns: ''
+  ip: 127.0.0.1
+  port: 8080
+  preference: 1
+  priority: 100
+  last_updated: 2025-02-04 19:19:18 UTC
+- name: phone
+  ddns: ''
+  ip: 100.11.111.111
+  port: 8081
+  preference: 1
+  priority: 20
+  last_updated: 2025-01-09 16:45:00 UTC
+config_metadata:
+  name: pc
+  last_updated: 2025-01-11 10:00:00 UTC
+execution:
+  instructions: ./test-script.sh
+  last_updated: 2025-01-11 10:00:00 UTC

src/config.rs

@@ -0,0 +1,46 @@
+use crate::timestamp::Timestamp;
+use serde::{Deserialize, Serialize};
+
+// Config
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct ProviderNode {
+    pub name: String,
+    pub ddns: String,
+    pub ip: String,
+    pub port: u32,
+    pub preference: u8,
+    pub priority: u32,
+    pub last_updated: Timestamp,
+}
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
+pub struct ConfigMetadata {
+    pub name: String,
+    pub last_updated: Timestamp,
+}
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
+pub struct ExecutionInstructions {
+    pub instructions: String,
+    pub last_updated: Timestamp,
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct Config {
+    pub ddns: Vec<ProviderNode>,
+    pub config_metadata: ConfigMetadata,
+    pub execution: ExecutionInstructions,
+}
+
+impl Config {
+    pub fn write(&self) {
+        let config_path = std::env::var("P2P_CONFIG_PATH")
+            .unwrap_or_else(|_| "p2p-failover.config.yaml".to_string());
+
+        let s = serde_yaml::to_string(&self).unwrap();
+        match std::fs::write(config_path, s) {
+            Ok(_) => (),
+            Err(e) => eprintln!("Failed to write config file: {:?}", e),
+        }
+    }
+}

src/debug.rs

@@ -0,0 +1,21 @@
+use std::sync::OnceLock;
+
+static DEBUG_ENABLED: OnceLock<bool> = OnceLock::new();
+
+pub fn is_debug_enabled() -> bool {
+    *DEBUG_ENABLED.get_or_init(|| {
+        std::env::var("DEBUG")
+            .map(|val| val == "1" || val.to_lowercase() == "true")
+            .unwrap_or(false)
+    })
+}
+
+#[macro_export]
+macro_rules! debug {
+    ($($arg:tt)*) => {
+        if $crate::debug::is_debug_enabled() {
+            println!($($arg)*);
+            // println!("DEBUG: {}", format!($($arg)*));
+        }
+    };
+}

src/file_watcher.rs

@@ -0,0 +1,73 @@
+use crate::config::Config;
+use crate::parser::Parser;
+use crate::{debug, log};
+use notify::{RecommendedWatcher, Watcher};
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::{fs::File, path::Path};
+
+pub fn start_file_watcher(config: Arc<Mutex<Config>>, config_string: Arc<Mutex<String>>) {
+    thread::spawn(move || {
+        let config_path = std::env::var("P2P_CONFIG_PATH")
+            .unwrap_or_else(|_| "p2p-failover.config.yaml".to_string());
+
+        let (tx, rx) = std::sync::mpsc::channel();
+
+        let mut watcher: RecommendedWatcher =
+            match notify::Watcher::new(tx, notify::Config::default()) {
+                Ok(w) => w,
+                Err(e) => {
+                    eprintln!("Failed to create watcher: {:?}", e);
+                    return;
+                }
+            };
+
+        if let Err(e) = watcher.watch(
+            Path::new(&config_path.clone()),
+            notify::RecursiveMode::NonRecursive,
+        ) {
+            eprintln!("Failed to watch config file: {:?}", e);
+            return;
+        }
+
+        // Block forever, printing out events as they come in
+        for res in rx {
+            if let Err(e) = res {
+                log!("watch error: {:?}", e);
+                continue;
+            }
+
+            let event = res.unwrap();
+            debug!("event: {:?}", event);
+
+            if let notify::EventKind::Modify(_) = event.kind {
+                // Refresh config
+                let config_file = get_file(&config_path);
+                let mut p = Parser::new(config_file);
+                let cfg = p.parse(Some(config_string.clone()));
+                if let Ok(cfg) = cfg {
+                    let mut config_guard = config.lock().unwrap();
+                    *config_guard = cfg;
+                    log!("Config updated: {:#?}", config_guard);
+                }
+            }
+        }
+    });
+}
+
+fn get_file(filename: &String) -> File {
+    match File::open(filename) {
+        Ok(file) => file,
+        Err(error) => {
+            // if file not created
+            if error.kind() == std::io::ErrorKind::NotFound {
+                match File::create(filename) {
+                    Ok(file) => file,
+                    Err(error) => panic!("Problem creating the file: {:?}", error),
+                }
+            } else {
+                panic!("Problem opening the file: {:?}", error);
+            }
+        }
+    }
+}

src/lib.rs

@@ -0,0 +1,11 @@
+pub mod config;
+pub mod debug;
+pub mod file_watcher;
+pub mod log;
+pub mod node;
+pub mod node_connections;
+pub mod parser;
+pub mod pending_verification;
+pub mod process;
+pub mod tcp_listener;
+pub mod timestamp;

src/log.rs

@@ -0,0 +1,26 @@
+use std::sync::OnceLock;
+
+static VERBOSE_ENABLED: OnceLock<bool> = OnceLock::new();
+
+pub fn is_verbose_enabled() -> bool {
+    *VERBOSE_ENABLED.get_or_init(|| {
+        std::env::var("VERBOSE")
+            .map(|val| val == "1" || val.to_lowercase() == "true")
+            .unwrap_or(
+                // Verbose by default when debugging
+                std::env::var("DEBUG")
+                    .map(|val| val == "1" || val.to_lowercase() == "true")
+                    .unwrap_or(false),
+            )
+    })
+}
+
+#[macro_export]
+macro_rules! log {
+    ($($arg:tt)*) => {
+        if $crate::log::is_verbose_enabled() {
+            println!($($arg)*);
+            // println!("LOG: {}", format!($($arg)*));
+        }
+    };
+}

src/main.rs

@@ -0,0 +1,48 @@
+use p2p_failover::{file_watcher, node::Node, parser::Parser, tcp_listener};
+use std::{
+    fs::File,
+    sync::{Arc, Mutex},
+    thread, time,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let config_path =
+        std::env::var("P2P_CONFIG_PATH").unwrap_or_else(|_| "p2p-failover.config.yaml".to_string());
+
+    let config_file = get_file(&config_path);
+    let mut p = Parser::new(config_file);
+
+    let config_string = Arc::new(Mutex::new(String::new()));
+    let config = {
+        let cfg = p.parse(Some(config_string.clone()))?;
+        Arc::new(Mutex::new(cfg))
+    };
+
+    let mut node = Node::new(config.clone());
+
+    file_watcher::start_file_watcher(config.clone(), config_string.clone());
+    tcp_listener::start_tcp_listener(config.clone(), config_string.clone());
+
+    loop {
+        node.heartbeat().await;
+        thread::sleep(time::Duration::from_secs(1))
+    }
+}
+
+fn get_file(filename: &String) -> File {
+    match File::open(filename) {
+        Ok(file) => file,
+        Err(error) => {
+            // if file not created
+            if error.kind() == std::io::ErrorKind::NotFound {
+                match File::create(filename) {
+                    Ok(file) => file,
+                    Err(error) => panic!("Problem creating the file: {:?}", error),
+                }
+            } else {
+                panic!("Problem opening the file: {:?}", error);
+            }
+        }
+    }
+}

src/node.rs

@@ -0,0 +1,178 @@
+use crate::{config::Config, log, node_connections::NodeConnections, process::Process};
+use std::sync::{Arc, Mutex};
+
+pub struct Node {
+    alive: bool,
+    pub config: Arc<Mutex<Config>>,
+    alives: Vec<bool>,
+    process: Option<Process>,
+    pub node_connections: NodeConnections,
+    // tick: u8,
+    // tick_dir: u8,
+}
+
+impl Node {
+    pub fn new(config: Arc<Mutex<Config>>) -> Node {
+        let alives = vec![false; config.lock().unwrap().ddns.len()];
+
+        Node {
+            alive: false,
+            config,
+            alives,
+            process: None,
+            node_connections: NodeConnections::new(),
+            // tick: 0,
+            // tick_dir: 1,
+        }
+    }
+
+    /// Returns the amount of alive hosts
+    pub async fn check_hosts(&mut self) -> u8 {
+        let mut alives: u8 = 0;
+        let config = self.config.lock().unwrap();
+        for host in config.ddns.iter().enumerate() {
+            if host.1.name == config.config_metadata.name {
+                continue;
+            }
+
+            self.alives[host.0] = false;
+
+            log!(
+                "Checking: {}:{}",
+                if host.1.preference == 0 {
+                    &host.1.ddns
+                } else {
+                    &host.1.ip
+                },
+                &host.1.port,
+            );
+            let alive = self.node_connections.ping(host.1);
+            if alive {
+                log!(
+                    "-> Alive: host \"{}\" with priority {}",
+                    host.1.name,
+                    host.1.priority
+                );
+                alives += 1;
+            } else {
+                log!(
+                    "-> Host \"{}\" with priority {} is dead",
+                    host.1.name,
+                    host.1.priority
+                );
+            }
+            self.alives[host.0] = alive;
+        }
+        alives
+    }
+
+    fn spawn(&mut self) {
+        let process = Process::new(&self.config.lock().unwrap());
+        self.process = Some(process);
+    }
+
+    /// Check for config updates and update
+    #[allow(dead_code)]
+    async fn check_config_diffs(&mut self) -> bool {
+        let c = self.config.lock().unwrap().clone();
+        'outer: for host in c.ddns.iter().enumerate() {
+            if host.1.name == c.config_metadata.name {
+                continue;
+            }
+
+            if !self.alives[host.0] {
+                continue;
+            }
+
+            // Connection
+            let connection_mutex = if let Some(conn) = self
+                .node_connections
+                .get_node_connection(host.1.name.clone())
+            {
+                conn
+            } else if let Some(conn) = self.node_connections.create_node_connection(host.1) {
+                conn
+            } else {
+                // If no connection can be established, continue the outer loop
+                continue 'outer;
+            };
+
+            log!("Checking for config updates");
+            let mut connection = connection_mutex.lock().unwrap();
+            let _ = connection.update_config(self.config.clone());
+        }
+        false
+    }
+
+    pub async fn heartbeat(&mut self) {
+        log!("\n====> Heartbeat");
+
+        let alives = self.check_hosts().await;
+        log!("\nAll hosts checked!");
+
+        log!("-> Alives: {}", alives);
+        if !self.alive
+            && (alives == 0 || {
+                // There are nodes alive with less priority
+                let config_guard = self.config.lock().unwrap();
+                assert!(config_guard.ddns.len() == self.alives.len());
+                let local_priority = config_guard
+                    .ddns
+                    .iter()
+                    .find(|d| d.name == config_guard.config_metadata.name)
+                    .map(|d| d.priority)
+                    .unwrap_or(0);
+
+                !config_guard
+                    .ddns
+                    .iter()
+                    .zip(self.alives.iter())
+                    .any(|(host, &alive)| alive && host.priority > local_priority)
+            })
+        {
+            log!("\n-> Node switching to alive");
+            self.alive = true;
+            self.spawn();
+        } else {
+            // Hosts alive
+            // First check configs and then kill or otherwise?
+            let config_guard = self.config.lock().unwrap();
+            let local_priority = config_guard
+                .ddns
+                .iter()
+                .find(|d| d.name == config_guard.config_metadata.name)
+                .map(|d| d.priority)
+                .unwrap_or(0);
+
+            if self.process.is_some()
+                && config_guard
+                    .ddns
+                    .iter()
+                    .any(|d| d.priority > local_priority)
+            {
+                // Clean up
+                self.alive = false;
+                if let Some(ref mut p) = self.process {
+                    p.kill();
+                    self.process = None;
+                }
+            }
+        }
+
+        // if alives != 0 && self.tick % 5 == 0 {
+        // self.check_config_diffs().await;
+        // }
+
+        // if self.tick == 0 {
+        //     self.tick_dir = 1;
+        // } else if self.tick == 5 {
+        //     self.tick_dir = 0;
+        // }
+        // if self.tick_dir == 1 {
+        //     self.tick += 1
+        // } else {
+        //     self.tick -= 1
+        // };
+        log!("====> Hearbeat end");
+    }
+}

src/node_connections.rs

@@ -0,0 +1,407 @@
+use std::{
+    io::{BufRead, BufReader},
+    sync::mpsc,
+    thread,
+    time::Duration,
+};
+#[allow(unused_imports)]
+use std::{
+    io::{Read, Write},
+    net::TcpStream,
+    sync::{Arc, Mutex},
+};
+
+use crate::{
+    config::{Config, ProviderNode},
+    debug, log,
+    parser::Parser,
+    timestamp::Timestamp,
+};
+
+#[derive(Debug)]
+pub struct NodeInfo {
+    pub target_name: String,
+    /// Either the IP or the DDNS
+    pub target: String,
+    pub port: u32,
+    pub preference: u8,
+    stream: Option<TcpStream>,
+}
+
+impl NodeInfo {
+    pub fn new(
+        target_name: String,
+        target: String,
+        port: u32,
+        preference: u8,
+        stream: Option<TcpStream>,
+    ) -> NodeInfo {
+        NodeInfo {
+            target_name,
+            target,
+            port,
+            preference,
+            stream,
+        }
+    }
+
+    pub fn update_config(
+        &mut self,
+        config_self_mutex: Arc<Mutex<Config>>,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        if let Some(ref mut stream) = self.stream {
+            let (tx, rx) = mpsc::channel();
+            let read_stream = stream.try_clone().unwrap();
+
+            thread::spawn(move || {
+                let mut reader = BufReader::new(read_stream);
+                let mut response = String::new();
+                match reader.read_line(&mut response) {
+                    Ok(_) => {
+                        debug!("Inside: Received response: {:?}", response);
+                        tx.send(response.trim().to_string()).unwrap_or_default();
+                    }
+                    Err(e) => {
+                        debug!("Error reading response: {:?}", e);
+                        tx.send(String::new()).unwrap_or_default();
+                    }
+                }
+                // let mut response = String::new();
+                // for line in reader.lines().map_while(Result::ok) {
+                //     if line == "DONE" {
+                //         break;
+                //     }
+                //     response.push_str(&line);
+                // }
+                // tx.send(response).unwrap_or_default();
+            });
+
+            stream.write_all(b"GET CONFIG\n")?;
+
+            let s = match rx.recv_timeout(Duration::from_secs(2)) {
+                Ok(response) => {
+                    debug!("Received response: {:?}", response);
+                    response.replace("\\n", "\n")
+                }
+                Err(e) => {
+                    debug!("Timeout waiting for response: {:?}", e);
+                    String::new()
+                }
+            };
+            if s.is_empty() {
+                debug!("Empty response: {:?}", s);
+                return Err(Box::new(std::io::Error::new(
+                    std::io::ErrorKind::Other,
+                    "No response",
+                )));
+            }
+
+            let cfg: Config = match serde_yaml::from_str(&s) {
+                Ok(cfg) => cfg,
+                Err(e) => {
+                    debug!("Error parsing config: {:?}", e);
+                    return Err(Box::new(e));
+                }
+            };
+
+            let mut config_self = config_self_mutex.lock().unwrap();
+            if config_self.config_metadata.last_updated > cfg.config_metadata.last_updated {
+                debug!("Local config is newer, aborting");
+                return Ok(());
+            }
+
+            // Update the config
+            // Execution instructions
+            config_self.execution.instructions = cfg.execution.instructions;
+
+            let node_self_name = config_self.config_metadata.name.clone();
+
+            // Add new Nodes (that do not exist in our config, but exist in the other config)
+            for ddns in &cfg.ddns {
+                if ddns.name == node_self_name {
+                    continue;
+                }
+                if !config_self.ddns.iter().any(|d| d.name == ddns.name) {
+                    config_self.ddns.push(ddns.clone());
+                }
+            }
+
+            config_self.config_metadata.last_updated = cfg.config_metadata.last_updated.clone();
+            // Wondering if we should update the last updated
+            config_self
+                .ddns
+                .iter_mut()
+                .find(|d| d.name == node_self_name)
+                .unwrap()
+                .last_updated = Timestamp::now();
+
+            config_self.write();
+            log!("Updated config successfully");
+
+            return Ok(());
+        }
+
+        debug!("No stream for {}", self.target_name);
+        Err(Box::new(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "No stream",
+        )))
+    }
+}
+
+#[allow(dead_code)]
+pub struct NodeConnections {
+    connections: Vec<Arc<Mutex<NodeInfo>>>,
+}
+
+impl Default for NodeConnections {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl NodeConnections {
+    pub fn new() -> NodeConnections {
+        NodeConnections {
+            connections: vec![],
+        }
+    }
+
+    pub fn get_node_connection(&self, node_name: String) -> Option<Arc<Mutex<NodeInfo>>> {
+        for connection in &self.connections {
+            let conn = connection.lock().unwrap();
+            if conn.target_name == node_name && conn.stream.is_some() {
+                return Some(connection.clone());
+            }
+        }
+        None
+    }
+
+    pub fn get_alive_connections(&self) -> &Vec<Arc<Mutex<NodeInfo>>> {
+        &self.connections
+    }
+
+    pub fn ping(&mut self, ddns: &ProviderNode) -> bool {
+        let target = {
+            if ddns.preference == 0 {
+                ddns.ddns.clone()
+            } else {
+                ddns.ip.clone()
+            }
+        };
+        let mut connection: Option<Arc<Mutex<NodeInfo>>> =
+            self.get_node_connection(ddns.name.clone());
+
+        if connection.is_none()
+            || (connection.is_some() && !is_connection_alive(connection.clone().unwrap()))
+        {
+            if connection.is_some() {
+                self.remove_node_connection(ddns.name.clone());
+            }
+            connection = self.create_node_connection(ddns);
+            if connection.is_none() {
+                return false;
+            }
+        }
+
+        let connection = connection.unwrap();
+        let connection_guard = connection.lock().unwrap();
+
+        if connection_guard.stream.is_none() {
+            return false;
+        }
+
+        let mut stream = connection_guard
+            .stream
+            .as_ref()
+            .unwrap()
+            .try_clone()
+            .unwrap();
+
+        let (tx, rx) = mpsc::channel();
+        let read_stream = stream.try_clone().unwrap();
+
+        thread::spawn(move || {
+            let mut reader = BufReader::new(read_stream);
+            let mut response = String::new();
+            match reader.read_line(&mut response) {
+                Ok(_) => {
+                    let is_pong = response.trim() == "PONG";
+                    tx.send(is_pong as i8).unwrap_or_default();
+                }
+                Err(e) => {
+                    debug!("Error reading response (in fn `ping`): {:?}", e);
+                    tx.send(-1).unwrap_or_default();
+                }
+            }
+        });
+
+        // Write PING
+        let _ = stream.write_all(b"PING\n");
+
+        let _ = stream.flush();
+
+        let reply = rx.recv_timeout(Duration::from_secs(2)).unwrap_or_default();
+        if reply == -1 {
+            self.remove_node_connection(target.clone());
+        }
+        reply == 1
+    }
+
+    pub fn create_node_connection(&mut self, node: &ProviderNode) -> Option<Arc<Mutex<NodeInfo>>> {
+        // TODO: DDNS
+        let stream = TcpStream::connect_timeout(
+            &std::net::SocketAddr::new(node.ip.clone().parse().unwrap(), node.port as u16),
+            Duration::from_millis(500),
+        );
+
+        match stream {
+            Ok(stream) => {
+                let connection = Arc::new(Mutex::new(NodeInfo::new(
+                    node.name.clone(),
+                    if node.preference == 0 {
+                        node.ddns.clone()
+                    } else {
+                        node.ip.clone()
+                    },
+                    node.port,
+                    node.preference,
+                    Some(stream),
+                )));
+
+                self.connections.push(connection.clone());
+                Some(connection.clone())
+            }
+
+            Err(error) => {
+                if error.kind() != std::io::ErrorKind::ConnectionRefused {
+                    log!("-> Problem creating the stream: {:?}", error);
+                }
+                None
+            }
+        }
+    }
+
+    pub fn remove_node_connection(&mut self, target_name: String) {
+        if let Some(pos) = self
+            .connections
+            .iter()
+            .position(|conn| conn.lock().unwrap().target_name == target_name)
+        {
+            self.connections.remove(pos);
+        }
+    }
+
+    pub fn confirm(&mut self, source: &str, is_ip: bool) -> Option<String> {
+        for connection in &self.connections {
+            let conn = connection.lock().unwrap();
+            if conn.stream.is_none() {
+                continue;
+            }
+
+            let mut stream = conn.stream.as_ref().unwrap();
+            stream
+                .write_all(format!("CONFIRM:{}:{}\n", is_ip as u8, source).as_bytes())
+                .unwrap();
+
+            let reader = BufReader::new(stream);
+            // let mut writer = &stream;
+
+            let sis_ip = is_ip.to_string();
+
+            for line in reader.lines() {
+                if line.is_err() {
+                    log!("Error reading line: {:?}", line.err());
+                    return None;
+                };
+                // Template: CONFIRM:_:_:bool
+                // bool is 0/1
+                let line = line.unwrap();
+                let parts: Vec<&str> = line.split(':').collect();
+                if parts.len() != 4 {
+                    log!("Invalid response: {}", line);
+                    return None;
+                }
+
+                if parts[0] != "CONFIRM" {
+                    log!("Invalid response: {}", line);
+                    return None;
+                }
+
+                if parts[1] != sis_ip {
+                    log!("Invalid response: {}", line);
+                    return None;
+                }
+
+                if parts[2] != source {
+                    log!("Invalid response: {}", line);
+                    return None;
+                }
+
+                if parts[3] == "1" {
+                    return Some(conn.target_name.clone());
+                }
+            }
+        }
+
+        None
+    }
+
+    pub fn get_config_for(
+        &mut self,
+        source: &str,
+        is_ip: bool,
+        target_name: String,
+    ) -> Option<ProviderNode> {
+        for connection in &self.connections {
+            let conn = connection.lock().unwrap();
+            if conn.stream.is_none() || conn.target_name != target_name {
+                continue;
+            }
+
+            let mut stream = conn.stream.as_ref().unwrap();
+            stream.write_all(b"GET CONFIG\n").unwrap();
+
+            let reader = BufReader::new(stream);
+
+            for line in reader.lines() {
+                if line.is_err() {
+                    log!("Error reading line: {:?}", line.err());
+                    continue;
+                };
+                let line = line.unwrap();
+                let mut parser = Parser::new(line.as_bytes());
+
+                let cfg = match parser.parse(None) {
+                    Ok(cfg) => cfg,
+                    Err(_) => {
+                        stream.write_all(b"AUTH FAIL: BAD CONFIG\n").unwrap();
+                        continue;
+                    }
+                };
+
+                if let Some(provider) =  cfg.ddns.iter().find(|d| {if is_ip { d.ip.clone()  } else { d.ddns.clone() } } == source) {
+                    return Some(provider.clone());
+                } else {
+                    return None;
+                }
+            }
+        }
+        None
+    }
+}
+
+fn is_connection_alive(connection: Arc<Mutex<NodeInfo>>) -> bool {
+    let connection_guard = connection.lock().unwrap();
+    if connection_guard.stream.is_none() {
+        return false;
+    }
+
+    let mut stream = connection_guard.stream.as_ref().unwrap();
+    match stream.write(&[]) {
+        Ok(_) => true,
+        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => true,
+        Err(e) if e.kind() == std::io::ErrorKind::BrokenPipe => false,
+        Err(_) => false,
+    }
+}

src/parser.rs

@@ -0,0 +1,68 @@
+use std::{
+    io::Read,
+    sync::{Arc, Mutex},
+};
+
+use crate::config::Config;
+
+pub struct Parser<R: Read> {
+    src: R,
+}
+
+impl<R: Read> Parser<R> {
+    pub fn new(src: R) -> Parser<R> {
+        Parser { src }
+    }
+
+    pub fn parse(
+        &mut self,
+        config_str: Option<Arc<Mutex<String>>>,
+    ) -> Result<Config, std::io::Error> {
+        let mut contents = String::new();
+        self.src.read_to_string(&mut contents)?;
+
+        // parse
+        let cfg: Config = serde_yaml::from_str(&contents).unwrap();
+        if let Some(config_str) = config_str {
+            *config_str.lock().unwrap() = contents;
+        }
+
+        Ok(cfg)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::io::Cursor;
+
+    #[test]
+    fn test_parser() {
+        let yaml = r#"
+ddns:
+- name: test
+  ddns: ''
+  ip: 127.0.0.1
+  port: 8080
+  preference: 1
+  priority: 100
+  last_updated: 2024-03-20 00:00:00 UTC
+config_metadata:
+  name: test
+  last_updated: 2024-03-20 00:00:00 UTC
+execution:
+  instructions: ./test.sh
+  last_updated: 2024-03-20 00:00:00 UTC
+"#;
+        let mut parser = Parser::new(Cursor::new(yaml));
+        let result = parser.parse(None);
+        assert!(result.is_ok());
+
+        let config = result.unwrap();
+        assert_eq!(config.ddns.len(), 1);
+        assert_eq!(config.ddns[0].ip, "127.0.0.1");
+        assert_eq!(config.ddns[0].priority, 100);
+        assert_eq!(config.ddns[0].name, "test");
+        assert_eq!(config.ddns[0].name, config.config_metadata.name);
+    }
+}

src/pending_verification.rs

@@ -0,0 +1,6 @@
+pub struct PendingVerification {
+    pub source: String,
+    pub remote_addr: String,
+    pub redirect_node: String,
+    pub is_ip: bool,
+}

src/process.rs

@@ -0,0 +1,22 @@
+use crate::{config::Config, log};
+
+pub struct Process {
+    pub child: std::process::Child,
+}
+
+impl Process {
+    pub fn new(cfg: &Config) -> Process {
+        let args: Vec<&str> = cfg.execution.instructions.split(" ").collect();
+        let child = std::process::Command::new(args[0])
+            .args(&args[1..])
+            .spawn()
+            .expect("Couldn't spawn the process.");
+
+        Process { child }
+    }
+
+    pub fn kill(&mut self) {
+        log!("Killing process {}", self.child.id());
+        self.child.kill().expect("!kill");
+    }
+}

src/tcp_listener.rs

@@ -0,0 +1,191 @@
+use crate::config::Config;
+use crate::{debug, log};
+use std::io::{BufRead, BufReader, Write};
+use std::net::TcpListener;
+use std::sync::{Arc, Mutex};
+use std::thread;
+
+pub fn start_tcp_listener(config: Arc<Mutex<Config>>, config_string: Arc<Mutex<String>>) {
+    thread::spawn(move || {
+        //let trustkey_path =
+        //std::env::var("P2P_TRUSTKEY_PATH").unwrap_or_else(|_| ".p2p-trustkey".to_string());
+        //let mut trustkey = String::new();
+        //let _ = get_file(&trustkey_path).read_to_string(&mut trustkey);
+
+        let config = config.clone();
+
+        let port = {
+            let cfg = config.lock().unwrap();
+            let self_name = &cfg.config_metadata.name;
+            cfg.ddns.iter().find(|d| d.name == *self_name).unwrap().port
+        };
+
+        let listener = match TcpListener::bind(format!("0.0.0.0:{}", port)) {
+            Ok(listener) => listener,
+            Err(error) => {
+                panic!("TcpListener can't bind to port {port}, {:?}", error);
+            }
+        };
+
+        log!("Rocking on port {port}!");
+
+        for stream in listener.incoming() {
+            debug!("CONNECTION established");
+
+            if let Ok(stream) = stream {
+                let reader = BufReader::new(&stream);
+                let mut writer = &stream;
+
+                for line in reader.lines().map_while(Result::ok) {
+                    let remote_addr = stream.peer_addr().unwrap().ip().to_string();
+                    let line = line.as_str();
+
+                    debug!(
+                        "Received line: {:?} (l:{}) from {}",
+                        line,
+                        line.len(),
+                        remote_addr
+                    );
+
+                    if line.len() == 4 && &line[0..4] == "PING" {
+                        let _ = writer.write_all(b"PONG\n");
+                        let _ = writer.flush();
+                    } else if line.len() >= 10 && &line[0..10] == "GET CONFIG" {
+                        let config_str = config_string.lock().unwrap();
+                        let _ = writer
+                            .write_all(format!("{}\n", config_str.replace("\n", "\\n")).as_bytes());
+                        let _ = writer.flush();
+                        debug!("Sent config to {}", remote_addr);
+                    }
+                    // else if line.len() > 8 && &line[0..8] == "CONFIRM:" {
+                    //     // Template: CONFIRM:is_ip:source
+                    //     // is_ip is either 0 or 1
+                    //     let parts = line.split(":").collect::<Vec<&str>>();
+                    //     if parts.len() != 3 || !(parts[1] == "0" || parts[1] == "1") {
+                    //         debug!("CONFIRM FAIL: BAD REQUEST");
+                    //         let _ = writer.write_all(b"AUTH FAIL: BAD REQUEST\n");
+                    //         continue;
+                    //     }
+                    //
+                    //     let is_ip = parts[1] == "1";
+                    //     let source = parts[2];
+                    //
+                    //     let config_guard = config.lock().unwrap();
+                    //     let found = config_guard.ddns.iter().any(|d| {if is_ip {&d.ip} else {&d.ddns}} == source);
+                    //     let _ = writer
+                    //         .write_all({
+                    //             if found {
+                    //                 b"1\n"
+                    //             } else {
+                    //                 b"0\n"
+                    //             }
+                    //         });
+                    // }
+                    // else if line.len() >= 8 && &line[0..8] == "AUTH REQ" {
+                    //     // Template: AUTH:source:port:trustkey:redirect_node
+                    //     let parts = line.split(":").collect::<Vec<&str>>();
+                    //     if parts.len() < 4 || parts.len() > 5 {
+                    //         writer.write_all(b"AUTH FAIL: BAD REQUEST\n").unwrap();
+                    //         continue;
+                    //     }
+                    //
+                    //     let source = parts[2];
+                    //     let _source_port = match parts[3].parse::<u32>() {
+                    //         Ok(p) => p,
+                    //         Err(_) => continue,
+                    //     };
+                    //     let is_ip = source.chars().all(|c: char| c == '.' || c.is_ascii_digit());
+                    //
+                    //     // Check if other Nodes have it
+                    //     {
+                    //         let mut node_guard = node.lock().unwrap();
+                    //         let node_confirmed = node_guard.node_connections.confirm(source, is_ip);
+                    //         if let Some(node_confirmed) = node_confirmed {
+                    //             let provider = node_guard.node_connections.get_config_for(
+                    //                 source,
+                    //                 is_ip,
+                    //                 node_confirmed,
+                    //             );
+                    //             if let Some(provider) = provider {
+                    //                 let mut config_guard = config.lock().unwrap();
+                    //                 config_guard.ddns.push(provider);
+                    //             }
+                    //         }
+                    //     };
+                    //
+                    //     // DDNS; Verification
+                    //     if is_ip && remote_addr != source {
+                    //         writer.write_all(b"AUTH FAIL: SOURCE MISMATCH\\nn").unwrap();
+                    //         continue;
+                    //     }
+                    //
+                    //     let request_trustkey = parts[4];
+                    //     if request_trustkey != trustkey {
+                    //         writer.write_all(b"AUTH FAIL: TRUSTKEY MISMATCH\n").unwrap();
+                    //         continue;
+                    //     }
+                    //
+                    //     let config_guard = config.lock().unwrap();
+                    //     let ddns = config_guard.ddns.iter().find(|d| d.name == source);
+                    //     if ddns.is_some() {
+                    //         writer.write_all(b"AUTH SUCCESS: ALREADY EXISTS\n").unwrap();
+                    //         continue;
+                    //     }
+                    //
+                    //     // TODO: A) This. Search for `TODO: A)`
+                    //     verifications.push(PendingVerification {
+                    //         source: source.to_string(),
+                    //         remote_addr,
+                    //         redirect_node: {
+                    //             if parts.len() >= 5 {
+                    //                 parts[5]
+                    //             } else {
+                    //                 ""
+                    //             }
+                    //         }
+                    //         .to_string(),
+                    //         is_ip,
+                    //     });
+                    //
+                    //     writer.write_all(b"GET CONFIG\n").unwrap();
+                    // }
+                    // else if line.len() > 12 && &line[0..12] == "AUTH PENDING" {
+                    //     // Template: AUTH PENDING:config
+                    //     if !verifications.iter().any(|v| remote_addr == v.remote_addr) {
+                    //         writer.write_all(b"AUTH FAIL: NOT PENDING\n").unwrap();
+                    //         continue;
+                    //     };
+                    //
+                    //     let config_incoming = &line[13..line.len()];
+                    //     let mut parser_incoming = Parser::new(config_incoming.as_bytes());
+                    //
+                    //     let config_incoming = match parser_incoming.parse(None) {
+                    //         Ok(cfg) => cfg,
+                    //         Err(_) => {
+                    //             writer.write_all(b"AUTH FAIL: BAD CONFIG\n").unwrap();
+                    //             continue;
+                    //         }
+                    //     };
+                    //
+                    //     let ddns_incoming = match config_incoming
+                    //         .ddns
+                    //         .iter()
+                    //         .find(|d| d.name == config_incoming.config_metadata.name)
+                    //     {
+                    //         Some(d) => d,
+                    //         None => {
+                    //             writer
+                    //                 .write_all(b"AUTH FAIL: SELF ABSENT IN DDNS\n")
+                    //                 .unwrap();
+                    //             continue;
+                    //         }
+                    //     };
+                    //
+                    //     let mut config_guard = config.lock().unwrap();
+                    //     config_guard.ddns.push(ddns_incoming.clone());
+                    // }
+                }
+            }
+        }
+    });
+}

src/timestamp.rs

@@ -0,0 +1,45 @@
+use dateparser::DateTimeUtc;
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+
+#[derive(Debug, Clone)]
+pub struct Timestamp(pub(crate) DateTimeUtc);
+
+impl Timestamp {
+    pub fn now() -> Timestamp {
+        Timestamp(DateTimeUtc(chrono::Utc::now()))
+    }
+}
+
+impl PartialOrd for Timestamp {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        self.0 .0.partial_cmp(&other.0 .0)
+    }
+}
+
+impl PartialEq for Timestamp {
+    fn eq(&self, other: &Self) -> bool {
+        self.0 .0 == other.0 .0
+    }
+}
+
+impl Serialize for Timestamp {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        // let s = format!("{}", self.);
+        let s = self.0 .0.to_string();
+        serializer.serialize_str(&s)
+    }
+}
+
+impl<'de> Deserialize<'de> for Timestamp {
+    fn deserialize<D>(deserializer: D) -> Result<Timestamp, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        let dt = s.parse::<DateTimeUtc>().map_err(serde::de::Error::custom)?;
+        Ok(Timestamp(dt))
+    }
+}