Begin HTTP server, SNS message receiver

This commit introduces the HTTP interface for the dynamic K8s node
provisioner.  It will serve as the main communication point between the
ephemeral nodes in the cloud, sharing the keys and tokens they require
in order to join the Kubernetes cluster.

The initial functionality is simply an Amazon SNS notification receiver.
SNS notifications will be used to manage the lifecycle of the dynamic
nodes.

For now, the notification receiver handles subscription confirmation
messages by following the link provided to confirm the subscription.
All other messages are simply written to the filesystem; these will be
used to implement and test future functionality.

Cargo.toml

@@ -6,6 +6,8 @@ edition = "2021"
 [dependencies]
 base64 = "0.13.0"
 log = "0.4.17"
+reqwest = "0.11.11"
+rocket = { version = "0.5.0-rc.2", features = ["json"] }
 rsa = "0.6.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0.85"

src/error.rs

@@ -0,0 +1,12 @@
+//! Application error types
+use serde::Serialize;
+
+/// Basic type for errors that can be returned in HTTP responses
+#[derive(Debug, Serialize)]
+pub struct ApiError {
+    /// The error message
+    pub error: String,
+    /// Additional details about the message, if any
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub detail: Option<String>,
+}

src/main.rs

@@ -1,7 +1,19 @@
+mod error;
 mod model;
+mod routes;
 mod sns;
 
 #[doc(hidden)]
-fn main() {
-    println!("Hello, world!");
+#[rocket::launch]
+fn rocket() -> _ {
+    rocket::build().mount(
+        "/",
+        rocket::routes![
+            routes::health::get_health,
+            routes::sns::post_sns_notify,
+            routes::sns::get_sns_notify,
+            routes::sns::put_sns_notify,
+            routes::sns::patch_sns_notify,
+        ],
+    )
 }

src/routes/health.rs

@@ -0,0 +1,4 @@
+#[rocket::get("/")]
+pub async fn get_health() -> &'static str {
+    "UP"
+}

src/routes/mod.rs

@@ -0,0 +1,3 @@
+//! Rocket route handlers
+pub mod health;
+pub mod sns;

src/routes/sns.rs

