// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

//

// SPDX-License-Identifier: Apache-2.0

//! Asynchronous OpenStack client

use std::collections::HashMap;

use std::convert::TryInto;

use std::fmt::{self, Debug};

use std::ops::{Deref, DerefMut};

use std::sync::{Arc, Weak};

use std::time::{Duration, SystemTime};

use std::{fs::File, io::Read};

use arc_swap::ArcSwap;

use async_trait::async_trait;

use bytes::Bytes;

use chrono::TimeDelta;

use futures::io::{Error as IoError, ErrorKind as IoErrorKind};

use futures::stream::TryStreamExt;

use http::{HeaderMap, HeaderValue, Response as HttpResponse, StatusCode, header};

use parking_lot::RwLock;

use reqwest::{Certificate, Request, Response};

use secrecy::{ExposeSecret, SecretString};

use tokio_util::codec;

use tokio_util::compat::FuturesAsyncReadCompatExt;

use tracing::{Level, debug, enabled, error, event, info, instrument, trace, warn};

use openstack_sdk_auth_core::{

Auth, AuthError, AuthPluginRegistration, AuthToken, OpenStackAuthType,

authtoken::AuthTokenError,

authtoken_scope::AuthTokenScope,

types::{AuthResponse, Project, ServiceEndpoints},

};

// Private use of auth plugins for token refresh and receipt handling.

use openstack_sdk_auth_receipt as token_receipt;

use openstack_sdk_auth_token as token_auth;

use crate::auth::authtoken::{AuthType, build_token_info_endpoint};

use crate::session;

use openstack_sdk_core::api::{self, AsyncClient, QueryAsync, query, raw};

use openstack_sdk_core::auth::{

AuthState,

auth_helper::{AuthHelper, Dialoguer, Noop},

gather_auth_data,

};

use openstack_sdk_core::catalog::{Catalog, CatalogError, ServiceEndpoint};

use openstack_sdk_core::config::CloudConfig;

use openstack_sdk_core::error::{OpenStackError, OpenStackResult, RestError};

use openstack_sdk_core::session::AuthSnapshot;

use openstack_sdk_core::types::{ApiVersion, BoxedAsyncRead, ServiceType};

use openstack_sdk_core::utils::expand_tilde;

/// Asynchronous client for the OpenStack API for a single user

///

/// Separate Identity (not the scope) should use separate instances of this.

/// ```rust

/// use openstack_sdk::api::{paged, Pagination, QueryAsync};

/// use openstack_sdk::{AsyncOpenStack, config::ConfigFile, OpenStackError};

/// use openstack_sdk::types::ServiceType;

/// use openstack_sdk::api::compute::v2::flavor::list_20;

///

/// async fn list_flavors() -> Result<(), OpenStackError> {

/// // Get the builder for the listing Flavors Endpoint

/// let mut ep_builder = list_20::Request::builder();

/// // Set the `min_disk` query param

/// ep_builder.min_disk("15");

/// let ep = ep_builder.build().unwrap();

///

/// let cfg = ConfigFile::new().unwrap();

/// // Get connection config from clouds.yaml/secure.yaml

/// let profile = cfg.get_cloud_config("devstack").unwrap().unwrap();

/// // Establish connection

/// let mut session = AsyncOpenStack::new(&profile).await?;

///

/// // Invoke service discovery when desired.

/// session.discover_service_endpoint(&ServiceType::Compute).await?;

///

/// // Execute the call with pagination limiting maximum amount of entries to 1000

/// let data: Vec<serde_json::Value> = paged(ep, Pagination::Limit(1000))

/// .query_async(&session)

/// .await.unwrap();

///

/// println!("Data = {:?}", data);

/// Ok(())

/// }

/// ```

