feat: Initial implementation for OIDC with machines

Author: johannwagner

Cargo.lock

@@ -47,6 +47,7 @@ TODO @yu-re-ka: Add Information
     + Add own machine token to configuration
         + This is needed for `fairy` later.
     + Add OIDC authentication provider to configuration
+        + See OIDC provider section.
 + Enter `vicky`
 + Run `cargo run --bin vicky`
 
@@ -89,3 +90,25 @@ Options:
   -V, --version  Print version
 ```
 
+## OIDC Provider
+
+Since implementing user, role and account management is timeconsuming, we settled on fully using OIDC flows for this application.
+Therefore, there is some configuration required.
+This is tested against Keycloak instances. Your mileage may vary on other implementations.
+
+### Configuration
+
+Configuration is done via a well-known OIDC endpoint, e.g. `https://my-nice-keycloak-instance.com/realms/wobcom/.well-known/openid-configuration`. 
+
+You need two different clients, one client which acts as a service account for your backend services and one client to authenticate your users against using the web interface. Every user authenticating with the backend client gets the role `vicky:machine`, everyone else gets the role `vicky:user`.
+
+We expected the following keys in the userinfo endpoint:
++ `vicky:user`
+    + TBD
+    + `vicky_roles`
+        + List of assigned roles, some of `vicky:machine` or `vicky:user`.
++ `vicky:machine`
+    + `sub`
+    + `preferred_username`
+    + `vicky_roles`
+        + List of assigned roles, some of `vicky:machine` or `vicky:user`.

README.md

@@ -18,3 +18,6 @@ tokio = { version = "1.32.0", features = ["rt", "macros", "process"] }
 tokio-util = { version = "0.7.9", features = ["codec"] }
 uuid = { version = "1.4.1", features = ["serde"] }
 rocket = { version="0.5.0", features = ["json", "secrets"] } 
+openidconnect = "3.5.0"
+reqwest = { version="0.12.4", features = ["json"]}
+chrono = "0.4.38"

fairy/Cargo.toml

@@ -2,5 +2,9 @@
 
 vicky_url = "http://localhost:8000"
 vicky_external_url = "https://vicky.lab.wobcom.de"
-machine_token = ""
-features = []
+features = []
+
+[default.oidc_config]
+authority = "https://id.lab.wobcom.de/realms/wobcom"
+client_id = "vicky-dev-api"
+client_secret = ""

fairy/Rocket.example.toml