@@ -0,0 +1,120 @@
+use rocket::http::Status;
+use rocket::response::status;
+use rocket::serde::json::Json;
+
+use crate::error::ApiError;
+use crate::model::sns::Message;
+use crate::sns::*;
+
+#[rocket::post("/sns/notify", data = "<message>")]
+pub async fn post_sns_notify(
+    message: Json<Message>,
+) -> Result<status::NoContent, status::BadRequest<Json<ApiError>>> {
+    match message.into_inner() {
+        Message::SubscriptionConfirmation(m) => handle_subscribe(m)
+            .await
+            .map_err(|e| status::BadRequest(Some(Json(e.into()))))?,
+        Message::UnsubscribeConfirmation(m) => handle_unsubscribe(m)
+            .await
+            .map_err(|e| status::BadRequest(Some(Json(e.into()))))?,
+        Message::Notification(m) => handle_notify(m)
+            .await
+            .map_err(|e| status::BadRequest(Some(Json(e.into()))))?,
+    };
+    Ok(status::NoContent)
+}
+
+#[rocket::get("/sns/notify")]
+pub async fn get_sns_notify() -> Status {
+    Status::MethodNotAllowed
+}
+
+#[rocket::put("/sns/notify")]
+pub async fn put_sns_notify() -> Status {
+    Status::MethodNotAllowed
+}
+
+#[rocket::patch("/sns/notify")]
+pub async fn patch_sns_notify() -> Status {
+    Status::MethodNotAllowed
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::rocket;
+    use rocket::local::blocking::Client;
+    use rocket::uri;
+
+    #[test]
+    fn test_sub_conf_msg() {
+        let client = Client::tracked(rocket()).unwrap();
+        let data = std::fs::read_to_string(
+            "test/data/sns/subscriptionconfirmation.json",
+        )
+        .unwrap();
+        let res = client.post(uri!(post_sns_notify)).body(&data).dispatch();
+        assert_eq!(res.status(), Status::NoContent);
+    }
+
+    #[test]
+    fn test_sub_conf_msg_invalid() {
+        let client = Client::tracked(rocket()).unwrap();
+        let res = client.post(uri!(post_sns_notify)).body("{}").dispatch();
+        assert_eq!(res.status(), Status::UnprocessableEntity);
+    }
+
+    #[test]
+    fn test_sub_conf_msg_bad() {
+        let client = Client::tracked(rocket()).unwrap();
+        let data = std::fs::read_to_string(
+            "test/data/sns/subscriptionconfirmation-bad.json",
+        )
+        .unwrap();
+        let res = client.post(uri!(post_sns_notify)).body(&data).dispatch();
+        assert_eq!(res.status(), Status::BadRequest);
+        assert_eq!(
+            res.into_string().unwrap(),
+            concat!(
+                "{",
+                r#""error":"Could not verify signature","#,
+                r#""detail":"verification error""#,
+                "}",
+            )
+        );
+    }
+
+    #[test]
+    fn test_sub_conf_msg_bad_cert_url() {
+        let client = Client::tracked(rocket()).unwrap();
+        let data = std::fs::read_to_string(
+            "test/data/sns/subscriptionconfirmation-bad-url.json",
+        )
+        .unwrap();
+        let res = client.post(uri!(post_sns_notify)).body(&data).dispatch();
+        assert_eq!(res.status(), Status::BadRequest);
+        assert_eq!(
+            res.into_string().unwrap(),
+            concat!(
+                "{",
+                r#""error":"Could not retrieve signing certificate","#,
+                r#""detail":"Unrecognized host""#,
+                "}",
+            )
+        );
+    }
+
+    #[test]
+    fn test_sub_conf_msg_wrong_method() {
+        let client = Client::tracked(rocket()).unwrap();
+
+        let res = client.get(uri!(post_sns_notify)).dispatch();
+        assert_eq!(res.status(), Status::MethodNotAllowed);
+
+        let res = client.put(uri!(post_sns_notify)).dispatch();
+        assert_eq!(res.status(), Status::MethodNotAllowed);
+
+        let res = client.patch(uri!(post_sns_notify)).dispatch();
+        assert_eq!(res.status(), Status::MethodNotAllowed);
+    }
+}

src/sns/error.rs

@@ -0,0 +1,49 @@
+//! Error types for Amazon Simple Notification Service
+use serde::Serialize;
+use std::fmt;
+
+use super::sig::SignatureError;
+use crate::error::ApiError;
+
+/// Basic type for SNS errors
+#[derive(Debug, Serialize)]
+pub enum SnsError {
+    /// An error occurred while verifying a message signature
+    SignatureError(String),
+    /// An error occurred while attempting to fetch a signing certificate
+    CertificateError(String),
+}
+
+impl fmt::Display for SnsError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::SignatureError(s) => {
+                write!(f, "Could not verify signature: {}", s)
+            }
+            Self::CertificateError(s) => {
+                write!(f, "Could not retrieve signing certificate: {}", s)
+            }
+        }
+    }
+}
+
+impl From<SignatureError> for SnsError {
+    fn from(e: SignatureError) -> Self {
+        Self::SignatureError(e.to_string())
+    }
+}
+
+impl Into<ApiError> for SnsError {
+    fn into(self) -> ApiError {
+        match self {
+            Self::SignatureError(s) => ApiError {
+                error: "Could not verify signature".into(),
+                detail: Some(s),
+            },
+            Self::CertificateError(s) => ApiError {
+                error: "Could not retrieve signing certificate".into(),
+                detail: Some(s),
+            },
+        }
+    }
+}

src/sns/mod.rs

@@ -1,2 +1,143 @@
 //! Amazon Simple Notification Service (SNS) utilities
+pub mod error;
 pub mod sig;