pub struct AsyncOpenStack {

/// The client to use for API calls.

client: reqwest::Client,

/// Cloud configuration

config: CloudConfig,

/// Session context (auth, catalog, state)

session: Arc<RwLock<session::SessionContext>>,

/// Lock-free snapshot of auth/catalog/region for the per-request hot

/// path, republished by `SessionWriteGuard::drop` on every session write.

auth_snapshot: Arc<ArcSwap<AuthSnapshot>>,

/// Auth helper for re-authentication on 401.

auth_helper: Option<Arc<dyn AuthHelper>>,

/// Max retries on 401.

max_auth_retries: u32,

/// Single-flight guard for re-authentication, shared across all clones.

/// Held for the duration of `handle_401_retry` so concurrent 401s queue

/// on one re-auth instead of each running their own. Also intended to be

/// shared with any future proactive token-renewal task.

reauth_lock: Arc<tokio::sync::Mutex<()>>,

}

impl Clone for AsyncOpenStack {

fn clone(&self) -> Self {

Self {

client: self.client.clone(),

config: self.config.clone(),

session: Arc::clone(&self.session),

auth_snapshot: Arc::clone(&self.auth_snapshot),

auth_helper: self.auth_helper.clone(),

max_auth_retries: self.max_auth_retries,

reauth_lock: Arc::clone(&self.reauth_lock),

}

}

}

impl Debug for AsyncOpenStack {

fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {

f.debug_struct("OpenStack")

.field("service_endpoints", &self.auth_snapshot.load().catalog)

.finish()

}

}

#[async_trait]

impl api::RestClient for AsyncOpenStack {

type Error = RestError;

fn get_current_project(&self) -> Option<Project> {

let session = self.session.read();

if let Auth::AuthToken(token) = &session.auth {

return token.auth_info.clone().and_then(|x| x.token.project);

}

None

}

}

#[async_trait]

impl api::AsyncClient for AsyncOpenStack {

async fn rest_async(

&self,

request: http::request::Builder,

body: Vec<u8>,

) -> Result<HttpResponse<Bytes>, api::ApiError<<Self as api::RestClient>::Error>> {

let (mut auth, mut generation) = {

let snap = self.auth_snapshot.load();

(snap.auth.clone(), snap.auth_generation)

};

let mut retries = 0;

let orig_method = request.method_ref().cloned().unwrap_or(http::Method::GET);

let orig_uri = request.uri_ref().cloned().unwrap_or_default();

let orig_headers = request.headers_ref().cloned().unwrap_or_default();

let mut request = Some(request);

loop {

let current_request = match request.take() {

Some(r) => r,

None => unreachable!("request missing after retry setup"),

};

let result = self

.rest_with_auth_async(current_request, body.clone(), &auth)

.await;

let is_401 = match &result {

Ok(rsp) => rsp.status() == http::StatusCode::UNAUTHORIZED,

Err(_) => false,

};

if is_401 && retries < self.max_auth_retries {

warn!(

"Received 401 Unauthorized, retry {} of {}",

retries + 1,

self.max_auth_retries

);

match self.handle_401_retry(generation).await {

Ok(_) => {

(auth, generation) = {

let snap = self.auth_snapshot.load();

(snap.auth.clone(), snap.auth_generation)

};

retries += 1;

let mut new_builder = http::Request::builder()

.method(orig_method.clone())

.uri(orig_uri.clone());

for (name, value) in &orig_headers {

new_builder = new_builder.header(name.clone(), value.clone());

}

request = Some(new_builder);

continue;

}

Err(e) => {

warn!("401 retry failed: {}", e);

return result;

}

}

}

return result;

}

}

async fn rest_read_body_async(

&self,

request: http::request::Builder,

body: BoxedAsyncRead,

) -> Result<HttpResponse<Bytes>, api::ApiError<<Self as api::RestClient>::Error>> {

let auth = self.auth_snapshot.load().auth.clone();

self.rest_with_auth_read_body_async(request, body, &auth)

.await

}

async fn download_async(

&self,

request: http::request::Builder,

body: Vec<u8>,

) -> Result<(HeaderMap, BoxedAsyncRead), api::ApiError<<Self as api::RestClient>::Error>> {

let auth = self.auth_snapshot.load().auth.clone();

self.download_with_auth_async(request, body, &auth).await

}

async fn get_service_endpoint(

&self,

service_type: &ServiceType,

version: Option<&ApiVersion>,

) -> Result<ServiceEndpoint, api::ApiError<Self::Error>> {

let lookup = || {

let snap = self.auth_snapshot.load();

let region = snap

.region_name

.as_ref()

.or(self.config.region_name.as_ref());

snap.catalog

.get_service_endpoint(

service_type.to_string(),

version,

region,

self.config.interface.as_ref(),

)

.cloned()

.map_err(api::ApiError::catalog)

};

let ep = match lookup() {

Ok(ep) => return Ok(ep),

Err(api::ApiError::Catalog { source }) => {

if matches!(source, CatalogError::ServiceNotConfigured { .. }) {

info!(

"Service '{}' not in catalog, attempting refresh via auth/catalog",

service_type

);

if let Err(e) = self.refresh_catalog().await {

warn!("Catalog refresh failed: {}", e);

}

match lookup() {

Ok(ep) => {

debug!("Service '{}' found after catalog refresh", service_type);

self.persist_catalog_to_cache();

ep

}

Err(e) => return Err(e),

}

} else {

return Err(api::ApiError::Catalog { source });

}

}

Err(e) => return Err(e),

};

Ok(ep)

}

}

