[02/02] service-k8s: reconcile identity-separated backup and restore Jobs with truthful status

crates/service-k8s/src/api/render.rs

@@ -18,6 +18,7 @@
 //! read the root as legacy, and do not "finish" the split without moving the
 //! callers in the same change.
 
+pub use crate::infrastructure::manifest::projected_token::ProjectedServiceAccountToken;
 pub use crate::infrastructure::manifest::{
     client_service, client_service_with_ports, cron_job, dedicated_node_affinity,
     guaranteed_resources, headless_service, headless_service_with_ports, horizontal_pod_autoscaler,
@@ -28,11 +29,11 @@ pub use crate::infrastructure::manifest::{
     DaemonSetPlan, DeploymentPlan, FqdnMatchPlan, FqdnNetworkPolicyPlan, HorizontalPodAutoscaler,
     LabelSet, NetworkPeerPlan, NetworkPolicyPlan, NetworkPortPlan, NetworkRulePlan,
     PersistentVolumeClaimPlan, PodDisruptionBudgetPlan, PodPlan, PodRuntimePolicy, RbacRulePlan,
-    RenderCtx, RoleBindingPlan, RolePlan, ServiceAccountPlan, ServiceAccountSubjectPlan,
-    ServicePlan, ServicePortPlan, ServiceStatefulSet, ShardedStatefulSet, StatefulSetPlan,
-    TokenRegistryProjection, TokenRegistrySource, WorkloadPlan, WorkloadPlanError,
-    WorkloadVolumeClaim, DEFAULT_TOKEN_REGISTRY_CSI_DRIVER, ENV_POD_NAME, ENV_POD_NAMESPACE,
-    ENV_REPLICAS_PER_SHARD, ENV_SHARD_COUNT, ENV_VOTER_COUNT,
+    RenderCtx, RoleBindingPlan, RolePlan, ServiceAccountIdentityPlan, ServiceAccountPlan,
+    ServiceAccountSubjectPlan, ServicePlan, ServicePortPlan, ServiceStatefulSet,
+    ShardedStatefulSet, StatefulSetPlan, TokenRegistryProjection, TokenRegistrySource,
+    WorkloadPlan, WorkloadPlanError, WorkloadVolumeClaim, DEFAULT_TOKEN_REGISTRY_CSI_DRIVER,
+    ENV_POD_NAME, ENV_POD_NAMESPACE, ENV_REPLICAS_PER_SHARD, ENV_SHARD_COUNT, ENV_VOTER_COUNT,
 };
 
 pub mod common {
@@ -74,8 +75,8 @@ pub mod workload_plan {
         ClusterRoleBindingPlan, ContainerPlan, CronJobPlan, DaemonSetPlan, DeploymentPlan,
         FqdnMatchPlan, FqdnNetworkPolicyPlan, LabelSet, NetworkPeerPlan, NetworkPolicyPlan,
         NetworkPortPlan, NetworkRulePlan, PersistentVolumeClaimPlan, PodDisruptionBudgetPlan,
-        PodPlan, PodRuntimePolicy, RbacRulePlan, RoleBindingPlan, RolePlan, ServiceAccountPlan,
-        ServiceAccountSubjectPlan, ServicePlan, ServicePortPlan, StatefulSetPlan, WorkloadPlan,
-        WorkloadPlanError,
+        PodPlan, PodRuntimePolicy, RbacRulePlan, RoleBindingPlan, RolePlan,
+        ServiceAccountIdentityPlan, ServiceAccountPlan, ServiceAccountSubjectPlan, ServicePlan,
+        ServicePortPlan, StatefulSetPlan, WorkloadPlan, WorkloadPlanError,
     };
 }

crates/service-k8s/src/infrastructure/manifest/workload_plan.rs

@@ -14,6 +14,7 @@ mod access;
 mod network;
 mod pod;
 mod service;
+mod service_account_identity;
 mod workload;
 
 pub use access::{
@@ -25,6 +26,7 @@ pub use network::{
 };
 pub use pod::{ContainerPlan, PodPlan, PodRuntimePolicy};
 pub use service::{ServiceAccountPlan, ServicePlan, ServicePortPlan};
+pub use service_account_identity::ServiceAccountIdentityPlan;
 pub use workload::{
     CronJobPlan, DaemonSetPlan, DeploymentPlan, PersistentVolumeClaimPlan, PodDisruptionBudgetPlan,
     StatefulSetPlan,
@@ -42,6 +44,7 @@ fn merge_string_labels(mut base: Value, extra: &LabelSet) -> Value {
 
 enum PlannedObject {
     ServiceAccount(ServiceAccountPlan),
+    ServiceAccountIdentity(ServiceAccountIdentityPlan),
     Service(ServicePlan),
     StatefulSet(StatefulSetPlan),
     Deployment(DeploymentPlan),
@@ -59,6 +62,8 @@ enum PlannedObject {
 pub enum WorkloadPlanError {
     #[error("workload {0} must have at least one replica")]
     ZeroReplicas(String),
+    #[error("service account {0} has invalid workload identity email '{1}': must match <name>@<project>.iam.gserviceaccount.com")]
+    InvalidWorkloadIdentity(String, String),
 }
 
 /// One ordered set of typed Kubernetes children for a managed service.
@@ -79,6 +84,11 @@ impl<'a> WorkloadPlan<'a> {
         self.objects.push(PlannedObject::ServiceAccount(plan));
     }
 
+    pub fn add_service_account_identity(&mut self, plan: ServiceAccountIdentityPlan) {
+        self.objects
+            .push(PlannedObject::ServiceAccountIdentity(plan));
+    }
+
     pub fn add_service(&mut self, plan: ServicePlan) {
         self.objects.push(PlannedObject::Service(plan));
     }
@@ -133,6 +143,9 @@ impl<'a> WorkloadPlan<'a> {
                     "metadata": self.cx.meta(&plan.name, &plan.component),
                     "automountServiceAccountToken": plan.automount_service_account_token,
                 })),
+                PlannedObject::ServiceAccountIdentity(plan) => {
+                    rendered.push(plan.render(self.cx)?);
+                }
                 PlannedObject::Service(plan) => rendered.push(plan.render(self.cx)),
                 PlannedObject::StatefulSet(plan) => rendered.push(plan.render(self.cx)?),
                 PlannedObject::Deployment(plan) => rendered.extend(plan.render(self.cx)?),

crates/service-k8s/src/infrastructure/manifest/workload_plan/service_account_identity.rs

@@ -0,0 +1,117 @@
+//! Typed ServiceAccount plan with metadata annotations (e.g. Workload Identity).
+
+use std::collections::BTreeMap;
+
+use serde_json::{json, Value};
+
+use super::{RenderCtx, WorkloadPlanError};
+
+/// The annotation key for Google Cloud Workload Identity on Kubernetes ServiceAccounts.
+pub const WORKLOAD_IDENTITY_ANNOTATION: &str = "iam.gke.io/gcp-service-account";
+
+/// Validate that a GCP service account email matches `<name>@<project>.iam.gserviceaccount.com`.
+pub fn is_valid_workload_identity(email: &str) -> bool {
+    if email.is_empty() {
+        return false;
+    }
+    let Some((name, domain)) = email.split_once('@') else {
+        return false;
+    };
+    if name.is_empty() || name.contains('@') {
+        return false;
+    }
+    let Some(project) = domain.strip_suffix(".iam.gserviceaccount.com") else {
+        return false;
+    };
+    if project.is_empty() || project.contains('@') {
+        return false;
+    }
+    true
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ServiceAccountIdentityPlan {
+    pub name: String,
+    pub component: String,
+    pub annotations: BTreeMap<String, String>,
+    pub automount_service_account_token: bool,
+}
+
+impl ServiceAccountIdentityPlan {
+    pub fn new(name: impl Into<String>, component: impl Into<String>) -> Self {
+        Self {
+            name: name.into(),
+            component: component.into(),
+            annotations: BTreeMap::new(),
+            automount_service_account_token: false,
+        }
+    }
+
+    pub fn with_annotation(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
+        self.annotations.insert(key.into(), value.into());
+        self
+    }
+
+    pub fn with_workload_identity(mut self, gcp_service_account_email: impl Into<String>) -> Self {
+        let email = gcp_service_account_email.into();
+        self.annotations
+            .insert(WORKLOAD_IDENTITY_ANNOTATION.into(), email);
+        self
+    }
+
+    pub fn try_with_workload_identity(
+        mut self,
+        gcp_service_account_email: impl Into<String>,
+    ) -> Result<Self, WorkloadPlanError> {
+        let email = gcp_service_account_email.into();
+        if !is_valid_workload_identity(&email) {
+            return Err(WorkloadPlanError::InvalidWorkloadIdentity(self.name, email));
+        }
+        self.annotations
+            .insert(WORKLOAD_IDENTITY_ANNOTATION.into(), email);
+        Ok(self)
+    }
+
+    pub fn name(&self) -> &str {
+        &self.name
+    }
+
+    pub fn component(&self) -> &str {
+        &self.component
+    }
+
+    pub fn annotations(&self) -> &BTreeMap<String, String> {
+        &self.annotations
+    }
+
+    pub fn automount_service_account_token(&self) -> bool {
+        self.automount_service_account_token
+    }
+
+    pub fn workload_identity(&self) -> Option<&str> {
+        self.annotations
+            .get(WORKLOAD_IDENTITY_ANNOTATION)
+            .map(String::as_str)
+    }
+
+    pub(super) fn render(self, cx: &RenderCtx<'_>) -> Result<Value, WorkloadPlanError> {
+        for (key, value) in &self.annotations {
+            if key == WORKLOAD_IDENTITY_ANNOTATION && !is_valid_workload_identity(value) {
+                return Err(WorkloadPlanError::InvalidWorkloadIdentity(
+                    self.name,
+                    value.clone(),
+                ));
+            }
+        }
+        let mut metadata = cx.meta(&self.name, &self.component);
+        if !self.annotations.is_empty() {
+            metadata["annotations"] = json!(self.annotations);
+        }
+        Ok(json!({
+            "apiVersion": "v1",
+            "kind": "ServiceAccount",
+            "metadata": metadata,
+            "automountServiceAccountToken": self.automount_service_account_token,
+        }))
+    }
+}