[PATCH proxmox v2 05/16] notify: smtp: Add state handling logic

[Arthur Bied-Charreton <a.bied-charreton@proxmox.com>]

Create new state file in add_endpoint, and create/update existing one in
update_endpoint.

Add trigger_state_refresh to the Endpoint trait, with no-op default
implementation. Override trigger_state_refresh in SmtpEndpoint's
Endpoint impl to trigger an OAuth2 token exchange, in order to rotate
an existing token, or extend its lifetime.

Since trigger_state_refresh is called in pveupdate, it may be called
multiple times in quick succession by the different nodes in the
cluster. In order to avoid unnecessary churn on the state files, the
last_refreshed field is used to check if the state has been refreshed
shortly before, and skip the update if it is the case.

Signed-off-by: Arthur Bied-Charreton <a.bied-charreton@proxmox.com>
---
 proxmox-notify/src/api/common.rs     | 16 ++++++
 proxmox-notify/src/api/smtp.rs       | 37 ++++++++++++-
 proxmox-notify/src/endpoints/smtp.rs | 79 +++++++++++++++++++++++++++-
 proxmox-notify/src/lib.rs            | 21 +++++++-
 4 files changed, 149 insertions(+), 4 deletions(-)

diff --git a/proxmox-notify/src/api/common.rs b/proxmox-notify/src/api/common.rs
index fa2356e2..1e6b7d46 100644
--- a/proxmox-notify/src/api/common.rs
+++ b/proxmox-notify/src/api/common.rs
@@ -3,6 +3,22 @@ use proxmox_http_error::HttpError;
 use super::http_err;
 use crate::{Bus, Config, Notification};
 
+/// Refresh all notification targets' internal state.
+///
+/// The caller is responsible for any needed permission checks.
+pub fn trigger_state_refresh(config: &Config) -> Result<(), HttpError> {
+    let bus = Bus::from_config(config).map_err(|err| {
+        http_err!(
+            INTERNAL_SERVER_ERROR,
+            "Could not instantiate notification bus: {err}"
+        )
+    })?;
+
+    bus.trigger_state_refresh();
+
+    Ok(())
+}
+
 /// Send a notification to a given target.
 ///
 /// The caller is responsible for any needed permission checks.
diff --git a/proxmox-notify/src/api/smtp.rs b/proxmox-notify/src/api/smtp.rs
index 4231cdae..1b25e5d3 100644
--- a/proxmox-notify/src/api/smtp.rs
+++ b/proxmox-notify/src/api/smtp.rs
@@ -1,8 +1,9 @@
 use proxmox_http_error::HttpError;
 
 use crate::api::{http_bail, http_err};
+use crate::context::context;
 use crate::endpoints::smtp::{
-    DeleteableSmtpProperty, SmtpConfig, SmtpConfigUpdater, SmtpPrivateConfig,
+    self, DeleteableSmtpProperty, SmtpConfig, SmtpConfigUpdater, SmtpPrivateConfig,
     SmtpPrivateConfigUpdater, SMTP_TYPENAME,
 };
 use crate::Config;
@@ -69,6 +70,20 @@ pub fn add_endpoint(
         &endpoint_config.name,
     )?;
 
+    if oauth2_refresh_token.is_some() {
+        let oauth_state = smtp::State::from(oauth2_refresh_token);
+
+        context()
+            .save_oauth_state(&endpoint_config.name, Some(oauth_state))
+            .map_err(|e| {
+                http_err!(
+                    INTERNAL_SERVER_ERROR,
+                    "could not create state file for '{}': {e}",
+                    &endpoint_config.name
+                )
+            })?;
+    }
+
     config
         .config
         .set_data(&endpoint_config.name, SMTP_TYPENAME, &endpoint_config)
