Add Authorization policy

sdk/core/src/headers/mod.rs

@@ -61,8 +61,8 @@ impl Headers {
     }
 
     /// Get a header value given a specific header name
-    pub fn get(&self, key: &HeaderName) -> Option<&HeaderValue> {
-        self.0.get(key)
+    pub fn get<T: Into<HeaderName>>(&self, key: T) -> Option<&HeaderValue> {
+        self.0.get(&key.into())
     }
 
     /// Insert a header name/value pair

sdk/data_cosmos/src/authorization_policy.rs

@@ -14,15 +14,18 @@ use url::form_urlencoded;
 const AZURE_VERSION: &str = "2018-12-31";
 const VERSION: &str = "1.0";
 
-/// The `AuthorizationPolicy` takes care to authenticate your calls to Azure CosmosDB. Currently it
-/// supports two type of authorization: one at service level and another at resource level (see
-/// [AuthorizationToken] for more info). The policy must be added just before the transport policy
+/// The `AuthorizationPolicy` takes care to authenticate your calls to Azure CosmosDB.
+///
+/// Currently it supports two type of authorization: one at service level and another at resource level (see
+/// [`AuthorizationToken`] for more info). The policy must be added just before the transport policy
 /// because it needs to inspect the values that are about to be sent to the transport and inject
 /// the proper authorization token.
-/// The `AuthorizationPolicy` is the only owner of the passed credentials so if you want to
-/// authenticate the same operation with different credentials all you have to do is to swap the
+///
+/// The `AuthorizationPolicy` is the only owner of the passed credentials, so if you want to
+/// authenticate the same operation with different credentials, all you have to do is to swap the
 /// `AuthorizationPolicy`.
-/// This struct is `Debug` but secrets are encrypted by `AuthorizationToken` so there is no risk of
+///
+/// This struct implements `Debug` but secrets are encrypted by `AuthorizationToken` so there is no risk of
 /// leaks in debug logs (secrets are stored in cleartext in memory: dumps are still leaky).
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct AuthorizationPolicy {
@@ -100,9 +103,7 @@ impl Policy for AuthorizationPolicy {
 /// 2. Find if the uri **is** the ending string (without the leading slash). If so return an empty
 ///    string. This covers the exception of the rule above.
 /// 3. Return the received uri unchanged.
-// TODO: will become private as soon as client will be migrated
-// to pipeline arch.
-pub(crate) fn generate_resource_link(uri: &str) -> &str {
+fn generate_resource_link(uri: &str) -> &str {
     static ENDING_STRINGS: &[&str] = &[
         "/dbs",
         "/colls",
@@ -139,12 +140,11 @@ pub(crate) fn generate_resource_link(uri: &str) -> &str {
     }
 }
 
-/// The CosmosDB authorization can either be "primary" (ie one of the two service-level tokens) or
-/// "resource" (ie a single database). In the first case the signature must be constructed by
+/// The CosmosDB authorization can either be "primary" (i.e., one of the two service-level tokens) or
+/// "resource" (i.e., a single database). In the first case the signature must be constructed by
 /// signing the HTTP method, resource type, resource link (the relative URI) and the current time.
 /// In the second case, the signature is just the resource key.
-// TODO: make it private after pipeline migration
-pub(crate) fn generate_authorization(
+fn generate_authorization(
     auth_token: &AuthorizationToken,
     http_method: &http::Method,
     resource_type: &ResourceType,

sdk/storage/src/authorization_policy.rs

@@ -0,0 +1,268 @@
+use azure_core::error::{ErrorKind, ResultExt};
+use azure_core::{headers::*, Context, Policy, PolicyResult, Request};
+use http::header::AUTHORIZATION;
+use http::{Method, Uri};
+use std::sync::Arc;
+
+use crate::clients::{ServiceType, StorageCredentials};
+
+const STORAGE_TOKEN_SCOPE: &str = "https://storage.azure.com/";
+
+#[derive(Debug, Clone)]
+pub struct AuthorizationPolicy {
+    credentials: StorageCredentials,
+}
+
+impl AuthorizationPolicy {
+    pub(crate) fn new(credentials: StorageCredentials) -> Self {
+        Self { credentials }
+    }
+}
+
+#[async_trait::async_trait]
+impl Policy for AuthorizationPolicy {
+    async fn send(
+        &self,
+        ctx: &Context,
+        request: &mut Request,
+        next: &[Arc<dyn Policy>],
+    ) -> PolicyResult {
+        trace!("called AuthorizationPolicy::send. self == {:#?}", self);
+
+        assert!(
+            !next.is_empty(),
+            "Authorization policies cannot be the last policy of a pipeline"
+        );
+        let request = match &self.credentials {
+            StorageCredentials::Key(account, key) => {
+                if !request
+                    .uri()
+                    .query()
+                    .unwrap_or_default()
+                    .split("&")
+                    .any(|pair| matches!(pair.trim().split_once("="), Some(("sig", _))))
+                {
+                    let auth = generate_authorization(
+                        request.headers(),
+                        request.uri(),
+                        &request.method(),
+                        account,
+                        key,
+                        ctx.get()
+                            .expect("ServiceType must be in the Context at this point"),
+                    );
+                    request.headers_mut().insert(AUTHORIZATION, auth)
+                }
+                request
+            }
+            StorageCredentials::SASToken(_query_pairs) => {
+                // no headers to add here, the authentication is in the URL
+                request
+            }
+            StorageCredentials::BearerToken(token) => {
+                request
+                    .headers_mut()
+                    .insert(AUTHORIZATION, format!("Bearer {}", token));
+                request
+            }
+            StorageCredentials::TokenCredential(token_credential) => {
+                let bearer_token_future = token_credential.get_token(STORAGE_TOKEN_SCOPE);
+                let bearer_token = futures::executor::block_on(bearer_token_future)
+                    .context(ErrorKind::Credential, "failed to get bearer token")?;
+
+                request.headers_mut().insert(
+                    AUTHORIZATION,
+                    format!("Bearer {}", bearer_token.token.secret()),
+                );
+                request
+            }
+        };
+
+        next[0].send(ctx, request, &next[1..]).await
+    }
+}
+
+fn generate_authorization(
+    h: &Headers,
+    u: &Uri,
+    method: &Method,
+    account: &str,
+    key: &str,
+    service_type: &ServiceType,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account, &service_type);
+    let auth = crate::hmac::sign(&str_to_sign, key).unwrap();
+    format!("SharedKey {}:{}", account, auth)
+}
+
+fn add_if_exists<'a>(h: &'a Headers, key: &'static str) -> &'a str {
+    h.get(key).map(|ce| ce.as_str()).unwrap_or_default()
+}
+
+#[allow(unknown_lints)]
+fn string_to_sign(
+    h: &Headers,
+    u: &Uri,
+    method: &Method,
+    account: &str,
+    service_type: &ServiceType,
+) -> String {
+    match service_type {
+        ServiceType::Table => {
+            format!(
+                "{}\n{}\n{}\n{}\n{}",
+                method.as_str(),
+                add_if_exists(h, CONTENT_MD5),
+                add_if_exists(h, CONTENT_TYPE),
+                add_if_exists(h, MS_DATE),
+                canonicalized_resource_table(account, u)
+            )
+        }
+        _ => {
+            // content lenght must only be specified if != 0
+            // this is valid from 2015-02-21
+            let content_length = h
+                .get(CONTENT_LENGTH)
+                .map(|v| if v.as_str() == "0" { "" } else { v.as_str() })
+                .unwrap_or_default();
+            format!(
+                "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+                method.as_str(),
+                add_if_exists(h, CONTENT_ENCODING),
+                add_if_exists(h, CONTENT_LANGUAGE),
+                content_length,
+                add_if_exists(h, CONTENT_MD5),
+                add_if_exists(h, CONTENT_TYPE),
+                add_if_exists(h, DATE),
+                add_if_exists(h, IF_MODIFIED_SINCE),
+                add_if_exists(h, IF_MATCH),
+                add_if_exists(h, IF_NONE_MATCH),
+                add_if_exists(h, IF_UNMODIFIED_SINCE),
+                add_if_exists(h, RANGE),
+                canonicalize_header(h),
+                canonicalized_resource(account, u)
+            )
+        }
+    }
+
+    // expected
+    // GET\n /*HTTP Verb*/
+    // \n    /*Content-Encoding*/
+    // \n    /*Content-Language*/
+    // \n    /*Content-Length (include value when zero)*/
+    // \n    /*Content-MD5*/
+    // \n    /*Content-Type*/
+    // \n    /*Date*/
+    // \n    /*If-Modified-Since */
+    // \n    /*If-Match*/
+    // \n    /*If-None-Match*/
+    // \n    /*If-Unmodified-Since*/
+    // \n    /*Range*/
+    // x-ms-date:Sun, 11 Oct 2009 21:49:13 GMT\nx-ms-version:2009-09-19\n
+    //                                  /*CanonicalizedHeaders*/
+    // /myaccount /mycontainer\ncomp:metadata\nrestype:container\ntimeout:20
+    //                                  /*CanonicalizedResource*/
+    //
+    //
+}
+
+fn canonicalize_header(h: &Headers) -> String {
+    let mut v_headers = h
+        .iter()
+        .filter(|(k, _)| k.as_str().starts_with("x-ms"))
+        .map(|(k, _)| k.as_str().to_owned())
+        .collect::<Vec<_>>();
+    v_headers.sort_unstable();
+
+    let mut can = String::new();
+
+    for header_name in v_headers {
+        let s = h.get(header_name.clone()).unwrap().as_str();
+        can = format!("{can}{header_name}:{s}\n");
+    }
+    can
+}
+
+fn canonicalized_resource_table(account: &str, u: &Uri) -> String {
+    format!("/{}{}", account, u.path())
+}
+
+fn canonicalized_resource(account: &str, uri: &Uri) -> String {
+    let mut can_res: String = String::new();
+    can_res += "/";
+    can_res += account;
+
+    let path = uri.path();
+
+    for p in path.split("/") {
+        can_res.push('/');
+        can_res.push_str(&*p);
+    }
+    can_res += "\n";
+
+    // query parameters
+    let query_pairs = uri
+        .query()
+        .unwrap_or_default()
+        .split("&")
+        .filter_map(|p| p.split_once("="));
+    {
+        let mut qps = Vec::new();
+        for (q, _p) in query_pairs.clone() {
+            if !(qps.iter().any(|x| x == q)) {
+                qps.push(q.to_owned());
+            }
+        }
+
+        qps.sort();
+
+        for qparam in qps {
+            // find correct parameter
+            let ret = lexy_sort(query_pairs.clone(), &qparam);
+
+            can_res = can_res + &qparam.to_lowercase() + ":";
+
+            for (i, item) in ret.iter().enumerate() {
+                if i > 0 {
+                    can_res += ","
+                }
+                can_res += item;
+            }
+
+            can_res += "\n";
+        }
+    };
+
+    can_res[0..can_res.len() - 1].to_owned()
+}
+
+fn lexy_sort<'a>(
+    vec: impl Iterator<Item = (&'a str, &'a str)> + 'a,
+    query_param: &str,
+) -> Vec<&'a str> {
+    let mut v_values = Vec::new();
+
+    for (_, v) in vec.filter(|(k, _)| *k == query_param) {
+        v_values.push(v);
+    }
+    v_values.sort();
+
+    v_values
+}
+
+// TODO: handle Sas token
+// fn get_sas_token_parms(sas_token: &str) -> azure_core::error::Result<Vec<(String, String)>> {
+//     let url = if sas_token.starts_with('?') {
+//         Uri::parse(sas_token)
+//     } else {
+//         Uri::parse(&format!("?{}", sas_token))
+//     }
+//     .with_context(ErrorKind::DataConversion, || {
+//         format!("failed to parse SAS token: {sas_token}")
+//     })?;
+
+//     Ok(url
+//         .query_pairs()
+//         .map(|(k, v)| (String::from(k), String::from(v)))
+//         .collect())
+// }

sdk/storage/src/core/clients/storage_account_client.rs

@@ -1,6 +1,7 @@
+use crate::authorization_policy::AuthorizationPolicy;
 use crate::headers::CONTENT_MD5;
 use crate::{
-    core::{ConnectionString, No},
+    core::No,
     hmac::sign,
     shared_access_signature::account_sas::{
         AccountSharedAccessSignatureBuilder, ClientAccountSharedAccessSignature,
@@ -533,13 +534,7 @@ fn generate_authorization(
     service_type: ServiceType,
 ) -> String {
     let str_to_sign = string_to_sign(h, u, method, account, service_type);
-
-    // debug!("\nstr_to_sign == {:?}\n", str_to_sign);
-    // debug!("str_to_sign == {}", str_to_sign);
-
     let auth = sign(&str_to_sign, key).unwrap();
-    // debug!("auth == {:?}", auth);
-
     format!("SharedKey {}:{}", account, auth)
 }
 
@@ -720,7 +715,7 @@ fn get_endpoint_uri(url: Option<&str>, account: &str, endpoint_type: &str) -> Re
 
 /// Create a Pipeline from CosmosOptions
 fn new_pipeline_from_options(options: StorageOptions, credentials: StorageCredentials) -> Pipeline {
-    let auth_policy: Arc<dyn azure_core::Policy> = todo!();
+    let auth_policy: Arc<dyn azure_core::Policy> = Arc::new(AuthorizationPolicy::new(credentials));
 
     // The `AuthorizationPolicy` must be the **last** retry policy.
     // Policies can change the url and/or the headers, and the `AuthorizationPolicy`

sdk/storage/src/lib.rs

@@ -13,7 +13,9 @@ extern crate azure_core;
 
 #[cfg(feature = "account")]
 pub mod account;
+mod authorization_policy;
 pub mod core;
+
 pub use crate::core::*;
 
 #[cfg(feature = "account")]