Server: Worker refactor (#704)

* Add webhook-http-client, which is based on hyper, enabling several
  critical improvements:
  * Default filtering of endpoints that resolve to private
    address space, with ability to add CIDR exceptions as needed
  * Case-sensitive HTTP1 headers -- although HTTP spec states that
    headers are not case-sensitive, in practice many widely
    deployed web-servers expect to support case-sensitive headers.
    This requires a fork of hyper, which currently has but does
    not expose support for case-sensitive headers.
  * Assurance that the `host` header appears first in outgoing HTTP requests.
    Although HTTP spec does not prescribe an order for headers, in
    our experience, many web-servers will fail if the `host` header does
    not appear first in request headers. We ensure this by manually
    adding the header ourselves rather than relying on hyper to do so.
* Use `openssl` instead of `rusttls` for outbound webhooks. In our experience,
  many widely deployed webservers use "weak" ciphers that rusttls refuses to
  support, leaving us with no choice but to rely on openssl libraries.
* Support for soft-limiting of concurrent worker tasks. Unbounded concurrent worker
  tasks can easily overwhelm a system. This sets a default of 500 tasks with support
  for configuring other limits or setting unlimited (`0`) concurrent tasks
* Refactor message dispatch code for clarity. Dispatch code is now broken into
  separate methods: `prepare_dispatch`, `make_http_call`,
  `handle_successful_dispatch`, and `handle_failed_dispatch` with the dispatch itself
  having different types depending on its stage in the dispatch process.
  Other function names have also been changed for improved clarity.
* Prevent worker from shutting down when there are active tasks.
* Avoid multiple DB calls for message-destination insert/retrieval

server/Cargo.toml

@@ -1,2 +1,5 @@
 [workspace]
 members = ["svix-server", "svix-server_derive"]
+
+[patch.crates-io]
+hyper = { git = "https://github.com/svix/hyper/", rev = "b901ca7c" }

server/run-tests.sh

@@ -8,6 +8,7 @@ TEST_COMMAND="cargo test --all --all-features --all-targets"
 export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/postgres"
 export SVIX_JWT_SECRET="test value"
 export SVIX_LOG_LEVEL="info"
+export SVIX_WHITELIST_SUBNETS="[127.0.0.1/32]"
 
 echo "*********** RUN 1 ***********"
 SVIX_QUEUE_TYPE="redis" \

server/svix-server/Cargo.toml

@@ -23,6 +23,8 @@ clap = { version = "3.2.1", features = ["derive"] }
 axum = { version = "0.6.1", features = ["headers"] }
 base64 = "0.13.0"
 hyper = { version = "0.14.16", features = ["full"] }
+hyper-openssl = "0.9.2"
+openssl = "0.10.45"
 tokio = { version = "1.23.1", features = ["full"] }
 tower = "0.4.11"
 tower-http = { version = "0.3.4", features = ["trace", "cors", "request-id"] }
@@ -66,6 +68,11 @@ jsonschema = "0.16.1"
 aide = { version = "0.9.0", features = ["axum", "redoc", "macros"] }
 schemars = { version = "0.8.11", features = ["chrono", "url"] }
 indexmap = "1.9.2"
+trust-dns-resolver = "0.22.0"
+ipnet = { version = "2.5", features = ["serde"] }
+urlencoding = "2.1.2"
+strum_macros = "0.24"
+strum = { version = "0.24", features = ["derive"] }
 
 [dev-dependencies]
 anyhow = "1.0.56"

server/svix-server/config.default.toml

@@ -94,3 +94,11 @@ api_enabled = true
 
 # Should this instance run the message worker
 worker_enabled = true
+
+# Subnets to whitelist for outbound webhooks. Note that allowing endpoints in private IP space
+# is a security risk and should only be allowed if you are using the service internally or for
+# testing purposes. Should be specified in CIDR notation, e.g., `[127.0.0.1/32, 172.17.0.0/16, 192.168.0.0/16]`
+# whitelist_subnets = []
+
+# Maximum number of concurrent worker tasks to spawn (0 is unlimited)
+worker_max_tasks = 500

server/svix-server/src/cfg.rs

@@ -7,6 +7,7 @@ use figment::{
     providers::{Env, Format, Toml},
     Figment,
 };
+use ipnet::IpNet;
 use std::time::Duration;
 
 use crate::{core::cryptography::Encryption, core::security::Keys, error::Result};
@@ -173,6 +174,14 @@ pub struct ConfigurationInner {
     /// Should this instance run the message worker
     pub worker_enabled: bool,
 
+    /// Subnets to whitelist for outbound webhooks. Note that allowing endpoints in private IP space
+    /// is a security risk and should only be allowed if you are using the service internally or for
+    /// testing purposes. Should be specified in CIDR notation, e.g., `[127.0.0.1/32, 172.17.0.0/16, 192.168.0.0/16]`
+    pub whitelist_subnets: Option<Arc<Vec<IpNet>>>,
+
+    /// Maximum number of concurrent worker tasks to spawn (0 is unlimited)
+    pub worker_max_tasks: u16,
+
     #[serde(flatten)]
     pub internal: InternalConfig,
 }

server/svix-server/src/core/message_app.rs

@@ -67,7 +67,7 @@ impl CreateMessageApp {
     /// exists or from PostgreSQL otherwise. If the RedisCache is Some, but does not contain the
     /// requisite information, fetch it from PostgreSQL and insert the data into the cache.
     pub async fn layered_fetch(
-        cache: Cache,
+        cache: &Cache,
         pg: &DatabaseConnection,
         app: Option<application::Model>,
         org_id: OrganizationId,

server/svix-server/src/core/mod.rs

@@ -11,6 +11,7 @@ pub mod permissions;
 pub mod run_with_retries;
 pub mod security;
 pub mod types;
+pub mod webhook_http_client;
 
 #[cfg(test)]
 mod tests {