use crate::cloud::cloud_browse::Environment;
use crate::cloud::cloud_command::CommandError;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, SystemTime};
pub const STORAGE_SCOPE: &str = "https://storage.azure.com/.default";
pub const MANAGEMENT_SCOPE: &str = "https://management.azure.com/.default";
pub const API_VERSION: &str = "2023-11-03";
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum AzureAuth {
#[default]
None,
AzCli,
PowerShell,
ServicePrincipal(ServicePrincipal),
ManagedIdentity,
KeyCommand(String),
Bearer(String),
Key(String),
Sas(String),
}
#[derive(Clone, PartialEq, Eq, Default)]
pub struct ServicePrincipal {
pub tenant: String,
pub client_id: String,
pub secret: Option<String>,
pub token_file: Option<PathBuf>,
pub authority: String,
}
impl std::fmt::Debug for ServicePrincipal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ServicePrincipal")
.field("tenant", &self.tenant)
.field("client_id", &self.client_id)
.field("secret", &self.secret.as_ref().map(|_| "…"))
.field("token_file", &self.token_file)
.finish()
}
}
impl AzureAuth {
pub fn is_identity(&self) -> bool {
matches!(
self,
AzureAuth::AzCli
| AzureAuth::PowerShell
| AzureAuth::ServicePrincipal(_)
| AzureAuth::ManagedIdentity
)
}
pub fn describe(&self) -> &'static str {
match self {
AzureAuth::None => "none",
AzureAuth::AzCli => "az login",
AzureAuth::PowerShell => "Azure PowerShell",
AzureAuth::ServicePrincipal(sp) if sp.token_file.is_some() => "workload identity",
AzureAuth::ServicePrincipal(_) => "service principal",
AzureAuth::ManagedIdentity => "managed identity",
AzureAuth::KeyCommand(_) => "secret_command",
AzureAuth::Bearer(_) => "token",
AzureAuth::Key(_) => "access key",
AzureAuth::Sas(_) => "SAS token",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct AzureSettings {
pub account: Option<String>,
pub auth: AzureAuth,
pub blob_endpoint: Option<String>,
pub use_emulator: bool,
pub identity: Option<AzureAuth>,
}
impl AzureSettings {
pub fn blob_endpoint_for(&self, account: &str) -> String {
match (&self.blob_endpoint, self.use_emulator) {
(Some(endpoint), _) => format!("{}/", endpoint.trim_end_matches('/')),
(None, true) => format!("http://127.0.0.1:10000/{account}/"),
(None, false) => format!("https://{account}.blob.core.windows.net/"),
}
}
pub fn with_token(mut self, env: &Environment<'_>) -> Result<Self, String> {
if let AzureAuth::KeyCommand(command) = &self.auth {
self.auth = AzureAuth::Key(crate::cloud::cloud_command::secret(command, env)?);
}
if self.auth.is_identity() {
let token = identity_token(&self.auth, STORAGE_SCOPE, env)?;
self.identity = Some(std::mem::replace(&mut self.auth, AzureAuth::Bearer(token)));
}
Ok(self)
}
}
const AZURITE_ACCOUNT: &str = "devstoreaccount1";
const AZURITE_KEY: &str =
"Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
pub fn parse_connection_string(text: &str) -> Option<AzureSettings> {
let pairs: HashMap<String, String> = text
.split(';')
.filter_map(|pair| pair.split_once('='))
.map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string()))
.collect();
if pairs
.get("usedevelopmentstorage")
.is_some_and(|v| v.eq_ignore_ascii_case("true"))
{
return Some(AzureSettings {
account: Some(AZURITE_ACCOUNT.to_string()),
auth: AzureAuth::Key(AZURITE_KEY.to_string()),
blob_endpoint: None,
use_emulator: true,
identity: None,
});
}
let account = pairs.get("accountname").cloned();
let auth = match pairs.get("accountkey") {
Some(key) => AzureAuth::Key(key.clone()),
None => AzureAuth::Sas(
pairs
.get("sharedaccesssignature")?
.trim_start_matches('?')
.to_string(),
),
};
let blob_endpoint = pairs.get("blobendpoint").cloned().or_else(|| {
let account = account.as_ref()?;
let suffix = pairs.get("endpointsuffix")?;
let protocol = pairs
.get("defaultendpointsprotocol")
.map(String::as_str)
.unwrap_or("https");
Some(format!("{protocol}://{account}.blob.{suffix}"))
});
let account = account.or_else(|| {
let endpoint = blob_endpoint.as_deref()?;
let host = endpoint.split_once("://")?.1.split('/').next()?;
host.split_once(".blob.").map(|(a, _)| a.to_string())
})?;
Some(AzureSettings {
account: Some(account),
auth,
blob_endpoint,
use_emulator: false,
identity: None,
})
}
pub fn from_environment(var: &dyn Fn(&str) -> Option<String>) -> Option<(AzureSettings, String)> {
let set = |key: &str| {
var(key)
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
};
if let Some(text) = set("AZURE_STORAGE_CONNECTION_STRING") {
return parse_connection_string(&text)
.map(|s| (s, "AZURE_STORAGE_CONNECTION_STRING".to_string()));
}
let account = set("AZURE_STORAGE_ACCOUNT_NAME");
let key = set("AZURE_STORAGE_ACCOUNT_KEY").or_else(|| set("AZURE_STORAGE_ACCESS_KEY"));
let sas = set("AZURE_STORAGE_SAS_TOKEN").or_else(|| set("AZURE_STORAGE_SAS_KEY"));
let (auth, origin) = match (&account, key, sas, service_principal(var)) {
(Some(_), Some(key), _, _) => (AzureAuth::Key(key), "AZURE_STORAGE_ACCOUNT_KEY"),
(Some(_), None, Some(sas), _) => (
AzureAuth::Sas(sas.trim_start_matches('?').to_string()),
"AZURE_STORAGE_SAS_TOKEN",
),
(_, _, _, Some(sp)) => {
let origin = if sp.token_file.is_some() {
"AZURE_FEDERATED_TOKEN_FILE"
} else {
"AZURE_CLIENT_SECRET"
};
(AzureAuth::ServicePrincipal(sp), origin)
}
_ => return None,
};
Some((
AzureSettings {
account,
auth,
blob_endpoint: None,
use_emulator: false,
identity: None,
},
origin.to_string(),
))
}
pub fn service_principal(var: &dyn Fn(&str) -> Option<String>) -> Option<ServicePrincipal> {
let set = |key: &str| {
var(key)
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
};
let tenant = set("AZURE_TENANT_ID")?;
let client_id = set("AZURE_CLIENT_ID")?;
let secret = set("AZURE_CLIENT_SECRET");
let token_file = set("AZURE_FEDERATED_TOKEN_FILE").map(PathBuf::from);
if secret.is_none() && token_file.is_none() {
return None;
}
Some(ServicePrincipal {
tenant,
client_id,
secret,
token_file,
authority: set("AZURE_AUTHORITY_HOST")
.unwrap_or_else(|| "https://login.microsoftonline.com".to_string()),
})
}
pub fn az_login_evidence(env: &Environment<'_>) -> bool {
let dir = match (env.var)("AZURE_CONFIG_DIR").filter(|v| !v.trim().is_empty()) {
Some(dir) => PathBuf::from(dir.trim()),
None => match &env.home {
Some(home) => home.join(".azure"),
None => return false,
},
};
(env.exists)(&dir)
}
pub fn powershell_login_evidence(env: &Environment<'_>) -> bool {
powershell_context(env).is_some_and(|path| (env.exists)(&path))
}
fn powershell_context(env: &Environment<'_>) -> Option<PathBuf> {
Some(
env.home
.as_ref()?
.join(".Azure")
.join("AzureRmContext.json"),
)
}
pub fn not_signed_in(env: &Environment<'_>) -> Option<String> {
if az_login_evidence(env) || powershell_login_evidence(env) {
return None;
}
let on_path = |names: &[&str]| {
(env.var)("PATH").is_some_and(|path| {
std::env::split_paths(&path)
.any(|dir| names.iter().any(|name| (env.exists)(&dir.join(name))))
})
};
let az = if env.windows {
on_path(&["az.cmd", "az.exe"])
} else {
on_path(&["az"])
};
let az_accounts = (env.var)("PSModulePath").is_some_and(|paths| {
std::env::split_paths(&paths).any(|dir| (env.exists)(&dir.join("Az.Accounts")))
});
match (az, az_accounts) {
(false, false) => None,
(true, false) => Some("not signed in: run az login".to_string()),
(false, true) => Some("not signed in: run Connect-AzAccount in PowerShell".to_string()),
(true, true) => {
Some("not signed in: run az login, or Connect-AzAccount in PowerShell".to_string())
}
}
}
pub fn identity_token(
auth: &AzureAuth,
scope: &str,
env: &Environment<'_>,
) -> Result<String, String> {
match auth {
AzureAuth::AzCli => match token(scope, env) {
Ok(token) => Ok(token),
Err(az) if powershell_login_evidence(env) => {
powershell_token(scope, env).map_err(|ps| format!("{az}; {ps}"))
}
Err(e) => Err(e),
},
AzureAuth::PowerShell => powershell_token(scope, env),
AzureAuth::ServicePrincipal(sp) => service_principal_token(sp, scope, env),
AzureAuth::ManagedIdentity => managed_identity_token(scope, env),
AzureAuth::Bearer(token) => Ok(token.clone()),
AzureAuth::None | AzureAuth::Key(_) | AzureAuth::Sas(_) | AzureAuth::KeyCommand(_) => {
Err("this login has no tokens".to_string())
}
}
}
fn tokens() -> &'static crate::cloud::cloud_command::Expiring<String> {
static TOKENS: OnceLock<crate::cloud::cloud_command::Expiring<String>> = OnceLock::new();
TOKENS.get_or_init(Default::default)
}
pub fn token(scope: &str, env: &Environment<'_>) -> Result<String, String> {
if let Some(token) = tokens().get(scope) {
return Ok(token);
}
let output = (env.run)(
"az",
&[
"account",
"get-access-token",
"--scope",
scope,
"--output",
"json",
],
)
.map_err(|e| match e {
CommandError::Missing(_) => "needs the Azure CLI".to_string(),
CommandError::Failed(message) if message.contains("az login") => {
format!("not logged in: run az login ({})", message.trim())
}
other => other.to_string(),
})?;
let (token, expires) =
parse_token(&output).ok_or_else(|| "az returned no token".to_string())?;
cache_token(scope, token.clone(), expires);
Ok(token)
}
pub fn parse_token(text: &str) -> Option<(String, Option<SystemTime>)> {
let value: serde_json::Value = serde_json::from_str(text.trim()).ok()?;
let token = value.get("accessToken")?.as_str()?.to_string();
if token.is_empty() {
return None;
}
let expires = value
.get("expires_on")
.and_then(|v| v.as_u64().or_else(|| v.as_str()?.parse().ok()))
.map(|secs| SystemTime::UNIX_EPOCH + Duration::from_secs(secs));
Some((token, expires))
}
const POWERSHELL_SCRIPT: &str = "$ErrorActionPreference = 'Stop'; \
Import-Module Az.Accounts; \
$out = @{}; \
foreach ($r in @('https://storage.azure.com/', 'https://management.azure.com/')) { \
$t = Get-AzAccessToken -ResourceUrl $r; \
$tok = $t.Token; \
if ($tok -is [System.Security.SecureString]) { $tok = [System.Net.NetworkCredential]::new('', $tok).Password }; \
$out[$r] = @{ token = $tok; expires = $t.ExpiresOn.ToUnixTimeSeconds() } \
}; \
$out | ConvertTo-Json -Compress";
fn powershell_token(scope: &str, env: &Environment<'_>) -> Result<String, String> {
let key = |scope: &str| format!("powershell {scope}");
if let Some(token) = tokens().get(&key(scope)) {
return Ok(token);
}
let args = [
"-NoProfile",
"-NonInteractive",
"-Command",
POWERSHELL_SCRIPT,
];
let mut output = (env.run)("pwsh", &args);
if env.windows && matches!(output, Err(CommandError::Missing(_))) {
output = (env.run)("powershell", &args);
}
let output = output.map_err(|e| match e {
CommandError::Missing(_) => "needs PowerShell".to_string(),
CommandError::Failed(message) if message.contains("Connect-AzAccount") => {
format!("not signed in: run Connect-AzAccount ({})", message.trim())
}
CommandError::Failed(message) if message.contains("Az.Accounts") => {
"needs the Az.Accounts PowerShell module".to_string()
}
other => other.to_string(),
})?;
let tokens = parse_powershell_tokens(&output)
.ok_or_else(|| "Azure PowerShell returned no token".to_string())?;
let mut wanted = None;
for (resource, token, expires) in tokens {
let scope_of = format!("{resource}.default");
if scope_of == scope {
wanted = Some(token.clone());
}
cache_token(&key(&scope_of), token, Some(expires));
}
wanted.ok_or_else(|| format!("Azure PowerShell returned no token for {scope}"))
}
pub fn parse_powershell_tokens(text: &str) -> Option<Vec<(String, String, SystemTime)>> {
let value: serde_json::Value = serde_json::from_str(text.trim()).ok()?;
let tokens: Vec<_> = value
.as_object()?
.iter()
.filter_map(|(resource, entry)| {
let token = entry.get("token")?.as_str()?.to_string();
let expires = entry.get("expires")?.as_u64()?;
(!token.is_empty()).then(|| {
(
resource.clone(),
token,
SystemTime::UNIX_EPOCH + Duration::from_secs(expires),
)
})
})
.collect();
(!tokens.is_empty()).then_some(tokens)
}
fn service_principal_token(
sp: &ServicePrincipal,
scope: &str,
env: &Environment<'_>,
) -> Result<String, String> {
let key = format!("sp {} {} {scope}", sp.tenant, sp.client_id);
if let Some(token) = tokens().get(&key) {
return Ok(token);
}
let mut form: Vec<(&str, String)> = vec![
("client_id", sp.client_id.clone()),
("scope", scope.to_string()),
("grant_type", "client_credentials".to_string()),
];
match (&sp.token_file, &sp.secret) {
(Some(file), _) => {
let assertion = (env.read)(file).ok_or_else(|| {
format!("cannot read AZURE_FEDERATED_TOKEN_FILE {}", file.display())
})?;
form.push((
"client_assertion_type",
"urn:ietf:params:oauth:client-assertion-type:jwt-bearer".to_string(),
));
form.push(("client_assertion", assertion.trim().to_string()));
}
(None, Some(secret)) => form.push(("client_secret", secret.clone())),
(None, None) => return Err("the service principal has no secret".to_string()),
}
let body = form
.iter()
.map(|(k, v)| format!("{k}={}", crate::cloud::cloud_browse::urlencode(v)))
.collect::<Vec<_>>()
.join("&");
let url = format!(
"{}/{}/oauth2/v2.0/token",
sp.authority.trim_end_matches('/'),
crate::cloud::cloud_browse::urlencode(&sp.tenant)
);
let mut response = crate::cloud::cloud_browse::http_agent()
.post(&url)
.config()
.http_status_as_error(false)
.build()
.header("Content-Type", "application/x-www-form-urlencoded")
.send(body)
.map_err(|e| format!("{e}"))?;
let status = response.status().as_u16();
let text = response
.body_mut()
.read_to_string()
.map_err(|e| format!("could not read the response: {e}"))?;
let (token, expires) = parse_entra_token(&text).ok_or_else(|| {
let value: Option<serde_json::Value> = serde_json::from_str(&text).ok();
let description = value
.as_ref()
.and_then(|v| v.get("error_description"))
.and_then(|d| d.as_str())
.and_then(|d| d.lines().next())
.unwrap_or("no token in the response");
format!("not logged in: the service principal was refused ({status}): {description}")
})?;
cache_token(&key, token.clone(), Some(expires));
Ok(token)
}
fn managed_identity_token(scope: &str, env: &Environment<'_>) -> Result<String, String> {
let key = format!("managed {scope}");
if let Some(token) = tokens().get(&key) {
return Ok(token);
}
let resource = crate::cloud::cloud_browse::urlencode(scope.trim_end_matches(".default"));
let client = (env.var)("AZURE_CLIENT_ID")
.map(|id| format!("&client_id={}", crate::cloud::cloud_browse::urlencode(&id)))
.unwrap_or_default();
let set = |k: &str| (env.var)(k).filter(|v| !v.trim().is_empty());
let (url, header) = if let (Some(endpoint), Some(secret)) =
(set("IDENTITY_ENDPOINT"), set("IDENTITY_HEADER"))
{
(
format!("{endpoint}?api-version=2019-08-01&resource={resource}{client}"),
("X-IDENTITY-HEADER", secret),
)
} else if let Some(endpoint) = set("MSI_ENDPOINT") {
(
format!("{endpoint}?api-version=2017-09-01&resource={resource}{client}"),
match set("MSI_SECRET") {
Some(secret) => ("secret", secret),
None => ("Metadata", "true".to_string()),
},
)
} else {
(
format!(
"http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource={resource}{client}"
),
("Metadata", "true".to_string()),
)
};
let mut response = crate::cloud::user_agent::ureq_config()
.timeout_global(Some(Duration::from_secs(5)))
.http_status_as_error(false)
.build()
.new_agent()
.get(&url)
.header(header.0, &header.1)
.call()
.map_err(|e| format!("no managed identity answered: {e}"))?;
let status = response.status().as_u16();
let text = response
.body_mut()
.read_to_string()
.map_err(|e| format!("could not read the response: {e}"))?;
let (token, expires) = parse_entra_token(&text)
.ok_or_else(|| format!("the managed identity returned no token ({status})"))?;
cache_token(&key, token.clone(), Some(expires));
Ok(token)
}
pub fn parse_entra_token(text: &str) -> Option<(String, SystemTime)> {
let value: serde_json::Value = serde_json::from_str(text).ok()?;
let token = value.get("access_token")?.as_str()?.to_string();
let number = |key: &str| {
value
.get(key)
.and_then(|v| v.as_u64().or_else(|| v.as_str()?.parse().ok()))
};
let expires = match (number("expires_on"), number("expires_in")) {
(Some(on), _) => SystemTime::UNIX_EPOCH + Duration::from_secs(on),
(None, Some(within)) => SystemTime::now() + Duration::from_secs(within),
(None, None) => SystemTime::now() + Duration::from_secs(300),
};
(!token.is_empty()).then_some((token, expires))
}
fn cache_token(key: &str, token: String, expires: Option<SystemTime>) {
crate::logging::keep_out_of_log(&token);
tokens().put(key, token, expires);
}
fn account_keys() -> &'static Mutex<HashMap<String, String>> {
static KEYS: OnceLock<Mutex<HashMap<String, String>>> = OnceLock::new();
KEYS.get_or_init(Default::default)
}
#[cfg(test)]
pub(crate) fn remember_key_for_test(account: &str, key: &str) {
if let Ok(mut keys) = account_keys().lock() {
keys.insert(account.to_string(), key.to_string());
}
}
pub fn remembered_key(account: &str) -> Option<String> {
account_keys().lock().ok()?.get(account).cloned()
}
fn token_readable_accounts() -> &'static Mutex<std::collections::HashSet<String>> {
static ACCOUNTS: OnceLock<Mutex<std::collections::HashSet<String>>> = OnceLock::new();
ACCOUNTS.get_or_init(Default::default)
}
pub fn token_reads(account: &str) -> bool {
token_readable_accounts()
.lock()
.is_ok_and(|accounts| accounts.contains(account))
}
pub fn remember_token_reads(account: &str) {
if let Ok(mut accounts) = token_readable_accounts().lock() {
accounts.insert(account.to_string());
}
}
pub fn check_read(
account: &str,
container: &str,
path: &str,
settings: &AzureSettings,
) -> Result<(), String> {
let directory = match path.trim_start_matches('/').rsplit_once('/') {
Some((directory, _)) => format!("{directory}/"),
None => String::new(),
};
let url = format!(
"{}{}?restype=container&comp=list&maxresults=1&prefix={}",
settings.blob_endpoint_for(account),
crate::cloud::cloud_browse::urlencode(container),
crate::cloud::cloud_browse::urlencode(&directory)
);
send_signed(&url, account, settings).map(|_| ())
}
pub fn is_permission_mismatch(error: &str) -> bool {
error.contains("AuthorizationPermissionMismatch")
}
pub fn fetch_account_key(
account: &str,
identity: &AzureAuth,
env: &Environment<'_>,
) -> Result<String, String> {
if let Some(key) = remembered_key(account) {
return Ok(key);
}
let management = identity_token(identity, MANAGEMENT_SCOPE, env)?;
let query = format!(
"resources | where type =~ 'microsoft.storage/storageaccounts' and name =~ '{}' \
| project id, name, sharedKey=properties.allowSharedKeyAccess",
account.replace('\'', "")
);
let body = serde_json::json!({ "query": query });
let text = crate::cloud::cloud_browse::http_agent()
.post("https://management.azure.com/providers/Microsoft.ResourceGraph/resources?api-version=2024-04-01")
.header("Authorization", &format!("Bearer {management}"))
.header("Content-Type", "application/json")
.send(body.to_string())
.map_err(|e| format!("{e}"))?
.body_mut()
.read_to_string()
.map_err(|e| format!("could not read the response: {e}"))?;
let (id, shared_key) = parse_account_id(&text)
.ok_or_else(|| format!("{account} is not an account this login can manage"))?;
if !shared_key {
return Err("shared-key access is disabled on the account".to_string());
}
let mut response = crate::cloud::cloud_browse::http_agent()
.post(&format!(
"https://management.azure.com{id}/listKeys?api-version=2023-01-01"
))
.config()
.http_status_as_error(false)
.build()
.header("Authorization", &format!("Bearer {management}"))
.header("Content-Length", "0")
.send_empty()
.map_err(|e| format!("{e}"))?;
let status = response.status().as_u16();
let text = response
.body_mut()
.read_to_string()
.map_err(|e| format!("could not read the response: {e}"))?;
if status != 200 {
return Err(format!(
"the login may not list the account's keys ({status})"
));
}
let key = parse_keys(&text).ok_or_else(|| "the account returned no keys".to_string())?;
crate::logging::keep_out_of_log(&key);
if let Ok(mut keys) = account_keys().lock() {
keys.insert(account.to_string(), key.clone());
}
Ok(key)
}
pub fn parse_account_id(text: &str) -> Option<(String, bool)> {
let value: serde_json::Value = serde_json::from_str(text).ok()?;
let row = value.get("data")?.as_array()?.first()?;
let id = row.get("id")?.as_str()?.to_string();
let shared_key = row.get("sharedKey").and_then(|v| v.as_bool()) != Some(false);
Some((id, shared_key))
}
pub fn parse_keys(text: &str) -> Option<String> {
let value: serde_json::Value = serde_json::from_str(text).ok()?;
value
.get("keys")?
.as_array()?
.iter()
.find_map(|k| k.get("value")?.as_str().map(str::to_string))
.filter(|k| !k.is_empty())
}
pub fn with_account_key(
account: &str,
settings: &AzureSettings,
refusal: &str,
enabled: bool,
env: &Environment<'_>,
) -> Result<AzureSettings, String> {
let identity = match &settings.identity {
Some(identity) if enabled && is_permission_mismatch(refusal) => identity,
_ => return Err(refusal.to_string()),
};
match fetch_account_key(account, identity, env) {
Ok(key) => Ok(AzureSettings {
auth: AzureAuth::Key(key),
..settings.clone()
}),
Err(why) => Err(format!(
"{refusal}. The account's keys were not used: {why}"
)),
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct Account {
pub name: String,
pub subscription: Option<String>,
pub location: Option<String>,
pub hierarchical_namespace: bool,
pub blob_endpoint: Option<String>,
pub private_network: bool,
pub shared_key_access: bool,
}
const ACCOUNTS_QUERY: &str = "resources \
| where type =~ 'microsoft.storage/storageaccounts' \
| join kind=leftouter (resourcecontainers \
| where type =~ 'microsoft.resources/subscriptions' \
| project subscriptionId, subscription=name) on subscriptionId \
| project id, name, subscription, location, \
hns=properties.isHnsEnabled, \
blob=properties.primaryEndpoints.blob, \
publicNetworkAccess=properties.publicNetworkAccess, \
defaultAction=properties.networkAcls.defaultAction, \
sharedKey=properties.allowSharedKeyAccess \
| order by name asc";
const MAX_ACCOUNT_PAGES: usize = 20;
pub fn discover_accounts(
identity: &AzureAuth,
env: &Environment<'_>,
) -> Result<Vec<Account>, String> {
let token = identity_token(identity, MANAGEMENT_SCOPE, env)?;
crate::cloud::cloud_command::paged(MAX_ACCOUNT_PAGES, |skip| {
let mut options = serde_json::json!({ "$top": 1000 });
if let Some(skip) = skip {
options["$skipToken"] = serde_json::Value::String(skip.to_string());
}
let body = serde_json::json!({ "query": ACCOUNTS_QUERY, "options": options });
let text = crate::cloud::cloud_browse::http_agent()
.post("https://management.azure.com/providers/Microsoft.ResourceGraph/resources?api-version=2024-04-01")
.header("Authorization", &format!("Bearer {token}"))
.header("Content-Type", "application/json")
.send(body.to_string())
.map_err(|e| format!("{e}"))?
.body_mut()
.read_to_string()
.map_err(|e| format!("could not read the response: {e}"))?;
parse_accounts(&text)
})
}
pub fn parse_accounts(text: &str) -> Result<(Vec<Account>, Option<String>), String> {
let value: serde_json::Value =
serde_json::from_str(text).map_err(|e| format!("not JSON: {e}"))?;
if let Some(message) = value
.get("error")
.and_then(|e| e.get("message"))
.and_then(|m| m.as_str())
{
return Err(message.to_string());
}
let rows = value
.get("data")
.and_then(|d| d.as_array())
.ok_or_else(|| "no data in the response".to_string())?;
let text_of = |row: &serde_json::Value, key: &str| {
row.get(key)
.and_then(|v| v.as_str())
.map(str::to_string)
.filter(|v| !v.is_empty())
};
let accounts = rows
.iter()
.filter_map(|row| {
Some(Account {
name: text_of(row, "name")?,
subscription: text_of(row, "subscription"),
location: text_of(row, "location"),
hierarchical_namespace: row.get("hns").and_then(|v| v.as_bool()) == Some(true),
blob_endpoint: text_of(row, "blob"),
private_network: text_of(row, "publicNetworkAccess").as_deref() == Some("Disabled")
|| text_of(row, "defaultAction").as_deref() == Some("Deny"),
shared_key_access: row.get("sharedKey").and_then(|v| v.as_bool()) != Some(false),
})
})
.collect();
let next = value
.get("$skipToken")
.and_then(|v| v.as_str())
.map(str::to_string);
Ok((accounts, next))
}
pub fn list_containers(account: &str, settings: &AzureSettings) -> Result<Vec<String>, String> {
let endpoint = settings.blob_endpoint_for(account);
let mut names = crate::cloud::cloud_command::paged(MAX_ACCOUNT_PAGES, |marker| {
let mut url = format!("{endpoint}?comp=list&maxresults=5000");
if let Some(marker) = marker {
url.push_str(&format!(
"&marker={}",
crate::cloud::cloud_browse::urlencode(marker)
));
}
parse_containers(&send_signed(&url, account, settings)?)
})?;
names.sort();
Ok(names)
}
fn send_signed(url: &str, account: &str, settings: &AzureSettings) -> Result<String, String> {
let url = match &settings.auth {
AzureAuth::Sas(sas) => format!("{url}&{sas}"),
_ => url.to_string(),
};
let mut request = http::Request::builder()
.method("GET")
.uri(&url)
.header("x-ms-version", API_VERSION)
.body(object_store::client::HttpRequestBody::empty())
.map_err(|e| format!("could not build the request: {e}"))?;
match &settings.auth {
AzureAuth::Bearer(token) => {
let credential = object_store::azure::AzureCredential::BearerToken(token.clone());
object_store::azure::AzureAuthorizer::new(&credential, account).authorize(&mut request);
}
AzureAuth::Key(key) => {
let key = object_store::azure::AzureAccessKey::try_new(key)
.map_err(|e| format!("the account key is not valid: {e}"))?;
let credential = object_store::azure::AzureCredential::AccessKey(key);
object_store::azure::AzureAuthorizer::new(&credential, account).authorize(&mut request);
}
AzureAuth::Sas(_)
| AzureAuth::None
| AzureAuth::AzCli
| AzureAuth::PowerShell
| AzureAuth::ServicePrincipal(_)
| AzureAuth::ManagedIdentity
| AzureAuth::KeyCommand(_) => {}
}
let mut call = crate::cloud::cloud_browse::http_agent()
.get(&url)
.config()
.http_status_as_error(false)
.build();
for (name, value) in request.headers() {
if let Ok(value) = value.to_str() {
call = call.header(name.as_str(), value);
}
}
let mut response = call.call().map_err(|e| format!("{e}"))?;
let status = response.status();
let text = response
.body_mut()
.read_to_string()
.map_err(|e| format!("could not read the response: {e}"))?;
if status.is_success() {
Ok(text)
} else {
Err(describe_error(status.as_u16(), &text))
}
}
pub fn describe_error(status: u16, body: &str) -> String {
let field = |name: &str| {
let open = format!("<{name}>");
let close = format!("</{name}>");
let start = body.find(&open)? + open.len();
let end = body[start..].find(&close)? + start;
Some(body[start..end].trim().to_string())
};
let code = field("Code").unwrap_or_default();
let message = field("Message")
.and_then(|m| m.lines().next().map(str::to_string))
.unwrap_or_default();
let mut text = format!("{status} {code}");
if !message.is_empty() {
text.push_str(&format!(": {message}"));
}
if code == "AuthorizationPermissionMismatch" {
text.push_str(
". Reading blobs with a sign-in needs the Storage Blob Data Reader role on the \
account, or an ACL on the path when it has hierarchical namespace",
);
}
text
}
pub fn parse_containers(text: &str) -> Result<(Vec<String>, Option<String>), String> {
let mut names = Vec::new();
let mut rest = text;
while let Some(start) = rest.find("<Container>") {
rest = &rest[start + "<Container>".len()..];
let end = rest.find("</Container>").unwrap_or(rest.len());
let block = &rest[..end];
if let Some(name_start) = block.find("<Name>") {
let after = &block[name_start + "<Name>".len()..];
if let Some(name_end) = after.find("</Name>") {
let name = after[..name_end].trim();
if !name.is_empty() {
names.push(name.to_string());
}
}
}
rest = &rest[end..];
}
if names.is_empty() && !text.contains("<EnumerationResults") {
return Err("not a container listing".to_string());
}
let next = text
.find("<NextMarker>")
.and_then(|start| {
let after = &text[start + "<NextMarker>".len()..];
after
.find("</NextMarker>")
.map(|end| after[..end].trim().to_string())
})
.filter(|m| !m.is_empty());
Ok((names, next))
}
pub fn store(
account: &str,
container: &str,
settings: &AzureSettings,
) -> Result<object_store::azure::MicrosoftAzure, String> {
let mut builder = object_store::azure::MicrosoftAzureBuilder::new()
.with_account(account)
.with_container_name(container)
.with_config(
object_store::azure::AzureConfigKey::Client(crate::cloud::user_agent::CLIENT_KEY),
crate::cloud::user_agent::get(),
);
if settings.use_emulator {
builder = builder.with_use_emulator(true);
}
if let Some(endpoint) = &settings.blob_endpoint {
builder = builder
.with_endpoint(endpoint.trim_end_matches('/').to_string())
.with_allow_http(endpoint.starts_with("http://"));
}
builder = match &settings.auth {
AzureAuth::Bearer(token) => builder.with_bearer_token_authorization(token.clone()),
AzureAuth::Key(key) => builder.with_access_key(key.clone()),
AzureAuth::Sas(sas) => {
builder.with_config(object_store::azure::AzureConfigKey::SasKey, sas.clone())
}
AzureAuth::None
| AzureAuth::AzCli
| AzureAuth::PowerShell
| AzureAuth::ServicePrincipal(_)
| AzureAuth::ManagedIdentity
| AzureAuth::KeyCommand(_) => builder.with_skip_signature(true),
};
builder
.build()
.map_err(|e| format!("Azure is not configured: {e}"))
}
pub fn polars_options(
account: &str,
settings: &AzureSettings,
) -> Vec<(object_store::azure::AzureConfigKey, String)> {
use object_store::azure::AzureConfigKey;
let mut options = vec![(AzureConfigKey::AccountName, account.to_string())];
if settings.use_emulator {
options.push((AzureConfigKey::UseEmulator, "true".to_string()));
}
if let Some(endpoint) = &settings.blob_endpoint {
options.push((AzureConfigKey::Endpoint, endpoint.clone()));
}
match &settings.auth {
AzureAuth::Bearer(token) => options.push((AzureConfigKey::Token, token.clone())),
AzureAuth::Key(key) => options.push((AzureConfigKey::AccessKey, key.clone())),
AzureAuth::Sas(sas) => options.push((AzureConfigKey::SasKey, sas.clone())),
AzureAuth::None
| AzureAuth::AzCli
| AzureAuth::PowerShell
| AzureAuth::ServicePrincipal(_)
| AzureAuth::ManagedIdentity
| AzureAuth::KeyCommand(_) => {
options.push((AzureConfigKey::SkipSignature, "true".to_string()))
}
}
options
}
pub fn is_folder_marker(name: &str, size: u64, prefixes: &[String]) -> bool {
size == 0 && prefixes.iter().any(|p| p.trim_end_matches('/') == name)
}
#[cfg(test)]
mod tests;