added connection acceptance filters

Cargo.lock

@@ -63,6 +63,12 @@ version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693"
 
+[[package]]
+name = "bitstring"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3e54f7b7a46d7b183eb41e2d82965261fa8a1597c68b50aced268ee1fc70272d"
+
 [[package]]
 name = "block-buffer"
 version = "0.7.3"
@@ -160,6 +166,16 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
 
+[[package]]
+name = "cidr"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c6316c62053228eddd526a5e6deb6344c80bf2bc1e9786e7f90b3083e73197c1"
+dependencies = [
+ "bitstring",
+ "serde",
+]
+
 [[package]]
 name = "cloudabi"
 version = "0.0.3"
@@ -621,6 +637,7 @@ dependencies = [
  "async-compression",
  "bzip2-sys",
  "cfg-if 1.0.0",
+ "cidr",
  "futures",
  "hyper",
  "lazy_static",

Cargo.toml

@@ -68,6 +68,7 @@ smallmap = "1.1.5"
 lazy_static = "1.4.0"
 once_cell = "1.4.1"
 bzip2-sys = {version = "0.1.9", optional = true}
+cidr = {version = "0.1.1", features = ["serde"]}
 
 [build-dependencies]
 rustc_version = "0.2"

markov.toml

@@ -14,5 +14,6 @@ outbound = ''
 backlog = 32
 internal_backlog = 8
 capacity = 4
-timeout_ms = 5000
+
-throttle_ms = 50
+[mask]
+default = 'Accept'

src/config.rs

@@ -15,6 +15,7 @@ use tokio::{
     time::Duration,
     io::BufReader,
 };
+use ipfilt::IpFilter;
 
 pub const DEFAULT_FILE_LOCATION: &'static str = "markov.toml";
 
@@ -33,11 +34,14 @@ pub struct Config
     pub filter: FilterConfig,
     #[serde(default)]
     pub writer: WriterConfig,
+    #[serde(default)]
+    pub mask: IpFilter,
 }
 
 #[derive(Debug, Default, Clone, PartialEq, Eq, PartialOrd, Hash, Serialize, Deserialize)]
 pub struct FilterConfig
 {
+    #[serde(default)]
     inbound: String,
     #[serde(default)]
     outbound: String,
@@ -123,6 +127,7 @@ impl Default for Config
 	    filter: Default::default(),
 	    feed_bounds: "2..".to_owned(),
 	    writer: Default::default(),
+	    mask: Default::default(),
 	}
     }
 }

src/ipfilt.rs

