add basic protocol interaction

4 files changed, 322 insertions, 4 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 2cae607..8fb027c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -18,6 +18,41 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
 
 [[package]]
+name = "async-io"
+version = "2.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "10da8f3146014722c89e7859e1d7bb97873125d7346d10ca642ffab794355828"
+dependencies = [
+ "async-lock",
+ "cfg-if",
+ "concurrent-queue",
+ "futures-io",
+ "futures-lite",
+ "parking",
+ "polling",
+ "rustix",
+ "slab",
+ "tracing",
+ "waker-fn",
+ "windows-sys",
+]
+
+[[package]]
+name = "async-lock"
+version = "2.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "287272293e9d8c41773cec55e365490fe034813a2f172f502d6ddcf75b2f582b"
+dependencies = [
+ "event-listener",
+]
+
+[[package]]
+name = "autocfg"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
+
+[[package]]
 name = "backtrace"
 version = "0.3.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -39,6 +74,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2d7e60934ceec538daadb9d8432424ed043a904d8e0243f3c6446bce549a46ac"
 
 [[package]]
+name = "bitflags"
+version = "2.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "327762f6e5a765692301e5bb513e0d9fef63be86bbc14528052b1cd3e6f03e07"
+
+[[package]]
 name = "cc"
 version = "1.0.83"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -54,6 +95,24 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
 [[package]]
+name = "concurrent-queue"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f057a694a54f12365049b0958a1685bb52d567f5593b355fbf685838e873d400"
+dependencies = [
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-utils"
+version = "0.8.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a22b2d63d4d1dc0b7f1b6b2747dd0088008a9be28b6ddf0b1e7d335e3037294"
+dependencies = [
+ "cfg-if",
+]
+
+[[package]]
 name = "darling"
 version = "0.20.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -89,12 +148,50 @@ dependencies = [
 ]
 
 [[package]]
+name = "errno"
+version = "0.3.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac3e13f66a2f95e32a39eaa81f6b95d42878ca0e1db0c7543723dfe12557e860"
+dependencies = [
+ "libc",
+ "windows-sys",
+]
+
+[[package]]
+name = "event-listener"
+version = "2.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0"
+
+[[package]]
 name = "fnv"
 version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
 [[package]]
+name = "futures-core"
+version = "0.3.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eb1d22c66e66d9d72e1758f0bd7d4fd0bee04cad842ee34587d68c07e45d088c"
+
+[[package]]
+name = "futures-io"
+version = "0.3.29"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8bf34a163b5c4c52d0478a4d757da8fb65cabef42ba90515efee0f6f9fa45aaa"
+
+[[package]]
+name = "futures-lite"
+version = "1.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49a9d51ce47660b1e808d3c990b4709f2f415d928835a17dfd16991515c46bce"
+dependencies = [
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
 name = "getrandom"
 version = "0.2.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -152,6 +249,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a08173bc88b7955d1b3145aa561539096c421ac8debde8cbc3612ec635fee29b"
 
 [[package]]
+name = "linux-raw-sys"
+version = "0.4.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da2479e8c062e40bf0066ffa0bc823de0a9368974af99c9f6df941d2c231e03f"
+
+[[package]]
 name = "memchr"
 version = "2.6.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -186,12 +289,32 @@ dependencies = [
 ]
 
 [[package]]
+name = "parking"
+version = "2.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bb813b8af86854136c6922af0598d719255ecb2179515e6e7730d468f05c9cae"
+
+[[package]]
 name = "pin-project-lite"
 version = "0.2.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8afb450f006bf6385ca15ef45d71d2288452bc3683ce2e2cacc0d18e4be60b58"
 
 [[package]]
+name = "polling"
+version = "3.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e53b6af1f60f36f8c2ac2aad5459d75a5a9b4be1e8cdd40264f315d78193e531"
+dependencies = [
+ "cfg-if",
+ "concurrent-queue",
+ "pin-project-lite",
+ "rustix",
+ "tracing",
+ "windows-sys",
+]
+
+[[package]]
 name = "ppproperly"
 version = "0.1.0"
 source = "git+https://github.com/rsdsl/ppproperly.git#ab2277d70070369d57ba4854af5f1c03b19f8ae3"
