mod corporate_action;
pub use corporate_action::{IbkrFlexCorporateAction, IbkrReorgType, parse_corporate_actions};
use crate::exchange::http::{MAX_ERROR_BODY_DOWNLOAD_BYTES, read_body_capped, truncate_str};
use quick_xml::{Reader, events::Event};
use serde::Deserialize;
use smol_str::SmolStr;
use std::{env, fmt, time::Duration};
use thiserror::Error;
use tracing::debug;
const FLEX_BASE_URL: &str =
"https://ndcdyn.interactivebrokers.com/AccountManagement/FlexWebService";
const FLEX_VERSION: &str = "3";
const REQUEST_TIMEOUT_SECS: u64 = 30;
const POLL_MAX_ATTEMPTS: u32 = 12;
const POLL_INTERVAL: Duration = Duration::from_secs(5);
const POLL_INITIAL_DELAY: Duration = Duration::from_secs(5);
const GENERATION_IN_PROGRESS_CODE: &str = "1019";
const MAX_ERROR_BODY_BYTES: usize = 1024;
const _: () = assert!(
MAX_ERROR_BODY_DOWNLOAD_BYTES >= MAX_ERROR_BODY_BYTES * 4,
"MAX_ERROR_BODY_DOWNLOAD_BYTES must stay well above MAX_ERROR_BODY_BYTES so a token straddling \
the download-cap boundary is truncated away by the retained-cap bound (see sanitize_error_body)"
);
#[non_exhaustive]
#[derive(Debug, Error)]
pub enum IbkrFlexError {
#[error("environment variable error: {0}")]
EnvVar(String),
#[error("invalid credential: {0}")]
InvalidCredential(String),
#[error("HTTP error: {0}")]
Http(reqwest::Error),
#[error("HTTP status {status}: {body}")]
HttpStatus {
status: u16,
body: String,
},
#[error("Flex service error ({code}): {message}")]
Flex {
code: String,
message: String,
},
#[error("Flex statement not ready after {attempts} poll attempts")]
PollTimeout {
attempts: u32,
},
#[error("Flex XML parse error: {0}")]
Parse(String),
}
impl From<reqwest::Error> for IbkrFlexError {
fn from(error: reqwest::Error) -> Self {
IbkrFlexError::Http(error.without_url())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FlexPollPolicy {
pub initial_delay: Duration,
pub interval: Duration,
pub max_attempts: u32,
}
impl Default for FlexPollPolicy {
fn default() -> Self {
Self {
initial_delay: POLL_INITIAL_DELAY,
interval: POLL_INTERVAL,
max_attempts: POLL_MAX_ATTEMPTS,
}
}
}
#[derive(Clone)]
pub struct IbkrFlexConfig {
token: String,
query_id: String,
}
impl fmt::Debug for IbkrFlexConfig {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("IbkrFlexConfig")
.field("token", &"[REDACTED]")
.field("query_id", &self.query_id)
.finish()
}
}
impl IbkrFlexConfig {
pub fn new(
token: impl Into<String>,
query_id: impl Into<String>,
) -> Result<Self, IbkrFlexError> {
Ok(Self {
token: Self::require_nonempty_trimmed("token", token.into())?,
query_id: Self::require_nonempty_trimmed("query_id", query_id.into())?,
})
}
pub fn from_env() -> Result<Self, IbkrFlexError> {
let token = env::var("IBKR_FLEX_TOKEN")
.map_err(|e| IbkrFlexError::EnvVar(format!("IBKR_FLEX_TOKEN: {e}")))?;
let query_id = env::var("IBKR_FLEX_QUERY_ID")
.map_err(|e| IbkrFlexError::EnvVar(format!("IBKR_FLEX_QUERY_ID: {e}")))?;
let token = token.trim();
let query_id = query_id.trim();
if token.is_empty() {
return Err(IbkrFlexError::EnvVar(
"IBKR_FLEX_TOKEN is set but empty".to_owned(),
));
}
if query_id.is_empty() {
return Err(IbkrFlexError::EnvVar(
"IBKR_FLEX_QUERY_ID is set but empty".to_owned(),
));
}
Self::new(token, query_id)
}
fn require_nonempty_trimmed(
field: &'static str,
value: String,
) -> Result<String, IbkrFlexError> {
let trimmed = value.trim();
if trimmed.is_empty() {
return Err(IbkrFlexError::InvalidCredential(format!(
"{field} is empty after trimming whitespace"
)));
}
if trimmed.len() == value.len() {
Ok(value)
} else {
Ok(trimmed.to_owned())
}
}
}
#[derive(Clone)]
pub struct IbkrFlexClient {
http: reqwest::Client,
token: String,
query_id: String,
base_url: String,
poll_policy: FlexPollPolicy,
}
impl fmt::Debug for IbkrFlexClient {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("IbkrFlexClient")
.field("query_id", &self.query_id)
.field("base_url", &self.base_url)
.field("poll_policy", &self.poll_policy)
.finish_non_exhaustive()
}
}
impl IbkrFlexClient {
pub fn new(config: IbkrFlexConfig) -> Result<Self, IbkrFlexError> {
let http = reqwest::Client::builder()
.timeout(Duration::from_secs(REQUEST_TIMEOUT_SECS))
.https_only(true)
.redirect(reqwest::redirect::Policy::none())
.build()?;
Ok(Self {
http,
token: config.token,
query_id: config.query_id,
base_url: FLEX_BASE_URL.to_owned(),
poll_policy: FlexPollPolicy::default(),
})
}
#[must_use]
pub fn with_poll_policy(mut self, policy: FlexPollPolicy) -> Self {
self.poll_policy = policy;
self
}
pub fn from_env() -> Result<Self, IbkrFlexError> {
Self::new(IbkrFlexConfig::from_env()?)
}
pub async fn fetch_statement_xml(&self) -> Result<String, IbkrFlexError> {
let send_url = format!("{}/SendRequest", self.base_url);
debug!("Requesting IBKR Flex statement generation");
let (status, body) = self
.get_with_query(&send_url, self.query_id.as_str())
.await?;
let SendRequestOk {
reference_code,
url,
} = interpret_send_response(status, &body, &self.token)?;
poll_until_ready(&self.poll_policy, &self.token, async || {
self.get_with_query(&url, reference_code.as_str()).await
})
.await
}
async fn get_with_query(
&self,
url: &str,
q: &str,
) -> Result<(reqwest::StatusCode, String), IbkrFlexError> {
let response = self
.http
.get(url)
.query(&[("t", self.token.as_str()), ("q", q), ("v", FLEX_VERSION)])
.send()
.await?;
read_status_and_body(response).await
}
pub async fn fetch_corporate_actions(
&self,
) -> Result<Vec<IbkrFlexCorporateAction>, IbkrFlexError> {
let xml = self.fetch_statement_xml().await?;
corporate_action::parse_corporate_actions_scrubbed(&xml, Some(&self.token))
}
}
async fn read_status_and_body(
response: reqwest::Response,
) -> Result<(reqwest::StatusCode, String), IbkrFlexError> {
let status = response.status();
let body = if status.is_success() {
response.text().await?
} else {
read_body_capped(response, MAX_ERROR_BODY_DOWNLOAD_BYTES).await?
};
Ok((status, body))
}
async fn poll_until_ready<F>(
policy: &FlexPollPolicy,
token: &str,
mut poll_once: F,
) -> Result<String, IbkrFlexError>
where
F: AsyncFnMut() -> Result<(reqwest::StatusCode, String), IbkrFlexError>,
{
if policy.max_attempts > 0 {
tokio::time::sleep(policy.initial_delay).await;
}
for attempt in 1..=policy.max_attempts {
let (status, body) = poll_once().await?;
match interpret_poll_response(status, &body, token)? {
GetStatementOutcome::Ready => return Ok(body),
GetStatementOutcome::InProgress => {
debug!(
attempt,
max = policy.max_attempts,
"Flex statement still generating"
);
if attempt < policy.max_attempts {
tokio::time::sleep(policy.interval).await;
}
}
}
}
Err(IbkrFlexError::PollTimeout {
attempts: policy.max_attempts,
})
}
#[derive(Debug)]
struct SendRequestOk {
reference_code: String,
url: String,
}
#[derive(Debug, PartialEq, Eq)]
enum GetStatementOutcome {
Ready,
InProgress,
}
#[derive(Debug, Deserialize)]
struct FlexStatementResponse {
#[serde(rename = "Status", default)]
status: String,
#[serde(rename = "ReferenceCode", default)]
reference_code: Option<String>,
#[serde(rename = "Url", default)]
url: Option<String>,
#[serde(rename = "ErrorCode", default)]
error_code: Option<String>,
#[serde(rename = "ErrorMessage", default)]
error_message: Option<String>,
}
fn parse_send_request_response(xml: &str) -> Result<SendRequestOk, IbkrFlexError> {
match root_element_name(xml).as_deref() {
Some("FlexStatementResponse") => {}
Some(other) => {
return Err(IbkrFlexError::Parse(format!(
"unexpected Flex SendRequest root element <{other}>"
)));
}
None => {
return Err(IbkrFlexError::Parse(
"empty or unreadable Flex SendRequest response".to_owned(),
));
}
}
let resp: FlexStatementResponse = quick_xml::de::from_str(xml).map_err(|e| {
IbkrFlexError::Parse(format!("failed to parse Flex SendRequest response: {e}"))
})?;
if resp.status.eq_ignore_ascii_case("Success") {
match (nonempty(resp.reference_code), nonempty(resp.url)) {
(Some(reference_code), Some(url)) => {
if !url.starts_with("https://") {
return Err(IbkrFlexError::Parse(format!(
"Flex SendRequest returned a non-HTTPS poll URL; refusing to send credentials over it: {url}"
)));
}
Ok(SendRequestOk {
reference_code,
url,
})
}
_ => Err(IbkrFlexError::Parse(
"Flex SendRequest reported success but is missing ReferenceCode/Url".to_owned(),
)),
}
} else {
Err(flex_error(resp, "SendRequest"))
}
}
fn classify_get_statement(xml: &str) -> Result<GetStatementOutcome, IbkrFlexError> {
match root_element_name(xml).as_deref() {
Some("FlexQueryResponse") => Ok(GetStatementOutcome::Ready),
Some("FlexStatementResponse") => {
let resp: FlexStatementResponse = quick_xml::de::from_str(xml).map_err(|e| {
IbkrFlexError::Parse(format!(
"failed to parse Flex GetStatement status response: {e}"
))
})?;
if resp.error_code.as_deref() == Some(GENERATION_IN_PROGRESS_CODE)
&& resp.status.eq_ignore_ascii_case("warn")
{
Ok(GetStatementOutcome::InProgress)
} else {
Err(flex_error(resp, "GetStatement"))
}
}
Some(other) => Err(IbkrFlexError::Parse(format!(
"unexpected Flex GetStatement root element <{other}>"
))),
None => Err(IbkrFlexError::Parse(
"empty or unreadable Flex GetStatement response".to_owned(),
)),
}
}
fn flex_error(resp: FlexStatementResponse, stage: &str) -> IbkrFlexError {
IbkrFlexError::Flex {
code: resp.error_code.unwrap_or_default(),
message: resp
.error_message
.unwrap_or_else(|| format!("Flex {stage} returned non-success status {}", resp.status)),
}
}
fn scrub_flex_error(code: String, message: String, token: &str) -> IbkrFlexError {
IbkrFlexError::Flex {
code: sanitize_error_body(&code, token),
message: sanitize_error_body(&message, token),
}
}
fn interpret_send_response(
status: reqwest::StatusCode,
body: &str,
token: &str,
) -> Result<SendRequestOk, IbkrFlexError> {
match parse_send_request_response(body) {
Err(IbkrFlexError::Parse(_)) if !status.is_success() => {
Err(http_status_error(status, body, token))
}
Err(IbkrFlexError::Parse(message)) => Err(finish_parse_error(message, Some(token))),
Err(IbkrFlexError::Flex { code, message }) => Err(scrub_flex_error(code, message, token)),
other => other,
}
}
fn interpret_poll_response(
status: reqwest::StatusCode,
body: &str,
token: &str,
) -> Result<GetStatementOutcome, IbkrFlexError> {
match classify_get_statement(body) {
Err(IbkrFlexError::Parse(_)) if !status.is_success() => {
Err(http_status_error(status, body, token))
}
Err(IbkrFlexError::Parse(message)) => Err(finish_parse_error(message, Some(token))),
Err(IbkrFlexError::Flex { code, message }) => Err(scrub_flex_error(code, message, token)),
other => other,
}
}
fn finish_parse_error(message: String, token: Option<&str>) -> IbkrFlexError {
IbkrFlexError::Parse(match token {
Some(token) => sanitize_error_body(&message, token),
None => truncate_str(&message, MAX_ERROR_BODY_BYTES),
})
}
fn http_status_error(status: reqwest::StatusCode, body: &str, token: &str) -> IbkrFlexError {
IbkrFlexError::HttpStatus {
status: status.as_u16(),
body: sanitize_error_body(body, token),
}
}
fn sanitize_error_body(body: &str, token: &str) -> String {
let body = if body.len() > MAX_ERROR_BODY_DOWNLOAD_BYTES {
&body[..body.floor_char_boundary(MAX_ERROR_BODY_DOWNLOAD_BYTES)]
} else {
body
};
let mut scrubbed = redact(body, token);
if !token.is_empty() {
let wire_form: String = url::form_urlencoded::byte_serialize(token.as_bytes()).collect();
if wire_form.as_str() != token {
scrubbed = redact(&scrubbed, &wire_form);
}
}
if scrubbed.len() > MAX_ERROR_BODY_BYTES {
scrubbed.truncate(scrubbed.floor_char_boundary(MAX_ERROR_BODY_BYTES));
}
scrubbed
}
fn redact(body: &str, needle: &str) -> String {
if !needle.is_empty() && body.contains(needle) {
body.replace(needle, "[REDACTED]")
} else {
body.to_owned()
}
}
fn root_element_name(xml: &str) -> Option<SmolStr> {
let mut reader = Reader::from_str(xml);
loop {
match reader.read_event() {
Ok(Event::Start(e)) | Ok(Event::Empty(e)) => {
return Some(SmolStr::new(e.name().as_ref()));
}
Ok(Event::Eof) | Err(_) => return None,
Ok(_) => {}
}
}
}
fn nonempty(value: Option<String>) -> Option<String> {
value.and_then(|v| {
let trimmed = v.trim();
if trimmed.is_empty() {
None
} else if trimmed.len() == v.len() {
Some(v)
} else {
Some(trimmed.to_owned())
}
})
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
use super::*;
const SEND_SUCCESS: &str = r#"<FlexStatementResponse timestamp="15 January, 2025 03:00 PM EST">
<Status>Success</Status>
<ReferenceCode>1234567890</ReferenceCode>
<Url>https://ndcdyn.interactivebrokers.com/AccountManagement/FlexWebService/GetStatement</Url>
</FlexStatementResponse>"#;
const SEND_FAIL: &str = r#"<FlexStatementResponse timestamp="15 January, 2025 03:00 PM EST">
<Status>Fail</Status>
<ErrorCode>1003</ErrorCode>
<ErrorMessage>Statement could not be generated at this time. Please try again shortly.</ErrorMessage>
</FlexStatementResponse>"#;
const GET_IN_PROGRESS: &str = r#"<FlexStatementResponse timestamp="15 January, 2025 03:00 PM EST">
<Status>Warn</Status>
<ErrorCode>1019</ErrorCode>
<ErrorMessage>Statement generation in progress. Please try again shortly.</ErrorMessage>
</FlexStatementResponse>"#;
const GET_FAIL: &str = r#"<FlexStatementResponse>
<Status>Fail</Status>
<ErrorCode>1020</ErrorCode>
<ErrorMessage>Invalid request or unable to validate request.</ErrorMessage>
</FlexStatementResponse>"#;
const GET_READY: &str = r#"<?xml version="1.0"?>
<FlexQueryResponse queryName="Activity" type="AF">
<FlexStatements count="1">
<FlexStatement accountId="U1"><CorporateActions /></FlexStatement>
</FlexStatements>
</FlexQueryResponse>"#;
#[test]
fn send_request_success_extracts_reference_and_url() {
let ok = parse_send_request_response(SEND_SUCCESS).unwrap();
assert_eq!(ok.reference_code, "1234567890");
assert_eq!(
ok.url,
"https://ndcdyn.interactivebrokers.com/AccountManagement/FlexWebService/GetStatement"
);
}
#[test]
fn send_request_failure_is_a_flex_error() {
match parse_send_request_response(SEND_FAIL) {
Err(IbkrFlexError::Flex { code, message }) => {
assert_eq!(code, "1003");
assert!(message.contains("could not be generated"));
}
other => panic!("expected Flex error, got {other:?}"),
}
}
#[test]
fn send_request_success_without_url_is_a_parse_error() {
let xml = r#"<FlexStatementResponse><Status>Success</Status>
<ReferenceCode>123</ReferenceCode></FlexStatementResponse>"#;
assert!(matches!(
parse_send_request_response(xml),
Err(IbkrFlexError::Parse(_))
));
}
#[test]
fn send_request_non_https_url_is_rejected() {
let xml = r#"<FlexStatementResponse><Status>Success</Status>
<ReferenceCode>123</ReferenceCode>
<Url>http://ndcdyn.interactivebrokers.com/AccountManagement/FlexWebService/GetStatement</Url></FlexStatementResponse>"#;
assert!(matches!(
parse_send_request_response(xml),
Err(IbkrFlexError::Parse(_))
));
}
#[test]
fn send_request_non_envelope_root_is_a_parse_error() {
for body in [
"<html><body>502 Bad Gateway</body></html>",
"<not-flex/>",
"",
] {
assert!(
matches!(
parse_send_request_response(body),
Err(IbkrFlexError::Parse(_))
),
"expected Parse for body {body:?}"
);
}
}
#[test]
fn get_statement_ready_is_classified_ready() {
assert_eq!(
classify_get_statement(GET_READY).unwrap(),
GetStatementOutcome::Ready
);
}
#[test]
fn get_statement_generation_in_progress_is_classified_in_progress() {
assert_eq!(
classify_get_statement(GET_IN_PROGRESS).unwrap(),
GetStatementOutcome::InProgress
);
}
#[test]
fn get_statement_in_progress_code_is_case_insensitive_on_status() {
const WARN_UPPER_1019: &str = r#"<FlexStatementResponse>
<Status>WARN</Status>
<ErrorCode>1019</ErrorCode>
<ErrorMessage>Statement generation in progress. Please try again shortly.</ErrorMessage>
</FlexStatementResponse>"#;
assert_eq!(
classify_get_statement(WARN_UPPER_1019).unwrap(),
GetStatementOutcome::InProgress
);
}
#[test]
fn get_statement_in_progress_code_under_fail_status_is_a_flex_error() {
const FAIL_1019: &str = r#"<FlexStatementResponse>
<Status>Fail</Status>
<ErrorCode>1019</ErrorCode>
<ErrorMessage>Statement generation in progress. Please try again shortly.</ErrorMessage>
</FlexStatementResponse>"#;
match classify_get_statement(FAIL_1019) {
Err(IbkrFlexError::Flex { code, .. }) => assert_eq!(code, "1019"),
other => panic!("expected terminal Flex error, got {other:?}"),
}
}
#[test]
fn get_statement_1019_without_status_is_a_flex_error() {
const NO_STATUS_1019: &str = r#"<FlexStatementResponse>
<ErrorCode>1019</ErrorCode>
<ErrorMessage>Statement generation in progress. Please try again shortly.</ErrorMessage>
</FlexStatementResponse>"#;
match classify_get_statement(NO_STATUS_1019) {
Err(IbkrFlexError::Flex { code, .. }) => assert_eq!(code, "1019"),
other => panic!("expected terminal Flex error, got {other:?}"),
}
}
#[test]
fn get_statement_1019_under_success_status_is_a_flex_error() {
const SUCCESS_1019: &str = r#"<FlexStatementResponse>
<Status>Success</Status>
<ErrorCode>1019</ErrorCode>
<ErrorMessage>Statement generation in progress. Please try again shortly.</ErrorMessage>
</FlexStatementResponse>"#;
match classify_get_statement(SUCCESS_1019) {
Err(IbkrFlexError::Flex { code, .. }) => assert_eq!(code, "1019"),
other => panic!("expected terminal Flex error, got {other:?}"),
}
}
#[test]
fn get_statement_terminal_failure_is_a_flex_error() {
match classify_get_statement(GET_FAIL) {
Err(IbkrFlexError::Flex { code, message }) => {
assert_eq!(code, "1020");
assert!(message.contains("Invalid request"));
}
other => panic!("expected Flex error, got {other:?}"),
}
}
#[test]
fn root_element_name_skips_prolog() {
assert_eq!(
root_element_name(GET_READY).as_deref(),
Some("FlexQueryResponse")
);
assert_eq!(
root_element_name(SEND_SUCCESS).as_deref(),
Some("FlexStatementResponse")
);
assert_eq!(root_element_name("").as_deref(), None);
}
#[test]
fn config_debug_redacts_token() {
let config =
IbkrFlexConfig::new("super-secret-token", "987654").expect("valid credentials");
let debug = format!("{config:?}");
assert!(
!debug.contains("super-secret-token"),
"token must be redacted"
);
assert!(debug.contains("987654"), "query_id is not a secret");
}
#[test]
fn client_debug_hides_token() {
let client = IbkrFlexClient::new(
IbkrFlexConfig::new("super-secret-token", "987654").expect("valid credentials"),
)
.expect("HTTP client should build");
let debug = format!("{client:?}");
assert!(
!debug.contains("super-secret-token"),
"token must not leak via Debug"
);
assert!(debug.contains("987654"));
}
#[test]
fn new_rejects_empty_or_whitespace_token() {
for token in ["", " ", "\t\n"] {
match IbkrFlexConfig::new(token, "987654") {
Err(IbkrFlexError::InvalidCredential(msg)) => assert!(
msg.contains("token"),
"error should name the `token` field, got: {msg}"
),
other => panic!("expected InvalidCredential for token {token:?}, got: {other:?}"),
}
}
}
#[test]
fn new_rejects_empty_or_whitespace_query_id() {
for query_id in ["", " ", "\t\n"] {
match IbkrFlexConfig::new("valid-token", query_id) {
Err(IbkrFlexError::InvalidCredential(msg)) => assert!(
msg.contains("query_id"),
"error should name the `query_id` field, got: {msg}"
),
other => {
panic!("expected InvalidCredential for query_id {query_id:?}, got: {other:?}")
}
}
}
}
#[test]
fn new_trims_surrounding_whitespace() {
let config = IbkrFlexConfig::new(" super-secret-token ", " 987654 ")
.expect("padded-but-nonempty credentials are valid");
let debug = format!("{config:?}");
assert!(
debug.contains("\"987654\""),
"query_id should be trimmed to `987654`, got: {debug}"
);
assert!(
!debug.contains(" 987654"),
"surrounding whitespace should not survive into the stored query_id, got: {debug}"
);
}
#[test]
fn interpret_send_response_success_parses() {
let ok = interpret_send_response(reqwest::StatusCode::OK, SEND_SUCCESS, "tok").unwrap();
assert_eq!(ok.reference_code, "1234567890");
}
#[test]
fn interpret_send_response_flex_error_preserved_under_non_2xx() {
match interpret_send_response(reqwest::StatusCode::UNAUTHORIZED, SEND_FAIL, "tok") {
Err(IbkrFlexError::Flex { code, .. }) => assert_eq!(code, "1003"),
other => panic!("expected Flex error, got {other:?}"),
}
}
#[test]
fn interpret_send_response_non_flex_body_under_non_2xx_is_http_status() {
let body = "<html><body>502 Bad Gateway</body></html>";
match interpret_send_response(reqwest::StatusCode::BAD_GATEWAY, body, "tok") {
Err(IbkrFlexError::HttpStatus { status, body }) => {
assert_eq!(status, 502);
assert!(body.contains("Bad Gateway"), "diagnostic body preserved");
}
other => panic!("expected HttpStatus, got {other:?}"),
}
}
#[test]
fn interpret_send_response_garbage_under_2xx_is_parse_error() {
assert!(matches!(
interpret_send_response(reqwest::StatusCode::OK, "<not-flex/>", "tok"),
Err(IbkrFlexError::Parse(_))
));
}
#[test]
fn interpret_poll_response_ready_and_in_progress() {
assert_eq!(
interpret_poll_response(reqwest::StatusCode::OK, GET_READY, "tok").unwrap(),
GetStatementOutcome::Ready
);
assert_eq!(
interpret_poll_response(reqwest::StatusCode::OK, GET_IN_PROGRESS, "tok").unwrap(),
GetStatementOutcome::InProgress
);
}
#[test]
fn interpret_poll_response_1019_under_non_2xx_is_still_retryable() {
assert_eq!(
interpret_poll_response(
reqwest::StatusCode::SERVICE_UNAVAILABLE,
GET_IN_PROGRESS,
"tok"
)
.unwrap(),
GetStatementOutcome::InProgress
);
}
#[test]
fn interpret_poll_response_terminal_flex_error_preserved_under_non_2xx() {
match interpret_poll_response(reqwest::StatusCode::BAD_REQUEST, GET_FAIL, "tok") {
Err(IbkrFlexError::Flex { code, .. }) => assert_eq!(code, "1020"),
other => panic!("expected Flex error, got {other:?}"),
}
}
#[test]
fn interpret_send_response_flex_error_message_is_token_scrubbed() {
let token = "super-secret-token";
let body = format!(
"<FlexStatementResponse><Status>Fail</Status><ErrorCode>1003</ErrorCode>\
<ErrorMessage>invalid token for GET /SendRequest?t={token}&q=1&v=3</ErrorMessage>\
</FlexStatementResponse>"
);
match interpret_send_response(reqwest::StatusCode::OK, &body, token) {
Err(IbkrFlexError::Flex { code, message }) => {
assert_eq!(code, "1003", "the Flex code is preserved for matching");
assert!(
!message.contains(token),
"token must be redacted, got: {message}"
);
assert!(message.contains("[REDACTED]"));
}
other => panic!("expected Flex, got {other:?}"),
}
}
#[test]
fn interpret_poll_response_flex_error_message_is_token_scrubbed() {
let token = "super-secret-token";
let body = format!(
"<FlexStatementResponse><Status>Fail</Status><ErrorCode>1020</ErrorCode>\
<ErrorMessage>bad reference for GET /GetStatement?t={token}&q=1&v=3</ErrorMessage>\
</FlexStatementResponse>"
);
match interpret_poll_response(reqwest::StatusCode::OK, &body, token) {
Err(IbkrFlexError::Flex { code, message }) => {
assert_eq!(code, "1020");
assert!(
!message.contains(token),
"token must be redacted, got: {message}"
);
assert!(message.contains("[REDACTED]"));
}
other => panic!("expected Flex, got {other:?}"),
}
}
#[test]
fn interpret_send_response_flex_error_code_is_token_scrubbed() {
let token = "super-secret-token";
let body = format!(
"<FlexStatementResponse><Status>Fail</Status>\
<ErrorCode>bogus t={token}</ErrorCode>\
<ErrorMessage>rejected</ErrorMessage></FlexStatementResponse>"
);
match interpret_send_response(reqwest::StatusCode::OK, &body, token) {
Err(IbkrFlexError::Flex { code, message }) => {
assert!(
!code.contains(token),
"token must be redacted from code, got: {code}"
);
assert!(code.contains("[REDACTED]"));
assert_eq!(message, "rejected");
}
other => panic!("expected Flex, got {other:?}"),
}
}
#[test]
fn flex_error_message_is_bounded() {
let body = format!(
"<FlexStatementResponse><Status>Fail</Status><ErrorCode>1003</ErrorCode>\
<ErrorMessage>{}</ErrorMessage></FlexStatementResponse>",
"€".repeat(1000) );
match interpret_send_response(reqwest::StatusCode::OK, &body, "tok") {
Err(IbkrFlexError::Flex { message, .. }) => assert!(
message.len() <= MAX_ERROR_BODY_BYTES,
"message bounded to the cap, got {} bytes",
message.len()
),
other => panic!("expected Flex, got {other:?}"),
}
}
#[test]
fn interpret_poll_response_non_flex_body_under_non_2xx_is_http_status() {
match interpret_poll_response(
reqwest::StatusCode::GATEWAY_TIMEOUT,
"upstream request timed out",
"tok",
) {
Err(IbkrFlexError::HttpStatus { status, body }) => {
assert_eq!(status, 504);
assert!(body.contains("timed out"));
}
other => panic!("expected HttpStatus, got {other:?}"),
}
}
fn synthetic_response(status: u16, body: &'static str) -> reqwest::Response {
reqwest::Response::from(
http::Response::builder()
.status(status)
.body(body)
.expect("status and body are valid"),
)
}
#[tokio::test]
async fn read_status_and_body_keeps_a_1019_body_under_a_non_2xx_status() {
let (status, body) = read_status_and_body(synthetic_response(503, GET_IN_PROGRESS))
.await
.expect("a non-success status must still yield its body");
assert_eq!(status, reqwest::StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(
classify_get_statement(&body).expect("body is a Flex envelope"),
GetStatementOutcome::InProgress,
"a 1019 arriving under a non-2xx status must still classify as retryable"
);
}
#[tokio::test]
async fn read_status_and_body_reads_a_success_body_in_full() {
let (status, body) = read_status_and_body(synthetic_response(200, GET_READY))
.await
.expect("success body reads");
assert_eq!(status, reqwest::StatusCode::OK);
assert_eq!(body, GET_READY);
}
#[test]
fn http_status_body_is_token_scrubbed() {
let token = "super-secret-token";
let body = format!("400 Bad Request: GET /SendRequest?t={token}&q=1&v=3");
match interpret_send_response(reqwest::StatusCode::BAD_REQUEST, &body, token) {
Err(IbkrFlexError::HttpStatus { body, .. }) => {
assert!(!body.contains(token), "token must be redacted, got: {body}");
assert!(body.contains("[REDACTED]"));
}
other => panic!("expected HttpStatus, got {other:?}"),
}
}
#[test]
fn parse_message_is_token_scrubbed_under_2xx() {
let token = "super-secret-token";
let body = format!("<{token}/>");
match interpret_send_response(reqwest::StatusCode::OK, &body, token) {
Err(IbkrFlexError::Parse(message)) => {
assert!(
!message.contains(token),
"token must be redacted, got: {message}"
);
assert!(message.contains("[REDACTED]"));
}
other => panic!("expected Parse, got {other:?}"),
}
match interpret_poll_response(reqwest::StatusCode::OK, &body, token) {
Err(IbkrFlexError::Parse(message)) => {
assert!(
!message.contains(token),
"token must be redacted, got: {message}"
);
assert!(message.contains("[REDACTED]"));
}
other => panic!("expected Parse, got {other:?}"),
}
}
#[test]
fn parse_message_is_bounded_under_2xx() {
let body = format!("<{}/>", "€".repeat(1000)); match interpret_send_response(reqwest::StatusCode::OK, &body, "tok") {
Err(IbkrFlexError::Parse(message)) => {
assert!(
message.starts_with("unexpected"),
"fixture must reach the root-element path, got: {message}"
);
assert!(
message.len() <= MAX_ERROR_BODY_BYTES,
"message bounded to the cap, got {} bytes",
message.len()
);
}
other => panic!("expected Parse, got {other:?}"),
}
}
#[test]
fn http_status_body_is_bounded_on_char_boundary() {
let body = "€".repeat(1000); match interpret_send_response(reqwest::StatusCode::BAD_GATEWAY, &body, "tok") {
Err(IbkrFlexError::HttpStatus { body, .. }) => {
assert!(
body.len() <= MAX_ERROR_BODY_BYTES,
"body bounded to the cap"
);
assert!(
body.chars().all(|c| c == '€'),
"truncation must land on a char boundary (valid UTF-8)"
);
}
other => panic!("expected HttpStatus, got {other:?}"),
}
}
#[test]
fn http_status_body_scrubs_token_straddling_the_cap() {
let token = "SECRETTOKEN1234567890"; let prefix = "x".repeat(MAX_ERROR_BODY_BYTES - 9); let body = format!("{prefix}{token}{}", "y".repeat(100)); match interpret_send_response(reqwest::StatusCode::BAD_REQUEST, &body, token) {
Err(IbkrFlexError::HttpStatus { body, .. }) => {
assert!(
!body.contains(token),
"full token must be redacted, got: {body}"
);
assert!(
!body.contains("SECRET"),
"no token fragment may survive the boundary, got: {body}"
);
assert!(
body.len() <= MAX_ERROR_BODY_BYTES,
"redaction must not push the body back over the cap, len = {}",
body.len()
);
}
other => panic!("expected HttpStatus, got {other:?}"),
}
}
#[test]
fn corporate_action_parse_scrubs_token_straddling_the_cap() {
let token = "SECRETTOKEN1234567890"; let mut straddled = false;
for pad in (MAX_ERROR_BODY_BYTES - 80)..(MAX_ERROR_BODY_BYTES - 40) {
let xml = format!("<{}{token}/>", "x".repeat(pad));
let Err(IbkrFlexError::Parse(plain)) =
corporate_action::parse_corporate_actions_scrubbed(&xml, None)
else {
panic!("expected Parse from the token-less parse at pad {pad}");
};
straddled |= (1..token.len()).any(|cut| plain.ends_with(&token[..cut]));
match corporate_action::parse_corporate_actions_scrubbed(&xml, Some(token)) {
Err(IbkrFlexError::Parse(message)) => {
assert!(
!message.contains(token),
"full token must be redacted at pad {pad}"
);
for cut in 1..token.len() {
assert!(
!message.ends_with(&token[..cut]),
"a {cut}-byte token fragment survived the boundary at pad {pad}, got: {message}"
);
}
assert!(
message.len() <= MAX_ERROR_BODY_BYTES,
"redaction must not push the message over the cap at pad {pad}, len = {}",
message.len()
);
}
other => panic!("expected Parse, got {other:?}"),
}
}
assert!(
straddled,
"the swept range no longer places the token across the {MAX_ERROR_BODY_BYTES}-byte cap \
— widen it, or this test proves nothing"
);
}
#[test]
fn corporate_action_parse_without_a_token_is_still_bounded() {
let xml = format!("<{}/>", "€".repeat(1000)); match parse_corporate_actions(&xml) {
Err(IbkrFlexError::Parse(message)) => assert!(
message.len() <= MAX_ERROR_BODY_BYTES,
"len = {}",
message.len()
),
other => panic!("expected Parse, got {other:?}"),
}
}
#[test]
fn max_error_body_bytes_is_the_documented_cap() {
assert_eq!(MAX_ERROR_BODY_BYTES, 1024);
}
#[test]
fn flex_poll_policy_default_is_the_documented_budget() {
assert_eq!(
FlexPollPolicy::default(),
FlexPollPolicy {
initial_delay: Duration::from_secs(5),
interval: Duration::from_secs(5),
max_attempts: 12,
}
);
}
#[test]
fn sanitize_error_body_scrubs_reqwests_wire_encoded_token() {
let token = "abc def+g/h~i";
let request = reqwest::Client::new()
.get("https://ndcdyn.interactivebrokers.com/AccountManagement/FlexWebService/GetStatement")
.query(&[("t", token), ("q", "1234567890"), ("v", FLEX_VERSION)])
.build()
.expect("request builds without sending");
let wire_query = request.url().query().expect("query present");
let wire_form: String = url::form_urlencoded::byte_serialize(token.as_bytes()).collect();
assert_ne!(
wire_form, token,
"test token must actually encode to something different, else it proves nothing"
);
assert!(
wire_query.contains(&format!("t={wire_form}")),
"byte_serialize must match reqwest's actual query encoding, else the wire-form scrub \
silently stops matching; query = {wire_query}, wire_form = {wire_form}"
);
let reflected = format!("502 Bad Gateway while proxying GET /GetStatement?{wire_query}");
let scrubbed = sanitize_error_body(&reflected, token);
assert!(
!scrubbed.contains(&wire_form),
"wire-encoded token must be redacted, got: {scrubbed}"
);
assert!(scrubbed.contains("[REDACTED]"));
}
#[tokio::test]
async fn poll_until_ready_zero_max_attempts_times_out_without_polling() {
let policy = FlexPollPolicy {
initial_delay: Duration::ZERO,
interval: Duration::ZERO,
max_attempts: 0,
};
let calls = std::cell::Cell::new(0u32);
let result = poll_until_ready(&policy, "tok", async || {
calls.set(calls.get() + 1);
Ok((reqwest::StatusCode::OK, GET_READY.to_owned()))
})
.await;
assert_eq!(calls.get(), 0, "max_attempts == 0 must not issue any poll");
assert!(matches!(
result,
Err(IbkrFlexError::PollTimeout { attempts: 0 })
));
}
#[tokio::test]
async fn poll_until_ready_returns_as_soon_as_statement_is_ready() {
let policy = FlexPollPolicy {
initial_delay: Duration::ZERO,
interval: Duration::ZERO,
max_attempts: 5,
};
let calls = std::cell::Cell::new(0u32);
let result = poll_until_ready(&policy, "tok", async || {
calls.set(calls.get() + 1);
let body = if calls.get() < 2 {
GET_IN_PROGRESS
} else {
GET_READY
};
Ok((reqwest::StatusCode::OK, body.to_owned()))
})
.await;
assert_eq!(calls.get(), 2, "must stop polling once Ready is seen");
assert_eq!(result.expect("ready"), GET_READY);
}
#[tokio::test]
async fn poll_until_ready_exhausts_budget_and_reports_attempt_count() {
let policy = FlexPollPolicy {
initial_delay: Duration::ZERO,
interval: Duration::ZERO,
max_attempts: 3,
};
let calls = std::cell::Cell::new(0u32);
let result = poll_until_ready(&policy, "tok", async || {
calls.set(calls.get() + 1);
Ok((reqwest::StatusCode::OK, GET_IN_PROGRESS.to_owned()))
})
.await;
assert_eq!(calls.get(), 3, "must poll exactly max_attempts times");
assert!(matches!(
result,
Err(IbkrFlexError::PollTimeout { attempts: 3 })
));
}
#[tokio::test(start_paused = true)]
async fn poll_until_ready_sleeps_initial_delay_first_then_interval_between_polls() {
let policy = FlexPollPolicy {
initial_delay: Duration::from_secs(5),
interval: Duration::from_secs(2),
max_attempts: 3,
};
let start = tokio::time::Instant::now();
let elapsed_at_poll = std::cell::RefCell::new(Vec::<Duration>::new());
let result = poll_until_ready(&policy, "tok", async || {
elapsed_at_poll.borrow_mut().push(start.elapsed());
Ok((reqwest::StatusCode::OK, GET_IN_PROGRESS.to_owned()))
})
.await;
assert!(matches!(
result,
Err(IbkrFlexError::PollTimeout { attempts: 3 })
));
assert_eq!(
*elapsed_at_poll.borrow(),
vec![
Duration::from_secs(5),
Duration::from_secs(7),
Duration::from_secs(9),
],
);
}
}