use crate::auth_http_client;
use async_trait::async_trait;
use base64::Engine;
use faucet_core::{AuthProvider, Credential, CredentialPlacement, FaucetError, RequestAuth};
use hmac::{Hmac, Mac};
use jsonpath_rust::JsonPath;
use serde::Deserialize;
use serde_json::Value;
use sha2::Sha256;
use std::collections::HashMap;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::Mutex;
use tokio::time::{Duration, Instant};
type HmacSha256 = Hmac<Sha256>;
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct FlowRequest {
#[serde(default = "default_method")]
method: String,
url: String,
#[serde(default)]
headers: HashMap<String, String>,
#[serde(default)]
query: HashMap<String, String>,
#[serde(default)]
form: Option<HashMap<String, String>>,
#[serde(default)]
json: Option<Value>,
#[serde(default)]
sign: Option<SignSpec>,
}
fn default_method() -> String {
"GET".to_owned()
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
enum CaptureFrom {
#[default]
Json,
Xml,
Header,
SetCookie,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct CaptureSource {
from: CaptureFrom,
#[serde(default)]
name: Option<String>,
#[serde(default)]
path: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(untagged)]
enum CaptureSpec {
Json(String),
Source(CaptureSource),
}
impl CaptureSpec {
fn kind(&self) -> CaptureFrom {
match self {
CaptureSpec::Json(_) => CaptureFrom::Json,
CaptureSpec::Source(s) => s.from,
}
}
fn validate(&self) -> Result<(), &'static str> {
match self {
CaptureSpec::Json(p) if p.trim().is_empty() => Err("empty JSONPath"),
CaptureSpec::Json(_) => Ok(()),
CaptureSpec::Source(s) => match s.from {
CaptureFrom::Json | CaptureFrom::Xml => match s.path.as_deref().map(str::trim) {
Some(p) if !p.is_empty() => Ok(()),
_ => Err("`from: json|xml` requires a non-empty `path`"),
},
CaptureFrom::Header | CaptureFrom::SetCookie => {
match s.name.as_deref().map(str::trim) {
Some(n) if !n.is_empty() => Ok(()),
_ => Err("`from: header|set_cookie` requires a non-empty `name`"),
}
}
},
}
}
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct FlowStep {
request: FlowRequest,
#[serde(default)]
capture: HashMap<String, CaptureSpec>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
enum PlaceTarget {
Header,
Query,
Cookie,
Body,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
enum SignAlg {
HmacSha256,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "snake_case")]
enum SigEncoding {
#[default]
Hex,
Base64,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct SignInto {
header: String,
#[serde(default = "default_sig_format")]
format: String,
}
fn default_sig_format() -> String {
"${sig}".to_owned()
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct SignSpec {
alg: SignAlg,
key: String,
template: String,
#[serde(default)]
encoding: SigEncoding,
into: SignInto,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(untagged, deny_unknown_fields)]
enum ApplySpec {
Place {
into: PlaceTarget,
name: String,
value: String,
},
Sign { sign: SignSpec },
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct FlowConfig {
#[serde(default)]
steps: Vec<FlowStep>,
#[serde(default)]
apply: Vec<ApplySpec>,
#[serde(default)]
base_url_from: Option<String>,
#[serde(default)]
ttl_secs: Option<u64>,
#[serde(default)]
reauth_on: Vec<u16>,
}
impl FlowConfig {
fn validate(&self) -> Result<(), FaucetError> {
if self.steps.is_empty() && self.apply.is_empty() {
return Err(FaucetError::Config(
"flow auth: at least one of `steps` or `apply` is required".to_owned(),
));
}
for (i, step) in self.steps.iter().enumerate() {
if step.request.url.trim().is_empty() {
return Err(FaucetError::Config(format!(
"flow auth: step {i} has an empty `url`"
)));
}
if step.request.form.is_some() && step.request.json.is_some() {
return Err(FaucetError::Config(format!(
"flow auth: step {i} sets both `form` and `json`; pick one"
)));
}
if let Some(sign) = &step.request.sign
&& sign.into.header.trim().is_empty()
{
return Err(FaucetError::Config(format!(
"flow auth: step {i} sign.into.header must not be empty"
)));
}
for (name, cap) in &step.capture {
if let Err(msg) = cap.validate() {
return Err(FaucetError::Config(format!(
"flow auth: step {i} capture '{name}': {msg}"
)));
}
}
}
for (i, a) in self.apply.iter().enumerate() {
match a {
ApplySpec::Place { name, .. } if name.trim().is_empty() => {
return Err(FaucetError::Config(format!(
"flow auth: apply[{i}] has an empty `name`"
)));
}
ApplySpec::Sign { sign } if sign.into.header.trim().is_empty() => {
return Err(FaucetError::Config(format!(
"flow auth: apply[{i}].sign.into.header must not be empty"
)));
}
_ => {}
}
}
Ok(())
}
}
fn render(template: &str, ctx: &HashMap<String, String>) -> String {
let mut out = String::with_capacity(template.len());
let mut rest = template;
while let Some(start) = rest.find("${") {
out.push_str(&rest[..start]);
let after = &rest[start + 2..];
if let Some(end) = after.find('}') {
let key = &after[..end];
match ctx.get(key) {
Some(v) => out.push_str(v),
None => {
out.push_str("${");
out.push_str(key);
out.push('}');
}
}
rest = &after[end + 1..];
} else {
out.push_str(&rest[start..]);
rest = "";
}
}
out.push_str(rest);
out
}
fn value_to_string(v: &Value) -> String {
match v {
Value::String(s) => s.clone(),
Value::Bool(b) => b.to_string(),
Value::Number(n) => n.to_string(),
Value::Null => String::new(),
other => other.to_string(),
}
}
fn jsonpath_first(body: &Value, path: &str) -> Option<Value> {
let results = body.query(path).ok()?;
results.first().map(|v| (*v).clone())
}
fn to_hex(bytes: &[u8]) -> String {
let mut s = String::with_capacity(bytes.len() * 2);
for b in bytes {
s.push_str(&format!("{b:02x}"));
}
s
}
fn hmac_sign(key: &str, message: &str, encoding: SigEncoding) -> String {
let mut mac =
HmacSha256::new_from_slice(key.as_bytes()).expect("HMAC accepts a key of any length");
mac.update(message.as_bytes());
let bytes = mac.finalize().into_bytes();
match encoding {
SigEncoding::Hex => to_hex(&bytes),
SigEncoding::Base64 => base64::engine::general_purpose::STANDARD.encode(bytes),
}
}
fn sign_header(sign: &SignSpec, ctx: &HashMap<String, String>) -> (String, String) {
let base = render(&sign.template, ctx);
let sig = match sign.alg {
SignAlg::HmacSha256 => hmac_sign(&sign.key, &base, sign.encoding),
};
let mut sig_ctx = ctx.clone();
sig_ctx.insert("sig".to_owned(), sig);
let value = render(&sign.into.format, &sig_ctx);
(sign.into.header.clone(), value)
}
fn header_value(headers: &reqwest::header::HeaderMap, name: &str) -> Option<String> {
headers
.get(name)
.and_then(|v| v.to_str().ok())
.map(|s| s.to_owned())
}
fn set_cookie_value(headers: &reqwest::header::HeaderMap, name: &str) -> Option<String> {
for v in headers.get_all("set-cookie") {
let Ok(s) = v.to_str() else { continue };
let pair = s.split(';').next().unwrap_or("");
if let Some((k, val)) = pair.split_once('=')
&& k.trim() == name
{
return Some(val.trim().to_owned());
}
}
None
}
#[derive(Debug)]
struct XmlNode {
tag: String,
text: String,
children: Vec<XmlNode>,
}
enum XmlToken {
Start(String),
End,
SelfClose(String),
Text(String),
}
fn xml_local_name(raw: &str) -> String {
let name = raw.split_whitespace().next().unwrap_or("");
match name.split_once(':') {
Some((_, local)) => local.to_owned(),
None => name.to_owned(),
}
}
fn xml_unescape(s: &str) -> String {
s.replace("<", "<")
.replace(">", ">")
.replace(""", "\"")
.replace("'", "'")
.replace("&", "&")
}
fn xml_tokenize(input: &str) -> Vec<XmlToken> {
let mut out = Vec::new();
let mut rest = input;
while let Some(lt) = rest.find('<') {
let text = &rest[..lt];
if !text.trim().is_empty() {
out.push(XmlToken::Text(xml_unescape(text)));
}
let after = &rest[lt..];
if let Some(cdata) = after.strip_prefix("<![CDATA[") {
if let Some(end) = cdata.find("]]>") {
let content = &cdata[..end];
if !content.trim().is_empty() {
out.push(XmlToken::Text(content.to_owned()));
}
rest = &cdata[end + 3..];
continue;
}
break;
}
let Some(gt_rel) = after.find('>') else { break };
let inner = &after[1..gt_rel]; rest = &after[gt_rel + 1..];
if inner.starts_with('?') || inner.starts_with('!') {
continue;
}
if let Some(close) = inner.strip_prefix('/') {
let _ = close;
out.push(XmlToken::End);
} else if let Some(sc) = inner.strip_suffix('/') {
out.push(XmlToken::SelfClose(xml_local_name(sc)));
} else {
out.push(XmlToken::Start(xml_local_name(inner)));
}
}
out
}
fn xml_build_tree(tokens: Vec<XmlToken>) -> XmlNode {
let mut stack: Vec<XmlNode> = vec![XmlNode {
tag: String::new(),
text: String::new(),
children: Vec::new(),
}];
for tok in tokens {
match tok {
XmlToken::Start(tag) => stack.push(XmlNode {
tag,
text: String::new(),
children: Vec::new(),
}),
XmlToken::SelfClose(tag) => {
if let Some(parent) = stack.last_mut() {
parent.children.push(XmlNode {
tag,
text: String::new(),
children: Vec::new(),
});
}
}
XmlToken::Text(t) => {
if let Some(node) = stack.last_mut() {
node.text.push_str(&t);
}
}
XmlToken::End => {
if stack.len() > 1 {
let node = stack.pop().unwrap();
stack.last_mut().unwrap().children.push(node);
}
}
}
}
while stack.len() > 1 {
let node = stack.pop().unwrap();
stack.last_mut().unwrap().children.push(node);
}
stack.pop().unwrap()
}
fn xml_dot_path(xml: &str, path: &str) -> Option<String> {
let root = xml_build_tree(xml_tokenize(xml));
let mut current = &root;
for seg in path.split('.') {
let seg = xml_local_name(seg.trim());
if seg.is_empty() {
continue;
}
current = current.children.iter().find(|c| c.tag == seg)?;
}
Some(current.text.trim().to_owned())
}
fn build_request_auth(
apply: &[ApplySpec],
base_url_from: &Option<String>,
ctx: &HashMap<String, String>,
) -> RequestAuth {
let mut out = RequestAuth::new();
for a in apply {
match a {
ApplySpec::Place { into, name, value } => {
let value = render(value, ctx);
let name = name.clone();
let placement = match into {
PlaceTarget::Header => CredentialPlacement::Header { name, value },
PlaceTarget::Query => CredentialPlacement::Query { name, value },
PlaceTarget::Cookie => CredentialPlacement::Cookie { name, value },
PlaceTarget::Body => CredentialPlacement::BodyField { name, value },
};
out = out.with_placement(placement);
}
ApplySpec::Sign { sign } => {
let (name, value) = sign_header(sign, ctx);
out = out.with_placement(CredentialPlacement::Header { name, value });
}
}
}
if let Some(tmpl) = base_url_from {
let rendered = render(tmpl, ctx);
if !rendered.is_empty() {
out = out.with_base_url(rendered);
}
}
out = out.with_captured(ctx.iter().map(|(k, v)| (k.clone(), v.clone())).collect());
out
}
struct Session {
ctx: HashMap<String, String>,
expires_at: Option<Instant>,
}
impl Session {
fn valid(&self) -> bool {
match self.expires_at {
Some(exp) => Instant::now() < exp,
None => true,
}
}
}
pub struct FlowProvider {
http: reqwest::Client,
config: FlowConfig,
state: Mutex<Option<Session>>,
}
impl std::fmt::Debug for FlowProvider {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("FlowProvider")
.field("steps", &self.config.steps.len())
.field("apply", &self.config.apply.len())
.finish()
}
}
impl FlowProvider {
pub fn from_config(config: &Value) -> Result<Self, FaucetError> {
let config: FlowConfig = serde_json::from_value(config.clone())
.map_err(|e| FaucetError::Config(format!("flow auth: invalid config: {e}")))?;
config.validate()?;
Ok(Self {
http: auth_http_client(),
config,
state: Mutex::new(None),
})
}
fn clock_ctx() -> HashMap<String, String> {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.subsec_nanos())
.unwrap_or(0);
let mut ctx = HashMap::new();
ctx.insert("ts".to_owned(), now.to_string());
ctx.insert("nonce".to_owned(), format!("{now}{nanos}"));
ctx
}
async fn run_login(&self) -> Result<HashMap<String, String>, FaucetError> {
let mut ctx: HashMap<String, String> = HashMap::new();
for (i, step) in self.config.steps.iter().enumerate() {
let req = &step.request;
let method = reqwest::Method::from_bytes(req.method.to_uppercase().as_bytes())
.map_err(|_| {
FaucetError::Config(format!(
"flow auth: step {i} has an invalid HTTP method '{}'",
req.method
))
})?;
let url = render(&req.url, &ctx);
let mut builder = self.http.request(method, &url);
for (k, v) in &req.headers {
builder = builder.header(k.as_str(), render(v, &ctx));
}
if !req.query.is_empty() {
let q: Vec<(String, String)> = req
.query
.iter()
.map(|(k, v)| (k.clone(), render(v, &ctx)))
.collect();
builder = builder.query(&q);
}
if let Some(form) = &req.form {
let f: Vec<(String, String)> = form
.iter()
.map(|(k, v)| (k.clone(), render(v, &ctx)))
.collect();
builder = builder.form(&f);
} else if let Some(json) = &req.json {
let rendered = render(&serde_json::to_string(json).unwrap_or_default(), &ctx);
let body: Value = serde_json::from_str(&rendered).unwrap_or_else(|_| json.clone());
builder = builder.json(&body);
}
if let Some(sign) = &req.sign {
let mut sign_ctx = ctx.clone();
sign_ctx.extend(Self::clock_ctx());
let (name, value) = sign_header(sign, &sign_ctx);
builder = builder.header(name.as_str(), value);
}
let resp = builder.send().await.map_err(|e| {
FaucetError::Auth(format!("flow auth: step {i} request failed: {e}"))
})?;
let status = resp.status();
if !status.is_success() {
return Err(FaucetError::Auth(format!(
"flow auth: step {i} returned HTTP {}",
status.as_u16()
)));
}
let headers = resp.headers().clone();
let needs_json = step.capture.values().any(|c| c.kind() == CaptureFrom::Json);
let needs_xml = step.capture.values().any(|c| c.kind() == CaptureFrom::Xml);
let text: Option<String> = if needs_json || needs_xml {
Some(resp.text().await.map_err(|e| {
FaucetError::Auth(format!(
"flow auth: step {i} failed to read response body: {e}"
))
})?)
} else {
None
};
let json_body: Option<Value> = if needs_json {
Some(
serde_json::from_str(text.as_deref().unwrap_or("")).map_err(|e| {
FaucetError::Auth(format!(
"flow auth: step {i} response was not valid JSON: {e}"
))
})?,
)
} else {
None
};
for (name, cap) in &step.capture {
let extracted = match cap {
CaptureSpec::Json(path) => json_body
.as_ref()
.and_then(|b| jsonpath_first(b, path))
.map(|v| value_to_string(&v)),
CaptureSpec::Source(s) => match s.from {
CaptureFrom::Json => json_body
.as_ref()
.and_then(|b| jsonpath_first(b, s.path.as_deref().unwrap_or("")))
.map(|v| value_to_string(&v)),
CaptureFrom::Xml => text
.as_deref()
.and_then(|t| xml_dot_path(t, s.path.as_deref().unwrap_or(""))),
CaptureFrom::Header => {
header_value(&headers, s.name.as_deref().unwrap_or(""))
}
CaptureFrom::SetCookie => {
set_cookie_value(&headers, s.name.as_deref().unwrap_or(""))
}
},
};
match extracted {
Some(v) => {
ctx.insert(name.clone(), v);
}
None => {
return Err(FaucetError::Auth(format!(
"flow auth: step {i} capture '{name}' matched nothing"
)));
}
}
}
}
Ok(ctx)
}
async fn ensure_ctx(&self) -> Result<HashMap<String, String>, FaucetError> {
let mut guard = self.state.lock().await;
if let Some(s) = guard.as_ref()
&& s.valid()
{
return Ok(s.ctx.clone());
}
let ctx = self.run_login().await?;
let expires_at = self
.config
.ttl_secs
.map(|s| Instant::now() + Duration::from_secs(s));
*guard = Some(Session {
ctx: ctx.clone(),
expires_at,
});
Ok(ctx)
}
fn request_auth_from_ctx(&self, captured: &HashMap<String, String>) -> RequestAuth {
let mut ctx = captured.clone();
ctx.extend(Self::clock_ctx());
build_request_auth(&self.config.apply, &self.config.base_url_from, &ctx)
}
}
#[async_trait]
impl AuthProvider for FlowProvider {
async fn credential(&self) -> Result<Credential, FaucetError> {
let ctx = self.ensure_ctx().await?;
let auth = self.request_auth_from_ctx(&ctx);
for p in &auth.placements {
if let CredentialPlacement::Header { name, value } = p {
return Ok(Credential::Header {
name: name.clone(),
value: value.clone(),
});
}
}
Err(FaucetError::Auth(
"flow auth: no header credential to apply via credential(); this flow places its \
credential in a query/cookie/body, which requires a connector that consumes \
request_auth() (e.g. the REST source)"
.to_owned(),
))
}
async fn invalidate(&self, _stale: &Credential) -> Result<Credential, FaucetError> {
*self.state.lock().await = None;
self.credential().await
}
async fn request_auth(
&self,
_method: &str,
_url: &str,
_query: &std::collections::BTreeMap<String, String>,
) -> Result<RequestAuth, FaucetError> {
let ctx = self.ensure_ctx().await?;
Ok(self.request_auth_from_ctx(&ctx))
}
fn reauth_statuses(&self) -> &[u16] {
&self.config.reauth_on
}
fn provider_name(&self) -> &'static str {
"flow"
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn ctx(pairs: &[(&str, &str)]) -> HashMap<String, String> {
pairs
.iter()
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect()
}
#[test]
fn render_substitutes_known_and_leaves_unknown() {
let c = ctx(&[("token", "abc"), ("region", "eu")]);
assert_eq!(render("Bearer ${token}", &c), "Bearer abc");
assert_eq!(render("https://${region}.api", &c), "https://eu.api");
assert_eq!(render("${missing}", &c), "${missing}");
assert_eq!(render("no tokens", &c), "no tokens");
assert_eq!(render("${token}${region}", &c), "abceu");
}
#[test]
fn render_handles_unterminated_and_utf8() {
let c = ctx(&[("x", "✓")]);
assert_eq!(render("prefix ${x} ünïcode", &c), "prefix ✓ ünïcode");
assert_eq!(render("dangling ${x", &c), "dangling ${x");
}
#[test]
fn value_to_string_covers_kinds() {
assert_eq!(value_to_string(&json!("s")), "s");
assert_eq!(value_to_string(&json!(7)), "7");
assert_eq!(value_to_string(&json!(true)), "true");
assert_eq!(value_to_string(&json!(null)), "");
assert_eq!(value_to_string(&json!([1, 2])), "[1,2]");
}
#[test]
fn jsonpath_first_extracts_scalar() {
let body = json!({"data": {"BhRestToken": "tok", "restUrl": "https://host/x"}});
assert_eq!(
jsonpath_first(&body, "$.data.BhRestToken"),
Some(json!("tok"))
);
assert_eq!(
jsonpath_first(&body, "$.data.restUrl"),
Some(json!("https://host/x"))
);
assert_eq!(jsonpath_first(&body, "$.data.missing"), None);
}
#[test]
fn hmac_hex_and_base64_are_deterministic() {
let hex = hmac_sign("key", "message", SigEncoding::Hex);
assert_eq!(
hex,
"6e9ef29b75fffc5b7abae527d58fdadb2fe42e7219011976917343065f58ed4a"
);
let b64 = hmac_sign("key", "message", SigEncoding::Base64);
assert_eq!(b64, "bp7ym3X//Ft6uuUn1Y/a2y/kLnIZARl2kXNDBl9Y7Uo=");
}
#[test]
fn build_request_auth_places_header_query_cookie_body() {
let apply: Vec<ApplySpec> = serde_json::from_value(json!([
{ "into": "header", "name": "X-Tok", "value": "${tok}" },
{ "into": "query", "name": "access_token", "value": "${tok}" },
{ "into": "cookie", "name": "sid", "value": "${sid}" },
{ "into": "body", "name": "auth", "value": "${tok}" }
]))
.unwrap();
let c = ctx(&[("tok", "T"), ("sid", "S")]);
let ra = build_request_auth(&apply, &None, &c);
assert_eq!(ra.placements.len(), 4);
assert!(
matches!(&ra.placements[0], CredentialPlacement::Header { name, value } if name=="X-Tok" && value=="T")
);
assert!(
matches!(&ra.placements[1], CredentialPlacement::Query { name, value } if name=="access_token" && value=="T")
);
assert!(
matches!(&ra.placements[2], CredentialPlacement::Cookie { name, value } if name=="sid" && value=="S")
);
assert!(
matches!(&ra.placements[3], CredentialPlacement::BodyField { name, value } if name=="auth" && value=="T")
);
assert!(ra.base_url.is_none());
assert_eq!(ra.captured.get("tok").map(String::as_str), Some("T"));
assert_eq!(ra.captured.get("sid").map(String::as_str), Some("S"));
}
#[test]
fn build_request_auth_signs_and_sets_base_url() {
let apply: Vec<ApplySpec> = serde_json::from_value(json!([
{ "sign": { "alg": "hmac_sha256", "key": "secret", "template": "${client}:${ts}",
"encoding": "hex", "into": { "header": "Authorization", "format": "SS ${sig}" } } }
]))
.unwrap();
let mut c = ctx(&[
("client", "abc"),
("ts", "100"),
("base_url", "https://eu.host"),
]);
c.insert("base_url".to_owned(), "https://eu.host".to_owned());
let ra = build_request_auth(&apply, &Some("${base_url}".to_owned()), &c);
assert_eq!(ra.placements.len(), 1);
let expected = format!("SS {}", hmac_sign("secret", "abc:100", SigEncoding::Hex));
assert!(
matches!(&ra.placements[0], CredentialPlacement::Header { name, value } if name=="Authorization" && *value==expected)
);
assert_eq!(ra.base_url.as_deref(), Some("https://eu.host"));
}
#[test]
fn base_url_from_empty_render_is_ignored() {
let ra = build_request_auth(&[], &Some("${missing_and_stripped}".to_owned()), &ctx(&[]));
assert!(ra.base_url.is_some());
let ra2 = build_request_auth(&[], &Some("${e}".to_owned()), &ctx(&[("e", "")]));
assert!(ra2.base_url.is_none());
}
#[test]
fn config_validate_requires_steps_or_apply() {
let empty: FlowConfig = serde_json::from_value(json!({})).unwrap();
assert!(empty.validate().is_err());
}
#[test]
fn config_rejects_form_and_json_together() {
let cfg: FlowConfig = serde_json::from_value(json!({
"steps": [ { "request": { "url": "https://x", "form": {"a":"b"}, "json": {"c":"d"} } } ]
}))
.unwrap();
assert!(cfg.validate().is_err());
}
#[test]
fn config_rejects_empty_place_name_and_sign_header() {
let bad_place: FlowConfig = serde_json::from_value(json!({
"apply": [ { "into": "query", "name": "", "value": "${x}" } ]
}))
.unwrap();
assert!(bad_place.validate().is_err());
let bad_sign: FlowConfig = serde_json::from_value(json!({
"apply": [ { "sign": { "alg": "hmac_sha256", "key": "k", "template": "${ts}",
"into": { "header": "" } } } ]
}))
.unwrap();
assert!(bad_sign.validate().is_err());
}
#[test]
fn from_config_rejects_unknown_field() {
assert!(FlowProvider::from_config(&json!({ "bogus": 1 })).is_err());
}
#[test]
fn build_provider_dispatches_flow() {
let p = crate::build_provider(&json!({
"type": "flow",
"config": { "apply": [ { "into": "header", "name": "X", "value": "v" } ] }
}))
.unwrap();
assert_eq!(p.provider_name(), "flow");
}
#[test]
fn debug_redacts_and_summarizes() {
let p = FlowProvider::from_config(&json!({
"apply": [ { "into": "header", "name": "X", "value": "${t}" } ]
}))
.unwrap();
let s = format!("{p:?}");
assert!(s.contains("FlowProvider"));
assert!(s.contains("apply"));
}
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use wiremock::matchers::{body_json, header, method, path};
use wiremock::{Mock, MockServer, Respond, ResponseTemplate};
#[tokio::test]
async fn login_chain_captures_placements_and_base_url() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/token"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"access_token": "AT"})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/login"))
.respond_with(
ResponseTemplate::new(200).set_body_json(
json!({"BhRestToken": "BRT", "restUrl": "https://data.example"}),
),
)
.mount(&server)
.await;
let cfg = json!({
"steps": [
{ "request": { "method": "POST", "url": format!("{}/token", server.uri()),
"form": {"grant_type": "refresh_token"} },
"capture": { "access_token": "$.access_token" } },
{ "request": { "method": "GET", "url": format!("{}/login", server.uri()),
"query": {"access_token": "${access_token}"} },
"capture": { "bh_rest_token": "$.BhRestToken", "base_url": "$.restUrl" } }
],
"apply": [ { "into": "query", "name": "BhRestToken", "value": "${bh_rest_token}" } ],
"base_url_from": "${base_url}",
"reauth_on": [401]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let ra = p
.request_auth("GET", "https://data.example/x", &BTreeMap::new())
.await
.unwrap();
assert_eq!(ra.base_url.as_deref(), Some("https://data.example"));
assert!(
matches!(&ra.placements[0], CredentialPlacement::Query { name, value } if name == "BhRestToken" && value == "BRT")
);
assert_eq!(p.reauth_statuses(), &[401]);
assert!(p.credential().await.is_err());
}
struct CountingLogin(Arc<AtomicUsize>);
impl Respond for CountingLogin {
fn respond(&self, _: &wiremock::Request) -> ResponseTemplate {
let n = self.0.fetch_add(1, Ordering::SeqCst) + 1;
ResponseTemplate::new(200).set_body_json(json!({ "sid": format!("S{n}") }))
}
}
#[tokio::test]
async fn header_credential_caches_then_reloads_on_invalidate() {
let server = MockServer::start().await;
let hits = Arc::new(AtomicUsize::new(0));
Mock::given(method("POST"))
.and(path("/login"))
.respond_with(CountingLogin(hits.clone()))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "method": "POST", "url": format!("{}/login", server.uri()) },
"capture": { "sid": "$.sid" } } ],
"apply": [ { "into": "header", "name": "X-Session", "value": "${sid}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let c1 = p.credential().await.unwrap();
assert!(
matches!(&c1, Credential::Header { name, value } if name == "X-Session" && value == "S1")
);
let _ = p.credential().await.unwrap();
assert_eq!(hits.load(Ordering::SeqCst), 1);
let c2 = p.invalidate(&c1).await.unwrap();
assert!(matches!(&c2, Credential::Header { value, .. } if value == "S2"));
assert_eq!(hits.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn login_capture_miss_is_an_error() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/x"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"other": 1})))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "url": format!("{}/x", server.uri()) },
"capture": { "tok": "$.access_token" } } ],
"apply": [ { "into": "header", "name": "X", "value": "${tok}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
assert!(p.credential().await.is_err());
}
#[tokio::test]
async fn login_non_success_is_an_error() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/x"))
.respond_with(ResponseTemplate::new(500))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "url": format!("{}/x", server.uri()) } } ],
"apply": [ { "into": "header", "name": "X", "value": "static" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
assert!(
p.request_auth("GET", "https://x", &BTreeMap::new())
.await
.is_err()
);
}
#[test]
fn config_rejects_empty_step_url() {
let cfg: FlowConfig = serde_json::from_value(json!({
"steps": [ { "request": { "url": " " } } ]
}))
.unwrap();
assert!(cfg.validate().is_err());
}
#[tokio::test]
async fn session_ttl_caches_within_window() {
let server = MockServer::start().await;
let hits = Arc::new(AtomicUsize::new(0));
Mock::given(method("POST"))
.and(path("/login"))
.respond_with(CountingLogin(hits.clone()))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "method": "POST", "url": format!("{}/login", server.uri()) },
"capture": { "sid": "$.sid" } } ],
"apply": [ { "into": "header", "name": "X", "value": "${sid}" } ],
"ttl_secs": 3600
});
let p = FlowProvider::from_config(&cfg).unwrap();
let _ = p.credential().await.unwrap();
let _ = p.credential().await.unwrap();
assert_eq!(hits.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn login_sends_headers_and_json_body() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/login"))
.and(header("x-tenant", "acme"))
.and(body_json(json!({"scope": "read"})))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"tok": "T"})))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "method": "POST", "url": format!("{}/login", server.uri()),
"headers": {"X-Tenant": "acme"}, "json": {"scope": "read"} },
"capture": { "t": "$.tok" } } ],
"apply": [ { "into": "header", "name": "Authorization", "value": "Bearer ${t}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
assert_eq!(p.provider_name(), "flow");
let c = p.credential().await.unwrap();
assert!(
matches!(&c, Credential::Header { name, value } if name == "Authorization" && value == "Bearer T")
);
}
#[tokio::test]
async fn login_invalid_method_errors() {
let cfg = json!({
"steps": [ { "request": { "method": "BAD METHOD", "url": "https://x/login" } } ],
"apply": [ { "into": "header", "name": "X", "value": "static" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
assert!(
p.request_auth("GET", "https://x", &BTreeMap::new())
.await
.is_err()
);
}
#[tokio::test]
async fn login_send_failure_errors() {
let cfg = json!({
"steps": [ { "request": { "url": "http://127.0.0.1:1/x" }, "capture": { "t": "$.t" } } ],
"apply": [ { "into": "header", "name": "X", "value": "${t}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
assert!(
p.request_auth("GET", "https://x", &BTreeMap::new())
.await
.is_err()
);
}
#[tokio::test]
async fn login_non_json_response_errors() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/x"))
.respond_with(ResponseTemplate::new(200).set_body_string("not json"))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "url": format!("{}/x", server.uri()) }, "capture": { "t": "$.t" } } ],
"apply": [ { "into": "header", "name": "X", "value": "${t}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
assert!(
p.request_auth("GET", "https://x", &BTreeMap::new())
.await
.is_err()
);
}
#[tokio::test]
async fn login_step_with_empty_capture_succeeds() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/ping"))
.respond_with(ResponseTemplate::new(200).set_body_string("ok"))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/token"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"t": "T"})))
.mount(&server)
.await;
let cfg = json!({
"steps": [
{ "request": { "url": format!("{}/ping", server.uri()) } },
{ "request": { "url": format!("{}/token", server.uri()) }, "capture": { "t": "$.t" } }
],
"apply": [ { "into": "header", "name": "X", "value": "${t}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let c = p.credential().await.unwrap();
assert!(matches!(&c, Credential::Header { value, .. } if value == "T"));
}
#[tokio::test]
async fn no_steps_apply_only_flow_works() {
let cfg = json!({
"apply": [ { "sign": { "alg": "hmac_sha256", "key": "k", "template": "msg",
"into": { "header": "Authorization", "format": "HMAC ${sig}" } } } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let c = p.credential().await.unwrap();
let expected = format!("HMAC {}", hmac_sign("k", "msg", SigEncoding::Hex));
assert!(
matches!(&c, Credential::Header { name, value } if name == "Authorization" && *value == expected)
);
}
#[test]
fn sign_header_computes_name_and_value_with_clock() {
let mut c = ctx(&[("client", "abc")]);
c.insert("ts".to_owned(), "999".to_owned());
let sign: SignSpec = serde_json::from_value(json!({
"alg": "hmac_sha256", "key": "k", "template": "${client}:${ts}",
"encoding": "hex", "into": { "header": "Authorization", "format": "SS ${sig}" }
}))
.unwrap();
let (name, value) = sign_header(&sign, &c);
assert_eq!(name, "Authorization");
assert_eq!(
value,
format!("SS {}", hmac_sign("k", "abc:999", SigEncoding::Hex))
);
}
#[tokio::test]
async fn signed_login_step_sends_computed_signature_header() {
let server = MockServer::start().await;
let key = "secret_key";
let expected = format!(
"SS access:{}",
hmac_sign(key, "client:secret", SigEncoding::Base64)
);
Mock::given(method("POST"))
.and(path("/auth/login"))
.and(header("authorization", expected.as_str()))
.respond_with(
ResponseTemplate::new(200).set_body_json(json!({"session": {"token": "SESS"}})),
)
.mount(&server)
.await;
let cfg = json!({
"steps": [ {
"request": {
"method": "POST",
"url": format!("{}/auth/login", server.uri()),
"json": {"clientId": "client"},
"sign": { "alg": "hmac_sha256", "key": key, "template": "client:secret",
"encoding": "base64",
"into": { "header": "Authorization", "format": "SS access:${sig}" } }
},
"capture": { "session": "$.session.token" }
} ],
"apply": [ { "into": "header", "name": "Session", "value": "${session}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let c = p.credential().await.unwrap();
assert!(
matches!(&c, Credential::Header { name, value } if name == "Session" && value == "SESS")
);
}
#[tokio::test]
async fn skyslope_style_signed_login_then_signed_data_requests() {
let server = MockServer::start().await;
let key = "base64secret";
let login_sig = format!(
"SS access:{}",
hmac_sign(key, "cid:csec", SigEncoding::Base64)
);
Mock::given(method("POST"))
.and(path("/auth/login"))
.and(header("authorization", login_sig.as_str()))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(json!({"session": {"token": "SESSION-TOK"}})),
)
.mount(&server)
.await;
let signer = json!({ "alg": "hmac_sha256", "key": key, "template": "cid:csec",
"encoding": "base64",
"into": { "header": "Authorization", "format": "SS access:${sig}" } });
let cfg = json!({
"steps": [ {
"request": { "method": "POST", "url": format!("{}/auth/login", server.uri()),
"json": {"clientId": "cid"}, "sign": signer },
"capture": { "session": "$.session.token" }
} ],
"apply": [
{ "sign": signer },
{ "into": "header", "name": "Session", "value": "${session}" }
]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let ra = p
.request_auth("GET", "https://data", &BTreeMap::new())
.await
.unwrap();
let has_auth = ra.placements.iter().any(|pl| {
matches!(pl, CredentialPlacement::Header { name, value } if name == "Authorization" && *value == login_sig)
});
let has_sess = ra.placements.iter().any(|pl| {
matches!(pl, CredentialPlacement::Header { name, value } if name == "Session" && value == "SESSION-TOK")
});
assert!(has_auth, "data request carries the signer header");
assert!(has_sess, "data request carries the captured Session header");
}
#[test]
fn config_rejects_empty_step_sign_header() {
let cfg: FlowConfig = serde_json::from_value(json!({
"steps": [ { "request": { "url": "https://x",
"sign": { "alg": "hmac_sha256", "key": "k", "template": "${ts}",
"into": { "header": "" } } } } ]
}))
.unwrap();
assert!(cfg.validate().is_err());
}
#[test]
fn xml_dot_path_extracts_nested_text() {
let xml = "<operation><result><data><api><sessionid>ABC123</sessionid></api></data></result></operation>";
assert_eq!(
xml_dot_path(xml, "operation.result.data.api.sessionid").as_deref(),
Some("ABC123")
);
assert_eq!(xml_dot_path(xml, "operation.result.missing"), None);
}
#[test]
fn xml_dot_path_ignores_declaration_attrs_and_namespaces() {
let xml = r#"<?xml version="1.0"?><ns:root xmlns:ns="urn:x"><ns:child id="1"> hi </ns:child></ns:root>"#;
assert_eq!(xml_dot_path(xml, "root.child").as_deref(), Some("hi"));
assert_eq!(xml_dot_path(xml, "ns:root.ns:child").as_deref(), Some("hi"));
}
#[test]
fn xml_dot_path_handles_cdata_and_entities() {
let xml = "<r><a><![CDATA[a&b]]></a><b>x & y</b></r>";
assert_eq!(xml_dot_path(xml, "r.a").as_deref(), Some("a&b"));
assert_eq!(xml_dot_path(xml, "r.b").as_deref(), Some("x & y"));
}
#[test]
fn set_cookie_and_header_helpers_select_by_name() {
let mut h = reqwest::header::HeaderMap::new();
h.append("set-cookie", "a=1; path=/".parse().unwrap());
h.append(
"set-cookie",
"ASP.NET_SessionId=SID; path=/; HttpOnly".parse().unwrap(),
);
h.insert("location", "https://x/next".parse().unwrap());
assert_eq!(
set_cookie_value(&h, "ASP.NET_SessionId").as_deref(),
Some("SID")
);
assert_eq!(set_cookie_value(&h, "missing"), None);
assert_eq!(
header_value(&h, "Location").as_deref(),
Some("https://x/next")
);
assert_eq!(header_value(&h, "absent"), None);
}
#[test]
fn capture_backcompat_string_form_parses_as_json() {
let step: FlowStep = serde_json::from_value(json!({
"request": { "url": "https://x" },
"capture": { "tok": "$.access_token" }
}))
.unwrap();
assert_eq!(step.capture["tok"].kind(), CaptureFrom::Json);
assert!(matches!(&step.capture["tok"], CaptureSpec::Json(p) if p == "$.access_token"));
}
#[test]
fn capture_struct_forms_parse_kinds() {
let step: FlowStep = serde_json::from_value(json!({
"request": { "url": "https://x" },
"capture": {
"h": { "from": "header", "name": "Location" },
"c": { "from": "set_cookie", "name": "sid" },
"x": { "from": "xml", "path": "a.b" },
"j": { "from": "json", "path": "$.tok" }
}
}))
.unwrap();
assert_eq!(step.capture["h"].kind(), CaptureFrom::Header);
assert_eq!(step.capture["c"].kind(), CaptureFrom::SetCookie);
assert_eq!(step.capture["x"].kind(), CaptureFrom::Xml);
assert_eq!(step.capture["j"].kind(), CaptureFrom::Json);
}
#[test]
fn config_rejects_capture_missing_selector() {
let cfg: FlowConfig = serde_json::from_value(json!({
"steps": [ { "request": { "url": "https://x" }, "capture": { "h": { "from": "header" } } } ]
}))
.unwrap();
assert!(cfg.validate().is_err());
let cfg2: FlowConfig = serde_json::from_value(json!({
"steps": [ { "request": { "url": "https://x" }, "capture": { "x": { "from": "xml" } } } ]
}))
.unwrap();
assert!(cfg2.validate().is_err());
}
#[tokio::test]
async fn capture_from_header_works() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/login"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("Location", "https://redirect.example/next"),
)
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "url": format!("{}/login", server.uri()) },
"capture": { "loc": { "from": "header", "name": "Location" } } } ],
"apply": [ { "into": "header", "name": "X-Loc", "value": "${loc}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let c = p.credential().await.unwrap();
assert!(
matches!(&c, Credential::Header { name, value } if name == "X-Loc" && value == "https://redirect.example/next")
);
}
#[tokio::test]
async fn capture_from_set_cookie_applies_as_cookie() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/entity/auth/login"))
.respond_with(
ResponseTemplate::new(204)
.append_header("Set-Cookie", "other=nope; path=/")
.append_header("Set-Cookie", "ASP.NET_SessionId=SID123; path=/; HttpOnly"),
)
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "method": "POST",
"url": format!("{}/entity/auth/login", server.uri()),
"json": {"name": "u", "password": "p"} },
"capture": { "session_cookie": { "from": "set_cookie", "name": "ASP.NET_SessionId" } } } ],
"apply": [ { "into": "cookie", "name": "ASP.NET_SessionId", "value": "${session_cookie}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let ra = p
.request_auth("GET", "https://data", &BTreeMap::new())
.await
.unwrap();
assert!(
matches!(&ra.placements[0], CredentialPlacement::Cookie { name, value } if name == "ASP.NET_SessionId" && value == "SID123")
);
}
#[tokio::test]
async fn capture_from_xml_body_works() {
let server = MockServer::start().await;
let xml = r#"<?xml version="1.0" encoding="UTF-8"?><response><operation><result><data><api><sessionid>XYZ-SESSION</sessionid></api></data></result></operation></response>"#;
Mock::given(method("POST"))
.and(path("/xml/xmlgw.phtml"))
.respond_with(ResponseTemplate::new(200).set_body_string(xml))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "method": "POST",
"url": format!("{}/xml/xmlgw.phtml", server.uri()) },
"capture": { "sess_id": { "from": "xml",
"path": "response.operation.result.data.api.sessionid" } } } ],
"apply": [ { "into": "header", "name": "X-Session", "value": "${sess_id}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
let c = p.credential().await.unwrap();
assert!(matches!(&c, Credential::Header { value, .. } if value == "XYZ-SESSION"));
}
#[tokio::test]
async fn capture_from_xml_miss_is_an_error() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/x"))
.respond_with(ResponseTemplate::new(200).set_body_string("<r><a>1</a></r>"))
.mount(&server)
.await;
let cfg = json!({
"steps": [ { "request": { "method": "POST", "url": format!("{}/x", server.uri()) },
"capture": { "s": { "from": "xml", "path": "r.missing" } } } ],
"apply": [ { "into": "header", "name": "X", "value": "${s}" } ]
});
let p = FlowProvider::from_config(&cfg).unwrap();
assert!(p.credential().await.is_err());
}
}