/// Write guard for `SessionContext` that republishes `AuthSnapshot` on drop,

/// while the write lock is still held (the manual `Drop::drop` below runs

/// before the `guard` field's own destructor, so the store happens with the

/// lock still taken). This guarantees a concurrent hot-path reader never

/// sees a released lock with a stale snapshot: every write path that

/// mutates `auth`, `auth_generation`, `catalog`, or `region_name` is

/// automatically reflected, with no per-call-site bookkeeping required.

///

/// Trade-off: writes that only touch `state` (auth-cache bookkeeping, e.g.

/// `get_scope_auth`/`clear_all_auth`) still pay a full `AuthSnapshot`

/// rebuild — including a `Catalog` clone — since the guard can't tell which

/// fields changed. These are auth-flow writes, not per-request, so the cost

/// is bounded; revisit if `Catalog` grows large enough to matter.

struct SessionWriteGuard<'a> {

guard: parking_lot::RwLockWriteGuard<'a, session::SessionContext>,

snapshot: &'a ArcSwap<AuthSnapshot>,

}

impl Deref for SessionWriteGuard<'_> {

type Target = session::SessionContext;

fn deref(&self) -> &Self::Target {

&self.guard

}

}

impl DerefMut for SessionWriteGuard<'_> {

fn deref_mut(&mut self) -> &mut Self::Target {

&mut self.guard

}

}

impl Drop for SessionWriteGuard<'_> {

fn drop(&mut self) {

self.snapshot

.store(Arc::new(AuthSnapshot::from(&*self.guard)));

}

}

/// Handle to a background token-renewal task started by

/// [`AsyncOpenStack::enable_auto_renew`]. Dropping it stops the task; the

/// task also stops on its own once every `AsyncOpenStack` sharing the

/// session it was spawned for has been dropped.

pub struct RenewHandle(tokio::task::JoinHandle<()>);

impl Drop for RenewHandle {

fn drop(&mut self) {

self.0.abort();

}

}

impl AsyncOpenStack {

/// Lock the session for reading. `location` is kept as a parameter for

/// call-site self-documentation (parking_lot never poisons, so it no

/// longer feeds an error message).

#[allow(unused_variables)]

fn session_read(

&self,

location: &'static str,

) -> parking_lot::RwLockReadGuard<'_, session::SessionContext> {

self.session.read()

}

/// Lock the session for writing. Returns a guard that republishes

/// `AuthSnapshot` on drop — see `SessionWriteGuard`.

#[allow(unused_variables)]

