Skip to content

Commit e0a9470

Browse files
JasonVranekJasonVranek
authored
Rewrite get_header relay communication to support SSZ and JSON (#467)
1 parent 26d62bc commit e0a9470

20 files changed

Lines changed: 1342 additions & 346 deletions

‎Cargo.lock‎

Lines changed: 2 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎crates/common/src/config/mux.rs‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -236,7 +236,6 @@ impl MuxKeysLoader {
236236
"Mux keys URL {url} is insecure; consider using HTTPS if possible instead"
237237
);
238238
}
239-
let url = url.as_str();
240239
let client = reqwest::ClientBuilder::new().timeout(http_timeout).build()?;
241240
let response = client.get(url).send().await?;
242241
let pubkey_bytes = safe_read_http_response(response, MUXER_HTTP_MAX_LENGTH).await?;

‎crates/common/src/pbs/error.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,9 @@ pub enum PbsError {
1515
#[error("json decode error: {err:?}, raw: {raw}")]
1616
JsonDecode { err: serde_json::Error, raw: String },
1717

18+
#[error("ssz decode error: {err:?}, fork: {fork}")]
19+
SSZDecode { err: String, fork: ForkName },
20+
1821
#[error("{0}")]
1922
ReadResponse(#[from] ResponseReadError),
2023

‎crates/common/src/wire.rs‎

Lines changed: 91 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -7,10 +7,10 @@ use bytes::Bytes;
77
use futures::StreamExt;
88
use headers_accept::Accept;
99
use lh_types::{BeaconBlock, ForkName};
10-
use mediatype::{MediaType, ReadParams};
10+
use mediatype::{MediaType, ReadParams, names};
1111
use reqwest::{
1212
Response,
13-
header::{ACCEPT, CONTENT_TYPE, HeaderMap},
13+
header::{ACCEPT, CONTENT_TYPE, HeaderMap, ToStrError},
1414
};
1515
use thiserror::Error;
1616

@@ -37,6 +37,18 @@ pub enum ResponseReadError {
3737
NonSuccess { status_code: u16, error_msg: String, request_url: String },
3838
}
3939

40+
#[derive(Debug, Error)]
41+
pub enum AcceptedEncodingsError {
42+
#[error("invalid header string: {0}")]
43+
InvalidString(#[from] ToStrError),
44+
45+
#[error("invalid accept header")]
46+
InvalidAccept,
47+
48+
#[error("unsupported accept type")]
49+
UnsupportedAcceptType,
50+
}
51+
4052
#[cfg(feature = "testing-flags")]
4153
thread_local! {
4254
static IGNORE_CONTENT_LENGTH: Cell<bool> = const { Cell::new(false) };
@@ -108,7 +120,7 @@ pub async fn read_chunked_body_with_max(
108120
/// Reads an HTTP response body with a size limit, erroring on non-success
109121
/// status or read failure.
110122
pub async fn safe_read_http_response(
111-
response: reqwest::Response,
123+
response: Response,
112124
max_size: usize,
113125
) -> Result<Vec<u8>, ResponseReadError> {
114126
let status_code = response.status();
@@ -140,12 +152,6 @@ pub fn get_user_agent_with_version(req_headers: &HeaderMap) -> eyre::Result<Head
140152
Ok(HeaderValue::from_str(&format!("commit-boost/{HEADER_VERSION_VALUE} {ua}"))?)
141153
}
142154

143-
/// Deterministic outbound `Accept` header used when PBS asks a relay for a
144-
/// response it will itself decode (validation mode On/Extra). SSZ is preferred
145-
/// for wire efficiency. Emitted verbatim so packet captures and support
146-
/// tickets are reproducible.
147-
pub const OUTBOUND_ACCEPT: &str = "application/octet-stream;q=1.0,application/json;q=0.9";
148-
149155
/// Default encoding used when the caller does not express a format
150156
/// preference. This covers both `Accept: */*` (see `get_accept_types`) and
151157
/// a missing Content-Type header on inbound or relay responses (see
@@ -190,24 +196,32 @@ impl IntoIterator for AcceptedEncodings {
190196
/// Parse the ACCEPT header into a q-value ordered [`AcceptedEncodings`]
191197
/// (highest preference first, deduplicated), defaulting to the request's
192198
/// Content-Type when no Accept header is present. Returns an error only if
193-
/// every media type in the header is malformed or unsupported. Supports
194-
/// requests with multiple ACCEPT headers or headers with multiple media
195-
/// types. `q=0` entries are treated as explicit rejections per RFC 7231
199+
/// every media type in the header is malformed or unsupported.
200+
/// Multiple Accept header fields are combined before parsing so q-value
201+
/// ordering is applied globally across all media ranges.
202+
/// `q=0` entries are treated as explicit rejections per RFC 7231
196203
/// §5.3.1 and are skipped.
197-
///
198204
/// The returned order honors the RFC 9110 §12.5.1 precedence rules already
199205
/// applied by `headers_accept::Accept::media_types()` (specificity, then
200206
/// q-value, then original order).
201-
pub fn get_accept_types(req_headers: &HeaderMap) -> eyre::Result<AcceptedEncodings> {
207+
pub fn get_accept_types(
208+
req_headers: &HeaderMap,
209+
) -> Result<AcceptedEncodings, AcceptedEncodingsError> {
202210
// Only two supported media types, so the ordered set is at most two
203211
// entries: primary + optional fallback.
204212
let mut primary: Option<EncodingType> = None;
205213
let mut fallback: Option<EncodingType> = None;
206214
let mut saw_any = false;
207215
let mut had_supported = false;
216+
let mut accept_values = Vec::new();
217+
208218
for header in req_headers.get_all(ACCEPT).iter() {
209-
let accept = Accept::from_str(header.to_str()?)
210-
.map_err(|e| eyre::eyre!("invalid accept header: {e}"))?;
219+
accept_values.push(header.to_str()?);
220+
}
221+
if !accept_values.is_empty() {
222+
let accept_str = accept_values.join(",");
223+
let accept =
224+
Accept::from_str(&accept_str).map_err(|_| AcceptedEncodingsError::InvalidAccept)?;
211225
for mt in accept.media_types() {
212226
saw_any = true;
213227

@@ -221,13 +235,7 @@ pub fn get_accept_types(req_headers: &HeaderMap) -> eyre::Result<AcceptedEncodin
221235
continue;
222236
}
223237

224-
let parsed = match mt.essence().to_string().as_str() {
225-
APPLICATION_OCTET_STREAM => Some(EncodingType::Ssz),
226-
APPLICATION_JSON => Some(EncodingType::Json),
227-
WILDCARD => Some(NO_PREFERENCE_DEFAULT),
228-
_ => None,
229-
};
230-
if let Some(enc) = parsed {
238+
if let Some(enc) = essence_encoding(&mt.essence()) {
231239
had_supported = true;
232240
match primary {
233241
None => primary = Some(enc),
@@ -243,14 +251,30 @@ pub fn get_accept_types(req_headers: &HeaderMap) -> eyre::Result<AcceptedEncodin
243251
}
244252

245253
if saw_any && !had_supported {
246-
eyre::bail!("unsupported accept type");
254+
return Err(AcceptedEncodingsError::UnsupportedAcceptType)
247255
}
248256

249257
// No accept header (or only q=0 rejections): fall back to the request
250258
// Content-Type, which mirrors the historical behavior.
251259
Ok(AcceptedEncodings::single(get_content_type(req_headers)))
252260
}
253261

262+
fn essence_encoding(mt: &MediaType) -> Option<EncodingType> {
263+
if mt.suffix.is_some() {
264+
return None;
265+
}
266+
267+
match () {
268+
_ if mt.ty == names::_STAR && mt.subty == names::_STAR => Some(NO_PREFERENCE_DEFAULT),
269+
_ if mt.ty == names::APPLICATION && mt.subty == names::OCTET_STREAM => {
270+
Some(EncodingType::Ssz)
271+
}
272+
_ if mt.ty == names::APPLICATION && mt.subty == names::JSON => Some(EncodingType::Json),
273+
_ if mt.ty == names::APPLICATION && mt.subty == names::_STAR => Some(NO_PREFERENCE_DEFAULT),
274+
_ => None,
275+
}
276+
}
277+
254278
/// Compute the q-value for the `index`-th preferred encoding when building an
255279
/// outbound `Accept` header. The first entry gets q=1.0, each subsequent entry
256280
/// decreases by 0.1, and the value is clamped to a minimum of 0.1 so we never
@@ -269,17 +293,19 @@ fn format_accept_entry(enc: EncodingType, q: f32) -> String {
269293
format!("{};q={:.1}", enc.content_type(), q)
270294
}
271295

272-
/// Build an `Accept` header string that mirrors the caller's preference order
296+
/// Build an `Accept` header that mirrors the caller's preference order
273297
/// so the relay sees the same priority the beacon node asked us for. Each
274298
/// subsequent entry receives a q-value 0.1 lower than the previous one,
275-
/// starting at 1.0.
276-
pub fn build_outbound_accept(preferred: AcceptedEncodings) -> String {
277-
preferred
299+
/// starting at 1.0. Returns a ready-to-use `HeaderValue` — the output is
300+
/// always valid ASCII, so infallible.
301+
pub fn build_outbound_accept(preferred: AcceptedEncodings) -> HeaderValue {
302+
let s = preferred
278303
.iter()
279304
.enumerate()
280305
.map(|(i, enc)| format_accept_entry(enc, accept_q_value_for_index(i)))
281306
.collect::<Vec<_>>()
282-
.join(",")
307+
.join(",");
308+
HeaderValue::from_str(&s).expect("build_outbound_accept produces valid header value")
283309
}
284310

285311
pub fn get_content_type(req_headers: &HeaderMap) -> EncodingType {
@@ -345,11 +371,7 @@ impl FromStr for EncodingType {
345371
// (e.g. `application/json; charset=utf-8`). Compare essence only.
346372
let parsed =
347373
MediaType::parse(value).map_err(|e| format!("invalid content type {value}: {e}"))?;
348-
match parsed.essence().to_string().to_ascii_lowercase().as_str() {
349-
APPLICATION_JSON => Ok(EncodingType::Json),
350-
APPLICATION_OCTET_STREAM => Ok(EncodingType::Ssz),
351-
_ => Err(format!("unsupported encoding type: {value}")),
352-
}
374+
essence_encoding(&parsed).ok_or_else(|| format!("unsupported encoding type: {value}"))
353375
}
354376
}
355377

@@ -430,7 +452,7 @@ mod test {
430452

431453
use super::{
432454
APPLICATION_JSON, APPLICATION_OCTET_STREAM, AcceptedEncodings, BodyDeserializeError,
433-
CONSENSUS_VERSION_HEADER, EncodingType, NO_PREFERENCE_DEFAULT, OUTBOUND_ACCEPT, WILDCARD,
455+
CONSENSUS_VERSION_HEADER, EncodingType, NO_PREFERENCE_DEFAULT, WILDCARD,
434456
accept_q_value_for_index, build_outbound_accept, deserialize_body, format_accept_entry,
435457
get_accept_types, get_consensus_version_header, get_content_type,
436458
parse_response_encoding_and_fork,
@@ -525,6 +547,15 @@ mod test {
525547
assert!(result.is_err());
526548
}
527549

550+
/// Test rejecting an unknown Accept: / type
551+
#[test]
552+
fn test_invalid_accept_header_type_slash() {
553+
let mut headers = HeaderMap::new();
554+
headers.append(ACCEPT, HeaderValue::from_str("/").unwrap());
555+
let result = get_accept_types(&headers);
556+
assert!(result.is_err());
557+
}
558+
528559
/// Test accepting one header with multiple values
529560
#[test]
530561
fn test_accept_header_invalid_parse() {
@@ -689,6 +720,24 @@ mod test {
689720
);
690721
}
691722

723+
/// Multiple Accept header fields must be combined before parsing so
724+
/// q-values are ordered globally across all media ranges, not per
725+
/// header field.
726+
#[test]
727+
fn test_multiple_accept_headers_q_value_ordering() {
728+
let mut headers = HeaderMap::new();
729+
730+
// SSZ appears in the first header field but has a lower q-value.
731+
// JSON appears in the second header field and should win globally.
732+
headers.append(ACCEPT, HeaderValue::from_str("application/octet-stream;q=0.1").unwrap());
733+
headers.append(ACCEPT, HeaderValue::from_str("application/json;q=1.0").unwrap());
734+
735+
assert_eq!(get_accept_types(&headers).unwrap(), AcceptedEncodings {
736+
primary: EncodingType::Json,
737+
fallback: Some(EncodingType::Ssz),
738+
});
739+
}
740+
692741
/// Once primary and fallback are filled, further supported entries must
693742
/// not overwrite fallback. (Belt-and-suspenders — only two supported
694743
/// variants exist today, so this is mostly a guard against future
@@ -746,13 +795,6 @@ mod test {
746795
);
747796
}
748797

749-
/// Snapshot test: constant emits exactly what we document in
750-
/// OUTBOUND_ACCEPT.
751-
#[test]
752-
fn test_outbound_accept_constant_snapshot() {
753-
assert_eq!(OUTBOUND_ACCEPT, "application/octet-stream;q=1.0,application/json;q=0.9");
754-
}
755-
756798
/// q-value ladder: first entry is 1.0, each subsequent entry drops by 0.1.
757799
#[test]
758800
fn test_accept_q_value_for_index_ladder() {
@@ -810,6 +852,13 @@ mod test {
810852
assert_eq!(get_content_type(&headers), EncodingType::Json);
811853
}
812854

855+
#[test]
856+
fn test_content_type_invalid_defaults_to_json() {
857+
let mut headers = HeaderMap::new();
858+
headers.append(CONTENT_TYPE, HeaderValue::from_str("/").unwrap());
859+
assert_eq!(get_content_type(&headers), EncodingType::Json);
860+
}
861+
813862
// ── get_consensus_version_header ─────────────────────────────────────────
814863

815864
#[test]

‎crates/pbs/src/error.rs‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,8 @@
1-
use axum::{http::StatusCode, response::IntoResponse};
2-
use cb_common::wire::BodyDeserializeError;
1+
use axum::{
2+
http::StatusCode,
3+
response::{IntoResponse, Response},
4+
};
5+
use cb_common::wire::{AcceptedEncodingsError, BodyDeserializeError};
36
use thiserror::Error;
47

58
#[derive(Debug, Error)]
@@ -13,6 +16,8 @@ pub enum PbsClientError {
1316
Internal,
1417
#[error("failed to deserialize body: {0}")]
1518
DecodeError(#[from] BodyDeserializeError),
19+
#[error("invalid accept types: {0}")]
20+
HeaderError(#[from] AcceptedEncodingsError),
1621
}
1722

1823
impl PbsClientError {
@@ -22,17 +27,19 @@ impl PbsClientError {
2227
PbsClientError::NoPayload => StatusCode::BAD_GATEWAY,
2328
PbsClientError::Internal => StatusCode::INTERNAL_SERVER_ERROR,
2429
PbsClientError::DecodeError(_) => StatusCode::BAD_REQUEST,
30+
PbsClientError::HeaderError(_) => StatusCode::NOT_ACCEPTABLE,
2531
}
2632
}
2733
}
2834

2935
impl IntoResponse for PbsClientError {
30-
fn into_response(self) -> axum::response::Response {
36+
fn into_response(self) -> Response {
3137
let msg = match &self {
3238
PbsClientError::NoResponse => "no response from relays".to_string(),
3339
PbsClientError::NoPayload => "no payload from relays".to_string(),
3440
PbsClientError::Internal => "internal server error".to_string(),
3541
PbsClientError::DecodeError(e) => format!("error decoding request: {e}"),
42+
PbsClientError::HeaderError(e) => format!("header error: {e}"),
3643
};
3744

3845
(self.status_code(), msg).into_response()

0 commit comments

Comments
 (0)