@@ -278,6 +401,7 @@ dependencies = [
 name = "rsdsl_pppoe3"
 version = "0.1.0"
 dependencies = [
+ "async-io",
  "ioctls",
  "libc",
  "ppproperly",
@@ -297,6 +421,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d626bb9dae77e28219937af045c257c28bfd3f69333c512553507f5f9798cb76"
 
 [[package]]
+name = "rustix"
+version = "0.38.21"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b426b0506e5d50a7d8dafcf2e81471400deb602392c7dd110815afb4eaf02a3"
+dependencies = [
+ "bitflags",
+ "errno",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys",
+]
+
+[[package]]
 name = "ryu"
 version = "1.0.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -334,6 +471,15 @@ dependencies = [
 ]
 
 [[package]]
+name = "slab"
+version = "0.4.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f92a496fb766b417c996b9c5e57daf2f7ad3b0bebe1ccfca4856390e3d3bb67"
+dependencies = [
+ "autocfg",
+]
+
+[[package]]
 name = "socket2"
 version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -404,12 +550,34 @@ dependencies = [
 ]
 
 [[package]]
+name = "tracing"
+version = "0.1.40"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3523ab5a71916ccf420eebdf5521fcef02141234bbc0b8a49f2fdc4544364ef"
+dependencies = [
+ "pin-project-lite",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-core"
+version = "0.1.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c06d3da6113f116aaee68e4d601191614c9053067f9ab7f6edbcb161237daa54"
+
+[[package]]
 name = "unicode-ident"
 version = "1.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3354b9ac3fae1ff6755cb6db53683adb661634f67557942dea4facebec0fee4b"
 
 [[package]]
+name = "waker-fn"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f3c4517f54858c779bbcbf228f4fca63d121bf85fbecb2dc578cdf4a39395690"
+
+[[package]]
 name = "wasi"
 version = "0.11.0+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
diff --git a/Cargo.toml b/Cargo.toml
index a552211..6689722 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,6 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+async-io = "2.1.0"
 ioctls = "0.6.1"
 libc = "0.2.149"
 ppproperly = { git = "https://github.com/rsdsl/ppproperly.git", version = "0.1.0" }
diff --git a/src/error.rs b/src/error.rs
index 9eed12a..7b53b10 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -1,5 +1,7 @@
 use std::{ffi, io};
 
+use tokio::sync::watch;
+
 use thiserror::Error;
 
 /// An external error that prevents a supervisor from functioning.
@@ -9,6 +11,9 @@ pub enum Error {
     Io(#[from] io::Error),
     #[error("interface name contains nul byte: {0}")]
     Nul(#[from] ffi::NulError),
+
+    #[error("error receiving from tokio watch channel: {0}")]
+    WatchRecv(#[from] watch::error::RecvError),
 }
 
 /// An alias for a [`std::result::Result`] with the [`enum@Error`] type of this crate.
diff --git a/src/supervisor.rs b/src/supervisor.rs
index c008a15..721f793 100644
--- a/src/supervisor.rs
+++ b/src/supervisor.rs
@@ -8,10 +8,13 @@ use crate::{
 
 use std::ffi::CString;
 use std::fs::{File, OpenOptions};
+use std::io::Read;
 use std::net::Ipv4Addr;
 use std::os::fd::AsRawFd;
+use std::time::Duration;
 use std::{io, mem};
 
+use async_io::Async;
 use ppproperly::{IpCompressionProtocol, IpcpOpt, Ipv6cpOpt, LcpOpt, MacAddr, QualityProtocol};
 use socket2::{SockAddr, Socket, Type};
 
@@ -90,6 +93,8 @@ pub struct Client {
     session_id: u16,
     remote: MacAddr,
 
+    authenticated: bool,
+
     pppoe: PppoeClient,
     lcp: NegotiationProtocol<LcpOpt>,
     pap: PapClient,
@@ -128,6 +133,8 @@ impl Client {
             session_id: 0,
             remote: MacAddr::BROADCAST,
 
+            authenticated: false,
+
             pppoe: PppoeClient::new(None, None),
             lcp: NegotiationProtocol::new(ProtocolConfig {
                 require: vec![LcpOpt::Mru(1492), LcpOpt::MagicNumber(peer_magic)],
@@ -206,13 +213,150 @@ impl Client {
 
     /// Tries to keep a session open at all costs.
     /// Blocks the caller forever unless a panic occurs.
-    pub fn run(&self) {
-        let sock_disc = self.new_discovery_socket();
+    pub async fn run(&mut self) -> Result<()> {
+        let sock_disc = self.new_discovery_socket()?;
+
+        let mut pppoe_buf = [0; 1522];
+
+        let mut ncp_check = tokio::time::interval(Duration::from_secs(20));
+
+        let mut pppoe_rx = self.pppoe.active();
+        let mut lcp_rx = self.lcp.opened();
+        let mut pap_rx = self.pap.opened();
+        let mut chap_rx = self.chap.opened();
+        let mut ipcp_rx = self.ipcp.opened();
+        let mut ipv6cp_rx = self.ipv6cp.opened();
+
+        let mut lcp_lower_rx = self.lcp.active();
+        let mut ipcp_lower_rx = self.ipcp.active();
+        let mut ipv6cp_lower_rx = self.ipv6cp.active();
+
+        self.pppoe.open();
+        self.lcp.open();
+        // Authentication protocols are opened as needed, see select! below.
+        self.ipcp.open();
+        self.ipv6cp.open();
+
+        loop {
+            tokio::select! {
+                result = pppoe_rx.changed() => {
+                    result?;
+
+                    let is_active = *pppoe_rx.borrow_and_update();
+                    if is_active {
+                        todo!("setup session (fds)");
+                        self.lcp.up();
+                    } else {
+                        todo!("erase session (fds)");
+                        self.lcp.down();
+                    }
+                }
+                result = lcp_rx.changed() => {
+                    result?;
+
+                    let is_opened = *lcp_rx.borrow_and_update();
+                    if is_opened {
+                        todo!("open auth as needed or skip straight to NCPs and set authenticated = true and reset ncp check timer")
+                    } else {
+                        self.authenticated = false;
+
+                        self.pap.down();
+                        self.chap.down();
+                        self.ipcp.down();
+                        self.ipv6cp.down();
+
+                        self.pap.close();
+                        self.chap.close();
+                    }
+                }
+                result = pap_rx.changed() => {
+                    result?;
+
+                    let is_opened = *pap_rx.borrow_and_update();
+                    if is_opened {
+                        self.authenticated = true;
+                        ncp_check.reset();
+
+                        self.ipcp.up();
+                        self.ipv6cp.up();
+                    } // PAP cannot go down once it has opened successfully.
+                }
+                result = chap_rx.changed() => {
+                    result?;
+
+                    let is_opened = *chap_rx.borrow_and_update();
+                    if is_opened {
+                        self.authenticated = true;
+                        ncp_check.reset();
+
+                        self.ipcp.up();
+                        self.ipv6cp.up();
+                    } else {
+                        self.authenticated = false;
+                        self.lcp.close();
+                    }
+                }
+                result = ipcp_rx.changed() => {
+                    result?;
+
+                    let is_opened = *ipcp_rx.borrow_and_update();
+                    if is_opened {
+                        todo!("write v4 success to ds config and inform netlinkd")
+                    } else {
+                        todo!("write v4 invalidation to ds config and inform netlinkd")
+                    }
+                }
+                result = ipv6cp_rx.changed() => {
+                    result?;
+
+                    let is_opened = *ipv6cp_rx.borrow_and_update();
+                    if is_opened {
+                        todo!("write v6 success to ds config and inform netlinkd and dhcp6")
+                    } else {
+                        todo!("write v6 invalidation to ds config and inform netlinkd and dhcp6")
+                    }
+                }
+
+                result = lcp_lower_rx.changed() => {
+                    result?;
+
+                    let is_active = *lcp_lower_rx.borrow_and_update();
+                    if !is_active { // LCP has gone down, a new PPPoE session is needed.
+                        self.lcp.down();
+
+                        self.pppoe.close();
+                        self.pppoe.open();
+
+                        self.lcp.open();
+                    }
+                }
+                result = ipcp_lower_rx.changed() => {
+                    result?;
+                    ncp_check.reset();
+                }
+                result = ipv6cp_lower_rx.changed() => {
+                    result?;
+                    ncp_check.reset();
+                }
+
+                _ = ncp_check.tick() => {
+                    if *lcp_rx.borrow() && self.authenticated && !*ipcp_rx.borrow() && !*ipv6cp_rx.borrow() {
+                        // No NCPs are up after 20 seconds, terminate the link.
+                        self.lcp.close();
+                    }
+                }
+
+                result = sock_disc.read_with(|mut io| io.read(&mut pppoe_buf)) => {
+                    let n = result?;
+                    let pppoe_buf = &pppoe_buf[..n];
+                }
+            }
+        }
     }
 
     /// Creates a new socket for PPPoE Discovery traffic.
     /// Used by the PPPoE implementation.
-    fn new_discovery_socket(&self) -> Result<Socket> {
+    fn new_discovery_socket(&self) -> Result<Async<Socket>> {
         use libc::{
             sockaddr_ll, sockaddr_storage, socklen_t, AF_PACKET, ETH_P_PPP_DISC, PF_PACKET,
             SOCK_RAW,
@@ -250,7 +394,7 @@ impl Client {
             )
         })?;
 
-        Ok(sock)
+        Ok(Async::new(sock)?)
     }
 
     /// Creates a control socket for the `ppp0` virtual network interface