fn session_write(&self, location: &'static str) -> SessionWriteGuard<'_> {

SessionWriteGuard {

guard: self.session.write(),

snapshot: &self.auth_snapshot,

}

}

/// Basic constructor — visible to the sync facade.

pub fn new_impl(config: &CloudConfig, auth: Auth) -> OpenStackResult<Self> {

// Ensure auth plugin registries are populated even if the linker

// would otherwise strip crates with no other references.

openstack_sdk_auth_core::anchor_plugins();

let mut client_builder = reqwest::Client::builder();

if let Some(cacert) = &config.cacert {

let mut buf = Vec::new();

File::open(expand_tilde(cacert).unwrap_or(cacert.into()))

.map_err(|e| OpenStackError::IOWithPath {

source: e,

path: cacert.into(),

})?

.read_to_end(&mut buf)

.map_err(|e| OpenStackError::IOWithPath {

source: e,

path: cacert.into(),

})?;

for cert in Certificate::from_pem_bundle(&buf)? {

client_builder = client_builder.add_root_certificate(cert);

}

}

if let Some(false) = &config.verify {

warn!(

"SSL Verification is disabled! Please consider using `cacert` for adding custom certificate instead."

);

client_builder = client_builder.danger_accept_invalid_certs(true);

}

client_builder = client_builder.pool_max_idle_per_host(10);

client_builder = client_builder.pool_idle_timeout(Duration::from_secs(30));

client_builder = client_builder.timeout(Duration::from_secs(

config

.options

.get("api_timeout")

.and_then(|val| val.clone().into_uint().ok())

.unwrap_or(30),

));

client_builder = client_builder.connect_timeout(Duration::from_secs(5));

client_builder = client_builder.tcp_keepalive(Duration::from_secs(60));

client_builder = client_builder.gzip(true);

client_builder = client_builder.deflate(true);

// Pass CloudConfig.auth_cache as override; SessionContext resolves priority chain.

let session_ctx = session::SessionContext::new(config, auth, config.auth_cache)?;

let auth_snapshot = Arc::new(ArcSwap::new(Arc::new(AuthSnapshot::from(&session_ctx))));

Ok(AsyncOpenStack {

client: client_builder.build()?,

config: config.clone(),

session: Arc::new(RwLock::new(session_ctx)),

auth_snapshot,

auth_helper: None,

max_auth_retries: 1,

reauth_lock: Arc::new(tokio::sync::Mutex::new(())),

})

}

/// Create a new OpenStack API session from CloudConfig

#[instrument(name = "connect", level = "trace", skip(config))]

pub async fn new(config: &CloudConfig) -> OpenStackResult<Self> {

let session = Self::new_impl(config, Auth::None)?;

// Ensure we resolve identity endpoint using version discovery

session

.discover_service_endpoint(&ServiceType::Identity)

.await?;

session.authorize(None, false, false).await?;

Ok(session)

}

/// Create a new OpenStack API session from CloudConfig

#[instrument(name = "connect", level = "trace", skip(config, auth_helper))]

pub async fn new_with_authentication_helper<A>(

config: &CloudConfig,

auth_helper: A,

renew_auth: bool,

) -> OpenStackResult<Self>

where

A: AuthHelper + Sync + Send + 'static,

{

let mut session = Self::new_impl(config, Auth::None)?;

// Ensure we resolve identity endpoint using version discovery

session

.discover_service_endpoint(&ServiceType::Identity)

.await?;

session

.authorize_with_auth_helper(None, &auth_helper, renew_auth)

.await?;

session.auth_helper = Some(Arc::new(auth_helper));

Ok(session)

}

/// Create a new OpenStack API session from `CloudConfig`, loading any

/// already-cached authorization but never performing network I/O to

/// discover endpoints or obtain a fresh token.

///

/// Intended for commands that only need to *report* on locally known

/// auth state (e.g. `osc auth status`) and must keep working even when

/// the cloud is unreachable or the cached token has expired.

#[instrument(name = "connect_cache_only", level = "trace", skip(config))]

pub fn new_cache_only(config: &CloudConfig) -> OpenStackResult<Self> {

let session = Self::new_impl(config, Auth::None)?;

let requested_scope = AuthTokenScope::try_from(&session.config).ok();

let cached = {

let mut inner = session.session_write("new_cache_only: get cache auth");

match &requested_scope {

Some(scope) => inner.state.get_scope_auth(scope),

None => inner.state.get_any_valid_auth(),

}

};

if let Some(auth) = cached {

session.set_auth(Auth::AuthToken(Box::new(auth)), true)?;

}

Ok(session)

}

#[instrument(name = "connect", level = "trace", skip(config))]

#[deprecated(

since = "0.22.0",

note = "please use `new_with_authentication_helper` instead"

)]

