Use etcd encryption if enabled

Signed-off-by: Michail Resvanis <[email protected]>

src/k8s_etcd.rs

@@ -1,4 +1,5 @@
 use crate::cluster_crypto::locations::K8sResourceLocation;
+use crate::encrypt::ResourceTransformers;
 use crate::etcd_encoding;
 use anyhow::{bail, ensure, Context, Result};
 use etcd_client::{Client as EtcdClient, GetOptions};
@@ -24,16 +25,24 @@ pub(crate) struct InMemoryK8sEtcd {
     etcd_keyvalue_hashmap: Mutex<HashMap<String, Vec<u8>>>,
     edited: Mutex<HashMap<String, Vec<u8>>>,
     deleted_keys: Mutex<HashSet<String>>,
+    decrypt_resource_transformers: Option<ResourceTransformers>,
+    encrypt_resource_transformers: Option<ResourceTransformers>,
 }
 
 impl InMemoryK8sEtcd {
     /// Pass a None etcd_client to disable actual etcd access (dummy mode, empty key list).
-    pub(crate) fn new(etcd_client: Option<EtcdClient>) -> Self {
+    pub(crate) fn new(
+        etcd_client: Option<EtcdClient>,
+        decrypt_resource_transformers: Option<ResourceTransformers>,
+        encrypt_resource_transformers: Option<ResourceTransformers>,
+    ) -> Self {
         Self {
             etcd_client: etcd_client.map(Arc::new),
             etcd_keyvalue_hashmap: Mutex::new(HashMap::new()),
             deleted_keys: Mutex::new(HashSet::new()),
             edited: Mutex::new(HashMap::new()),
+            decrypt_resource_transformers,
+            encrypt_resource_transformers,
         }
     }
 
@@ -98,7 +107,21 @@ impl InMemoryK8sEtcd {
             let key = key.clone();
             let value = value.clone();
             let etcd_client = Arc::clone(etcd_client);
-            let value = etcd_encoding::encode(value.as_slice()).await.context("encoding value")?;
+
+            let mut value = etcd_encoding::encode(value.as_slice()).await.context("encoding value")?;
+
+            if let Some(resource_transformers) = &self.encrypt_resource_transformers {
+                // https://github.com/kubernetes/apiserver/blob/master/pkg/storage/value/transformer.go#L172
+                if let Some(transformers) = resource_transformers
+                    .resource_to_prefix_transformers
+                    .get(&resource_from_key(key.to_string()))
+                {
+                    value = transformers[0]
+                        .encrypt(key.to_string(), value)
+                        .await
+                        .context("encrypting etcd value")?;
+                }
+            }
 
             etcd_client
                 .kv_client()
@@ -134,7 +157,27 @@ impl InMemoryK8sEtcd {
         if let Some(value) = get_result.kvs().first() {
             let raw_etcd_value = value.value();
 
-            let decoded_value = etcd_encoding::decode(raw_etcd_value).await.context("decoding value")?;
+            let mut decoded_value = etcd_encoding::decode(raw_etcd_value).await.context("decoding value")?;
+
+            if let Some(resource_transformers) = &self.decrypt_resource_transformers {
+                // https://github.com/kubernetes/apiserver/blob/master/pkg/storage/value/transformer.go#L110
+                if let Some(transformers) = resource_transformers
+                    .resource_to_prefix_transformers
+                    .get(&resource_from_key(key.to_string()))
+                {
+                    for transformer in transformers {
+                        if raw_etcd_value.to_vec().starts_with(transformer.get_prefix().as_bytes()) {
+                            let plaintext_value = transformer
+                                .decrypt(key.to_string(), raw_etcd_value.to_vec())
+                                .await
+                                .context("decrypting etcd value")?;
+                            decoded_value = etcd_encoding::decode(&plaintext_value).await.context("decoding value")?;
+                            break;
+                        }
+                    }
+                }
+            }
+
             self.etcd_keyvalue_hashmap
                 .lock()
                 .await
@@ -214,6 +257,10 @@ impl InMemoryK8sEtcd {
     }
 }
 
+fn resource_from_key(key: String) -> String {
+    key.split('/').collect::<Vec<_>>()[2].to_string()
+}
+
 fn is_too_many_requests_error(delete_response: &std::prelude::v1::Result<etcd_client::DeleteResponse, etcd_client::Error>) -> bool {
     match delete_response {
         Ok(_) => false,