+
+use log::{debug, error, info};
+use reqwest::Url;
+
+use crate::model::sns::*;
+use error::SnsError;
+use sig::SignatureVerifier;
+
+/// Handle a subscription confirmation message
+///
+/// After verifying the message signature, the subscription is confirmed using
+/// the URL provided in the message.
+pub async fn handle_subscribe(
+    msg: SubscriptionConfirmationMessage,
+) -> Result<(), SnsError> {
+    verify(&msg, &msg.signing_cert_url).await?;
+    confirm_subscription(&msg).await;
+    Ok(())
+}
+
+/// Handle an unsubscribe confirmation message
+///
+/// After verifying the message signature, the message contents are written to
+/// a file for later inspection.
+pub async fn handle_unsubscribe(
+    msg: UnsubscribeConfirmationMessage,
+) -> Result<(), SnsError> {
+    verify(&msg, &msg.signing_cert_url).await?;
+    Ok(())
+}
+
+/// Handle an notification message
+///
+/// After verifying the message signature, the message contents are written to
+/// a file for later inspection.
+pub async fn handle_notify(msg: NotificationMessage) -> Result<(), SnsError> {
+    verify(&msg, &msg.signing_cert_url).await?;
+    Ok(())
+}
+
+/// Verify the signature of an SNS message
+///
+/// This function verifies the signature of the specified SNS message, using
+/// the certificate found at the URL provided in the message.  If the signature
+/// verification fails, [`SnsError`] is returned.
+async fn verify<M: SignatureVerifier>(
+    msg: &M,
+    cert_url: &str,
+) -> Result<(), SnsError> {
+    let cert = fetch_cert(cert_url).await?;
+    match msg.verify(&cert) {
+        Ok(_) => {
+            info!("Successfully verified message signature");
+            Ok(())
+        }
+        Err(e) => {
+            error!("Could not verify message signature: {}", e);
+            Err(e.into())
+        }
+    }
+}
+
+/// Fetch the certificate from the given URL
+///
+/// This function returns the contents of the resource at the specified URL,
+/// which is presumably a PEM-encoded certificate file.
+///
+/// Some safety checks are performed before making the remote request:
+///
+/// 1. Only https:// URLs are supported
+/// 2. The host name portion of the URL must end with `.amazonaws.com`
+///
+/// If these conditions are not met, [`SnsError`] is returned.
+///
+/// If an error occurs while attempting to fetch the resource, [`SnsError`] is
+/// returned.
+async fn fetch_cert<U: AsRef<str>>(url: U) -> Result<String, SnsError> {
+    let url = match Url::parse(url.as_ref()) {
+        Ok(u) => u,
+        Err(e) => {
+            error!("Invalid certificate URL: {}", e);
+            return Err(SnsError::CertificateError(e.to_string()));
+        }
+    };
+    if url.scheme() != "https" {
+        return Err(SnsError::CertificateError(
+            "Unsupported URL scheme".into(),
+        ));
+    }
+    match url.host_str() {
+        Some(s) => {
+            if !s.ends_with(".amazonaws.com") {
+                return Err(SnsError::CertificateError(
+                    "Unrecognized host".into(),
+                ));
+            }
+        }
+        None => {
+            return Err(SnsError::CertificateError(
+                "Invalid certificate URL: no host".into(),
+            ));
+        }
+    };
+    info!("Fetching signing certificate from {}", url);
+    match reqwest::get(url).await {
+        Ok(res) => match res.text().await {
+            Ok(v) => Ok(v),
+            Err(e) => Err(SnsError::CertificateError(e.to_string())),
+        },
+        Err(e) => Err(SnsError::CertificateError(e.to_string())),
+    }
+}
+
+/// Confirm an SNS topic subscription
+///
+/// This function confirms the SNS subscription specified by the provided
+/// subscription confirmation message by making an HTTP GET request to the URL
+/// listed in the `subscribe_url` field of the message.
+///
+/// TODO retry requests that failed because of e.g. network errors
+async fn confirm_subscription(msg: &SubscriptionConfirmationMessage) {
+    debug!("Confirming subscription to topic {}", &msg.topic_arn);
+    #[cfg(not(test))]
+    match reqwest::get(&msg.subscribe_url).await {
+        Ok(res) => {
+            debug!("Got HTTP response: {}", res.status());
+            match res.error_for_status() {
+                Ok(_) => {
+                    info!("Successfully confirmed subscription");
+                }
+                Err(e) => {
+                    error!("Failed to confirm subscription: {}", e);
+                }
+            };
+        }
+        Err(e) => {
+            error!("Failed to confirm subscription: {}", e);
+        }
+    }
+}

