filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

RateLimit header fields for HTTP

Browse Source
This commit is contained in:
Maurus Decimus
2026-05-23 11:41:43 +02:00
parent 7aa4865a10
commit 777cf1252b
71 changed files with 825 additions and 274 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
tests/src/smtp/lookup/utils.rs
View File

@@ -18,7 +18,7 @@ use registry::{
enums::MtaIpStrategy,
structs::{
Expression, MtaConnectionIpHost, MtaConnectionStrategy, MtaOutboundStrategy, MtaRoute,
MtaRouteMx,
MtaRouteMx,
},
},
types::{ipaddr::IpAddr, list::List},

6
tests/src/store/blob.rs
View File

@@ -41,10 +41,10 @@ pub async fn blob_tests() {
);
// Test blob quota
assert!(test.server.blob_has_quota(0, 1024).await.unwrap());
assert!(!test.server.blob_has_quota(0, 1024).await.unwrap());
assert!(test.server.blob_has_quota(0, 1024).await.unwrap().allowed);
assert!(!test.server.blob_has_quota(0, 1024).await.unwrap().allowed);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
assert!(test.server.blob_has_quota(0, 1024).await.unwrap());
assert!(test.server.blob_has_quota(0, 1024).await.unwrap().allowed);
// Test and reset store
test_store(blob_store.clone()).await;

99
tests/src/system/quota.rs
View File

