Tbt/auth daemon patch

backend/auth-core/Cargo.toml

@@ -23,7 +23,7 @@ http-body-util = "0.1.3"
 migration = { version = "0.1.0", path = "migration" }
 oauth2 = "5.0.0"
 openidconnect = { workspace = true, features = ["timing-resistant-secret-traits"] }
-rustls = { workspace = true, features = ["ring"] }
+rustls = { workspace = true, features = ["aws-lc-rs"] }
 sea-orm = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

backend/auth-daemon/Cargo.toml

@@ -22,6 +22,8 @@ tracing.workspace = true
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 url.workspace = true
 chrono.workspace = true
+kube = { workspace = true }
+k8s-openapi = { workspace = true }
 
 [dev-dependencies]
 mockito.workspace = true

backend/auth-daemon/src/main.rs

@@ -10,14 +10,15 @@ use auth_core::openidconnect::SubjectIdentifier;
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
     process,
-    sync::Arc,
+    sync::{Arc, OnceLock},
 };
 
 use axum::{Router, http::Method, middleware, routing::get};
 use clap::Parser;
 use regex::Regex;
 use tokio::signal::unix::{SignalKind, signal};
 
+use anyhow::Context;
 use tower_http::cors::{AllowOrigin, CorsLayer};
 use tracing::{debug, info};
 use tracing_subscriber::EnvFilter;
@@ -27,9 +28,50 @@ mod state;
 
 use axum_reverse_proxy::ReverseProxy;
 use config::DaemonConfig;
+use kube::{
+    Api, Client,
+    api::{ApiResource, DynamicObject},
+    core::GroupVersionKind,
+};
 
 type Result<T> = std::result::Result<T, Error>;
 
+static CRYPTO_PROVIDER: OnceLock<()> = OnceLock::new();
+
+async fn resolve_subject() -> anyhow::Result<SubjectIdentifier> {
+    let workflow_name = std::env::var("ARGO_WORKFLOW_NAME")
+        .map_err(|_| anyhow::anyhow!("ARGO_WORKFLOW_NAME not set"))?;
+    let namespace = std::env::var("MY_POD_NAMESPACE")
+        .map_err(|_| anyhow::anyhow!("MY_POD_NAMESPACE not set"))?;
+
+    let client = Client::try_default()
+        .await
+        .context("failed to create Kubernetes client (check service account token mount)")?;
+    let gvk = GroupVersionKind::gvk("argoproj.io", "v1alpha1", "Workflow");
+    let api = Api::<DynamicObject>::namespaced_with(
+        client,
+        &namespace,
+        &ApiResource::from_gvk_with_plural(&gvk, "workflows"),
+    );
+    let workflow = api
+        .get(&workflow_name)
+        .await
+        .with_context(|| format!("failed to get workflow {}/{}", namespace, workflow_name))?;
+    let subject = workflow
+        .metadata
+        .labels
+        .as_ref()
+        .and_then(|l| l.get("workflows.argoproj.io/creator"))
+        .ok_or_else(|| {
+            anyhow::anyhow!(
+                "label workflows.argoproj.io/creator missing on workflow {}",
+                workflow_name
+            )
+        })?
+        .clone();
+    Ok(SubjectIdentifier::new(subject))
+}
+
 #[derive(Parser, Debug)]
 #[command(author, version, about)]
 struct ServeArgs {
@@ -41,8 +83,6 @@ struct ServeArgs {
         default_value = "config.yaml"
     )]
     config: String,
-    #[arg(env = "WORKFLOWS_AUTH_DAEMON_SUBJECT")]
-    subject: String,
 }
 
 #[derive(Debug, Parser)]