@@ -0,0 +1,181 @@
+//! Filter accepts and denies based on cidr masks.
+use super::*;
+use cidr::{
+    Cidr,
+    IpCidr,
+};
+use std::{
+    net::{
+	IpAddr,
+    },
+    error,
+    fmt,
+};
+
+#[derive(Debug)]
+pub struct IpFilterDeniedError(IpAddr, Option<IpCidr>);
+
+impl warp::reject::Reject for IpFilterDeniedError{}
+impl error::Error for IpFilterDeniedError{}
+impl fmt::Display for IpFilterDeniedError
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result
+    {
+	write!(f, "Denied {} due to ", self.0)?;
+	match &self.1 {
+	    Some(cidr) => write!(f, "matching rule {}", cidr),
+	    None => write!(f, "non-matching accept rule"),
+	}
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Copy, PartialOrd, Ord,  Serialize, Deserialize)]
+pub enum Rule
+{
+    Accept,
+    Deny,
+}
+
+impl Default for Rule
+{
+    #[inline]
+    fn default() -> Self
+    {
+	Self::Deny
+    }
+}
+
+impl Rule
+{
+    fn into_result<'a>(self, net: Option<&'a IpCidr>) -> Result<Option<&'a IpCidr>, Option<IpCidr>>
+    {
+	if let Self::Accept = self {
+	    Ok(net)
+	} else {
+	    Err(net.cloned())
+	}
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+pub struct IpFilter
+{
+    /// The default fallback rule
+    pub default: Rule,
+    
+    #[serde(default)]
+    accept: Vec<IpCidr>,
+    #[serde(default)]
+    deny: Vec<IpCidr>,
+}
+
+#[inline] fn find_in<'a>(needle: &IpAddr, haystack: &'a [IpCidr]) -> Option<&'a IpCidr>
+{
+    for x in haystack.iter()
+    {
+	if x.contains(needle) {
+	    return Some(x);
+	}
+    }
+    None
+}
+
+impl Default for IpFilter
+{
+    #[inline]
+    fn default() -> Self
+    {
+	Self {
+	    default: Rule::Deny,
+	    accept: vec![cidr::Cidr::new_host([127,0,0,1].into())],
+	    deny: Vec::default(),
+	}
+    }
+}
+
+impl IpFilter
+{
+    /// Create a new CIDR filter with thie default rule.
+    ///
+    /// Use `default()` to use with default rule.
+    pub fn new(fallback: Rule) -> Self
+    {
+	Self {
+	    default: fallback,
+	    accept: Vec::new(),
+	    deny: Vec::new(),
+	}
+    }
+
+    /// Checks the rule for this IP, returns a result if it should accept or not.
+    ///
+    /// If acceptance rule is met, return the CIDR match that caused the acceptance if applicable
+    ///
+    /// If acceptance rule is not met, return in the error which CIDR match cause the deny if applicable
+    pub fn check(&self, ip: &IpAddr) -> Result<Option<&'_ IpCidr>, IpFilterDeniedError>
+    {
+	let accept = find_in(ip, &self.accept[..]);
+	let deny = find_in(ip, &self.deny[..]);
+
+	let (rule, cidr) = match (accept, deny) {
+	    (None, Some(net)) => (Rule::Deny, Some(net)),
+	    (Some(net), None) => (Rule::Accept, Some(net)),
+	    (Some(ac), Some(den)) if ac != den => {
+		if ac.network_length() > den.network_length() {
+		    (Rule::Accept, Some(ac))
+		} else {
+		    (Rule::Deny, Some(den))
+		}
+	    },
+	    _ => (self.default, None)
+	};
+	rule.into_result(cidr)
+	    .map_err(|cidr| IpFilterDeniedError(*ip, cidr))
+    }
+
+    pub fn accept_mask(&self) -> &[IpCidr]
+    {
+	&self.accept[..]
+    }
+    pub fn deny_mask(&self) -> &[IpCidr]
+    {
+	&self.deny[..]
+    }
+    pub fn accept_range(&mut self, items: impl IntoIterator<Item = IpCidr>)
+    {
+	self.accept.extend(items)
+    }
+    pub fn deny_range(&mut self, items: impl IntoIterator<Item = IpCidr>)
+    {
+	self.deny.extend(items)
+    }
+    
+    pub fn accept_one(&mut self, item: IpCidr)
+    {
+	self.accept.push(item)
+    }
+    pub fn deny_one(&mut self, items: IpCidr)
+    {
+	self.deny.push(items)
+    }
+
+    /// Can any connection ever be accepted?
+    pub fn possible(&self) -> bool
+    {
+	//TODO: Test this
+	!(self.default == Rule::Deny && self.accept.len() == 0) &&
+	    !(self.deny.iter().find(|x| x.network_length() == 0).is_some() && self.accept.len() == 0)
+    }
+}
+
+pub async fn recover(err: warp::Rejection) -> Result<impl warp::Reply, warp::Rejection>
+{
+    if let Some(t) = err.find::<IpFilterDeniedError>() {
+	error!("Denying access to {} because of {:?} (403)", t.0, t.1);
+	Ok(warp::http::Response::builder()
+	   .status(status!(403))
+	   .body(format!("Access denied: {}", t)))
+    } else {
+	Err(err)
+    }
+}

src/main.rs

@@ -76,6 +76,7 @@ mod msg;
 mod state;
 use state::State;
 mod save;
+mod ipfilt;
 mod forwarded_list;
 use forwarded_list::XForwardedFor;
 mod handle;
@@ -172,11 +173,25 @@ async fn main() {
 	} else {
 	    warp::filters::addr::remote().and_then(|x: Option<SocketAddr>| async move {x.map(|x| x.ip()).ok_or_else(|| warp::reject::not_found())}).boxed()
 	};
+
+	let ipfilter = warp::any()
+	    .and(chain)
+	    .and(client_ip)
+	    .and_then(|state: State, host: IpAddr| {
+		async move {
+		    state.config().mask.check(&host)
+			.map(|ci| {
+			    trace!("Accepting from rule {:?}", ci);
+			    host
+			})
+			.map(move |host| (state, host))
+			.map_err(warp::reject::custom)
+		}
+	    }).untuple_one();
 	
 	let push = warp::put()
 	    .and(warp::path("put"))
-	    .and(chain.clone())
+	    .and(ipfilter.clone())
-	    .and(client_ip.clone())
 	    .and(warp::body::content_length_limit(state.config().max_content_length))
 	    .and(warp::body::stream())
 	    .and_then(|state: State, host: IpAddr, buf| {
@@ -186,6 +201,8 @@ async fn main() {
 			.map_err(|_| warp::reject::not_found()) //(warp::reject::custom) //TODO: Recover rejection filter down below for custom error return
 		}
 	    })
+	    
+	    .recover(ipfilt::recover)
 	    .with(warp::log("markov::put"));
 
 	
@@ -195,8 +212,8 @@ async fn main() {
 		    let single = {
 			let msz = state.config().max_gen_size;
 			warp::post()
+			    .and(ipfilter.clone())
 			    .and(warp::path("single"))
-			    .and(client_ip.clone())
 			    .and(warp::path::param()
 				 .map(move |sz: usize| {
 				     if sz == 0 || (2..=msz).contains(&sz) {
@@ -209,11 +226,13 @@ async fn main() {
 				 .unify())
 			    .and(warp::body::content_length_limit(state.config().max_content_length))
 			    .and(warp::body::aggregate())
+			    .map(|_, x, y, z| (x,y,z)).untuple_one()
 			    .and_then(api::single)
 			    .with(warp::log("markov::api::single"))
 		    };
 		    warp::path("api")
 			.and(single)
+			.recover(ipfilt::recover)
 			.recover(api::error::rejection)
 		};
 	    }
@@ -221,8 +240,7 @@ async fn main() {
 	
 	
 	let read = warp::get()
-	    .and(chain.clone())
+	    .and(ipfilter.clone())
-	    .and(client_ip.clone())
 	    .and(warp::path::param().map(|opt: usize| Some(opt))
 		 .or(warp::path::end().map(|| Option::<usize>::None)).unify())
 	    .and_then(|state: State, host: IpAddr, num: Option<usize>| {
@@ -240,12 +258,12 @@ async fn main() {
 		    }))))
 		}
 	    })
+	    .recover(ipfilt::recover)
 	    .with(warp::log("markov::read"));
 	
 	let sentance = warp::get()
 	    .and(warp::path("sentance")) //TODO: sanitise::Sentance::new_iter the body line
-	    .and(chain.clone())
+	    .and(ipfilter.clone())
-            .and(client_ip.clone())
 	    .and(warp::path::param().map(|opt: usize| Some(opt))
 		 .or(warp::path::end().map(|| Option::<usize>::None)).unify())
 	    .and_then(|state: State, host: IpAddr, num: Option<usize>| {
@@ -263,6 +281,7 @@ async fn main() {
 		    }))))
 		}
 	    })
+	    .recover(ipfilt::recover)
 	    .with(warp::log("markov::read::sentance"));
 
 	let read = warp::path("get").and(read.or(sentance));