pub async fn new_interactive(config: &CloudConfig, renew_auth: bool) -> OpenStackResult<Self> {

Self::new_with_authentication_helper(config, Dialoguer::default(), renew_auth).await

}

/// Set the authorization to be used by the client

fn set_auth(&self, auth: Auth, skip_cache_update: bool) -> Result<(), OpenStackError> {

let mut session = self.session_write("set_auth");

session.auth = auth;

session.auth_generation = session.auth_generation.wrapping_add(1);

if !skip_cache_update && let Auth::AuthToken(a) = &session.auth {

let a_clone = a.clone();

let scope = match &a_clone.auth_info {

Some(info) => {

if info.token.application_credential.is_some() {

AuthTokenScope::Unscoped

} else {

a_clone.get_scope()

}

}

_ => a_clone.get_scope(),

};

session.state.set_scope_auth(&scope, &a_clone);

}

Ok(())

}

/// Authorize against the cloud using provided credentials and get the session token.

pub async fn authorize(

&self,

scope: Option<AuthTokenScope>,

interactive: bool,

renew_auth: bool,

) -> Result<(), OpenStackError> {

if interactive {

self.authorize_with_auth_helper(scope, &Dialoguer::default(), renew_auth)

.await

} else {

self.authorize_with_auth_helper(scope, &Noop::default(), renew_auth)

.await

}

}

/// Re-authenticate with the existing auth for the given scope.

async fn reauth(

&self,

auth: &AuthToken,

scope: &AuthTokenScope,

) -> Result<Auth, OpenStackError> {

Ok(token_auth::PLUGIN

.auth(

&self.client,

self.get_service_endpoint(&ServiceType::Identity, Some(&ApiVersion::from((3, 0))))

.await?

.url(),

&HashMap::from([("token".into(), auth.token.clone())]),

Some(scope),

None,

)

.await?)

}

/// Handle 401 Unauthorized by re-authenticating.

///

/// `seen_generation` is the `auth_generation` observed by the caller

/// before it hit the 401. Re-auth is single-flighted via `reauth_lock`:

/// if another concurrent caller already completed a re-auth while this

/// one was waiting for the lock, `auth_generation` will have moved on

/// and this call skips its own re-auth.

async fn handle_401_retry(&self, seen_generation: u64) -> Result<(), OpenStackError> {

let _guard = self.reauth_lock.lock().await;

let current_generation = self.auth_snapshot.load().auth_generation;

if current_generation != seen_generation {

// Another task already re-authed while we waited for the lock.

return Ok(());

}

let scope = AuthTokenScope::try_from(&self.config)?;

// Re-authenticate with current config. First try to reauth with existing token,

// fall back to full authentication if auth_helper is available.

let auth_helper_opt = self.auth_helper.clone();

if let Some(auth_helper) = auth_helper_opt {

{

let mut session = self.session_write("handle_401_retry: clear state");

session.state.clear_all_auth();

}

self.authorize_with_auth_helper(Some(scope), &auth_helper, true)

.await?;

} else {

// No auth helper available - try to reauth with the current token

if let Auth::AuthToken(token) = {

let session = self.session_read("handle_401_retry: read auth");

session.auth.clone()

} && let Ok(new_auth) = self.reauth(&token, &scope).await

{

self.set_auth(new_auth, false)?;

return Ok(());

}

// Reauth failed, clear state and try full auth as fallback

{

let mut session = self.session_write("handle_401_retry: clear state fallback");

session.state.clear_all_auth();

}

self.authorize_with_auth_helper(Some(scope), &Noop::default(), true)

.await?;

}

Ok(())

}

/// Disable authentication caching for this session.

///

/// Clears any cached tokens and prevents further caching — both the

/// in-memory store and the on-disk state file. Useful for scenarios

/// requiring fresh authentication on every request.

pub fn disable_auth_cache(&self) -> &Self {

self.session.write().state.disable_auth_cache();

self

}

/// Spawn a background task that proactively re-authenticates `margin`

/// before the current token expires, so the request hot path rarely

/// hits a 401 to begin with.

///

/// This performs a full, non-interactive re-authentication — *not*