@@ -0,0 +1,112 @@
+use std::sync::Arc;
+
+use chrono::{Utc, DateTime, Duration};
+use log::info;
+use openidconnect::core::{CoreClient, CoreProviderMetadata};
+use openidconnect::{ClientId, ClientSecret, IssuerUrl, OAuth2TokenResponse, Scope};
+use serde::de::DeserializeOwned;
+use serde::{Serialize};
+use reqwest::{self, Method, RequestBuilder};
+use openidconnect::reqwest::async_http_client;
+
+
+
+use crate::AppConfig;
+
+#[derive(Debug)]
+pub enum HttpClientState {
+    Authenticated {
+        access_token: String,
+        expires_at: DateTime<Utc>,
+    },
+    Unauthenticated,
+}
+#[derive(Debug)]
+pub struct HttpClient {
+    app_config: Arc<AppConfig>,
+    http_client: reqwest::Client,
+    client_state: HttpClientState
+}
+
+impl HttpClient {
+    pub fn new(cfg: Arc<AppConfig>) -> HttpClient {
+        HttpClient {
+            app_config: cfg,
+            http_client: reqwest::Client::new(),
+            client_state: HttpClientState::Unauthenticated,
+        }
+    }
+
+    async fn renew_access_token(&mut self) -> anyhow::Result<String> {
+        let client_id = ClientId::new(self.app_config.oidc_config.client_id.clone());
+        let client_secret = ClientSecret::new(self.app_config.oidc_config.client_secret.clone());
+        let issuer_url = IssuerUrl::new(self.app_config.oidc_config.issuer_url.clone())?;
+
+        info!("Using {:?} as client_id to try to authorize to {:?}..", client_id, issuer_url);
+
+        let provider_metadata = CoreProviderMetadata::discover_async(issuer_url, &async_http_client).await?;
+        let client = CoreClient::from_provider_metadata(
+            provider_metadata,
+            client_id,
+            Some(client_secret),
+        );
+
+        let ccreq = client
+            .exchange_client_credentials()
+            .add_scope(Scope::new("openid".to_string()));
+
+        let ccres = ccreq.request_async(async_http_client).await?;
+
+        let access_token = ccres.access_token().secret();
+        let expires_at = Utc::now() + ccres.expires_in().unwrap() - Duration::seconds(5);
+
+        info!("Accquired access token, expiring at {:?} ..", expires_at);
+
+        self.client_state = HttpClientState::Authenticated { access_token: access_token.clone(), expires_at };
+        Ok(access_token.clone())
+    }
+
+    async fn create_request<U: reqwest::IntoUrl>(&mut self, method: Method, url: U) -> anyhow::Result<RequestBuilder> {
+
+        let now = Utc::now();
+
+        info!("client_state: {:?}", self.client_state);
+
+        let access_token_to_use = match &self.client_state {
+            HttpClientState::Authenticated { expires_at, access_token } => {
+                if expires_at > &now {
+                    access_token.to_string()
+                } else {
+                    self.renew_access_token().await?
+                }
+            },
+            HttpClientState::Unauthenticated => {
+                self.renew_access_token().await?
+            },
+        };
+
+        Ok(self.http_client.request(method, url).header("Authorization", format!("Bearer {}", access_token_to_use)))
+    }    
+
+
+    pub async fn do_request<BODY: Serialize, RESPONSE: DeserializeOwned>(
+        &mut self,
+        method: Method,
+        endpoint: &str,
+        q: &BODY,
+    ) -> anyhow::Result<RESPONSE> {
+    
+        let response = self
+            .create_request(method, format!("{}/{}", self.app_config.vicky_url, endpoint)).await?
+            .header("content-type", "application/json")
+            .json(q)
+            .send().await?;
+    
+    
+        if !response.status().is_success() {
+            anyhow::bail!("API error: {:?}", response);
+        }
+    
+        Ok(response.json().await?)
+    }
+}

fairy/src/api.rs

@@ -1,25 +1,36 @@
 use anyhow::anyhow;
+use api::HttpClient;
 use futures_util::{Sink, StreamExt, TryStreamExt};
-use hyper::{Body, Client, Method, Request};
-use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use std::process::Stdio;
 use std::sync::Arc;
 use tokio::process::Command;
 use tokio_util::codec::{FramedRead, LinesCodec};
 use uuid::Uuid;
+use reqwest::{self, Method};
 
 use rocket::figment::providers::{Env, Format, Toml};
 use rocket::figment::{Figment, Profile};
 
-#[derive(Deserialize)]
+mod api;
+
+
+#[derive(Deserialize, Debug)]
+pub struct OIDCConfig {
+    issuer_url: String,
+    client_id: String,
+    client_secret: String,
+}
+
+#[derive(Deserialize, Debug)]
 pub(crate) struct AppConfig {
     pub(crate) vicky_url: String,
     pub(crate) vicky_external_url: String,
-    pub(crate) machine_token: String,
     pub(crate) features: Vec<String>,
+    pub(crate) oidc_config: OIDCConfig,
 }
 
