add durable sse subscriptions

Cargo.toml

@@ -18,6 +18,7 @@ rand = "0.9"
 serde = "1"
 serde_json = "1"
 sha1 = "0.10"
+thiserror = "2"
 time = "0.3"
 
 [lints.rust]

src/admin.rs

@@ -16,8 +16,8 @@ fn text_response(status: StatusCode, text: &str) -> Response {
     Response::from_status(status).with_body_text_plain(&format!("{text}\n"))
 }
 
-pub fn post_keys(req: Request) -> Response {
-    if !req.fastly_key_is_valid() {
+pub fn post_keys(fastly_authed: bool, _req: Request) -> Response {
+    if !fastly_authed {
         return text_response(
             StatusCode::UNAUTHORIZED,
             "Fastly-Key header invalid or not specified",

src/auth.rs

@@ -77,6 +77,26 @@ fn validate_token(token: &str, key: &[u8]) -> Result<Capabilities, TokenError> {
     Ok(caps)
 }
 
+pub fn create_token(
+    readable: Vec<String>,
+    writable: Vec<String>,
+    key_id: &str,
+    key: &[u8],
+) -> String {
+    let key = HS256Key::from_bytes(key).with_key_id(key_id);
+
+    let claims = Claims::with_custom_claims(
+        CustomClaims {
+            x_fastly_read: readable,
+            x_fastly_write: writable,
+        },
+        Duration::from_hours(1),
+    );
+
+    key.authenticate(claims)
+        .expect("token creation should always succeed")
+}
+
 #[derive(Debug)]
 pub enum AuthorizationError {
     Token(TokenError),
@@ -92,7 +112,11 @@ impl From<TokenError> for AuthorizationError {
 }
 
 pub trait Authorizor {
-    fn validate_token(&self, token: &str) -> Result<Capabilities, AuthorizationError>;
+    fn validate_token(
+        &self,
+        token: &str,
+        internal_key: Option<&[u8]>,
+    ) -> Result<Capabilities, AuthorizationError>;
 }
 
 pub struct KVStoreAuthorizor {
@@ -108,7 +132,11 @@ impl KVStoreAuthorizor {
 }
 
 impl Authorizor for KVStoreAuthorizor {
-    fn validate_token(&self, token: &str) -> Result<Capabilities, AuthorizationError> {
+    fn validate_token(
+        &self,
+        token: &str,
+        internal_key: Option<&[u8]>,
+    ) -> Result<Capabilities, AuthorizationError> {
         let Ok(metadata) = Token::decode_metadata(token) else {
             return Err(AuthorizationError::Token(TokenError::Invalid));
         };
@@ -117,28 +145,40 @@ impl Authorizor for KVStoreAuthorizor {
             return Err(AuthorizationError::Token(TokenError::NoKeyId));
         };
 
-        let store = match kv_store::KVStore::open(&self.store_name) {
-            Ok(Some(store)) => store,
-            Ok(None) => return Err(AuthorizationError::StoreNotFound),
-            Err(_) => return Err(AuthorizationError::StoreError),
-        };
-
-        let v = match store.lookup(key_id) {
-            Ok(mut lookup) => lookup.take_body_bytes(),
-            Err(kv_store::KVStoreError::ItemNotFound) => {
-                return Err(AuthorizationError::KeyNotFound)
+        let key = if key_id == "internal" {
+            let Some(internal_key) = internal_key else {
+                return Err(AuthorizationError::KeyNotFound);
+            };
+
+            internal_key.to_vec()
+        } else {
+            let store = match kv_store::KVStore::open(&self.store_name) {
+                Ok(Some(store)) => store,
+                Ok(None) => return Err(AuthorizationError::StoreNotFound),
+                Err(_) => return Err(AuthorizationError::StoreError),
+            };
+
+            match store.lookup(key_id) {
+                Ok(mut lookup) => lookup.take_body_bytes(),
+                Err(kv_store::KVStoreError::ItemNotFound) => {
+                    return Err(AuthorizationError::KeyNotFound)
+                }
+                Err(_) => return Err(AuthorizationError::StoreError),
             }
-            Err(_) => return Err(AuthorizationError::StoreError),
         };
 
-        Ok(validate_token(token, &v)?)
+        Ok(validate_token(token, &key)?)
     }
 }
 
 pub struct TestAuthorizor;
 
 impl Authorizor for TestAuthorizor {
-    fn validate_token(&self, token: &str) -> Result<Capabilities, AuthorizationError> {
+    fn validate_token(
+        &self,
+        token: &str,
+        _internal_key: Option<&[u8]>,
+    ) -> Result<Capabilities, AuthorizationError> {
         Ok(validate_token(token, b"notasecret")?)
     }
 }
@@ -160,7 +200,7 @@ mod tests {
         let key = HS256Key::from_bytes(b"notasecret");
         let token = key.authenticate(claims).unwrap();
 
-        let caps = TestAuthorizor.validate_token(&token).unwrap();
+        let caps = TestAuthorizor.validate_token(&token, None).unwrap();
         assert!(caps.can_subscribe("readable"));
         assert!(!caps.can_subscribe("foo"));
         assert!(caps.can_publish("writable"));

src/config.rs

@@ -7,6 +7,7 @@ pub struct Config {
     pub mqtt_enabled: bool,
     pub admin_enabled: bool,
     pub publish_token: String,
+    pub internal_key: Vec<u8>,
 }
 
 impl Default for Config {
@@ -17,6 +18,7 @@ impl Default for Config {
             mqtt_enabled: true,
             admin_enabled: true,
             publish_token: String::new(),
+            internal_key: Vec::new(),
         }
     }
 }
@@ -101,6 +103,12 @@ impl Source for ConfigAndSecretStoreSource {
                 Ok(None) => {}
                 Err(_) => return Err(ConfigError::StoreError),
             }
+
+            match store.try_get("internal-key") {
+                Ok(Some(v)) => config.internal_key = v.plaintext().to_vec(),
+                Ok(None) => {}
+                Err(_) => return Err(ConfigError::StoreError),
+            }
         }
 
         Ok(config)