/// token-exchange (`Self::reauth`, method=`token`). Keystone bakes the

/// original token's expiry into any token minted via token-exchange

/// (`keystone/auth/plugins/token.py`: `response_data.setdefault(

/// 'expires_at', token.expires_at)`, deliberately, to stop callers from

/// bumping expiry without reproving identity), so exchanging a token

/// for a token never actually extends the session — it would produce a

/// new token id with the exact same expiry as the one it replaced.

/// Reproving identity is required, so this clears cached auth state and

/// replays the credentials in `CloudConfig` (password / application

/// credential / client credentials) through [`Self::authorize_with_auth_helper`]

/// with [`Noop`] — deliberately never the configured interactive

/// `auth_helper`, since a background task cannot block on a prompt.

/// Clouds that need MFA/SSO/interactive input therefore don't renew

/// here; they keep relying on the interactive 401 retry path when a

/// user is actually present.

///

/// Re-authentication is single-flighted with the 401 retry path through

/// `reauth_lock`, so a concurrent 401 and a renewal wakeup never both

/// re-authenticate at once.

///

/// The task exits once it observes there is no `Auth::AuthToken` to

/// renew (unauthenticated session), once the last `AsyncOpenStack`

/// sharing this session is dropped, or when the returned handle is

/// dropped.

pub fn enable_auto_renew(&self, margin: TimeDelta) -> RenewHandle {

// Only `session` is held weakly — it's the field whose refcount

// marks "is any AsyncOpenStack for this session still alive".

// Everything else is cloned strong: none of it keeps `session`

// alive, and the task needs it to build a short-lived client for

// each wake-up without holding a strong `session` ref across the

// (possibly hours-long) sleep in between.

let weak_session: Weak<RwLock<session::SessionContext>> = Arc::downgrade(&self.session);

let reqwest_client = self.client.clone();

let config = self.config.clone();

let auth_snapshot = Arc::clone(&self.auth_snapshot);

let auth_helper = self.auth_helper.clone();

let max_auth_retries = self.max_auth_retries;

let reauth_lock = Arc::clone(&self.reauth_lock);

let join = tokio::spawn(async move {

// Rebuilds a client around the currently-live `session` Arc, or

// `None` once the last real `AsyncOpenStack` has been dropped.

let upgrade = || {

weak_session.upgrade().map(|session| AsyncOpenStack {

client: reqwest_client.clone(),

config: config.clone(),

session,

auth_snapshot: Arc::clone(&auth_snapshot),

auth_helper: auth_helper.clone(),

max_auth_retries,

reauth_lock: Arc::clone(&reauth_lock),

})

};

loop {

let sleep_for = {

let Some(client) = upgrade() else { return };

let Auth::AuthToken(token) = &client.auth_snapshot.load().auth else {

return;

};

let Some(auth_info) = &token.auth_info else {

return;

};

(auth_info.token.expires_at - margin - chrono::Utc::now())

.to_std()

.unwrap_or(Duration::ZERO)

// `client` (and its strong `session` ref) drops here,

// before the sleep below.

};

tokio::time::sleep(sleep_for).await;

let Some(client) = upgrade() else { return };

let seen_generation = client.auth_snapshot.load().auth_generation;

let _guard = client.reauth_lock.lock().await;

if client.auth_snapshot.load().auth_generation != seen_generation {

// Another task (401 retry or a previous renewal wakeup)

// already refreshed the token while we waited.

continue;

}

let scope = match AuthTokenScope::try_from(&client.config) {

Ok(s) => s,

Err(e) => {

warn!("background token renewal: cannot determine scope: {e}");

return;

}

};

// `authorize_with_auth_helper` re-processes the token

// catalog, which wipes all previously-discovered service

// endpoint versions (identity excepted) since a rescope may

// land on a different project_id. This renewal keeps the

// same scope, so remember what was discovered and redo it

// below rather than silently losing it.

let discovered_services = {

let session =

client.session_read("enable_auto_renew: snapshot discovered services");

session.catalog.discovered_service_types()

};

// Clear cached auth first: `authorize_with_auth_helper`

// would otherwise happily rescope a still-cached token via

// token-exchange, which (see doc comment above) preserves

// the old expiry instead of extending it. Clearing forces

// it down the real-credential path.

{

let mut session = client.session_write("enable_auto_renew: clear state");

session.state.clear_all_auth();

}

match client

.authorize_with_auth_helper(Some(scope), &Noop::default(), true)

.await

{

Ok(()) => {

for service_type in &discovered_services {

if let Err(e) = client

.discover_service_endpoint(&ServiceType::from(

service_type.as_str(),

))

.await

{

warn!(

"background token renewal: failed to re-discover endpoint for {service_type}: {e}"

);

}

}

let expires_at =

if let Auth::AuthToken(t) = &client.auth_snapshot.load().auth {

t.auth_info

.as_ref()

.map(|i| i.token.expires_at.to_rfc3339())

} else {

None

};

event!(

name: "token_renewed",

Level::DEBUG,

expires_at,

"Background token renewal succeeded"

)

}

Err(e) => {

warn!("background token renewal failed: {e}");

// Rate-limit retries; a subsequent 401 still falls

// back to the normal retry path in the meantime.

tokio::time::sleep(Duration::from_secs(30)).await;

}

}

}

});

RenewHandle(join)

}