+
 fn main() -> anyhow::Result<()> {
     env_logger::builder()
         .filter_level(log::LevelFilter::Debug)
@@ -45,31 +56,7 @@ fn main() -> anyhow::Result<()> {
     run(app_config)
 }
 
-async fn api<BODY: Serialize, RESPONSE: DeserializeOwned>(
-    cfg: &AppConfig,
-    method: Method,
-    endpoint: &str,
-    q: &BODY,
-) -> anyhow::Result<RESPONSE> {
-    let client = Client::new();
-    let req_data = serde_json::to_vec(q)?;
-
-    let request = Request::builder()
-        .uri(format!("{}/{}", cfg.vicky_url, endpoint))
-        .method(method)
-        .header("content-type", "application/json")
-        .header("authorization", &cfg.machine_token)
-        .body(Body::from(req_data))?;
-
-    let response = client.request(request).await?;
-
-    if !response.status().is_success() {
-        anyhow::bail!("API error: {:?}", response);
-    }
 
-    let resp_data = hyper::body::to_bytes(response.into_body()).await?;
-    Ok(serde_json::from_slice(&resp_data)?)
-}
 
 #[derive(Debug, Deserialize)]
 pub struct FlakeRef {
@@ -104,11 +91,11 @@ fn log_sink(
     cfg: Arc<AppConfig>,
     task_id: Uuid,
 ) -> impl Sink<Vec<String>, Error = anyhow::Error> + Send {
-    futures_util::sink::unfold((), move |_, lines: Vec<String>| {
-        let cfg = cfg.clone();
+    let vicky_client_task = HttpClient::new(cfg.clone());
+
+    futures_util::sink::unfold(vicky_client_task, move |mut http_client, lines: Vec<String>| {
         async move {
-            let response = api::<_, ()>(
-                &cfg,
+            let response = http_client.do_request::<_, ()>(
                 Method::POST,
                 &format!("api/v1/tasks/{}/logs", task_id),
                 &serde_json::json!({ "lines": lines }),
@@ -118,7 +105,7 @@ fn log_sink(
             match response {
                 Ok(_) => {
                     log::info!("logged {} line(s) from task", lines.len());
-                    Ok(())
+                    Ok(http_client)
                 }
                 Err(e) => {
                     log::error!(
@@ -139,14 +126,13 @@ async fn try_run_task(cfg: Arc<AppConfig>, task: &Task) -> anyhow::Result<()> {
     let mut child = Command::new("nix")
         .args(args)
         .env("VICKY_API_URL", &cfg.vicky_external_url)
-        .env("VICKY_MACHINE_TOKEN", &cfg.machine_token)
         .kill_on_drop(true)
         .stdin(Stdio::null())
         .stdout(Stdio::piped())
         .stderr(Stdio::piped())
         .spawn()?;
 
-    let logger = log_sink(cfg.clone(), task.id);
+    let logger = log_sink(cfg.clone(),  task.id);
 
     let lines = futures_util::stream::select(
         FramedRead::new(child.stdout.take().unwrap(), LinesCodec::new()),
@@ -170,6 +156,9 @@ async fn try_run_task(cfg: Arc<AppConfig>, task: &Task) -> anyhow::Result<()> {
 }
 
 async fn run_task(cfg: Arc<AppConfig>, task: Task) {
+
+    let mut vicky_client_task = HttpClient::new(cfg.clone());
+
     let result = match try_run_task(cfg.clone(), &task).await {
         Err(e) => {
             log::info!("task failed: {} {} {:?}", task.id, task.display_name, e);
@@ -178,19 +167,17 @@ async fn run_task(cfg: Arc<AppConfig>, task: Task) {
         Ok(_) => TaskResult::Success,
     };
     tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-    let _ = api::<_, ()>(
-        &cfg,
+    let _ = vicky_client_task.do_request::<_, ()>(
         Method::POST,
         &format!("api/v1/tasks/{}/finish", task.id),
         &serde_json::json!({ "result": result }),
     )
     .await;
 }
 
-async fn try_claim(cfg: Arc<AppConfig>) -> anyhow::Result<()> {
+async fn try_claim(cfg: Arc<AppConfig>, vicky_client: &mut HttpClient) -> anyhow::Result<()> {
     log::debug!("trying to claim task...");
-    if let Some(task) = api::<_, Option<Task>>(
-        &cfg,
+    if let Some(task) = vicky_client.do_request::<_, Option<Task>>(
         Method::POST,
         "api/v1/tasks/claim",
         &serde_json::json!({ "features": cfg.features }),
@@ -210,12 +197,14 @@ async fn try_claim(cfg: Arc<AppConfig>) -> anyhow::Result<()> {
 
 #[tokio::main(flavor = "current_thread")]
 async fn run(cfg: AppConfig) -> anyhow::Result<()> {
+    let cfg = Arc::new(cfg);
+    let mut vicky_client_mgmt = HttpClient::new(cfg.clone());
+
     log::info!("config valid, starting communication with vicky");
     log::info!("waiting for tasks...");
 
-    let cfg = Arc::new(cfg);
     loop {
-        if let Err(e) = try_claim(cfg.clone()).await {
+        if let Err(e) = try_claim(cfg.clone(), &mut vicky_client_mgmt).await {
             log::error!("{}", e);
             tokio::time::sleep(std::time::Duration::from_secs(5)).await;
         }

fairy/src/main.rs

@@ -1,9 +1,5 @@
 [default]
 
-machines = [
-    "abc1234"
-]
-
 [default.databases]
 postgres_db = { url = "postgres://vicky:vicky@localhost/vicky" }