test/data/sns/subscriptionconfirmation-bad-url.json

@@ -0,0 +1,12 @@
+{
+  "Type" : "SubscriptionConfirmation",
+  "MessageId" : "5506049a-38a4-4af5-b1e2-e1c4de84c009",
+  "Token" : "2336412f37fb687f5d51e6e2425dacbba7f97a62baf67e3ffa0b3da7071d369eb88d52257631cc33d52b277c7f215a6160215838db9eb1332ba95cf0418bbb0cf7037c4ab353be5f46441cb837c0cc6877623cbde0238813d4a64fce01ef705d3b8f1c3bdc86aa2f183e960ec112c984",
+  "TopicArn" : "arn:aws:sns:us-east-2:566967686773:dchtest1",
+  "Message" : "You have chosen to subscribe to the topic arn:aws:sns:us-east-2:566967686773:dchtest1.\nTo confirm the subscription, visit the SubscribeURL included in this message.",
+  "SubscribeURL" : "https://sns.us-east-2.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:us-east-2:566967686773:dchtest1&Token=2336412f37fb687f5d51e6e2425dacbba7f97a62baf67e3ffa0b3da7071d369eb88d52257631cc33d52b277c7f215a6160215838db9eb1332ba95cf0418bbb0cf7037c4ab353be5f46441cb837c0cc6877623cbde0238813d4a64fce01ef705d3b8f1c3bdc86aa2f183e960ec112c984",
+  "Timestamp" : "2022-08-31T21:17:09.115Z",
+  "SignatureVersion" : "1",
+  "Signature" : "uPd8FxfVtenWUtcOF7iBNI6YQ6uHqGMlbc1U8/KesvnA9p2/3XEK6MJhaLcgOmLfpBhbljRFcEWPK2xBWsagX6uUk5d5mCQRkE/N+IfezLg/Q8vwTUw3rhVXZge7gl8NCCpqia1xQoSo8PbMkfAb9sw6YoytJopaPrRzvxHRTdUmyTVw2vrl8yxHD2OTRVYKpKv6Pg1Pf0VXdZq07xMRaqF2zTFK+LNYBJ74wrJRg1zLe6xfscwQytUpKf8vFHyPhP2QZWxi/mJ6YIowvXh0cElGyjox3jLxEoQ+K0jARrQSAhOBufHzd35BOAm0b7JES/YMYE58NxYHkXmoX3u1Cg==",
+  "SigningCertURL" : "https://sns.us-east-2.amazonaws.com.therealamazonaws.co/SimpleNotificationService-56e67fcb41f6fec09b0196692625d385.pem"
+}

test/data/sns/subscriptionconfirmation-bad.json

