From 891e8421af482ee4cdba3e55c93c5f1dc9ef1765 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:00:53 +0000 Subject: [PATCH 1/3] feat(http): OAuth2 password grant for sl.http sl.oauth2_password(token_url, scope=, client_auth=) posts grant_type=password with the connection's username and password, plus client_id and client_secret when the connection has them. Renewal works like the other OAuth2 kinds: before expiry and once after a 401, using the refresh token the server handed out when there is one, and the password grant again when there isn't or the server refuses it. The password is redacted like the other secrets, and wrong credentials fail with exit 1, not retryable. --- .../src/http_source/_helpers/endpoint.rs | 13 +++ .../sqlload/src/http_source/_helpers/job.rs | 6 +- .../sqlload/src/http_source/_helpers/oauth.rs | 83 ++++++++++++++----- crates/sqlload/src/http_source/models.rs | 10 +++ crates/sqlload/src/project/models.rs | 4 +- packages/sqlload/python/sqlload/__init__.py | 2 + packages/sqlload/python/sqlload/_http.py | 10 +++ packages/sqlload/tests/test_http_jobs.py | 52 ++++++++++-- .../src/content/docs/docs/configuration.mdx | 4 +- .../src/content/docs/docs/sources/python.mdx | 22 ++++- 10 files changed, 174 insertions(+), 32 deletions(-) diff --git a/crates/sqlload/src/http_source/_helpers/endpoint.rs b/crates/sqlload/src/http_source/_helpers/endpoint.rs index 36b91ab1..4bfda26e 100644 --- a/crates/sqlload/src/http_source/_helpers/endpoint.rs +++ b/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") diff --git a/crates/sqlload/src/http_source/_helpers/job.rs b/crates/sqlload/src/http_source/_helpers/job.rs index 5e278b9a..cdb4431b 100644 --- a/crates/sqlload/src/http_source/_helpers/job.rs +++ b/crates/sqlload/src/http_source/_helpers/job.rs @@ -88,9 +88,9 @@ fn request_problems(request: &Request) -> Vec { 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( diff --git a/crates/sqlload/src/http_source/_helpers/oauth.rs b/crates/sqlload/src/http_source/_helpers/oauth.rs index 785dd268..e4a8ff06 100644 --- a/crates/sqlload/src/http_source/_helpers/oauth.rs +++ b/crates/sqlload/src/http_source/_helpers/oauth.rs @@ -59,7 +59,7 @@ 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, @@ -67,18 +67,25 @@ fn token_source(token_url: Url, spec: &OAuth2Spec, grant: Grant) -> TokenSource "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 = 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, @@ -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,18 +178,44 @@ 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, + ) -> Result<(Issued, Option), 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 { let policy = RetryPolicy { first_delay: FIRST_RETRY_DELAY, @@ -171,16 +223,9 @@ impl TokenSource { 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)); } diff --git a/crates/sqlload/src/http_source/models.rs b/crates/sqlload/src/http_source/models.rs index 0c455710..57da2d7f 100644 --- a/crates/sqlload/src/http_source/models.rs +++ b/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>, /// `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, @@ -201,6 +205,12 @@ pub enum Grant { /// `client_id` and `client_secret`, when the connection has them. client: (Option, Option), }, + Password { + username: String, + password: String, + /// `client_id` and `client_secret`, when the connection has them. + client: (Option, Option), + }, } /// What a token response issued. diff --git a/crates/sqlload/src/project/models.rs b/crates/sqlload/src/project/models.rs index 1913c31a..8d8fcc70 100644 --- a/crates/sqlload/src/project/models.rs +++ b/crates/sqlload/src/project/models.rs @@ -260,12 +260,12 @@ pub enum ConnectionConfig { base_url: String, #[serde(default)] token: Option, - /// For `sl.basic()`. + /// For `sl.basic()` and `sl.oauth2_password(...)`. #[serde(default)] username: Option, #[serde(default)] password: Option, - /// 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, #[serde(default)] diff --git a/packages/sqlload/python/sqlload/__init__.py b/packages/sqlload/python/sqlload/__init__.py index 215372d2..f4feec0b 100644 --- a/packages/sqlload/python/sqlload/__init__.py +++ b/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", diff --git a/packages/sqlload/python/sqlload/_http.py b/packages/sqlload/python/sqlload/_http.py index 31a9ff15..0d66f564 100644 --- a/packages/sqlload/python/sqlload/_http.py +++ b/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 `, 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}') diff --git a/packages/sqlload/tests/test_http_jobs.py b/packages/sqlload/tests/test_http_jobs.py index 9f452cc7..4b54a9f5 100644 --- a/packages/sqlload/tests/test_http_jobs.py +++ b/packages/sqlload/tests/test_http_jobs.py @@ -22,6 +22,7 @@ TOKEN = "s3cret-token" CLIENT_SECRET = "cl1ent+secret/x~y" REFRESH_TOKEN = "r3fresh-first" +PASSWORD = "pa55+word/x" class Api(BaseHTTPRequestHandler): @@ -116,6 +117,8 @@ def token(self) -> None: oauth = server.oauth # type: ignore[attr-defined] if client != ["app", CLIENT_SECRET] or form.get("grant_type") == "refresh_token" and form.get("refresh_token") != server.refresh_token: # type: ignore[attr-defined] return self.reply(401, {"error": "invalid_client", "error_description": f"unknown client {client} or refresh token"}) + if form.get("grant_type") == "password" and (form.get("username"), form.get("password")) != ("loader", PASSWORD): + return self.reply(400, {"error": "invalid_grant", "error_description": f"wrong password {form.get('password')} for {form.get('username')}"}) access = f"access-{len(server.token_requests)}" # type: ignore[attr-defined] server.tokens[access] = {"issued": time.monotonic(), "uses": 0} # type: ignore[attr-defined] body: dict[str, Any] = {"access_token": access, "token_type": "Bearer"} @@ -123,6 +126,8 @@ def token(self) -> None: body["expires_in"] = oauth["lifetime"] if oauth.get("rotate"): server.refresh_token = body["refresh_token"] = f"r3fresh-{len(server.token_requests) + 1}" # type: ignore[attr-defined] + if oauth.get("stale_refresh"): + body["refresh_token"] = "stale" self.reply(200, body) def access_allowed(self, bearer: str) -> bool: @@ -270,6 +275,8 @@ def make_project(tmp_path: Path, server: ThreadingHTTPServer, tables: str) -> Pa client_id = "app" client_secret = "${{ENV:SHOP_CLIENT_SECRET}}" refresh_token = "{REFRESH_TOKEN}" + username = "loader" + password = "${{ENV:SHOP_PASSWORD}}" [targets.dev] connection = "wh" @@ -284,7 +291,7 @@ def make_project(tmp_path: Path, server: ThreadingHTTPServer, tables: str) -> Pa def sqlload(project: Path, *args: str) -> subprocess.CompletedProcess[str]: home = project / "home" home.mkdir(exist_ok=True) - env = {**os.environ, "HOME": str(home), "XDG_CONFIG_HOME": str(home / ".config"), "SHOP_TOKEN": TOKEN, "SHOP_CLIENT_SECRET": CLIENT_SECRET} + env = {**os.environ, "HOME": str(home), "XDG_CONFIG_HOME": str(home / ".config"), "SHOP_TOKEN": TOKEN, "SHOP_CLIENT_SECRET": CLIENT_SECRET, "SHOP_PASSWORD": PASSWORD} return subprocess.run( [sys.executable, "-c", "from sqlload._cli import main; main()", *args], cwd=project, env=env, capture_output=True, text=True, timeout=120 ) @@ -554,15 +561,22 @@ def oauth_project(tmp_path: Path, server: ThreadingHTTPServer, auth: str, job: s return make_project(tmp_path, server, tables) -def test_given_short_lived_tokens_when_pages_outlast_one_then_it_is_renewed_before_it_expires(tmp_path: Path, api: ThreadingHTTPServer) -> None: +@pytest.mark.parametrize( + ("auth", "expected_grant"), + [("sl.oauth2_client_credentials(TOKEN_URL, scope='orders:read')", "client_credentials"), ("sl.oauth2_password(TOKEN_URL, scope='orders:read')", "password")], + ids=["client-credentials", "password"], +) +def test_given_short_lived_tokens_when_pages_outlast_one_then_it_is_renewed_before_it_expires( + tmp_path: Path, api: ThreadingHTTPServer, auth: str, expected_grant: str +) -> None: api.oauth = {"lifetime": 2} # type: ignore[attr-defined] - project = oauth_project(tmp_path, api, "sl.oauth2_client_credentials(TOKEN_URL, scope='orders:read')", "paginate('/slow', paginator=sl.page_number('p'), select='$.items')") + project = oauth_project(tmp_path, api, auth, "paginate('/slow', paginator=sl.page_number('p'), select='$.items')") done = sqlload(project, "load") assert done.returncode == 0, done.stdout + done.stderr assert query(project, 'SELECT order_id, status FROM "RAW".shop__t ORDER BY order_id') == ALL_ORDERS assert api.refused == 0 # type: ignore[attr-defined] assert len(api.token_requests) >= 2 # type: ignore[attr-defined] - assert {(request["grant_type"], request.get("scope")) for request in api.token_requests} == {("client_credentials", "orders:read")} # type: ignore[attr-defined] + assert {(request["grant_type"], request.get("scope")) for request in api.token_requests} == {(expected_grant, "orders:read")} # type: ignore[attr-defined] @pytest.mark.parametrize( @@ -570,8 +584,9 @@ def test_given_short_lived_tokens_when_pages_outlast_one_then_it_is_renewed_befo [ ("sl.oauth2_client_credentials(TOKEN_URL)", [("client_credentials", None)] * 3), ("sl.oauth2_refresh_token(TOKEN_URL, client_auth='post')", [("refresh_token", REFRESH_TOKEN), ("refresh_token", "r3fresh-2"), ("refresh_token", "r3fresh-3")]), + ("sl.oauth2_password(TOKEN_URL)", [("password", None), ("refresh_token", "r3fresh-2"), ("refresh_token", "r3fresh-3")]), ], - ids=["client-credentials", "refresh-token-rotated-client-in-form"], + ids=["client-credentials", "refresh-token-rotated-client-in-form", "password-then-its-refresh-token"], ) def test_given_a_token_revoked_mid_pagination_when_the_api_answers_401_then_a_new_one_is_fetched_and_the_page_retried( tmp_path: Path, api: ThreadingHTTPServer, auth: str, expected_grant: list[tuple[str, str | None]] @@ -583,6 +598,7 @@ def test_given_a_token_revoked_mid_pagination_when_the_api_answers_401_then_a_ne assert query(project, 'SELECT order_id, status FROM "RAW".shop__t ORDER BY order_id') == ALL_ORDERS assert api.refused == 2 # type: ignore[attr-defined] assert [(request["grant_type"], request.get("refresh_token")) for request in api.token_requests] == expected_grant # type: ignore[attr-defined] + assert all(("username" in request, "password" in request) == ((request["grant_type"] == "password",) * 2) for request in api.token_requests) # type: ignore[attr-defined] assert all(request["client"] == ["app", CLIENT_SECRET] for request in api.token_requests) # type: ignore[attr-defined] assert {auth for path, _, auth in api.log if path == "/v1/offset"} == {"Bearer access-1", "Bearer access-2", "Bearer access-3"} # type: ignore[attr-defined] @@ -627,6 +643,32 @@ def test_given_an_oauth2_failure_when_loading_then_the_error_says_why_and_hides_ assert all(path in ("/v1/whoami", "/v1/leak", "/v1/notjson") for path, _, _ in api.log) # type: ignore[attr-defined] +def test_given_a_password_grant_without_a_refresh_token_or_with_a_refused_one_when_the_api_answers_401_then_it_asks_with_the_password_again( + tmp_path: Path, api: ThreadingHTTPServer +) -> None: + api.oauth = {"uses": 3, "stale_refresh": True} # type: ignore[attr-defined] + project = oauth_project(tmp_path, api, "sl.oauth2_password(TOKEN_URL, client_auth='post')", "paginate('/offset', paginator=sl.offset(limit=1))") + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + assert query(project, 'SELECT order_id, status FROM "RAW".shop__t ORDER BY order_id') == ALL_ORDERS + expected = [("password", "loader", PASSWORD, None), ("refresh_token", None, None, "stale")] * 2 + [("password", "loader", PASSWORD, None)] + sent = [(request["grant_type"], request.get("username"), request.get("password"), request.get("refresh_token")) for request in api.token_requests] # type: ignore[attr-defined] + assert sent == expected + assert all(request["client"] == ["app", CLIENT_SECRET] for request in api.token_requests) # type: ignore[attr-defined] + + +def test_given_a_wrong_password_when_loading_with_json_then_it_fails_for_good_and_hides_the_password(tmp_path: Path, api: ThreadingHTTPServer) -> None: + project = oauth_project(tmp_path, api, "sl.oauth2_password(TOKEN_URL)", "paginate('/offset', paginator=sl.offset(limit=10))") + (project / "sqlload.toml").write_text((project / "sqlload.toml").read_text().replace("${ENV:SHOP_PASSWORD}", "wr0ng+pass")) + done = sqlload(project, "load", "--json") + assert done.returncode == 1, done.stdout + done.stderr + result = json.loads(done.stdout) + assert (result["ok"], result["retryable"], result["tables"][0]["retryable"]) == (False, False, False) + assert "HTTP 400: invalid_grant: wrong password *** for loader" in done.stdout, done.stdout + assert "wr0ng" not in done.stdout + done.stderr + assert len(api.token_requests) == 1 and api.log == [] # type: ignore[attr-defined] + + def test_given_401_listed_as_skip_when_the_api_answers_it_then_the_request_is_skipped_without_a_new_token(tmp_path: Path, api: ThreadingHTTPServer) -> None: project = oauth_project(tmp_path, api, "sl.oauth2_client_credentials(TOKEN_URL)", "fan_out('/denied?k={k}', over=range(5), concurrency=1, paginator=sl.single(), on_status={401: 'skip'})") done = sqlload(project, "load") diff --git a/website/src/content/docs/docs/configuration.mdx b/website/src/content/docs/docs/configuration.mdx index b6b723be..53ffc297 100644 --- a/website/src/content/docs/docs/configuration.mdx +++ b/website/src/content/docs/docs/configuration.mdx @@ -178,8 +178,8 @@ For [Python HTTP sources](/docs/sources/python/#http-apis); its settings reach t |-----|---------| | `base_url` | The API's base URL. | | `token` | For `sl.bearer()` and `sl.api_key(...)`. | -| `username`, `password` | For `sl.basic()`. | -| `client_id`, `client_secret` | From 0.27.0. For `sl.oauth2_client_credentials(...)`, and for `sl.oauth2_refresh_token(...)` when the API wants the client too. | +| `username`, `password` | For `sl.basic()`, and from 0.36.0 (not released yet) for `sl.oauth2_password(...)`. | +| `client_id`, `client_secret` | From 0.27.0. For `sl.oauth2_client_credentials(...)`, and for `sl.oauth2_refresh_token(...)` or `sl.oauth2_password(...)` when the API wants the client too. | | `refresh_token` | From 0.27.0. For `sl.oauth2_refresh_token(...)`. | | `headers` | Headers sent with every request. | diff --git a/website/src/content/docs/docs/sources/python.mdx b/website/src/content/docs/docs/sources/python.mdx index 328cfa31..e3f06efd 100644 --- a/website/src/content/docs/docs/sources/python.mdx +++ b/website/src/content/docs/docs/sources/python.mdx @@ -239,6 +239,7 @@ thread-safe. Set `parallel_tables=True` on the source when its client is. | `sl.basic()` | Basic auth with the connection's `username` and `password`. | | `sl.oauth2_client_credentials(token_url, *, scope=None, client_auth="basic")` | From 0.27.0. `Authorization: Bearer `, with the token `token_url` issues for the connection's `client_id` and `client_secret`. See [OAuth2](#oauth2). | | `sl.oauth2_refresh_token(token_url, *, scope=None, client_auth="basic")` | From 0.27.0. The same, with the token issued for the connection's `refresh_token`. The connection's `client_id` and `client_secret` go along when it has them. | +| `sl.oauth2_password(token_url, *, scope=None, client_auth="basic")` | From 0.36.0 (not released yet). The same, with the token issued 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. | - Every `paginate` names its paginator; nothing is guessed. - Credentials go only to the connection's own scheme, host and port, never to a next-page link @@ -284,7 +285,8 @@ origin, so `json_body` never leaves the connection's origin. ### OAuth2 -From 0.27.0. `sl.oauth2_client_credentials` and `sl.oauth2_refresh_token` fetch short-lived +From 0.27.0. `sl.oauth2_client_credentials`, `sl.oauth2_refresh_token` and (from 0.36.0, not +released yet) `sl.oauth2_password` fetch short-lived access tokens from `token_url`. See [OAuth2](/docs/guides/python/#oauth2) in the guide for an example. @@ -304,6 +306,24 @@ example. does not save it, so each run starts from the connection's `refresh_token`. An API that revokes the old refresh token on rotation therefore fails on the next run; use client credentials there when the API offers them. +- From 0.36.0 (not released yet), `sl.oauth2_password` posts `grant_type=password` with the + connection's `username` and `password`. When the response includes a `refresh_token`, later + renewals in the run use it; when it doesn't, or the server refuses it, SQLLoad asks with the + password again. The password is shown as `***` like the other secrets, and wrong credentials end + the run with exit 1 (not retryable) and the server's `error`. + +```toml +[connections.shop] +kind = "http" +base_url = "https://api.example.com/v1" +username = "loader" +password = "${ENV:SHOP_PASSWORD}" +client_id = "sqlload" +``` + +```python +api = sl.http(shop, auth=sl.oauth2_password("https://auth.example.com/oauth/token", client_auth="post")) +``` ## Logged state From 4d5054f6c634cd8dd98e02a6cfdf059b4771698f Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:14:08 +0000 Subject: [PATCH 2/3] fix(http): redact JSON-escaped secrets in token errors, and test a refresh that fails once --- .../sqlload/src/http_source/_helpers/oauth.rs | 17 +++++++++------ packages/sqlload/tests/test_http_jobs.py | 21 ++++++++++++++++--- .../src/content/docs/docs/sources/python.mdx | 2 +- 3 files changed, 30 insertions(+), 10 deletions(-) diff --git a/crates/sqlload/src/http_source/_helpers/oauth.rs b/crates/sqlload/src/http_source/_helpers/oauth.rs index e4a8ff06..b3079ada 100644 --- a/crates/sqlload/src/http_source/_helpers/oauth.rs +++ b/crates/sqlload/src/http_source/_helpers/oauth.rs @@ -132,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 { self.secrets .lock() @@ -269,8 +269,7 @@ impl TokenSource { let secrets = self.secrets(); let failed = |status: u16| { let reason = match serde_json::from_slice::(&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) @@ -349,12 +348,18 @@ pub(crate) fn renewal_delay(expires_in: Option<&Value>) -> Result [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] } diff --git a/packages/sqlload/tests/test_http_jobs.py b/packages/sqlload/tests/test_http_jobs.py index 4b54a9f5..b09931fc 100644 --- a/packages/sqlload/tests/test_http_jobs.py +++ b/packages/sqlload/tests/test_http_jobs.py @@ -118,7 +118,10 @@ def token(self) -> None: if client != ["app", CLIENT_SECRET] or form.get("grant_type") == "refresh_token" and form.get("refresh_token") != server.refresh_token: # type: ignore[attr-defined] return self.reply(401, {"error": "invalid_client", "error_description": f"unknown client {client} or refresh token"}) if form.get("grant_type") == "password" and (form.get("username"), form.get("password")) != ("loader", PASSWORD): - return self.reply(400, {"error": "invalid_grant", "error_description": f"wrong password {form.get('password')} for {form.get('username')}"}) + return self.reply(400, {"message": f"wrong password {form.get('password')} for {form.get('username')}"}) + if form.get("grant_type") == "refresh_token" and oauth.get("refresh_503_once"): + oauth["refresh_503_once"] = False + return self.reply(503, {"error": "temporarily_unavailable"}, {"Retry-After": "0"}) access = f"access-{len(server.token_requests)}" # type: ignore[attr-defined] server.tokens[access] = {"issued": time.monotonic(), "uses": 0} # type: ignore[attr-defined] body: dict[str, Any] = {"access_token": access, "token_type": "Bearer"} @@ -657,14 +660,26 @@ def test_given_a_password_grant_without_a_refresh_token_or_with_a_refused_one_wh assert all(request["client"] == ["app", CLIENT_SECRET] for request in api.token_requests) # type: ignore[attr-defined] +def test_given_a_token_endpoint_down_once_when_refreshing_after_a_password_grant_then_the_refresh_is_retried_without_the_password( + tmp_path: Path, api: ThreadingHTTPServer +) -> None: + api.oauth = {"uses": 3, "rotate": True, "refresh_503_once": True} # type: ignore[attr-defined] + project = oauth_project(tmp_path, api, "sl.oauth2_password(TOKEN_URL)", "paginate('/offset', paginator=sl.offset(limit=1))") + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + assert query(project, 'SELECT order_id, status FROM "RAW".shop__t ORDER BY order_id') == ALL_ORDERS + expected = [("password", None), ("refresh_token", "r3fresh-2"), ("refresh_token", "r3fresh-2"), ("refresh_token", "r3fresh-4")] + assert [(request["grant_type"], request.get("refresh_token")) for request in api.token_requests] == expected # type: ignore[attr-defined] + + def test_given_a_wrong_password_when_loading_with_json_then_it_fails_for_good_and_hides_the_password(tmp_path: Path, api: ThreadingHTTPServer) -> None: project = oauth_project(tmp_path, api, "sl.oauth2_password(TOKEN_URL)", "paginate('/offset', paginator=sl.offset(limit=10))") - (project / "sqlload.toml").write_text((project / "sqlload.toml").read_text().replace("${ENV:SHOP_PASSWORD}", "wr0ng+pass")) + (project / "sqlload.toml").write_text((project / "sqlload.toml").read_text().replace("${ENV:SHOP_PASSWORD}", "wr0ng+p\u00e4ss/word")) done = sqlload(project, "load", "--json") assert done.returncode == 1, done.stdout + done.stderr result = json.loads(done.stdout) assert (result["ok"], result["retryable"], result["tables"][0]["retryable"]) == (False, False, False) - assert "HTTP 400: invalid_grant: wrong password *** for loader" in done.stdout, done.stdout + assert 'HTTP 400: {"message":"wrong password *** for loader"}' in result["tables"][0]["error"] assert "wr0ng" not in done.stdout + done.stderr assert len(api.token_requests) == 1 and api.log == [] # type: ignore[attr-defined] diff --git a/website/src/content/docs/docs/sources/python.mdx b/website/src/content/docs/docs/sources/python.mdx index e3f06efd..cededed5 100644 --- a/website/src/content/docs/docs/sources/python.mdx +++ b/website/src/content/docs/docs/sources/python.mdx @@ -298,7 +298,7 @@ example. auth, the method RFC 6749 requires servers to support. As the RFC asks, both are form-encoded first, so a secret with characters such as `+` or `/` reaches the server encoded. Use `"post"` for a server that rejects that, or that wants them as form fields. -- The client secret and refresh token go only to `token_url`, which must be `https` (`http` only on +- The client secret, refresh token and password go only to `token_url`, which must be `https` (`http` only on `localhost`). Access tokens go only to the connection's own origin, like every other credential. - Token errors name the token URL, the status and the server's `error` and `error_description`. The client secret, refresh token and every access token are shown as `***`. From 2718e078374d199d19fdcc4fde95ad6d13472db7 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 12 Nov 2026 15:17:27 +0000 Subject: [PATCH 3/3] test(http): pin the refresh fallback without retries and a backslash in an echoed password --- packages/sqlload/tests/test_http_jobs.py | 12 +++++++++++- website/src/content/docs/docs/sources/python.mdx | 7 ++++--- 2 files changed, 15 insertions(+), 4 deletions(-) diff --git a/packages/sqlload/tests/test_http_jobs.py b/packages/sqlload/tests/test_http_jobs.py index b09931fc..c6c2cf1b 100644 --- a/packages/sqlload/tests/test_http_jobs.py +++ b/packages/sqlload/tests/test_http_jobs.py @@ -672,9 +672,19 @@ def test_given_a_token_endpoint_down_once_when_refreshing_after_a_password_grant assert [(request["grant_type"], request.get("refresh_token")) for request in api.token_requests] == expected # type: ignore[attr-defined] +def test_given_a_token_endpoint_down_when_refreshing_without_retries_then_the_run_is_retryable_and_the_password_is_not_resent( + tmp_path: Path, api: ThreadingHTTPServer +) -> None: + api.oauth = {"uses": 3, "rotate": True, "refresh_503_once": True} # type: ignore[attr-defined] + project = oauth_project(tmp_path, api, "sl.oauth2_password(TOKEN_URL), retries=0", "paginate('/offset', paginator=sl.offset(limit=1))") + done = sqlload(project, "load") + assert done.returncode == 75, done.stdout + done.stderr + assert [(request["grant_type"], request.get("refresh_token")) for request in api.token_requests] == [("password", None), ("refresh_token", "r3fresh-2")] # type: ignore[attr-defined] + + def test_given_a_wrong_password_when_loading_with_json_then_it_fails_for_good_and_hides_the_password(tmp_path: Path, api: ThreadingHTTPServer) -> None: project = oauth_project(tmp_path, api, "sl.oauth2_password(TOKEN_URL)", "paginate('/offset', paginator=sl.offset(limit=10))") - (project / "sqlload.toml").write_text((project / "sqlload.toml").read_text().replace("${ENV:SHOP_PASSWORD}", "wr0ng+p\u00e4ss/word")) + (project / "sqlload.toml").write_text((project / "sqlload.toml").read_text().replace("${ENV:SHOP_PASSWORD}", "wr0ng+p\u00e4ss\\\\word")) done = sqlload(project, "load", "--json") assert done.returncode == 1, done.stdout + done.stderr result = json.loads(done.stdout) diff --git a/website/src/content/docs/docs/sources/python.mdx b/website/src/content/docs/docs/sources/python.mdx index cededed5..81ae6c0a 100644 --- a/website/src/content/docs/docs/sources/python.mdx +++ b/website/src/content/docs/docs/sources/python.mdx @@ -298,10 +298,11 @@ example. auth, the method RFC 6749 requires servers to support. As the RFC asks, both are form-encoded first, so a secret with characters such as `+` or `/` reaches the server encoded. Use `"post"` for a server that rejects that, or that wants them as form fields. -- The client secret, refresh token and password go only to `token_url`, which must be `https` (`http` only on - `localhost`). Access tokens go only to the connection's own origin, like every other credential. +- The client secret, refresh token and password go only to `token_url`, which must be `https` + (`http` only on `localhost`). Access tokens go only to the connection's own origin, like every + other credential. - Token errors name the token URL, the status and the server's `error` and `error_description`. - The client secret, refresh token and every access token are shown as `***`. + The client secret, refresh token, password and every access token are shown as `***`. - When the server rotates the refresh token, SQLLoad uses the new one for the rest of the run but does not save it, so each run starts from the connection's `refresh_token`. An API that revokes the old refresh token on rotation therefore fails on the next run; use client credentials there