@@ -54,6 +94,15 @@ enum Cli {
 
 #[tokio::main]
 async fn main() -> Result<()> {
+    // Must be installed before any TLS connection, setup_router duplicates this as a fallback for
+    // current or future tests that bypass main
+    // OnceLock ensures only one installation of which path is run
+    CRYPTO_PROVIDER.get_or_init(|| {
+        auth_core::rustls::crypto::aws_lc_rs::default_provider()
+            .install_default()
+            .expect("Failed to install rustls CryptoProvider");
+    });
+
     dotenvy::dotenv().ok();
     let args = Cli::parse();
     tracing_subscriber::fmt()
@@ -62,13 +111,20 @@ async fn main() -> Result<()> {
 
     match args {
         Cli::Serve(args) => {
-            let config = DaemonConfig::from_file(args.config)?;
+            let config = DaemonConfig::from_file(&args.config)
+                .map_err(|e| anyhow::anyhow!("failed to read config {}: {:?}", args.config, e))?;
             let requested_port = config.common.port;
-            let router_state =
-                Arc::new(RouterState::new(config, &SubjectIdentifier::new(args.subject)).await?);
+            let subject = resolve_subject().await?;
+            let router_state = Arc::new(
+                RouterState::new(config, &subject)
+                    .await
+                    .map_err(|e| anyhow::anyhow!("failed to initialise router state: {:?}", e))?,
+            );
 
             let router = setup_router(router_state, None)?;
-            serve(router, IpAddr::V4(Ipv4Addr::UNSPECIFIED), requested_port).await?;
+            serve(router, IpAddr::V4(Ipv4Addr::UNSPECIFIED), requested_port)
+                .await
+                .with_context(|| format!("failed to bind/serve on port {}", requested_port))?;
         }
     }
 
@@ -78,9 +134,11 @@ async fn main() -> Result<()> {
 fn setup_router(state: Arc<RouterState>, cors_allow: Option<Vec<Regex>>) -> anyhow::Result<Router> {
     debug!("Setting up the router");
 
-    auth_core::rustls::crypto::ring::default_provider()
-        .install_default()
-        .expect("Failed to install rust TLS cryptography");
+    CRYPTO_PROVIDER.get_or_init(|| {
+        auth_core::rustls::crypto::aws_lc_rs::default_provider()
+            .install_default()
+            .expect("Failed to install rustls CryptoProvider");
+    });
 
     let cors_origin = if let Some(cors_allow) = cors_allow {
         info!("Allowing CORS Origin(s) matching: {:?}", cors_allow);
@@ -212,7 +270,6 @@ mod tests {
         let params = [
             ("grant_type", "refresh_token"),
             ("scope", "openid offline_access"),
-            ("subject", "test-subject"),
             ("refresh_token", "test-refresh-token"),
             ("client_id", "test-client"),
         ];

charts/workflows/Chart.yaml

@@ -2,7 +2,7 @@ apiVersion: v2
 name: workflows
 description: Data Analysis workflow orchestration
 type: application
-version: 0.13.45
+version: 0.13.47
 dependencies:
   - name: argo-workflows
     repository: https://argoproj.github.io/argo-helm

charts/workflows/templates/argo-workflow-clusterrole.yaml

@@ -16,19 +16,22 @@ rules:
   resources: [
     "workflowtaskresults",
     ]
-  verbs: ["create","patch"] 
+  verbs: ["create","patch"]
+- apiGroups: ["argoproj.io"]
+  resources: ["workflows"]
+  verbs: ["get"]
 - apiGroups: ["argoproj.io"]
   resources: [
     "workflowtasksets",
     "workflowartifactgctasks",
     ]
-  verbs: ["list", "watch"] 
+  verbs: ["list", "watch"]
 - apiGroups: ["argoproj.io"]
   resources: [
     "workflowtasksets/status",
     "workflowartifactgctasks/status",
     ]
-  verbs: ["patch"] 
+  verbs: ["patch"]
 - apiGroups: ["kubeflow.org"]
   resources: ["mpijobs"]
   verbs: ["create", "get", "list", "watch", "update", "delete"]