@@ -189,6 +204,17 @@ pub fn update_endpoint(
     if let Some(oauth2_tenant_id) = updater.oauth2_tenant_id {
         endpoint.oauth2_tenant_id = Some(oauth2_tenant_id);
     }
+    if let Some(oauth2_refresh_token) = oauth2_refresh_token {
+        let oauth_state = context()
+            .load_oauth_state(name)
+            .map_err(|e| http_err!(INTERNAL_SERVER_ERROR, "{e}"))?
+            .set_oauth2_refresh_token(Some(oauth2_refresh_token))
+            .set_last_refreshed(proxmox_time::epoch_i64());
+
+        context()
+            .save_oauth_state(name, Some(oauth_state))
+            .map_err(|e| http_err!(INTERNAL_SERVER_ERROR, "{e}"))?;
+    }
 
     if endpoint.mailto.is_empty() && endpoint.mailto_user.is_empty() {
         http_bail!(
@@ -196,7 +222,6 @@ pub fn update_endpoint(
             "must at least provide one recipient, either in mailto or in mailto-user"
         );
     }
-
     update_private_config(config, name, |c| {
         if let Some(password) = private_endpoint_config_updater.password {
             c.password = Some(password);
@@ -231,6 +256,14 @@ pub fn delete_endpoint(config: &mut Config, name: &str) -> Result<(), HttpError>
     super::ensure_safe_to_delete(config, name)?;
 
     super::remove_private_config_entry(config, name)?;
+
+    context().save_oauth_state(name, None).map_err(|e| {
+        http_err!(
+            INTERNAL_SERVER_ERROR,
+            "could not delete state for '{name}': {e}"
+        )
+    })?;
+
     config.config.sections.remove(name);
 
     Ok(())
diff --git a/proxmox-notify/src/endpoints/smtp.rs b/proxmox-notify/src/endpoints/smtp.rs
index b92f96f0..3e7175d2 100644
--- a/proxmox-notify/src/endpoints/smtp.rs
+++ b/proxmox-notify/src/endpoints/smtp.rs
@@ -1,11 +1,13 @@
 use std::borrow::Cow;
-use std::time::Duration;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use lettre::message::header::{HeaderName, HeaderValue};
 use lettre::message::{Mailbox, MultiPart, SinglePart};
 use lettre::transport::smtp::client::{Tls, TlsParameters};
 use lettre::{message::header::ContentType, Message, SmtpTransport, Transport};
+use oauth2::{ClientId, ClientSecret, RefreshToken};
 use serde::{Deserialize, Serialize};
+use tracing::info;
 
 use proxmox_schema::api_types::COMMENT_SCHEMA;
 use proxmox_schema::{api, Updater};
@@ -22,6 +24,7 @@ const SMTP_PORT: u16 = 25;
 const SMTP_SUBMISSION_STARTTLS_PORT: u16 = 587;
 const SMTP_SUBMISSION_TLS_PORT: u16 = 465;
 const SMTP_TIMEOUT: u16 = 5;
+const SMTP_STATE_REFRESH_CUTOFF_SECONDS: Duration = Duration::from_secs(60 * 60 * 12);
 
 mod xoauth2;
 
@@ -204,6 +207,43 @@ pub struct SmtpEndpoint {
 }
 
 impl SmtpEndpoint {
+    fn get_access_token(
+        &self,
+        refresh_token: &str,
+        auth_method: &SmtpAuthMethod,
+    ) -> Result<xoauth2::TokenExchangeResult, Error> {
+        let client_id = ClientId::new(
+            self.config
+                .oauth2_client_id
+                .as_ref()
+                .ok_or_else(|| Error::Generic("oauth2-client-id not set".into()))?
+                .to_string(),
+        );
+        let client_secret = ClientSecret::new(
+            self.private_config
+                .oauth2_client_secret
+                .as_ref()
+                .ok_or_else(|| Error::Generic("oauth2-client-secret not set".into()))?
+                .to_string(),
+        );
+        let refresh_token = RefreshToken::new(refresh_token.into());
+
+        match auth_method {
+            SmtpAuthMethod::GoogleOAuth2 => {
+                xoauth2::get_google_token(client_id, client_secret, refresh_token)
+            }
+            SmtpAuthMethod::MicrosoftOAuth2 => xoauth2::get_microsoft_token(
+                client_id,
+                client_secret,
+                self.config.oauth2_tenant_id.as_ref().ok_or(Error::Generic(
+                    "tenant ID not set, required for Microsoft OAuth2".into(),
+                ))?,
+                refresh_token,
+            ),
+            _ => Err(Error::Generic("OAuth2 not configured".into())),
+        }
+    }
+
     fn build_transport(&self, tls: Tls, port: u16) -> Result<SmtpTransport, Error> {
         let mut transport_builder = SmtpTransport::builder_dangerous(&self.config.server)
             .tls(tls)
@@ -335,6 +375,43 @@ impl Endpoint for SmtpEndpoint {
     fn disabled(&self) -> bool {
         self.config.disable.unwrap_or_default()
     }
+
+    fn trigger_state_refresh(&self) -> Result<(), Error> {
+        let state = context().load_oauth_state(self.name())?;
+
+        let Some(refresh_token) = &state.oauth2_refresh_token else {
+            return Ok(());
+        };
+
+        // The refresh job is configured in pveupdate, which runs once for each node.
+        // Don't refresh if we already did it recently.
+        if SystemTime::now()
+            .duration_since(UNIX_EPOCH + Duration::from_secs(state.last_refreshed as u64))
+            .map_err(|e| Error::Generic(e.to_string()))?
+            < SMTP_STATE_REFRESH_CUTOFF_SECONDS
+        {
+            return Ok(());
+        }
+
+        let Some(auth_method) = self.config.auth_method.as_ref() else {
+            return Ok(());
+        };
+
+        let state = match self
+            .get_access_token(refresh_token, auth_method)?
+            .refresh_token
+        {
+            Some(tok) => state.set_oauth2_refresh_token(Some(tok.into_secret())), // New token was returned, rotate
+            None => state,
+        }
+        .set_last_refreshed(proxmox_time::epoch_i64());
+
+        context().save_oauth_state(self.name(), Some(state))?;
+
+        info!("OAuth2 state refreshed for endpoint `{}`", self.name());
+
+        Ok(())
+    }
 }
 
 /// Construct a lettre `Message` from a raw email message.
diff --git a/proxmox-notify/src/lib.rs b/proxmox-notify/src/lib.rs
index 619dd7db..2025bd64 100644
--- a/proxmox-notify/src/lib.rs
+++ b/proxmox-notify/src/lib.rs
@@ -9,7 +9,7 @@ use context::context;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use serde_json::Value;
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 use proxmox_schema::api;
 use proxmox_section_config::SectionConfigData;
@@ -169,6 +169,11 @@ pub trait Endpoint {
 
     /// Check if the endpoint is disabled
     fn disabled(&self) -> bool;
+
+    /// Refresh endpoint's state
+    fn trigger_state_refresh(&self) -> Result<(), Error> {
+        Ok(())
+    }
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -605,6 +610,20 @@ impl Bus {
 
         Ok(())
     }
+
+    /// Refresh all endpoints' internal state.
+    ///
+    /// This function works on a best effort basis, if an endpoint's state cannot
+    /// be updated for whatever reason, the error is logged and the next one(s)
+    /// are attempted.
+    pub fn trigger_state_refresh(&self) {
+        for (name, endpoint) in &self.endpoints {
+            match endpoint.trigger_state_refresh() {
+                Ok(()) => debug!("triggered state refresh for endpoint '{name}'"),
+                Err(e) => error!("could not trigger state refresh for endpoint '{name}': {e}"),
+            };
+        }
+    }
 }
 
 #[cfg(test)]
-- 
2.47.3