feat(net): make passive request/response callbacks detachable (#416)
on_request/on_response stored callbacks in append-only Vecs with no way to remove one, so a crawler that installed a response-body capture hook for one phase could never detach it: it fired for the shared client's whole lifetime, re-registering across navigations piled up duplicates, and (because the client is shared per context) it saw sibling pages' requests too. Tag each callback with a stable id. on_request/on_response now return the id and off_request/off_response (remove_on_request/remove_on_response on the client) detach a specific one. Storage is Vec<(u64, Callback)>; all fire sites destructure the tuple. Closes #408.
This commit is contained in:
1 parent
d0b6fc4ffd
commit
99ece0e12d
5 files changed
+147
-23
No files matched your search
@@ -1652,19 +1652,32 @@ impl Page {
|
||||
/// Register a passive callback fired for every JS `fetch()`/XHR (and
|
||||
/// navigation) request, once the method/headers/body are known and before it
|
||||
/// is sent. Non-blocking; use `enable_interception` to mutate or block.
|
||||
pub fn on_request(&mut self, cb: RequestCallback) {
|
||||
if let Ok(mut v) = self.http_client.on_request.try_write() {
|
||||
v.push(cb);
|
||||
}
|
||||
/// Register a passive request observer, returning a stable id. Pass the id
|
||||
/// to `off_request` to detach it (issue #408). Note the observer lives on
|
||||
/// the context's shared http client, so it also sees requests from sibling
|
||||
/// pages in the same context; detach it when the capturing phase ends.
|
||||
pub fn on_request(&mut self, cb: RequestCallback) -> u64 {
|
||||
self.http_client.register_on_request(cb)
|
||||
}
|
||||
|
||||
/// Register a passive callback fired with every JS `fetch()`/XHR (and
|
||||
/// navigation) response, including its body. Non-blocking. The main path for
|
||||
/// crawlers that need to capture API response payloads.
|
||||
pub fn on_response(&mut self, cb: ResponseCallback) {
|
||||
if let Ok(mut v) = self.http_client.on_response.try_write() {
|
||||
v.push(cb);
|
||||
}
|
||||
/// navigation) response, including its body. Non-blocking. Returns a stable
|
||||
/// id for `off_response`. The main path for crawlers that need to capture
|
||||
/// API response payloads. Shared across the context like `on_request`.
|
||||
pub fn on_response(&mut self, cb: ResponseCallback) -> u64 {
|
||||
self.http_client.register_on_response(cb)
|
||||
}
|
||||
|
||||
/// Detach a request observer registered with `on_request`. Returns true if
|
||||
/// one was removed.
|
||||
pub fn off_request(&mut self, id: u64) -> bool {
|
||||
self.http_client.remove_on_request(id)
|
||||
}
|
||||
|
||||
/// Detach a response observer registered with `on_response`. Returns true if
|
||||
/// one was removed.
|
||||
pub fn off_response(&mut self, id: u64) -> bool {
|
||||
self.http_client.remove_on_response(id)
|
||||
}
|
||||
|
||||
pub async fn process_pending_navigation(&mut self) -> Result<bool, PageError> {
|
||||
|
||||
@@ -792,7 +792,7 @@ async fn op_fetch_url(
|
||||
headers: custom_headers.clone(),
|
||||
resource_type: ResourceType::Fetch,
|
||||
};
|
||||
for cb in cbs.iter() {
|
||||
for (_, cb) in cbs.iter() {
|
||||
cb(&info);
|
||||
}
|
||||
}
|
||||
@@ -1038,7 +1038,7 @@ async fn op_fetch_url(
|
||||
headers: resp_headers.clone(),
|
||||
resource_type: ResourceType::Fetch,
|
||||
};
|
||||
for cb in cbs.iter() {
|
||||
for (_, cb) in cbs.iter() {
|
||||
cb(&info, &resp);
|
||||
}
|
||||
}
|
||||
@@ -1231,7 +1231,7 @@ async fn stealth_fetch_all(
|
||||
headers: resp_headers.clone(),
|
||||
resource_type: ResourceType::Fetch,
|
||||
};
|
||||
for cb in cbs.iter() {
|
||||
for (_, cb) in cbs.iter() {
|
||||
cb(&info, &resp);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -262,8 +262,13 @@ pub struct ObscuraHttpClient {
|
||||
pub user_agent: RwLock<String>,
|
||||
pub extra_headers: RwLock<HashMap<String, String>>,
|
||||
pub interceptor: RwLock<Option<Box<dyn RequestInterceptor + Send + Sync>>>,
|
||||
pub on_request: RwLock<Vec<RequestCallback>>,
|
||||
pub on_response: RwLock<Vec<ResponseCallback>>,
|
||||
/// Passive request/response observers, each tagged with a stable id so a
|
||||
/// specific one can be detached via `remove_on_request`/`remove_on_response`
|
||||
/// (issue #408). Without ids the vecs were append-only: no detach, and
|
||||
/// re-registering across navigations piled up duplicates.
|
||||
pub on_request: RwLock<Vec<(u64, RequestCallback)>>,
|
||||
pub on_response: RwLock<Vec<(u64, ResponseCallback)>>,
|
||||
callback_id_counter: std::sync::atomic::AtomicU64,
|
||||
pub timeout: Duration,
|
||||
pub in_flight: Arc<std::sync::atomic::AtomicU32>,
|
||||
pub block_trackers: bool,
|
||||
@@ -342,6 +347,7 @@ impl ObscuraHttpClient {
|
||||
interceptor: RwLock::new(None),
|
||||
on_request: RwLock::new(Vec::new()),
|
||||
on_response: RwLock::new(Vec::new()),
|
||||
callback_id_counter: std::sync::atomic::AtomicU64::new(1),
|
||||
in_flight: Arc::new(std::sync::atomic::AtomicU32::new(0)),
|
||||
timeout: Duration::from_secs(30),
|
||||
block_trackers: false,
|
||||
@@ -381,6 +387,50 @@ impl ObscuraHttpClient {
|
||||
self.fetch_with_method(Method::GET, url, None).await
|
||||
}
|
||||
|
||||
/// Register a passive request observer, returning a stable id. Pass the id
|
||||
/// to `remove_on_request` to detach it (issue #408).
|
||||
pub fn register_on_request(&self, cb: RequestCallback) -> u64 {
|
||||
let id = self
|
||||
.callback_id_counter
|
||||
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
if let Ok(mut v) = self.on_request.try_write() {
|
||||
v.push((id, cb));
|
||||
}
|
||||
id
|
||||
}
|
||||
|
||||
/// Register a passive response observer, returning a stable id. Pass the id
|
||||
/// to `remove_on_response` to detach it (issue #408).
|
||||
pub fn register_on_response(&self, cb: ResponseCallback) -> u64 {
|
||||
let id = self
|
||||
.callback_id_counter
|
||||
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
if let Ok(mut v) = self.on_response.try_write() {
|
||||
v.push((id, cb));
|
||||
}
|
||||
id
|
||||
}
|
||||
|
||||
/// Detach a request observer by its id. Returns true if one was removed.
|
||||
pub fn remove_on_request(&self, id: u64) -> bool {
|
||||
if let Ok(mut v) = self.on_request.try_write() {
|
||||
let before = v.len();
|
||||
v.retain(|(cid, _)| *cid != id);
|
||||
return v.len() != before;
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
/// Detach a response observer by its id. Returns true if one was removed.
|
||||
pub fn remove_on_response(&self, id: u64) -> bool {
|
||||
if let Ok(mut v) = self.on_response.try_write() {
|
||||
let before = v.len();
|
||||
v.retain(|(cid, _)| *cid != id);
|
||||
return v.len() != before;
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
pub async fn post_form(&self, url: &Url, body: &str) -> Result<Response, ObscuraNetError> {
|
||||
self.fetch_with_method(Method::POST, url, Some(body.as_bytes().to_vec())).await
|
||||
}
|
||||
@@ -442,7 +492,7 @@ impl ObscuraHttpClient {
|
||||
}
|
||||
}
|
||||
|
||||
for cb in self.on_request.read().await.iter() {
|
||||
for (_, cb) in self.on_request.read().await.iter() {
|
||||
cb(&request_info);
|
||||
}
|
||||
|
||||
@@ -588,7 +638,7 @@ impl ObscuraHttpClient {
|
||||
redirected_from: redirects,
|
||||
};
|
||||
|
||||
for cb in self.on_response.read().await.iter() {
|
||||
for (_, cb) in self.on_response.read().await.iter() {
|
||||
cb(&request_info, &response);
|
||||
}
|
||||
|
||||
|
||||
@@ -114,16 +114,31 @@ impl Page {
|
||||
/// Register a passive callback fired for every request the page makes
|
||||
/// (navigation and JS `fetch()`/XHR), once its method/headers/body are known
|
||||
/// and before it is sent. Non-blocking; use `enable_interception` to mutate
|
||||
/// or block.
|
||||
pub fn on_request(&mut self, cb: RequestCallback) {
|
||||
self.inner.on_request(cb);
|
||||
/// or block. Returns a stable id; pass it to `off_request` to detach.
|
||||
pub fn on_request(&mut self, cb: RequestCallback) -> u64 {
|
||||
self.inner.on_request(cb)
|
||||
}
|
||||
|
||||
/// Register a passive callback fired with every response the page receives
|
||||
/// (navigation and JS `fetch()`/XHR), including its body. Non-blocking. The
|
||||
/// main path for capturing API response payloads from SPAs.
|
||||
pub fn on_response(&mut self, cb: ResponseCallback) {
|
||||
self.inner.on_response(cb);
|
||||
/// main path for capturing API response payloads from SPAs. Returns a stable
|
||||
/// id; pass it to `off_response` to detach.
|
||||
pub fn on_response(&mut self, cb: ResponseCallback) -> u64 {
|
||||
self.inner.on_response(cb)
|
||||
}
|
||||
|
||||
/// Detach a request callback previously registered with `on_request`.
|
||||
/// Returns true if a callback with that id was removed. Callbacks live on
|
||||
/// the context's shared HTTP client, so detach them when a page or capture
|
||||
/// phase ends to avoid firing for sibling pages (issue #408).
|
||||
pub fn off_request(&mut self, id: u64) -> bool {
|
||||
self.inner.off_request(id)
|
||||
}
|
||||
|
||||
/// Detach a response callback previously registered with `on_response`.
|
||||
/// Returns true if a callback with that id was removed.
|
||||
pub fn off_response(&mut self, id: u64) -> bool {
|
||||
self.inner.off_response(id)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -165,3 +165,49 @@ async fn page_rewrites_request_url_via_interception() {
|
||||
body
|
||||
);
|
||||
}
|
||||
|
||||
// Issue #408: a callback registered with on_response must be detachable via the
|
||||
// returned id, so a crawler can stop capturing after a phase. Before the fix the
|
||||
// vecs were append-only and a callback fired for the client's whole lifetime.
|
||||
#[tokio::test]
|
||||
async fn on_response_callback_can_be_detached() {
|
||||
std::env::set_var("OBSCURA_ALLOW_PRIVATE_NETWORK", "1");
|
||||
let base = spawn_echo_server();
|
||||
|
||||
let browser = Browser::new().unwrap();
|
||||
let mut page = browser.new_page().await.unwrap();
|
||||
|
||||
let hits = Arc::new(AtomicU32::new(0));
|
||||
let h = hits.clone();
|
||||
let id = page.on_response(Arc::new(move |info, _resp| {
|
||||
if info.resource_type == ResourceType::Fetch {
|
||||
h.fetch_add(1, Ordering::SeqCst);
|
||||
}
|
||||
}));
|
||||
|
||||
// First navigation: the callback fires for the JS fetch('/api').
|
||||
page.goto(&base).await.unwrap();
|
||||
for _ in 0..20 {
|
||||
page.settle(200).await;
|
||||
if hits.load(Ordering::SeqCst) >= 1 {
|
||||
break;
|
||||
}
|
||||
}
|
||||
let after_first = hits.load(Ordering::SeqCst);
|
||||
assert!(after_first >= 1, "on_response should fire while attached");
|
||||
|
||||
// Detach it; off_response must report success and the id must be gone.
|
||||
assert!(page.off_response(id), "off_response must remove the callback");
|
||||
assert!(!page.off_response(id), "removing an already-removed id returns false");
|
||||
|
||||
// Second navigation: the detached callback must not fire again.
|
||||
page.goto(&base).await.unwrap();
|
||||
for _ in 0..10 {
|
||||
page.settle(200).await;
|
||||
}
|
||||
assert_eq!(
|
||||
hits.load(Ordering::SeqCst),
|
||||
after_first,
|
||||
"detached on_response callback must not fire after off_response"
|
||||
);
|
||||
}
|
||||
Reference in new issue
Block a user