/// Set the maximum number of retries on 401 responses.

pub fn set_max_auth_retries(&mut self, n: u32) -> &mut Self {

self.max_auth_retries = n;

self

}

/// Set the auth helper used for 401 re-authentication.

pub fn set_auth_helper<A>(&mut self, auth_helper: A)

where

A: AuthHelper + Sync + Send + 'static,

{

self.auth_helper = Some(Arc::new(auth_helper));

}

/// Authorize against the cloud using provided credentials and get the session token with the

/// auth helper that may be invoked to interactively ask for the credentials.

pub async fn authorize_with_auth_helper<A>(

&self,

scope: Option<AuthTokenScope>,

auth_helper: &A,

renew_auth: bool,

) -> Result<(), OpenStackError>

where

A: AuthHelper + Sync + Send + 'static,

{

let requested_scope =

scope.map_or_else(|| AuthTokenScope::try_from(&self.config), |v| Ok(v.clone()))?;

let auth_opt = {

let mut session = self.session_write("authorize_with_auth_helper: get cache auth");

session.state.get_scope_auth(&requested_scope)

};

if let Some(auth) = auth_opt

&& !renew_auth

{

// Valid authorization is already available and no renewal is required

trace!("Auth already available");

self.set_auth(Auth::AuthToken(Box::new(auth.clone())), true)?;

} else {

// No valid authorization data is available in the state or

// renewal is requested

let auth_type = AuthType::from_cloud_config(&self.config)?;

let force_new_auth = matches!(auth_type, AuthType::V3ApplicationCredential);

let available_auth_opt = {

let mut session = self.session_write("authorize_with_auth_helper: reauthz");

session.state.get_any_valid_auth()

};

if let (Some(available_auth), false) = (available_auth_opt, force_new_auth) {

// State contain valid authentication for different scope/unscoped. It is possible

// to request new authz using this other auth

trace!("Valid Auth is available for reauthz: {:?}", available_auth);

let token_auth = self.reauth(&available_auth, &requested_scope).await?;

self.set_auth(token_auth.clone(), false)?;

} else {

// No auth/authz information available or force_new_auth. Proceed with new auth

trace!("No Auth already available. Proceeding with new login");

let auth_type = auth_type.as_str();

// Find authenticator supporting the auth_type

if let Some(authenticator) = inventory::iter::<AuthPluginRegistration>

.into_iter()

.find(|x| x.method.get_supported_auth_methods().contains(&auth_type))

.map(|x| x.method)

{

// authenticate

let auth_hints = self

.config

.auth_methods

.as_ref()

.map(|methods| serde_json::json!({"auth_methods": methods}));

match authenticator

.auth(

&self.client,

self.get_service_endpoint(

&ServiceType::Identity,

Some(&ApiVersion::from(authenticator.api_version())),

)

.await?

.url(),

&gather_auth_data(

&authenticator.requirements(auth_hints.as_ref())?,

&self.config,

auth_helper,

)

.await?,

Some(&requested_scope),

auth_hints.as_ref(),

)

.await

{

Ok(token_auth) => {

self.set_auth(token_auth.clone(), false)?;

}

Err(AuthError::AuthReceipt(receipt)) => {

// Auth Receipt is received

// Find the receipt auth plugin

// Convert the receipt into auth hints

let auth_hints = serde_json::to_value(&receipt)?;

// Authenticate

let token_auth = token_receipt::PLUGIN

.auth(

&self.client,

self.get_service_endpoint(

&ServiceType::Identity,

Some(&ApiVersion::from(authenticator.api_version())),

)

.await?

.url(),

&gather_auth_data(

&token_receipt::PLUGIN.requirements(Some(&auth_hints))?,

&self.config,

auth_helper,

)

.await?,

Some(&requested_scope),

Some(&auth_hints),

)

.await?;

self.set_auth(token_auth.clone(), false)?;

}

Err(other) => {

return Err(other.into());

}

}

} else {

Err(AuthTokenError::IdentityMethod {

auth_type: auth_type.into(),

})?;

}

}

}

let auth_opt = {

let session = self.session_read("authorize_with_auth_helper: get auth");

session.auth.clone()

};

if let Auth::AuthToken(token_auth) = &auth_opt {

// When unscoped auth is requested (maybe by not specifying the scope) we expect that

// the Authenticator returns the necessary scope (i.e. for ApplicationCredentials it is

// not possible to request scope, but the Keystone manages the scope) or unscoped.

// In this case we just save the auth as unscoped (in addition to what was done above).

// Otherwise we should rescope.

if token_auth.auth_info.is_none() {

let mut resolved_token = token_auth.clone();

let token_info = self.fetch_token_info(token_auth.token.clone()).await?;

resolved_token.auth_info = Some(token_info.clone());

let scope = AuthTokenScope::from(&token_info);

// Save unscoped token in the cache

{

let mut session =

self.session_write("authorize_with_auth_helper: set unscoped cache");

session.state.set_scope_auth(&scope, &resolved_token);

}

}

if requested_scope != AuthTokenScope::Unscoped

&& !token_auth

.auth_info

.as_ref()

.map(AuthTokenScope::from)

.is_some_and(|scope| requested_scope.matches(&scope))

{

// And now time to rescope the token

let token_auth = self.reauth(token_auth, &requested_scope).await?;

self.set_auth(token_auth.clone(), false)?;

} else {

// Client may not specify the target scope expecting the mapping to set

// the proper token. Save the auth as unscope (similarly to the AppCred

// handling).

{

let mut session =

self.session_write("authorize_with_auth_helper: set unscoped");

session

.state

.set_scope_auth(&AuthTokenScope::Unscoped, token_auth);

}

}

} else {

Err(OpenStackError::NoAuth)?;

}

{

let mut session = self.session_write("authorize_with_auth_helper: process catalog");

let auth_data = match &session.auth {

Auth::AuthToken(token_data) => match &token_data.auth_info {

Some(ad) => ad.clone(),

_ => return Err(OpenStackError::NoAuth),

},

_ => return Err(OpenStackError::NoAuth),

};

if let Some(project) = &auth_data.token.project {

session.catalog.set_project_id(project.id.clone());

session.catalog.configure(&self.config)?;

}

if let Some(endpoints) = &auth_data.token.catalog {

session.catalog.process_catalog_endpoints(endpoints)?;

} else {

error!("No catalog information");

}

}

Ok(())

}

/// Perform version discovery of a service

#[instrument(skip(self))]

pub async fn discover_service_endpoint(

&self,

service_type: &ServiceType,

) -> Result<(), OpenStackError> {

let ep_opt = {

let session = self.session_read("discover_service_endpoint: get ep");

let region = session

.region_name

.as_ref()

.or(self.config.region_name.as_ref());

session

.catalog

.get_service_endpoint(

service_type.to_string(),

None,

region,

self.config.interface.as_ref(),

)

.ok()

.cloned()

};

let ep = match ep_opt {

Some(e) => e,

None => return Ok(()),