structured-proxy 8.1.0

//! Forwarded request headers reach a real tonic upstream as the client sent
//! them: every value, in order, over the actual HTTP/2 stream and in process.

#[macro_use]
mod common;

use std::convert::Infallible;
use std::future::{ready, Ready};
use std::pin::Pin;
use std::task::{Context, Poll};

use axum::body::Body;
use http::{HeaderValue, StatusCode};
use prost_reflect::{DescriptorPool, DynamicMessage, Value as PbValue};
use structured_proxy::transcode::codec::DynamicCodec;
use structured_proxy::transcode::error::ErrorDetailsPolicy;

const HEADERS_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";

message Req {}
message Reply {
  string name = 1;
}

service Headers {
  rpc Seen(Req) returns (Reply) {
    option (google.api.http) = { get: "/v1/seen" };
  }
}
"#;

fn pool() -> DescriptorPool {
    common::compile("test/v1/headers.proto", HEADERS_PROTO)
}

/// Answers with the `dpop` values it received, joined by `|`.
#[derive(Clone)]
struct Seen {
    pool: DescriptorPool,
}

impl tonic::server::NamedService for Seen {
    const NAME: &'static str = "test.v1.Headers";
}

impl tonic::server::UnaryService<DynamicMessage> for Seen {
    type Response = DynamicMessage;
    type Future = Ready<Result<tonic::Response<DynamicMessage>, tonic::Status>>;

    fn call(&mut self, request: tonic::Request<DynamicMessage>) -> Self::Future {
        let seen: Vec<String> = request
            .metadata()
            .get_all("dpop")
            .iter()
            .map(|v| String::from_utf8_lossy(v.as_encoded_bytes()).into_owned())
            .collect();
        let mut reply =
            DynamicMessage::new(self.pool.get_message_by_name("test.v1.Reply").unwrap());
        reply.set_field_by_name("name", PbValue::String(seen.join("|")));
        ready(Ok(tonic::Response::new(reply)))
    }
}

impl tower::Service<http::Request<tonic::body::Body>> for Seen {
    type Response = http::Response<tonic::body::Body>;
    type Error = Infallible;
    type Future =
        Pin<Box<dyn std::future::Future<Output = Result<Self::Response, Infallible>> + Send>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: http::Request<tonic::body::Body>) -> Self::Future {
        let service = self.clone();
        Box::pin(async move {
            let method = service
                .pool
                .get_service_by_name("test.v1.Headers")
                .unwrap()
                .methods()
                .find(|m| m.name() == "Seen")
                .unwrap();
            let mut grpc = tonic::server::Grpc::new(DynamicCodec::new(method.input()));
            Ok(grpc.unary(service, req).await)
        })
    }
}

async fn proxy(upstream: common::Upstream) -> common::App {
    common::proxy(
        upstream,
        Seen { pool: pool() },
        pool(),
        ErrorDetailsPolicy::default(),
    )
    .await
}

upstream_tests! {
async fn every_value_of_a_repeated_header_reaches_the_upstream_in_order() {
    // RFC 9449 §4.3 has the server reject a request with two DPoP headers;
    // behind the proxy it can only do that if both arrive.
    let app = proxy(UPSTREAM).await;
    let request = http::Request::get("/v1/seen")
        .header("dpop", "proof-a")
        .header("dpop", "proof-b")
        .body(Body::empty())
        .unwrap();
    let (status, body) = common::send(&app, request).await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(
        serde_json::from_str::<serde_json::Value>(&body).unwrap()["name"],
        "proof-a|proof-b"
    );
}

async fn a_value_grpc_metadata_cannot_carry_is_refused() {
    // gRPC lets a receiver drop an ASCII metadata value outside %x20-%x7E, which
    // would change how many DPoP headers the upstream counts; the request is
    // refused instead of reaching it altered.
    let app = proxy(UPSTREAM).await;
    let request = http::Request::get("/v1/seen")
        .header("dpop", "proof-a")
        .header("dpop", HeaderValue::from_bytes(b"caf\xe9").unwrap())
        .body(Body::empty())
        .unwrap();
    let (status, body) = common::send(&app, request).await;
    assert_eq!(status, StatusCode::BAD_REQUEST, "{body}");
    let body: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(body["error"], "INVALID_ARGUMENT");
    assert!(body["message"].as_str().unwrap().contains("dpop"), "{body}");
}

async fn a_single_value_is_unchanged() {
    let app = proxy(UPSTREAM).await;
    let request = http::Request::get("/v1/seen")
        .header("dpop", "proof-a")
        .body(Body::empty())
        .unwrap();
    let (status, body) = common::send(&app, request).await;
    assert_eq!(status, StatusCode::OK, "{body}");
    assert_eq!(
        serde_json::from_str::<serde_json::Value>(&body).unwrap()["name"],
        "proof-a"
    );
}
}

#[tokio::test]
async fn a_forwarded_name_grpc_cannot_carry_is_a_config_error() {
    // `+` is a valid HTTP field-name character but not a gRPC key one, so an
    // upstream may reject the call: the proxy refuses to start instead.
    let config = structured_proxy::config::ProxyConfig::from_yaml_str(
        "upstream:\n  default: \"http://127.0.0.1:1\"\nforwarded_headers: [\"x+proof\"]\n",
    )
    .unwrap();
    let Err(err) = structured_proxy::ProxyServer::from_config(config)
        .with_descriptors(pool())
        .router()
    else {
        panic!("a forwarded header outside the gRPC key grammar must be rejected");
    };
    assert!(err.to_string().contains("x+proof"), "{err}");
}