Code: Firewall filter

Code implementation of a Firewall filter that will allow/deny packets
based on their from address on both read and write.

Documentation to come next to finish off the below two tickets.

Work on #158
Work on #343

Cargo.toml

@@ -70,6 +70,7 @@ uuid = { version = "0.8.2", default-features = false, features = ["v4"] }
 thiserror = "1.0.30"
 eyre = "0.6.5"
 stable-eyre = "0.2.2"
+ipnetwork = "0.18.0"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 sys-info = "0.9.0"

build.rs

@@ -35,6 +35,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         "proto/quilkin/extensions/filters/load_balancer/v1alpha1/load_balancer.proto",
         "proto/quilkin/extensions/filters/local_rate_limit/v1alpha1/local_rate_limit.proto",
         "proto/quilkin/extensions/filters/token_router/v1alpha1/token_router.proto",
+        "proto/quilkin/extensions/filters/firewall/v1alpha1/firewall.proto",
     ]
     .iter()
     .map(|name| std::env::current_dir().unwrap().join(name))

proto/quilkin/extensions/filters/firewall/v1alpha1/firewall.proto

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+syntax = "proto3";
+
+package quilkin.extensions.filters.firewall.v1alpha1;
+
+message Firewall {
+  enum Action {
+    Allow = 0;
+    Deny = 1;
+  }
+
+  message PortRange {
+    uint32 min = 1;
+    uint32 max = 2;
+  }
+
+  message Rule {
+    Action action = 1;
+    string source = 2;
+    repeated PortRange ports = 3;
+  }
+
+  repeated Rule on_read = 1;
+  repeated Rule on_write = 2;
+}
+

src/filters.rs

@@ -30,6 +30,7 @@ pub mod capture_bytes;
 pub mod compress;
 pub mod concatenate_bytes;
 pub mod debug;
+pub mod firewall;
 pub mod load_balancer;
 pub mod local_rate_limit;
 pub mod metadata;

src/filters/firewall.rs