@@ -0,0 +1,12 @@
+{
+  "Type" : "SubscriptionConfirmation",
+  "MessageId" : "5506049a-38a4-4af5-b1e2-e1c4de84c009",
+  "Token" : "2336412f37fb687f5d51e6e2425dacbba7f97a62baf67e3ffa0b3da7071d369eb88d52257631cc33d52b277c7f215a6160215838db9eb1332ba95cf0418bbb0cf7037c4ab353be5f46441cb837c0cc6877623cbde0238813d4a64fce01ef705d3b8f1c3bdc86aa2f183e960ec112c948",
+  "TopicArn" : "arn:aws:sns:us-east-2:566967686773:dchtest1",
+  "Message" : "You have chosen to subscribe to the topic arn:aws:sns:us-east-2:566967686773:dchtest1.\nTo confirm the subscription, visit the SubscribeURL included in this message.",
+  "SubscribeURL" : "https://sns.us-east-2.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:us-east-2:566967686773:dchtest1&Token=2336412f37fb687f5d51e6e2425dacbba7f97a62baf67e3ffa0b3da7071d369eb88d52257631cc33d52b277c7f215a6160215838db9eb1332ba95cf0418bbb0cf7037c4ab353be5f46441cb837c0cc6877623cbde0238813d4a64fce01ef705d3b8f1c3bdc86aa2f183e960ec112c984",
+  "Timestamp" : "2022-08-31T21:17:09.115Z",
+  "SignatureVersion" : "1",
+  "Signature" : "uPd8FxfVtenWUtcOF7iBNI6YQ6uHqGMlbc1U8/KesvnA9p2/3XEK6MJhaLcgOmLfpBhbljRFcEWPK2xBWsagX6uUk5d5mCQRkE/N+IfezLg/Q8vwTUw3rhVXZge7gl8NCCpqia1xQoSo8PbMkfAb9sw6YoytJopaPrRzvxHRTdUmyTVw2vrl8yxHD2OTRVYKpKv6Pg1Pf0VXdZq07xMRaqF2zTFK+LNYBJ74wrJRg1zLe6xfscwQytUpKf8vFHyPhP2QZWxi/mJ6YIowvXh0cElGyjox3jLxEoQ+K0jARrQSAhOBufHzd35BOAm0b7JES/YMYE58NxYHkXmoX3u1Cg==",
+  "SigningCertURL" : "https://sns.us-east-2.amazonaws.com/SimpleNotificationService-56e67fcb41f6fec09b0196692625d385.pem"
+}

test/data/sns/subscriptionconfirmation.json

@@ -0,0 +1,12 @@
+{
+  "Type" : "SubscriptionConfirmation",
+  "MessageId" : "5506049a-38a4-4af5-b1e2-e1c4de84c009",
+  "Token" : "2336412f37fb687f5d51e6e2425dacbba7f97a62baf67e3ffa0b3da7071d369eb88d52257631cc33d52b277c7f215a6160215838db9eb1332ba95cf0418bbb0cf7037c4ab353be5f46441cb837c0cc6877623cbde0238813d4a64fce01ef705d3b8f1c3bdc86aa2f183e960ec112c984",
+  "TopicArn" : "arn:aws:sns:us-east-2:566967686773:dchtest1",
+  "Message" : "You have chosen to subscribe to the topic arn:aws:sns:us-east-2:566967686773:dchtest1.\nTo confirm the subscription, visit the SubscribeURL included in this message.",
+  "SubscribeURL" : "https://sns.us-east-2.amazonaws.com/?Action=ConfirmSubscription&TopicArn=arn:aws:sns:us-east-2:566967686773:dchtest1&Token=2336412f37fb687f5d51e6e2425dacbba7f97a62baf67e3ffa0b3da7071d369eb88d52257631cc33d52b277c7f215a6160215838db9eb1332ba95cf0418bbb0cf7037c4ab353be5f46441cb837c0cc6877623cbde0238813d4a64fce01ef705d3b8f1c3bdc86aa2f183e960ec112c984",
+  "Timestamp" : "2022-08-31T21:17:09.115Z",
+  "SignatureVersion" : "1",
+  "Signature" : "uPd8FxfVtenWUtcOF7iBNI6YQ6uHqGMlbc1U8/KesvnA9p2/3XEK6MJhaLcgOmLfpBhbljRFcEWPK2xBWsagX6uUk5d5mCQRkE/N+IfezLg/Q8vwTUw3rhVXZge7gl8NCCpqia1xQoSo8PbMkfAb9sw6YoytJopaPrRzvxHRTdUmyTVw2vrl8yxHD2OTRVYKpKv6Pg1Pf0VXdZq07xMRaqF2zTFK+LNYBJ74wrJRg1zLe6xfscwQytUpKf8vFHyPhP2QZWxi/mJ6YIowvXh0cElGyjox3jLxEoQ+K0jARrQSAhOBufHzd35BOAm0b7JES/YMYE58NxYHkXmoX3u1Cg==",
+  "SigningCertURL" : "https://sns.us-east-2.amazonaws.com/SimpleNotificationService-56e67fcb41f6fec09b0196692625d385.pem"
+}