@@ -4,7 +4,9 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::utils::{account::Account, jmap::JmapUtils, server::TestServer, smtp::SmtpConnection};
use crate::utils::{
account::Account, http::HttpRequest, jmap::JmapUtils, server::TestServer, smtp::SmtpConnection,
};
use email::{cache::MessageCacheFetch, mailbox::INBOX_ID};
use jmap::blob::upload::DISABLE_UPLOAD_QUOTA;
use jmap_client::{
@@ -102,6 +104,12 @@ pub async fn test(test: &mut TestServer) {
// Test temporary blob quota (3 files)
DISABLE_UPLOAD_QUOTA.store(false, std::sync::atomic::Ordering::Relaxed);
let client = account.jmap_client().await;
let raw_http = HttpRequest::with_credentials(
8899,
"user1@example.org",
"this is a very strong password1",
);
let upload_url = format!("/jmap/upload/{account_id}");
for i in 0..3 {
assert_eq!(
client
@@ -112,14 +120,45 @@ pub async fn test(test: &mut TestServer) {
1024
);
}
match client
.upload(None, vec![b'Z'; 1024], None)
.await
.unwrap_err()
{
jmap_client::Error::Problem(err) if err.detail().unwrap().contains("quota") => (),
other => panic!("Unexpected error: {:?}", other),
}
let resp = raw_http
.send_full(
hyper::Method::POST,
&upload_url,
Some(vec![b'Z'; 1024]),
Some("application/octet-stream"),
)
.await;
assert_eq!(
resp.status.as_u16(),
429,
"blob-files-quota body: {}",
resp.body
);
let policy = resp
.rate_limit_policy()
.unwrap_or_else(|| panic!("missing RateLimit-Policy on {:?}", resp.headers));
assert!(
policy.contains("\"blob-upload-files\";q=3"),
"RateLimit-Policy = {policy}"
);
assert!(
policy.contains("\"blob-upload-bytes\";q=50000")
&& policy.contains(r#"qu="content-bytes""#),
"RateLimit-Policy = {policy}"
);
let state = resp
.rate_limit()
.unwrap_or_else(|| panic!("missing RateLimit on {:?}", resp.headers));
assert!(
state.contains("\"blob-upload-files\";r=0") && state.contains("t="),
"RateLimit = {state}"
);
assert!(
resp.retry_after().is_some(),
"missing Retry-After on {:?}",
resp.headers
);
assert!(resp.body.contains("quota"), "body = {}", resp.body);
test.blob_expire_all().await;
// Test temporary blob quota (50000 bytes)
@@ -134,14 +173,40 @@ pub async fn test(test: &mut TestServer) {
25000
);
}
match client
.upload(None, vec![b'z'; 1024], None)
.await
.unwrap_err()
{
jmap_client::Error::Problem(err) if err.detail().unwrap().contains("quota") => (),
other => panic!("Unexpected error: {:?}", other),
}
let resp = raw_http
.send_full(
hyper::Method::POST,
&upload_url,
Some(vec![b'z'; 1024]),
Some("application/octet-stream"),
)
.await;
assert_eq!(
resp.status.as_u16(),
429,
"blob-bytes-quota body: {}",
resp.body
);
let policy = resp
.rate_limit_policy()
.unwrap_or_else(|| panic!("missing RateLimit-Policy on {:?}", resp.headers));
assert!(
policy.contains("\"blob-upload-bytes\";q=50000")
&& policy.contains(r#"qu="content-bytes""#),
"RateLimit-Policy = {policy}"
);
let state = resp
.rate_limit()
.unwrap_or_else(|| panic!("missing RateLimit on {:?}", resp.headers));
assert!(
state.contains("\"blob-upload-bytes\";r=0") && state.contains("t="),
"RateLimit = {state}"
);
assert!(
resp.retry_after().is_some(),
"missing Retry-After on {:?}",
resp.headers
);
test.blob_expire_all().await;
tokio::time::sleep(std::time::Duration::from_millis(1100)).await;

90
tests/src/system/security.rs
View File

@@ -7,6 +7,7 @@
use crate::{
system::authentication::validate_password_with_ip,
utils::{
http::HttpRequest,
imap::{ImapConnection, Type},
registry::UnwrapRegistryId,
server::TestServer,
@@ -271,6 +272,11 @@ pub async fn test(test: &mut TestServer) {
// Concurrent requests check
let client = Arc::new(client);
let raw_http = HttpRequest::with_credentials(
8899,
"user@example.org",
"this is a very strong password",
);
for _ in 0..8 {
let client_ = client.clone();
tokio::spawn(async move {
@@ -283,14 +289,48 @@ pub async fn test(test: &mut TestServer) {
});
}
tokio::time::sleep(Duration::from_millis(500)).await;
assert!(matches!(
client
.mailbox_query(
mailbox::query::Filter::name("__sleep").into(),
[mailbox::query::Comparator::name()].into(),
)
.await,
Err(jmap_client::Error::Problem(err)) if err.status() == Some(400)));
let body = serde_json::to_vec(&json!({
"using": ["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"],
"methodCalls": [
["Mailbox/query", {
"accountId": user_id.to_string(),
"filter": { "name": "__sleep" }
}, "c1"]
]
}))
.unwrap();
let resp = raw_http
.send_full(
hyper::Method::POST,
"/jmap/",
Some(body),
Some("application/json"),
)
.await;
assert_eq!(
resp.status.as_u16(),
400,
"concurrent-requests body: {}",
resp.body
);
let policy = resp
.rate_limit_policy()
.unwrap_or_else(|| panic!("missing RateLimit-Policy header on {:?}", resp.headers));
assert!(
policy.contains("\"concurrent-requests\"") && policy.contains("q=8"),
"RateLimit-Policy = {policy}"
);
assert!(
policy.contains(r#"qu="concurrent-requests""#),
"RateLimit-Policy = {policy}"
);
let state = resp
.rate_limit()
.unwrap_or_else(|| panic!("missing RateLimit header on {:?}", resp.headers));
assert!(
state.contains("\"concurrent-requests\"") && state.contains("r=0"),
"RateLimit = {state}"
);
// Wait for sleep to be done
tokio::time::sleep(Duration::from_millis(1000)).await;
@@ -303,9 +343,37 @@ pub async fn test(test: &mut TestServer) {
});
}
tokio::time::sleep(Duration::from_millis(500)).await;
assert!(matches!(
client.upload(None, b"sleep".to_vec(), None).await,
Err(jmap_client::Error::Problem(err)) if err.status() == Some(400)));
let resp = raw_http
.send_full(
hyper::Method::POST,
&format!("/jmap/upload/{user_id}"),
Some(b"sleep".to_vec()),
Some("application/octet-stream"),
)
.await;
assert_eq!(
resp.status.as_u16(),
400,
"concurrent-uploads body: {}",
resp.body
);
let policy = resp
.rate_limit_policy()
.unwrap_or_else(|| panic!("missing RateLimit-Policy header on {:?}", resp.headers));
assert!(
policy.contains("\"concurrent-uploads\"") && policy.contains("q=4"),
"RateLimit-Policy = {policy}"
);
let state = resp
.rate_limit()
.unwrap_or_else(|| panic!("missing RateLimit header on {:?}", resp.headers));
assert!(
state.contains("\"concurrent-uploads\"") && state.contains("r=0"),
"RateLimit = {state}"
);
// Wait for sleep to be done before continuing
tokio::time::sleep(Duration::from_millis(1000)).await;
// Disable X-Forwarded-For processing
admin

7
tests/src/system/task.rs
View File

@@ -158,7 +158,12 @@ async fn pagination_test(test: &mut TestServer) {
.await
.object_ids()
.collect();
assert_eq!(asc_order.len(), 12, "expected 12 tasks, got {}", asc_order.len());
assert_eq!(
asc_order.len(),
12,
"expected 12 tasks, got {}",
asc_order.len()
);
let desc_order: Vec<Id> = asc_order.iter().rev().copied().collect();

6
tests/src/telemetry/metrics.rs
View File

@@ -87,7 +87,11 @@ pub async fn test(test: &TestServer) {
.await
.object_ids()
.collect();
assert!(asc_order.len() > 100, "expected >100 metrics, got {}", asc_order.len());
assert!(
asc_order.len() > 100,
"expected >100 metrics, got {}",
asc_order.len()
);
let desc_order: Vec<Id> = asc_order.iter().rev().copied().collect();
let total = asc_order.len();
let limit = 25usize;

2
tests/src/telemetry/webhooks.rs
View File

@@ -6,6 +6,7 @@
use crate::utils::server::TestServer;
use crate::utils::smtp::SmtpConnection;
use aws_lc_rs::hmac;
use base64::{Engine, engine::general_purpose::STANDARD};
use common::{manager::application::Resource, telemetry::tracers::store::TracingStore};
use http_proto::{ToHttpResponse, request::fetch_body};
@@ -21,7 +22,6 @@ use registry::{
},
types::map::Map,
};
use aws_lc_rs::hmac;
use std::{
sync::{
Arc,

62
tests/src/utils/http.rs
View File

@@ -5,6 +5,7 @@
*/
use hyper::Method;
use reqwest::header::HeaderMap;
use serde::{Serialize, de::DeserializeOwned};
use std::time::Duration;
@@ -14,6 +15,30 @@ pub struct HttpRequest {
pub password: Option<String>,
}
pub struct HttpResponseFull {
pub status: reqwest::StatusCode,
pub headers: HeaderMap,
pub body: String,
}
impl HttpResponseFull {
pub fn header(&self, name: &str) -> Option<&str> {
self.headers.get(name).and_then(|v| v.to_str().ok())
}
pub fn rate_limit_policy(&self) -> Option<&str> {
self.header("RateLimit-Policy")
}
pub fn rate_limit(&self) -> Option<&str> {
self.header("RateLimit")
}
pub fn retry_after(&self) -> Option<u64> {
self.header("Retry-After").and_then(|v| v.parse().ok())
}
}
impl Default for HttpRequest {
fn default() -> Self {
Self {
@@ -94,6 +119,43 @@ impl HttpRequest {
})
}
pub async fn send_full(
&self,
method: Method,
query: &str,
body: Option<Vec<u8>>,
content_type: Option<&str>,
) -> HttpResponseFull {
let mut request = reqwest::Client::builder()
.timeout(Duration::from_secs(5))
.danger_accept_invalid_certs(true)
.build()
.unwrap()
.request(method, format!("https://127.0.0.1:{}{query}", self.port));
if let Some(body) = body {
request = request.body(body);
}
if let Some(ct) = content_type {
request = request.header(hyper::header::CONTENT_TYPE, ct);
}
if let (Some(username), Some(password)) = (&self.username, &self.password) {
request = request.basic_auth(username, Some(password));
}
let response = request.send().await.expect("HTTP request failed");
let status = response.status();
let headers = response.headers().clone();
let body = response.text().await.unwrap_or_default();
HttpResponseFull {
status,
headers,
body,
}
}
async fn request_raw(
&self,
method: Method,

5
tests/src/webdav/cal_query.rs
View File

@@ -19,7 +19,10 @@ use types::TimeRange;
pub async fn test(test: &TestServer) {
println!("Running REPORT calendar-query & free-busy-query tests...");
let client = test.account("john@example.com").webdav_client();
let cal_path = format!("{}/john%40example.com/default/", DavResourceName::Cal.base_path());
let cal_path = format!(
"{}/john%40example.com/default/",
DavResourceName::Cal.base_path()
);
#[allow(clippy::never_loop)]
for (num, ics) in [

5
tests/src/webdav/card_query.rs
View File

@@ -15,7 +15,10 @@ pub async fn test(test: &TestServer) {
let client = test.account("john@example.com").webdav_client();
// Create test data
let default_path = format!("{}/john%40example.com/default/", DavResourceName::Card.base_path());
let default_path = format!(
"{}/john%40example.com/default/",
DavResourceName::Card.base_path()
);
let mut hrefs = Vec::with_capacity(3);
for (i, vcard) in [VCARD1, VCARD2, VCARD3].iter().enumerate() {
let href = format!("{default_path}contact-{i}.vcf",);

31
tests/src/webdav/mkcol.rs
View File

@@ -10,7 +10,6 @@ use crate::webdav::{TEST_FILE_1, TEST_ICAL_1, TEST_VCARD_1, TEST_VTIMEZONE_1};
use crate::utils::server::TestServer;
pub async fn test(test: &TestServer) {
println!("Running MKCOL tests...");
let client = test.account("john@example.com").webdav_client();
@@ -42,9 +41,18 @@ pub async fn test(test: &TestServer) {
// Create resources under the newly created collections
for (path, content) in [
("/dav/file/john%40example.com/my-files/file1.txt", TEST_FILE_1),
("/dav/card/john%40example.com/my-cards/card1.vcf", TEST_VCARD_1),
("/dav/cal/john%40example.com/my-events/event1.ics", TEST_ICAL_1),
(
"/dav/file/john%40example.com/my-files/file1.txt",
TEST_FILE_1,
),
(
"/dav/card/john%40example.com/my-cards/card1.vcf",
TEST_VCARD_1,
),
(
"/dav/cal/john%40example.com/my-events/event1.ics",
TEST_ICAL_1,
),
] {
client
.request("PUT", path, content)
@@ -69,7 +77,10 @@ pub async fn test(test: &TestServer) {
// Creating a sub-collections is allowed in FileDAV but in CalDAV and CardDAV
for (path, expected_status) in [
("/dav/file/john%40example.com/my-files/my-sub-files", StatusCode::CREATED),
(
"/dav/file/john%40example.com/my-files/my-sub-files",
StatusCode::CREATED,
),
(
"/dav/card/john%40example.com/my-cards/my-sub-cards",
StatusCode::METHOD_NOT_ALLOWED,
@@ -87,9 +98,15 @@ pub async fn test(test: &TestServer) {
// Extended MKCOL with an unsupported resource types should fail
for (path, resource_type) in [
("/dav/file/john%40example.com/my-named-files", "B:addressbook"),
(
"/dav/file/john%40example.com/my-named-files",
"B:addressbook",
),
("/dav/card/john%40example.com/my-named-cards", "A:calendar"),
("/dav/cal/john%40example.com/my-named-events", "B:addressbook"),
(
"/dav/cal/john%40example.com/my-named-events",
"B:addressbook",
),
] {
client
.mkcol("MKCOL", path, ["D:collection", resource_type], [])

6
tests/src/webdav/multiget.rs
View File

@@ -21,7 +21,11 @@ pub async fn test(test: &TestServer) {
let mut paths = Vec::new();
for name in ["file1", "file2"] {
let contents = resource_type.generate();
let path = format!("{}/john%40example.com/default/{}", resource_type.base_path(), name);
let path = format!(
"{}/john%40example.com/default/{}",
resource_type.base_path(),
name
);
let etag = client
.request("PUT", &path, contents.as_str())
.await