Signing key integration (#6)

* Request signing key support

Implement support for secret-sourced keys and CSI secrets store sourced keys

* Plumb the volume actually into the statefulset

* Fix env var name

* Add to helm rbac

* Improve commenting

charts/restate-operator-helm/templates/rbac.yaml

@@ -48,6 +48,7 @@ rules:
       - persistentvolumeclaims
       - pods
       - securitygrouppolicies
+      - secretproviderclasses
     verbs:
       - get
       - list
@@ -59,18 +60,21 @@ rules:
       - apps
       - networking.k8s.io
       - vpcresources.k8s.aws
+      - secrets-store.csi.x-k8s.io
   - resources:
       - statefulsets
       - networkpolicies
       - pods
       - securitygrouppolicies
+      - secretproviderclasses
     verbs:
       - delete
     apiGroups:
       - ''
       - apps
       - networking.k8s.io
       - vpcresources.k8s.aws
+      - secrets-store.csi.x-k8s.io
   {{- if .Values.awsPodIdentityAssociationCluster }}
   - resources:
       - podidentityassociations

crd/RestateCluster.pkl

@@ -88,8 +88,13 @@ class Security {
   /// Set any of these to [] to allow all traffic - not recommended.
   networkPeers: NetworkPeers?
 
+  /// If set, configure the use of a private key to sign outbound requests from this cluster
+  requestSigningPrivateKey: RequestSigningPrivateKey?
+
+  /// Annotations to set on the ServiceAccount created for Restate
   serviceAccountAnnotations: Mapping<String, String>?
 
+  /// Annotations to set on the Service created for Restate
   serviceAnnotations: Mapping<String, String>?
 }
 
@@ -103,6 +108,39 @@ class NetworkPeers {
   metrics: Listing<NetworkPolicy.NetworkPolicyPeer>?
 }
 
+/// If set, configure the use of a private key to sign outbound requests from this cluster
+class RequestSigningPrivateKey {
+  /// A Kubernetes Secret source for the private key
+  secret: Secret?
+
+  /// A CSI secret provider source for the private key; will create a SecretProviderClass.
+  secretProvider: SecretProvider?
+
+  /// The version of Restate request signing that the key is for; currently only "v1" accepted.
+  version: String
+}
+
+/// A Kubernetes Secret source for the private key
+class Secret {
+  /// The key of the secret to select from. Must be a valid secret key.
+  key: String
+
+  /// Name of the secret.
+  secretName: String
+}
+
+/// A CSI secret provider source for the private key; will create a SecretProviderClass.
+class SecretProvider {
+  /// Configuration for specific provider
+  parameters: Mapping<String, String>?
+
+  /// The path of the private key relative to the root of the mounted volume
+  path: String
+
+  /// Configuration for provider name
+  provider: String?
+}
+
 /// Storage configuration
 class Storage {
   /// storageClassName is the name of the StorageClass required by the claim. More info:

crd/crd.yaml

@@ -606,14 +606,60 @@ spec:
                         type: array
                         x-kubernetes-list-type: atomic
                     type: object
+                  requestSigningPrivateKey:
+                    description: If set, configure the use of a private key to sign outbound requests from this cluster
+                    nullable: true
+                    properties:
+                      secret:
+                        description: A Kubernetes Secret source for the private key
+                        nullable: true
+                        properties:
+                          key:
+                            description: The key of the secret to select from.  Must be a valid secret key.
+                            type: string
+                          secretName:
+                            description: Name of the secret.
+                            type: string
+                        required:
+                        - key
+                        - secretName
+                        type: object
+                      secretProvider:
+                        description: A CSI secret provider source for the private key; will create a SecretProviderClass.
+                        nullable: true
+                        properties:
+                          parameters:
+                            additionalProperties:
+                              type: string
+                            description: Configuration for specific provider
+                            nullable: true
+                            type: object
+                          path:
+                            description: The path of the private key relative to the root of the mounted volume
+                            type: string
+                          provider:
+                            description: Configuration for provider name
+                            nullable: true
+                            type: string
+                        required:
+                        - path
+                        type: object
+                      version:
+                        description: The version of Restate request signing that the key is for; currently only "v1" accepted.
+                        type: string
+                    required:
+                    - version
+                    type: object
                   serviceAccountAnnotations:
                     additionalProperties:
                       type: string
+                    description: Annotations to set on the ServiceAccount created for Restate
                     nullable: true
                     type: object
                   serviceAnnotations:
                     additionalProperties:
                       type: string
+                    description: Annotations to set on the Service created for Restate
                     nullable: true
                     type: object
                 type: object

src/controller.rs

@@ -2,6 +2,7 @@ use std::borrow::Cow;
 use std::collections::hash_map::DefaultHasher;
 use std::collections::BTreeMap;
 use std::hash::{Hash, Hasher};
+use std::path::PathBuf;
 use std::sync::Arc;
 
 use chrono::{DateTime, Utc};
@@ -40,6 +41,8 @@ use crate::podidentityassociations::PodIdentityAssociation;
 use crate::reconcilers::compute::reconcile_compute;
 use crate::reconcilers::network_policies::reconcile_network_policies;
 use crate::reconcilers::object_meta;
+use crate::reconcilers::signing_key::reconcile_signing_key;
+use crate::secretproviderclasses::SecretProviderClass;
 use crate::securitygrouppolicies::SecurityGroupPolicy;
 use crate::{telemetry, Error, Metrics, Result};
 
@@ -198,7 +201,9 @@ fn env_schema(g: &mut schemars::gen::SchemaGenerator) -> Schema {
 #[derive(Deserialize, Serialize, Clone, Default, Debug, JsonSchema)]
 #[serde(rename_all = "camelCase")]
 pub struct RestateClusterSecurity {
+    /// Annotations to set on the Service created for Restate
     pub service_annotations: Option<BTreeMap<String, String>>,
+    /// Annotations to set on the ServiceAccount created for Restate
     pub service_account_annotations: Option<BTreeMap<String, String>>,
     /// If set, create an AWS PodIdentityAssociation using the ACK CRD in order to give the Restate pod access to this role and
     /// allow the cluster to reach the Pod Identity agent.
@@ -212,6 +217,8 @@ pub struct RestateClusterSecurity {
     /// of allowing public internet access and cluster DNS access. Providing a single empty rule will allow
     /// all outbound traffic - not recommended
     pub network_egress_rules: Option<Vec<NetworkPolicyEgressRule>>,
+    /// If set, configure the use of a private key to sign outbound requests from this cluster
+    pub request_signing_private_key: Option<RequestSigningPrivateKey>,
 }
 
 #[derive(Deserialize, Serialize, Clone, Default, Debug, JsonSchema)]
@@ -284,6 +291,38 @@ fn network_ports_schema(_: &mut schemars::gen::SchemaGenerator) -> Schema {
         .unwrap()
 }
 
+/// Configuration for request signing private keys. Exactly one source of 'secret', 'secretProvider'
+/// must be provided.
+#[derive(Deserialize, Serialize, Clone, Default, Debug, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct RequestSigningPrivateKey {
+    /// The version of Restate request signing that the key is for; currently only "v1" accepted.
+    pub version: String,
+    /// A Kubernetes Secret source for the private key
+    pub secret: Option<SecretSigningKeySource>,
+    /// A CSI secret provider source for the private key; will create a SecretProviderClass.
+    pub secret_provider: Option<SecretProviderSigningKeySource>,
+}
+
+#[derive(Deserialize, Serialize, Clone, Default, Debug, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct SecretSigningKeySource {
+    /// The key of the secret to select from.  Must be a valid secret key.
+    pub key: String,
+    /// Name of the secret.
+    pub secret_name: String,
+}
+
+#[derive(Deserialize, Serialize, Clone, Default, Debug, JsonSchema)]
+pub struct SecretProviderSigningKeySource {
+    /// Configuration for specific provider
+    pub parameters: Option<BTreeMap<String, String>>,
+    /// Configuration for provider name
+    pub provider: Option<String>,
+    /// The path of the private key relative to the root of the mounted volume
+    pub path: PathBuf,
+}
+
 /// Status of the RestateCluster.
 /// This is set and managed automatically.
 /// Read-only.
@@ -326,6 +365,8 @@ pub struct Context {
     pub aws_pod_identity_association_cluster: Option<String>,
     // Whether the EKS SecurityGroupPolicy CRD is installed
     pub security_group_policy_installed: bool,
+    // Whether the SecretProviderClass CRD is installed
+    pub secret_provider_class_installed: bool,
     /// Diagnostics read by the web server
     pub diagnostics: Arc<RwLock<Diagnostics>>,
     /// Prometheus metrics
@@ -436,7 +477,18 @@ impl RestateCluster {
         )
         .await?;
 
-        reconcile_compute(&ctx, name, &base_metadata, &self.spec).await?;
+        let signing_key = reconcile_signing_key(
+            &ctx,
+            name,
+            &base_metadata,
+            self.spec
+                .security
+                .as_ref()
+                .and_then(|s| s.request_signing_private_key.as_ref()),
+        )
+        .await?;
+
+        reconcile_compute(&ctx, name, &base_metadata, &self.spec, signing_key).await?;
 
         Ok(())
     }
@@ -616,13 +668,15 @@ impl State {
         pvc_meta_store: Store<PartialObjectMeta<PersistentVolumeClaim>>,
         ss_store: Store<StatefulSet>,
         security_group_policy_installed: bool,
+        secret_provider_class_installed: bool,
     ) -> Arc<Context> {
         Arc::new(Context {
             client,
             pvc_meta_store,
             ss_store,
             aws_pod_identity_association_cluster: self.aws_pod_identity_association_cluster.clone(),
             security_group_policy_installed,
+            secret_provider_class_installed,
             metrics: Metrics::default().register(&self.registry).unwrap(),
             diagnostics: self.diagnostics.clone(),
         })
@@ -643,17 +697,22 @@ pub async fn run(state: State) {
         }
     };
 
-    let (security_group_policy_installed, pod_identity_association_installed) = api_groups
+    let (
+        security_group_policy_installed,
+        pod_identity_association_installed,
+        secret_provider_class_installed,
+    ) = api_groups
         .groups
         .iter()
-        .fold((false, false), |(sgp, pia), group| {
+        .fold((false, false, false), |(sgp, pia, spc), group| {
             fn group_matches<R: Resource<DynamicType = ()>>(group: &APIGroup) -> bool {
                 group.name == R::group(&())
                     && group.versions.iter().any(|v| v.version == R::version(&()))
             }
             (
                 sgp || group_matches::<SecurityGroupPolicy>(group),
                 pia || group_matches::<PodIdentityAssociation>(group),
+                spc || group_matches::<SecretProviderClass>(group),
             )
         });
 
@@ -667,6 +726,7 @@ pub async fn run(state: State) {
     let pia_api = Api::<PodIdentityAssociation>::all(client.clone());
     let pod_api = Api::<Pod>::all(client.clone());
     let sgp_api = Api::<SecurityGroupPolicy>::all(client.clone());
+    let spc_api = Api::<SecretProviderClass>::all(client.clone());
 
     if state.aws_pod_identity_association_cluster.is_some() && !pod_identity_association_installed {
         error!("PodIdentityAssociation is not available on apiserver, but a pod identity association cluster was provided. Is the CRD installed?");
@@ -741,6 +801,16 @@ pub async fn run(state: State) {
     } else {
         controller
     };
+    let controller = if secret_provider_class_installed {
+        let spc_watcher = metadata_watcher(spc_api, cfg.clone())
+            .touched_objects()
+            // avoid apply loops that seem to happen with crds
+            .predicate_filter(changed_predicate);
+
+        controller.owns_stream(spc_watcher)
+    } else {
+        controller
+    };
     controller
         .run(
             reconcile,
@@ -750,6 +820,7 @@ pub async fn run(state: State) {
                 pvc_meta_store,
                 ss_store,
                 security_group_policy_installed,
+                secret_provider_class_installed,
             ),
         )
         .filter_map(|x| async move { Result::ok(x) })

src/lib.rs

@@ -1,3 +1,4 @@
+use reconcilers::signing_key::InvalidSigningKeyError;
 use std::time::Duration;
 use thiserror::Error;
 
@@ -23,6 +24,9 @@ pub enum Error {
         reason: String,
         requeue_after: Option<Duration>,
     },
+
+    #[error(transparent)]
+    InvalidSigningKeyError(#[from] InvalidSigningKeyError),
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -51,4 +55,5 @@ mod reconcilers;
 
 /// External CRDs
 mod podidentityassociations;
+mod secretproviderclasses;
 mod securitygrouppolicies;