@@ -0,0 +1,215 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+crate::include_proto!("quilkin.extensions.filters.firewall.v1alpha1");
+
+use self::quilkin::extensions::filters::firewall::v1alpha1::Firewall as ProtoConfig;
+use crate::filters::firewall::config::{Action, Config, Rule};
+use crate::filters::firewall::metrics::Metrics;
+use crate::filters::{
+    CreateFilterArgs, DynFilterFactory, Error, Filter, FilterFactory, FilterInstance, ReadContext,
+    ReadResponse, WriteContext, WriteResponse,
+};
+use slog::{debug, o, Logger};
+
+mod config;
+mod metrics;
+
+pub const NAME: &str = "quilkin.extensions.filters.compress.v1alpha1.Firewall";
+
+pub fn factory(base: &Logger) -> DynFilterFactory {
+    Box::from(FirewallFactory::new(base))
+}
+
+struct FirewallFactory {
+    log: Logger,
+}
+
+impl FirewallFactory {
+    pub fn new(base: &Logger) -> Self {
+        Self { log: base.clone() }
+    }
+}
+
+impl FilterFactory for FirewallFactory {
+    fn name(&self) -> &'static str {
+        NAME
+    }
+
+    fn create_filter(&self, args: CreateFilterArgs) -> Result<FilterInstance, Error> {
+        let (config_json, config) = self
+            .require_config(args.config)?
+            .deserialize::<Config, ProtoConfig>(self.name())?;
+
+        let filter = Firewall::new(&self.log, config, Metrics::new(&args.metrics_registry)?);
+        Ok(FilterInstance::new(
+            config_json,
+            Box::new(filter) as Box<dyn Filter>,
+        ))
+    }
+}
+
+struct Firewall {
+    log: Logger,
+    metrics: Metrics,
+    on_read: Vec<Rule>,
+    on_write: Vec<Rule>,
+}
+
+impl Firewall {
+    fn new(base: &Logger, config: Config, metrics: Metrics) -> Self {
+        Self {
+            log: base.new(o!("source" => "extensions::Firewall")),
+            metrics,
+            on_read: config.on_read,
+            on_write: config.on_write,
+        }
+    }
+}
+
+impl Filter for Firewall {
+    fn read(&self, ctx: ReadContext) -> Option<ReadResponse> {
+        for rule in &self.on_read {
+            if rule.contains(ctx.from) {
+                return match rule.action {
+                    Action::Allow => {
+                        debug!(self.log, "Allow"; "event" => "read", "from" =>  ctx.from.to_string());
+                        self.metrics.packets_allowed_on_read.inc();
+                        Some(ctx.into())
+                    }
+                    Action::Deny => {
+                        debug!(self.log, "Deny"; "event" => "read", "from" => ctx.from );
+                        self.metrics.packets_denied_on_read.inc();
+                        None
+                    }
+                };
+            }
+        }
+
+        debug!(self.log, "default: Deny"; "event" => "read", "from" => ctx.from.to_string());
+        self.metrics.packets_denied_on_read.inc();
+        None
+    }
+
+    fn write(&self, ctx: WriteContext) -> Option<WriteResponse> {
+        for rule in &self.on_write {
+            if rule.contains(ctx.from) {
+                return match rule.action {
+                    Action::Allow => {
+                        debug!(self.log, "Allow"; "event" => "write", "from" =>  ctx.from.to_string());
+                        self.metrics.packets_allowed_on_write.inc();
+                        Some(ctx.into())
+                    }
+                    Action::Deny => {
+                        debug!(self.log, "Deny"; "event" => "write", "from" => ctx.from );
+                        self.metrics.packets_denied_on_write.inc();
+                        None
+                    }
+                };
+            }
+        }
+
+        debug!(self.log, "default: Deny"; "event" => "write", "from" => ctx.from.to_string());
+        self.metrics.packets_denied_on_write.inc();
+        None
+    }
+}
+#[cfg(test)]
+mod tests {
+    use crate::endpoint::{Endpoint, Endpoints, UpstreamEndpoints};
+    use crate::filters::firewall::config::PortRange;
+    use crate::test_utils::logger;
+    use prometheus::Registry;
+
+    use super::*;
+
+    #[test]
+    fn read() {
+        let firewall = Firewall {
+            log: logger(),
+            metrics: Metrics::new(&Registry::default()).unwrap(),
+            on_read: vec![Rule {
+                action: Action::Allow,
+                source: "192.168.75.0/24".parse().unwrap(),
+                ports: vec![PortRange { min: 10, max: 100 }],
+            }],
+            on_write: vec![],
+        };
+
+        let ctx = ReadContext::new(
+            UpstreamEndpoints::from(
+                Endpoints::new(vec![Endpoint::new("127.0.0.1:8080".parse().unwrap())]).unwrap(),
+            ),
+            "192.168.75.20:80".parse().unwrap(),
+            vec![],
+        );
+        assert!(firewall.read(ctx).is_some());
+        assert_eq!(1, firewall.metrics.packets_allowed_on_read.get());
+        assert_eq!(0, firewall.metrics.packets_denied_on_read.get());
+
+        let ctx = ReadContext::new(
+            UpstreamEndpoints::from(
+                Endpoints::new(vec![Endpoint::new("127.0.0.1:8080".parse().unwrap())]).unwrap(),
+            ),
+            "192.168.75.20:2000".parse().unwrap(),
+            vec![],
+        );
+        assert!(firewall.read(ctx).is_none());
+        assert_eq!(1, firewall.metrics.packets_allowed_on_read.get());
+        assert_eq!(1, firewall.metrics.packets_denied_on_read.get());
+
+        assert_eq!(0, firewall.metrics.packets_allowed_on_write.get());
+        assert_eq!(0, firewall.metrics.packets_denied_on_write.get());
+    }
+
+    #[test]
+    fn write() {
+        let firewall = Firewall {
+            log: logger(),
+            metrics: Metrics::new(&Registry::default()).unwrap(),
+            on_read: vec![],
+            on_write: vec![Rule {
+                action: Action::Allow,
+                source: "192.168.75.0/24".parse().unwrap(),
+                ports: vec![PortRange { min: 10, max: 100 }],
+            }],
+        };
+
+        let endpoint = Endpoint::new("127.0.0.1:80".parse().unwrap());
+        let ctx = WriteContext::new(
+            &endpoint,
+            "192.168.75.20:80".parse().unwrap(),
+            "127.0.0.1:8081".parse().unwrap(),
+            vec![],
+        );
+        assert!(firewall.write(ctx).is_some());
+        assert_eq!(1, firewall.metrics.packets_allowed_on_write.get());
+        assert_eq!(0, firewall.metrics.packets_denied_on_write.get());
+
+        let ctx = WriteContext::new(
+            &endpoint,
+            "192.168.77.20:80".parse().unwrap(),
+            "127.0.0.1:8081".parse().unwrap(),
+            vec![],
+        );
+        assert!(!firewall.write(ctx).is_some());
+        assert_eq!(1, firewall.metrics.packets_allowed_on_write.get());
+        assert_eq!(1, firewall.metrics.packets_denied_on_write.get());
+
+        assert_eq!(0, firewall.metrics.packets_allowed_on_read.get());
+        assert_eq!(0, firewall.metrics.packets_denied_on_read.get());
+    }
+}