feat(http): OAuth2 password grant for sl.http

crates/sqlload/src/http_source/_helpers/endpoint.rs

@@ -89,6 +89,19 @@ pub(crate) fn endpoint(
                 grant,
             )?)
         }
+        Auth::Oauth2Password(spec) => {
+            let grant = Grant::Password {
+                username: needed("username")?,
+                password: needed("password")?,
+                client: (present("client_id"), present("client_secret")),
+            };
+            Credential::OAuth2(shared_token_source(
+                name,
+                &format!("{:?}", client.auth),
+                spec,
+                grant,
+            )?)
+        }
     };
     let mut headers = settings
         .get("headers")

crates/sqlload/src/http_source/_helpers/job.rs

@@ -88,9 +88,9 @@ fn request_problems(request: &Request) -> Vec<String> {
     problems.extend(paginator_problem(&request.paginator));
     problems.extend(body_problems(request));
     problems.extend(match &request.client.auth {
-        Auth::Oauth2ClientCredentials(spec) | Auth::Oauth2RefreshToken(spec) => {
-            token_url(spec).err()
-        }
+        Auth::Oauth2ClientCredentials(spec)
+        | Auth::Oauth2RefreshToken(spec)
+        | Auth::Oauth2Password(spec) => token_url(spec).err(),
         Auth::None | Auth::Bearer | Auth::ApiKey { .. } | Auth::Basic => None,
     });
     problems.extend(

crates/sqlload/src/http_source/_helpers/oauth.rs

@@ -59,26 +59,33 @@ pub(crate) fn shared_token_source(
 }
 
 fn token_source(token_url: Url, spec: &OAuth2Spec, grant: Grant) -> TokenSource {
-    let (grant_type, client, refresh_token) = match grant {
+    let (grant_type, client, refresh_token, password) = match grant {
         Grant::ClientCredentials {
             client_id,
             client_secret,
         } => (
             "client_credentials",
             (Some(client_id), Some(client_secret)),
             None,
+            None,
         ),
         Grant::RefreshToken {
             refresh_token,
             client,
-        } => ("refresh_token", client, Some(refresh_token)),
+        } => ("refresh_token", client, Some(refresh_token), None),
+        Grant::Password {
+            username,
+            password,
+            client,
+        } => ("password", client, None, Some((username, password))),
     };
-    let mut form = vec![("grant_type".to_owned(), grant_type.to_owned())];
-    form.extend(spec.scope.clone().map(|scope| ("scope".to_owned(), scope)));
+    let scope = spec.scope.clone().map(|scope| ("scope".to_owned(), scope));
     let mut secrets: Vec<String> = refresh_token
         .iter()
-        .flat_map(|token| secret_forms(token))
+        .chain(password.iter().map(|(_, password)| password))
+        .flat_map(|secret| secret_forms(secret))
         .collect();
+    let mut client_fields: Vec<(String, String)> = Vec::new();
     let basic = match (client, spec.client_auth) {
         ((Some(id), Some(secret)), ClientAuth::Basic) => {
             let pair = (encoded(&id), encoded(&secret));
@@ -89,16 +96,32 @@ fn token_source(token_url: Url, spec: &OAuth2Spec, grant: Grant) -> TokenSource
             Some(pair)
         }
         ((id, secret), _) => {
-            form.extend(id.map(|id| ("client_id".to_owned(), id)));
+            client_fields.extend(id.map(|id| ("client_id".to_owned(), id)));
             secrets.extend(secret.iter().flat_map(|secret| secret_forms(secret)));
-            form.extend(secret.map(|secret| ("client_secret".to_owned(), secret)));
+            client_fields.extend(secret.map(|secret| ("client_secret".to_owned(), secret)));
             None
         }
     };
     secrets.sort_by_key(|secret| std::cmp::Reverse(secret.len()));
+    let refresh_form = password.is_some().then(|| {
+        let mut form = vec![("grant_type".to_owned(), "refresh_token".to_owned())];
+        form.extend(scope.clone());
+        form.extend(client_fields.clone());
+        form
+    });
+    let mut form = vec![("grant_type".to_owned(), grant_type.to_owned())];
+    form.extend(scope);
+    form.extend(password.into_iter().flat_map(|(username, password)| {
+        [
+            ("username".to_owned(), username),
+            ("password".to_owned(), password),
+        ]
+    }));
+    form.extend(client_fields);
     TokenSource {
         token_url,
         form,
+        refresh_form,
         basic,
         state: tokio::sync::Mutex::new(TokenState {
             refresh_token,
@@ -109,7 +132,7 @@ fn token_source(token_url: Url, spec: &OAuth2Spec, grant: Grant) -> TokenSource
 }
 
 impl TokenSource {
-    /// Every secret of the token flow so far: the client secret, refresh tokens and each access token issued.
+    /// Every secret of the token flow so far: the client secret, password, refresh tokens and each access token issued.
     pub(crate) fn secrets(&self) -> Vec<String> {
         self.secrets
             .lock()
@@ -139,7 +162,10 @@ impl TokenSource {
             Some(access) if fresh && Some(access.as_str()) != refused => return Ok(access.clone()),
             Some(_) | None => {}
         }
-        let issued = self.request(http, retries, &state).await?;
+        let (issued, kept) = self
+            .request(http, retries, state.refresh_token.clone())
+            .await?;
+        state.refresh_token = kept;
         let renew_at = issued
             .renew_after
             .and_then(|delay| Instant::now().checked_add(delay));
@@ -152,35 +178,54 @@ impl TokenSource {
         state.access = Some(issued.access.clone());
         state.issued_at = Some(Instant::now());
         state.renew_at = renew_at;
-        if state.refresh_token.is_some() {
+        if state.refresh_token.is_some() || self.refresh_form.is_some() {
             state.refresh_token = issued.refresh_token.or(state.refresh_token.take());
         }
         Ok(issued.access)
     }
 
-    /// One token request, retried like a page on transport errors, 429 and 5xx.
+    /// A new token, and the refresh token to keep: from the refresh token a password grant was handed, falling back to the grant itself (and dropping that refresh token) when the server refuses it, else from `form` and any `refresh_token`.
     async fn request(
         &self,
         http: &reqwest::Client,
         retries: u32,
-        state: &TokenState,
+        refresh_token: Option<String>,
+    ) -> Result<(Issued, Option<String>), ReadError> {
+        let (Some(refresh_form), Some(token)) = (&self.refresh_form, refresh_token.clone()) else {
+            let mut form = self.form.clone();
+            form.extend(
+                refresh_token
+                    .clone()
+                    .map(|token| ("refresh_token".to_owned(), token)),
+            );
+            return Ok((self.post(http, retries, &form).await?, refresh_token));
+        };
+        let mut form = refresh_form.clone();
+        form.push(("refresh_token".to_owned(), token));
+        match self.post(http, retries, &form).await {
+            Err(error) if !error.retryable => {
+                Ok((self.post(http, retries, &self.form).await?, None))
+            }
+            other => Ok((other?, refresh_token)),
+        }
+    }
+
+    /// One token request posting `form`, retried like a page on transport errors, 429 and 5xx.
+    async fn post(
+        &self,
+        http: &reqwest::Client,
+        retries: u32,
+        form: &[(String, String)],
     ) -> Result<Issued, ReadError> {
         let policy = RetryPolicy {
             first_delay: FIRST_RETRY_DELAY,
             max_delay: MAX_RETRY_DELAY,
             max_attempts: retries + 1,
         };
         let shown = format!("POST {} (OAuth2 token)", self.token_url);
-        let mut form = self.form.clone();
-        form.extend(
-            state
-                .refresh_token
-                .clone()
-                .map(|token| ("refresh_token".to_owned(), token)),
-        );
         let mut attempt = 1;
         loop {
-            let mut request = http.post(self.token_url.clone()).form(&form);
+            let mut request = http.post(self.token_url.clone()).form(form);
             if let Some((id, secret)) = &self.basic {
                 request = request.basic_auth(id, Some(secret));
             }
@@ -224,8 +269,7 @@ impl TokenSource {
         let secrets = self.secrets();
         let failed = |status: u16| {
             let reason = match serde_json::from_slice::<Value>(&body) {
-                Ok(document) => oauth_error(&document)
-                    .unwrap_or_else(|| String::from_utf8_lossy(&body).into_owned()),
+                Ok(document) => oauth_error(&document).unwrap_or_else(|| document.to_string()),
                 Err(_) => String::from_utf8_lossy(&body).into_owned(),
             };
             let reason = redacted(&reason, &secrets)
@@ -304,12 +348,18 @@ pub(crate) fn renewal_delay(expires_in: Option<&Value>) -> Result<Option<Duratio
         }))
 }
 
-/// `secret` as sent and as it appears in a URL or a form body, for redaction.
-fn secret_forms(secret: &str) -> [String; 3] {
+/// `secret` as sent, as it appears in a URL or a form body, and inside a JSON string, for redaction; a JSON error body is re-serialised before it is shown, so any escaping the server chose (`\/`, `\u00e9`) becomes this one.
+fn secret_forms(secret: &str) -> [String; 4] {
     let percent = encoded(secret);
     let form = percent
         .replace("%20", "+")
         .replace("%2A", "*")
         .replace('~', "%7E");
-    [secret.to_owned(), percent, form]
+    let quoted = serde_json::to_string(secret).unwrap_or_default();
+    let json = quoted
+        .strip_prefix('"')
+        .and_then(|inner| inner.strip_suffix('"'))
+        .unwrap_or(secret)
+        .to_owned();
+    [secret.to_owned(), percent, form, json]
 }

crates/sqlload/src/http_source/models.rs

@@ -144,6 +144,8 @@ pub enum Auth {
     Oauth2ClientCredentials(OAuth2Spec),
     /// An access token from `token_url` for the connection's `refresh_token`.
     Oauth2RefreshToken(OAuth2Spec),
+    /// An access token from `token_url` for the connection's `username` and `password` (the resource-owner password grant).
+    Oauth2Password(OAuth2Spec),
 }
 
 /// Where an OAuth2 access token comes from and how the client authenticates there.
@@ -173,6 +175,8 @@ pub struct TokenSource {
     pub token_url: reqwest::Url,
     /// `grant_type` and its fields, plus the client's fields under `client_auth = "post"`.
     pub form: Vec<(String, String)>,
+    /// For a grant that starts from something other than a refresh token (the password grant): the `refresh_token` grant's form, used instead of `form` once a token response hands one out.
+    pub refresh_form: Option<Vec<(String, String)>>,
     /// `client_id` and `client_secret` for basic client auth, form-encoded as RFC 6749 asks.
     pub basic: Option<(String, String)>,
     pub state: tokio::sync::Mutex<TokenState>,
@@ -201,6 +205,12 @@ pub enum Grant {
         /// `client_id` and `client_secret`, when the connection has them.
         client: (Option<String>, Option<String>),
     },
+    Password {
+        username: String,
+        password: String,
+        /// `client_id` and `client_secret`, when the connection has them.
+        client: (Option<String>, Option<String>),
+    },
 }
 
 /// What a token response issued.

crates/sqlload/src/project/models.rs

@@ -260,12 +260,12 @@ pub enum ConnectionConfig {
         base_url: String,
         #[serde(default)]
         token: Option<crate::auth::models::Secret>,
-        /// For `sl.basic()`.
+        /// For `sl.basic()` and `sl.oauth2_password(...)`.
         #[serde(default)]
         username: Option<String>,
         #[serde(default)]
         password: Option<crate::auth::models::Secret>,
-        /// For `sl.oauth2_client_credentials(...)`, and `sl.oauth2_refresh_token(...)` when the API wants the client too.
+        /// For `sl.oauth2_client_credentials(...)`, and `sl.oauth2_refresh_token(...)` or `sl.oauth2_password(...)` when the API wants the client too.
         #[serde(default)]
         client_id: Option<String>,
         #[serde(default)]

packages/sqlload/python/sqlload/__init__.py

@@ -40,6 +40,7 @@ def orders(ctx: sl.Context):
 	link_header,
 	next_link,
 	oauth2_client_credentials,
+	oauth2_password,
 	oauth2_refresh_token,
 	offset,
 	page_number,
@@ -69,6 +70,7 @@ def orders(ctx: sl.Context):
 	"ndjson",
 	"next_link",
 	"oauth2_client_credentials",
+	"oauth2_password",
 	"oauth2_refresh_token",
 	"offset",
 	"page_number",

packages/sqlload/python/sqlload/_http.py

@@ -111,6 +111,16 @@ def oauth2_refresh_token(token_url: str, *, scope: str | None = None, client_aut
 	return Auth(_oauth2("oauth2_refresh_token", token_url, scope, client_auth))
 
 
+def oauth2_password(token_url: str, *, scope: str | None = None, client_auth: str = "basic") -> Auth:
+	"""`Authorization: Bearer <access token>`, the token `token_url` issues for the connection's `username` and `password`.
+
+	The resource-owner password grant. The connection's `client_id` and `client_secret` go along when it has them. The
+	token is renewed before it expires and once after a 401: with the refresh token the server handed out, if it did,
+	and by asking with the password again otherwise or when that refresh token is refused.
+	"""
+	return Auth(_oauth2("oauth2_password", token_url, scope, client_auth))
+
+
 def _oauth2(kind: str, token_url: str, scope: str | None, client_auth: str) -> dict[str, Any]:
 	if client_auth not in ("basic", "post"):
 		raise ValueError(f'client_auth must be "basic" or "post", not {client_auth!r}')