#[macro_use]
mod common;
use std::convert::Infallible;
use std::future::{ready, Ready};
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use axum::body::Body;
use http::{Method, StatusCode};
use prost_reflect::{DescriptorPool, DynamicMessage, MessageDescriptor, Value as PbValue};
use serde_json::{json, Value};
use structured_proxy::hooks::{ExtraRoute, ExtraRouteHandler, RouteRequest, RouteResponse};
use structured_proxy::transcode::codec::DynamicCodec;
use tower::ServiceExt;
const OPS_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg {
string name = 1;
string operation = 2;
string rpc = 3;
}
service Ops {
rpc Get(Msg) returns (Msg) {
option (google.api.http) = { get: "/v2/ops/{operation}" };
}
rpc Cancel(Msg) returns (Msg) {
option (google.api.http) = { post: "/v2/ops/{operation}:cancel" };
}
rpc Archive(Msg) returns (Msg) {
option (google.api.http) = { post: "/v2/ops/{operation}:archive" };
}
rpc Pause(Msg) returns (Msg) {
option (google.api.http) = { post: "/v2/ops/{name}:pause" };
}
rpc Drop(Msg) returns (Msg) {
option (google.api.http) = { delete: "/v2/ops/{name}" };
}
rpc Restore(Msg) returns (Msg) {
option (google.api.http) = { post: "/v1/{name=publishers/*/books/*}:restore" };
}
rpc Batch(Msg) returns (Msg) {
option (google.api.http) = { post: "/v1/nodes:batch" };
}
rpc Broken(Msg) returns (Msg) {
option (google.api.http) = { get: "/v1/broken/a{name}b" };
}
rpc Make(Msg) returns (Msg) {
option (google.api.http) = { post: "/v3/x/{name}" };
}
rpc Sweep(Msg) returns (Msg) {
option (google.api.http) = { post: "/v3/{operation}/{name}:sweep" };
}
rpc Purge(Msg) returns (Msg) {
option (google.api.http) = { post: "/v4/{name=**}:purge" };
}
rpc RunNow(Msg) returns (Msg) {
option (google.api.http) = { post: "/v2/jobs/{name}:run%3Anow" };
}
rpc RunSlash(Msg) returns (Msg) {
option (google.api.http) = { post: "/v6/{name=**}:run%2Fnow" };
}
rpc Fixed(Msg) returns (Msg) {
option (google.api.http) = { get: "/v7/{name=foo%2Fbar}" };
}
rpc FixedMid(Msg) returns (Msg) {
option (google.api.http) = { get: "/v8/{name=foo%2Fbar}/x" };
}
rpc Literal(Msg) returns (Msg) {
option (google.api.http) = { get: "/v9/{name=foo%2Fbar/*}/x" };
}
rpc Shelve(Msg) returns (Msg) {
option (google.api.http) = { post: "/v5/{name=publishers/*}/books/{operation}" };
}
}
"#;
const PEEK_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg {
string name = 1;
string operation = 2;
string rpc = 3;
}
service Ops {
rpc Peek(Msg) returns (Msg) {
option (google.api.http) = { custom: { kind: "HEAD", path: "/v1/peek" } };
}
}
"#;
const RANKED_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg {
string name = 1;
string operation = 2;
string rpc = 3;
}
service Ops {
rpc Special(Msg) returns (Msg) {
option (google.api.http) = { get: "/v1/a/b/{name=special}" };
}
rpc Below(Msg) returns (Msg) {
option (google.api.http) = { get: "/v1/{name}/b/{operation}" };
}
}
"#;
#[derive(Clone)]
struct Echo {
rpc: String,
}
impl tonic::server::UnaryService<DynamicMessage> for Echo {
type Response = DynamicMessage;
type Future = Ready<Result<tonic::Response<DynamicMessage>, tonic::Status>>;
fn call(&mut self, request: tonic::Request<DynamicMessage>) -> Self::Future {
let mut msg = request.into_inner();
msg.set_field_by_name("rpc", PbValue::String(self.rpc.clone()));
ready(Ok(tonic::Response::new(msg)))
}
}
#[derive(Clone)]
struct Ops {
msg: MessageDescriptor,
}
impl tonic::server::NamedService for Ops {
const NAME: &'static str = "test.v1.Ops";
}
impl tower::Service<http::Request<tonic::body::Body>> for Ops {
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 msg = self.msg.clone();
Box::pin(async move {
let rpc = req
.uri()
.path()
.strip_prefix("/test.v1.Ops/")
.expect("an Ops method")
.to_owned();
let mut grpc = tonic::server::Grpc::new(DynamicCodec::new(msg));
Ok(grpc.unary(Echo { rpc }, req).await)
})
}
}
struct Ping;
#[async_trait::async_trait]
impl ExtraRouteHandler for Ping {
async fn handle(&self, _req: RouteRequest) -> RouteResponse {
RouteResponse::new(StatusCode::OK, bytes::Bytes::from_static(b"pong"))
}
}
async fn proxy(upstream: common::Upstream) -> common::App {
let pool: DescriptorPool = common::compile("test/v1/ops.proto", OPS_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
common::proxy(upstream, Ops { msg }, pool, Default::default()).await
}
async fn call(app: &common::App, method: Method, uri: &str) -> (StatusCode, String, Value) {
let request = http::Request::builder()
.method(method)
.uri(uri)
.body(Body::empty())
.unwrap();
let resp = app.clone().oneshot(request).await.unwrap();
let status = resp.status();
let allow = resp
.headers()
.get(http::header::ALLOW)
.map(|v| v.to_str().unwrap().to_owned())
.unwrap_or_default();
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
let body = if bytes.is_empty() {
Value::Null
} else {
serde_json::from_slice(&bytes).unwrap()
};
(status, allow, body)
}
upstream_tests! {
async fn verb_after_a_variable_routes_with_the_variable_bound() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::POST, "/v2/ops/op-1:cancel").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body, json!({"name": "", "operation": "op-1", "rpc": "Cancel"}));
}
async fn bindings_differing_only_by_verb_reach_their_own_methods() {
let app = proxy(UPSTREAM).await;
for (verb, rpc) in [("cancel", "Cancel"), ("archive", "Archive")] {
let (status, _, body) =
call(&app, Method::POST, &format!("/v2/ops/op-7:{verb}")).await;
assert_eq!(status, StatusCode::OK, "{verb}: {body}");
assert_eq!(body["rpc"], rpc, "{verb}");
assert_eq!(body["operation"], "op-7", "{verb}");
}
let (status, _, body) = call(&app, Method::POST, "/v2/ops/op-7:pause").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body, json!({"name": "op-7", "operation": "", "rpc": "Pause"}));
}
async fn a_request_without_the_verb_does_not_reach_a_verb_binding() {
let app = proxy(UPSTREAM).await;
let (status, allow, _) = call(&app, Method::POST, "/v2/ops/op-1").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(allow, "GET, DELETE, HEAD");
let (status, _, body) = call(&app, Method::GET, "/v2/ops/op-1").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Get");
assert_eq!(body["operation"], "op-1");
}
async fn bindings_of_one_path_bind_their_own_variable_names() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::DELETE, "/v2/ops/op-3").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body, json!({"name": "op-3", "operation": "", "rpc": "Drop"}));
}
async fn head_is_answered_by_the_get_binding_without_a_body() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::HEAD, "/v2/ops/op-1").await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body, Value::Null);
}
async fn another_verb_does_not_reach_a_verb_binding() {
let app = proxy(UPSTREAM).await;
let (status, allow, _) = call(&app, Method::POST, "/v2/ops/op-1:delete").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(allow, "GET, DELETE, HEAD");
let (status, _, body) = call(&app, Method::GET, "/v2/ops/op-1:delete").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Get");
assert_eq!(body["operation"], "op-1:delete");
}
async fn an_encoded_colon_is_data_not_a_verb() {
let app = proxy(UPSTREAM).await;
let (status, _, _) = call(&app, Method::POST, "/v2/ops/op-1%3Acancel").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
let (status, _, body) = call(&app, Method::GET, "/v2/ops/op-1%3Acancel").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["operation"], "op-1:cancel");
}
async fn a_verb_with_an_empty_variable_does_not_match() {
let app = proxy(UPSTREAM).await;
let (status, _, _) = call(&app, Method::POST, "/v2/ops/:cancel").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
}
async fn a_bound_verb_owns_its_url_for_every_method() {
let app = proxy(UPSTREAM).await;
for method in [Method::GET, Method::PUT, Method::DELETE] {
let (status, allow, _) = call(&app, method.clone(), "/v2/ops/op-1:cancel").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED, "{method}");
assert_eq!(allow, "POST", "{method}");
}
}
async fn the_verb_starts_at_the_first_colon_of_the_last_segment() {
let app = proxy(UPSTREAM).await;
let (status, allow, _) = call(&app, Method::POST, "/v2/ops/op-1:b:cancel").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(allow, "GET, DELETE, HEAD");
let (status, _, body) = call(&app, Method::GET, "/v2/ops/op-1:b:cancel").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["operation"], "op-1:b:cancel");
}
async fn a_verb_of_another_path_shape_is_found() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::POST, "/v3/x/y:sweep").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body, json!({"name": "y", "operation": "x", "rpc": "Sweep"}));
let (status, _, body) = call(&app, Method::POST, "/v3/a/b:sweep").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body, json!({"name": "b", "operation": "a", "rpc": "Sweep"}));
let (status, _, body) = call(&app, Method::POST, "/v3/x/y").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body, json!({"name": "y", "operation": "", "rpc": "Make"}));
}
async fn a_double_wildcard_before_a_verb_may_match_no_segment() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::POST, "/v4/:purge").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Purge");
assert_eq!(body["name"], "");
let (status, _, body) = call(&app, Method::POST, "/v4/a/b:purge").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["name"], "a/b");
}
async fn an_encoded_character_in_the_verb_is_taken_off_decoded() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::POST, "/v2/jobs/j1:run%3Anow").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "RunNow");
assert_eq!(body["name"], "j1");
}
async fn a_variable_over_several_segments_keeps_an_encoded_slash() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::POST, "/v5/publishers/a%2Fb%20c/books/x%2Fy").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Shelve");
assert_eq!(body["name"], "publishers/a%2Fb c");
assert_eq!(body["operation"], "x/y");
let (status, _, body) = call(&app, Method::POST, "/v4/a%2fb%20c/d:purge").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["name"], "a%2fb c/d");
let (status, _, body) = call(&app, Method::POST, "/v1/publishers/p%2F1/books/b2:restore").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["name"], "publishers/p%2F1/books/b2");
let (status, _, body) = call(&app, Method::POST, "/v6/a/b:run%2Fnow").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "RunSlash");
assert_eq!(body["name"], "a/b");
let (status, _, body) = call(&app, Method::POST, "/v6/a/b:run%2fnow").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["name"], "a/b");
let (status, _, body) = call(&app, Method::GET, "/v9/foo%2fbar/p1/x").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Literal");
assert_eq!(body["name"], "foo%2fbar/p1");
}
async fn a_field_template_of_one_segment_is_decoded_in_full() {
let app = proxy(UPSTREAM).await;
for (uri, rpc) in [("/v7/foo%2Fbar", "Fixed"), ("/v8/foo%2fbar/x", "FixedMid")] {
let (status, _, body) = call(&app, Method::GET, uri).await;
assert_eq!(status, StatusCode::OK, "{uri}: {body}");
assert_eq!(body["rpc"], rpc, "{uri}");
assert_eq!(body["name"], "foo/bar", "{uri}");
}
}
async fn verb_after_a_multi_segment_field_template_routes() {
let app = proxy(UPSTREAM).await;
let (status, _, body) =
call(&app, Method::POST, "/v1/publishers/p1/books/b2:restore").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Restore");
assert_eq!(body["name"], "publishers/p1/books/b2");
for uri in ["/v1/publishers/p1/books/b2", "/v1/publishers/p1/books/b2:archive"] {
let (status, _, body) = call(&app, Method::POST, uri).await;
assert_eq!(status, StatusCode::NOT_FOUND, "{uri}: {body}");
}
for uri in [
"/v1/anything/else:restore",
"/v1/publishers/p1/shelves/b2:restore",
"/v1/publishers/p1/books:restore",
"/v1/publishers/p1/books/b2/c3:restore",
] {
let (status, _, body) = call(&app, Method::POST, uri).await;
assert_eq!(status, StatusCode::NOT_FOUND, "{uri}: {body}");
}
}
async fn verb_after_a_literal_routes_as_before() {
let app = proxy(UPSTREAM).await;
let (status, _, body) = call(&app, Method::POST, "/v1/nodes:batch").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Batch");
for uri in ["/v1/nodes", "/v1/nodes:other"] {
let (status, _, body) = call(&app, Method::POST, uri).await;
assert_eq!(status, StatusCode::NOT_FOUND, "{uri}: {body}");
}
}
async fn an_extra_route_shares_a_transcoded_path_on_another_method() {
let pool: DescriptorPool = common::compile("test/v1/ops.proto", OPS_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::GET,
"/v1/nodes:batch",
Arc::new(Ping),
)])
})
.await;
let request = http::Request::get("/v1/nodes:batch").body(Body::empty()).unwrap();
let (status, body) = common::send(&app, request).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body, "pong");
let (status, _, body) = call(&app, Method::POST, "/v1/nodes:batch").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Batch");
}
async fn a_refused_binding_passes_to_the_next_route_in_rank_not_the_next_transcoded() {
let pool: DescriptorPool = common::compile("test/v1/ranked.proto", RANKED_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::GET,
"/v1/a/{kind}/{id}",
Arc::new(Ping),
)])
})
.await;
let request = http::Request::get("/v1/a/b/ordinary")
.body(Body::empty())
.unwrap();
assert_eq!(
common::send(&app, request).await,
(StatusCode::OK, "pong".to_owned())
);
let (status, _, body) = call(&app, Method::GET, "/v1/a/b/special").await;
assert_eq!((status, &body["rpc"]), (StatusCode::OK, &json!("Special")));
let (status, _, body) = call(&app, Method::GET, "/v1/x/b/y").await;
assert_eq!((status, &body["rpc"]), (StatusCode::OK, &json!("Below")));
}
async fn an_extra_route_of_a_method_the_router_skips_takes_no_request() {
let pool: DescriptorPool = common::compile("test/v1/ranked.proto", RANKED_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::from_bytes(b"PURGE").unwrap(),
"/v1/x/b/y",
Arc::new(Ping),
)])
})
.await;
let (status, _, body) = call(&app, Method::GET, "/v1/x/b/y").await;
assert_eq!((status, &body["rpc"]), (StatusCode::OK, &json!("Below")));
}
async fn a_405_past_a_refusing_path_allows_the_methods_at_the_answering_one() {
let pool: DescriptorPool = common::compile("test/v1/ranked.proto", RANKED_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::PUT,
"/v1/{name}/b/{operation}",
Arc::new(Ping),
)])
})
.await;
let (status, allow, _) = call(&app, Method::DELETE, "/v1/a/b/ordinary").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(allow, "GET, HEAD, PUT");
}
async fn a_405_allows_the_methods_of_an_extra_route_at_the_same_path() {
let pool: DescriptorPool = common::compile("test/v1/ranked.proto", RANKED_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::POST,
"/v1/{name}/b/{operation}",
Arc::new(Ping),
)])
})
.await;
let (status, allow, _) = call(&app, Method::DELETE, "/v1/x/b/y").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(allow, "GET, HEAD, POST");
}
async fn a_successful_connect_answer_has_no_length_and_no_content() {
const TUNNEL_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; string operation = 2; string rpc = 3; }
service Ops {
rpc Tunnel(Msg) returns (Msg) {
option (google.api.http) = { custom: { kind: "CONNECT", path: "/v1/tunnel" } };
}
}
"#;
let pool: DescriptorPool = common::compile("test/v1/tunnel.proto", TUNNEL_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
})
.await;
let request = http::Request::builder()
.method(Method::CONNECT)
.uri("/v1/tunnel")
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert!(response.headers().get("content-length").is_none());
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert!(body.is_empty(), "{body:?}");
}
async fn a_transcoded_head_binding_keeps_head_beside_an_extra_get() {
let pool: DescriptorPool = common::compile("test/v1/peek.proto", PEEK_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(Method::GET, "/v1/peek", Arc::new(Ping))])
})
.await;
let request = http::Request::head("/v1/peek").body(Body::empty()).unwrap();
let response = app.clone().oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["content-type"], "application/json");
let request = http::Request::get("/v1/peek").body(Body::empty()).unwrap();
assert_eq!(
common::send(&app, request).await,
(StatusCode::OK, "pong".to_owned())
);
}
async fn a_request_goes_where_one_router_of_every_route_sends_it() {
let pool: DescriptorPool = common::compile("test/v1/ops.proto", OPS_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let app = common::app(UPSTREAM, Ops { msg }, |yaml| {
structured_proxy::ProxyServer::from_yaml_str(yaml)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([
ExtraRoute::new(Method::GET, "/v3/x/ping", Arc::new(Ping)),
ExtraRoute::new(Method::GET, "/v1/nodes:batch", Arc::new(Ping)),
])
})
.await;
let request = http::Request::get("/v3/x/ping").body(Body::empty()).unwrap();
assert_eq!(
common::send(&app, request).await,
(StatusCode::OK, "pong".to_owned())
);
let (status, _, _) = call(&app, Method::POST, "/v3/x/ping").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
let (status, _, body) = call(&app, Method::POST, "/v3/x/pong").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Make");
let request = http::Request::head("/v1/nodes:batch")
.body(Body::empty())
.unwrap();
assert_eq!(
common::send(&app, request).await,
(StatusCode::OK, String::new())
);
}
async fn a_template_the_router_cannot_serve_is_skipped_not_fatal() {
let app = proxy(UPSTREAM).await;
let request = http::Request::get("/v1/broken/axb").body(Body::empty()).unwrap();
let (status, _) = common::send(&app, request).await;
assert_eq!(status, StatusCode::NOT_FOUND);
let (status, _, body) = call(&app, Method::GET, "/v2/ops/op-1").await;
assert_eq!(status, StatusCode::OK, "{body}");
}
}
#[tokio::test]
async fn a_verb_bound_twice_for_one_method_is_refused_at_startup() {
const TWICE_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; string id = 2; }
service Twice {
rpc A(Msg) returns (Msg) { option (google.api.http) = { post: "/v1/x/{name}:cancel" }; }
rpc B(Msg) returns (Msg) { option (google.api.http) = { post: "/v1/x/{id}:cancel" }; }
}
"#;
let pool = common::compile("test/v1/twice.proto", TWICE_PROTO);
let err = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.router()
.expect_err("a repeated method, path and verb must be rejected");
assert!(err.to_string().contains("more than one endpoint"), "{err}");
}
#[tokio::test]
async fn an_extra_route_the_router_cannot_hold_beside_a_transcoded_one_is_refused() {
const ID_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; string id = 2; }
service Ids {
rpc Get(Msg) returns (Msg) { option (google.api.http) = { get: "/v1/{id}" }; }
}
"#;
let pool = common::compile("test/v1/ids.proto", ID_PROTO);
let err = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(Method::GET, "/v1/{*path}", Arc::new(Ping))])
.router()
.expect_err("routes the router cannot hold together must be rejected");
assert!(
err.to_string().contains("conflicts with another route"),
"{err}"
);
}
#[tokio::test]
async fn a_transcoded_route_the_router_skips_reserves_no_path() {
const SKIPPED_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string path = 1; string id = 2; }
service Ids {
rpc Get(Msg) returns (Msg) { option (google.api.http) = { get: "/v1/{id}" }; }
rpc Put(Msg) returns (Msg) { option (google.api.http) = { post: "/v1/{path=**}" }; }
}
"#;
let pool = common::compile("test/v1/skipped.proto", SKIPPED_PROTO);
let err = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(Method::POST, "/v1/{*path}", Arc::new(Ping))])
.router()
.expect_err("the extra route conflicts with the mounted `/v1/{id}`");
assert!(
err.to_string().contains("conflicts with another route"),
"{err}"
);
}
#[tokio::test]
async fn an_extra_route_on_a_path_with_verbs_is_refused_at_startup() {
const VERB_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; }
service Verb {
rpc Cancel(Msg) returns (Msg) { option (google.api.http) = { post: "/v1/x/{name}:cancel" }; }
}
"#;
let pool = common::compile("test/v1/verb.proto", VERB_PROTO);
let err = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(Method::GET, "/v1/x/{id}", Arc::new(Ping))])
.router()
.expect_err("an extra route on a path with verbs must be rejected");
assert!(err.to_string().contains("more than one endpoint"), "{err}");
}
#[tokio::test]
async fn a_guard_may_not_scope_a_method_only_a_skipped_route_binds() {
const SKIPPED_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; string id = 2; }
service Dav {
rpc Get(Msg) returns (Msg) { option (google.api.http) = { get: "/v1/{id}" }; }
rpc Inspect(Msg) returns (Msg) {
option (google.api.http) = { custom: { kind: "PROPFIND" path: "/v1/{name=**}" } };
}
}
"#;
let pool = common::compile("test/v1/skipped_dav.proto", SKIPPED_PROTO);
let err = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\nmaintenance:\n enabled: true\n scope:\n traffic: [transcoded]\n methods: [\"PROPFIND\"]\n",
)
.unwrap()
.with_descriptors(pool)
.router()
.expect_err("PROPFIND is answered by no mounted route");
assert!(err.to_string().contains("nor one a route answers"), "{err}");
}
#[tokio::test]
async fn a_guard_may_scope_an_extension_method_bound_with_a_verb() {
const DAV_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; }
service Dav {
rpc Inspect(Msg) returns (Msg) {
option (google.api.http) = { custom: { kind: "PROPFIND" path: "/v1/x/{name}:inspect" } };
}
}
"#;
let pool = common::compile("test/v1/dav.proto", DAV_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\nmaintenance:\n enabled: true\n scope:\n traffic: [transcoded]\n methods: [\"PROPFIND\"]\n",
)
.unwrap()
.with_descriptors(pool)
.router();
assert!(router.is_ok(), "PROPFIND is routed: {router:?}");
}
#[tokio::test]
async fn a_url_no_binding_answers_reaches_the_fallback() {
let pool: DescriptorPool = common::compile("test/v1/ops.proto", OPS_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let fallback = axum::Router::new().fallback(|| async { (StatusCode::IM_A_TEAPOT, "yours") });
let service = structured_proxy::ProxyServer::new()
.with_descriptors(pool)
.service(tonic::service::Routes::new(Ops { msg }))
.unwrap()
.with_fallback(fallback);
let app = common::App::new(service);
for uri in ["/v1/anything/else:restore", "/v2/jobs/j1:other"] {
let request = http::Request::post(uri).body(Body::empty()).unwrap();
let (status, body) = common::send(&app, request).await;
assert_eq!(status, StatusCode::IM_A_TEAPOT, "{uri}");
assert_eq!(body, "yours", "{uri}");
}
let (status, _, _) = call(&app, Method::GET, "/v2/jobs/j1:run%3Anow").await;
assert_eq!(status, StatusCode::METHOD_NOT_ALLOWED);
let (status, _, body) = call(&app, Method::POST, "/v1/publishers/p1/books/b2:restore").await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body["rpc"], "Restore");
}
#[tokio::test]
async fn a_preflight_for_a_url_no_binding_answers_reaches_the_fallback() {
let pool: DescriptorPool = common::compile("test/v1/ops.proto", OPS_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let fallback = axum::Router::new().fallback(|| async { (StatusCode::IM_A_TEAPOT, "yours") });
let service = structured_proxy::ProxyServer::from_yaml_str(
"cors:\n origins: [\"https://app.example\"]\n",
)
.unwrap()
.with_descriptors(pool)
.service(tonic::service::Routes::new(Ops { msg }))
.unwrap()
.with_fallback(fallback);
let app = common::App::new(service);
let request = http::Request::builder()
.method(Method::OPTIONS)
.uri("/v1/anything/else:restore")
.header(http::header::ORIGIN, "https://app.example")
.header(http::header::ACCESS_CONTROL_REQUEST_METHOD, "POST")
.body(Body::empty())
.unwrap();
let (status, body) = common::send(&app, request).await;
assert_eq!(status, StatusCode::IM_A_TEAPOT);
assert_eq!(body, "yours");
}
#[tokio::test]
async fn a_prefix_capture_survives_a_binding_of_another_path() {
const PREFIX_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; string operation = 2; string rpc = 3; }
service Ops {
rpc Get(Msg) returns (Msg) { option (google.api.http) = { get: "/v1/a/{operation}" }; }
rpc Cancel(Msg) returns (Msg) { option (google.api.http) = { post: "/v1/{name=**}:cancel" }; }
}
"#;
let pool: DescriptorPool = common::compile("test/v1/prefix.proto", PREFIX_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let url = common::serve(Ops { msg }).await;
let router =
structured_proxy::ProxyServer::from_yaml_str(&format!("upstream:\n default: \"{url}\"\n"))
.unwrap()
.with_descriptors(pool)
.router()
.unwrap();
let app = axum::Router::new().nest("/{operation}", router);
let request = http::Request::post("/outer/v1/a/x:cancel")
.body(Body::empty())
.unwrap();
let response = app.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let body: Value = serde_json::from_slice(&body).unwrap();
assert_eq!(
(&body["rpc"], &body["name"], &body["operation"]),
(&json!("Cancel"), &json!("a/x"), &json!("outer"))
);
}
#[tokio::test]
async fn bindings_of_one_shape_under_other_names_start_as_one_route() {
const NAMES_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string id = 1; string name = 2; }
service Names {
rpc Get(Msg) returns (Msg) { option (google.api.http) = { get: "/v1/{id}" }; }
rpc Post(Msg) returns (Msg) { option (google.api.http) = { post: "/v1/{name}" }; }
}
"#;
let pool = common::compile("test/v1/names.proto", NAMES_PROTO);
let built = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.router();
assert!(built.is_ok(), "{:?}", built.err());
}
#[tokio::test]
async fn an_extra_route_spelling_a_transcoded_path_otherwise_is_refused() {
let pool = common::compile("test/v1/ranked.proto", RANKED_PROTO);
let err = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::POST,
"/v1/{kind}/b/{id}",
Arc::new(Ping),
)])
.router()
.expect_err("a second spelling of one path must be rejected");
assert!(
err.to_string().contains("conflicts with another route"),
"{err}"
);
}
#[tokio::test]
async fn a_routed_405_allows_the_methods_of_an_extra_route_at_the_same_path() {
let pool = common::compile("test/v1/ranked.proto", RANKED_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::POST,
"/v1/{name}/b/{operation}",
Arc::new(Ping),
)])
.router()
.unwrap();
let request = http::Request::delete("/v1/x/b/y")
.body(Body::empty())
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::METHOD_NOT_ALLOWED);
assert_eq!(response.headers()[http::header::ALLOW], "GET, HEAD, POST");
}
#[tokio::test]
async fn a_routed_refused_binding_passes_to_the_next_route_in_rank() {
let pool = common::compile("test/v1/ranked.proto", RANKED_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(
Method::GET,
"/v1/a/{kind}/{id}",
Arc::new(Ping),
)])
.router()
.unwrap();
let request = http::Request::get("/v1/a/b/ordinary")
.body(Body::empty())
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"pong");
}
#[tokio::test]
async fn a_routed_transcoded_head_binding_keeps_head_beside_an_extra_get() {
let pool = common::compile("test/v1/peek.proto", PEEK_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(Method::GET, "/v1/peek", Arc::new(Ping))])
.router()
.unwrap();
let request = http::Request::head("/v1/peek").body(Body::empty()).unwrap();
let response = router.clone().oneshot(request).await.unwrap();
assert_ne!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["content-type"], "application/json");
let request = http::Request::get("/v1/peek").body(Body::empty()).unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[tokio::test]
async fn a_url_no_binding_answers_reaches_a_route_of_the_proxy_below_it() {
const SPECIAL_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; }
service Special {
rpc Get(Msg) returns (Msg) { option (google.api.http) = { get: "/v1/books/{name=special/*}" }; }
}
"#;
let pool = common::compile("test/v1/special.proto", SPECIAL_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(Method::GET, "/v1/{*path}", Arc::new(Ping))])
.router()
.unwrap();
let request = http::Request::get("/v1/books/ordinary/x")
.body(Body::empty())
.unwrap();
let response = router.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"pong");
}
#[tokio::test]
async fn a_url_no_binding_answers_reaches_a_route_below_it_in_a_nested_router() {
const SPECIAL_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; }
service Special {
rpc Get(Msg) returns (Msg) { option (google.api.http) = { get: "/v1/books/{name=special/*}" }; }
}
"#;
let pool = common::compile("test/v1/special.proto", SPECIAL_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.with_extra_routes([ExtraRoute::new(Method::GET, "/v1/{*path}", Arc::new(Ping))])
.router()
.unwrap();
let app = axum::Router::new().nest("/api", router);
let request = http::Request::get("/api/v1/books/ordinary/x")
.body(Body::empty())
.unwrap();
let response = app.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
assert_eq!(&body[..], b"pong");
}
#[tokio::test]
async fn the_fallback_routes_a_url_no_binding_answers_as_its_own() {
let pool: DescriptorPool = common::compile("test/v1/ops.proto", OPS_PROTO);
let msg = pool.get_message_by_name("test.v1.Msg").unwrap();
let fallback = axum::Router::new().route(
"/v1/{*rest}",
axum::routing::any(
|axum::extract::Path(rest): axum::extract::Path<String>,
path: axum::extract::MatchedPath| async move {
format!("{} {rest}", path.as_str())
},
),
);
let service = structured_proxy::ProxyServer::new()
.with_descriptors(pool)
.service(tonic::service::Routes::new(Ops { msg }))
.unwrap()
.with_fallback(fallback);
let app = common::App::new(service);
let request = http::Request::post("/v1/anything/else:restore")
.body(Body::empty())
.unwrap();
let (status, body) = common::send(&app, request).await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(body, "/v1/{*rest} anything/else:restore");
}
#[tokio::test]
async fn a_guard_may_scope_an_extension_method_a_star_rule_answers() {
const ANY_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; }
service Any {
rpc Inspect(Msg) returns (Msg) {
option (google.api.http) = { custom: { kind: "*" path: "/v1/x/{name}" } };
}
}
"#;
let pool = common::compile("test/v1/any.proto", ANY_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\nmaintenance:\n enabled: true\n scope:\n traffic: [transcoded]\n methods: [\"PROPFIND\"]\n",
)
.unwrap()
.with_descriptors(pool)
.router();
assert!(router.is_ok(), "the `*` rule answers PROPFIND: {router:?}");
}
#[tokio::test]
async fn a_star_rule_with_a_verb_starts_beside_another_verb() {
const STAR_PROTO: &str = r#"
syntax = "proto3";
package test.v1;
import "google/api/annotations.proto";
message Msg { string name = 1; }
service Star {
rpc Inspect(Msg) returns (Msg) {
option (google.api.http) = { custom: { kind: "*" path: "/v1/x/{name}:inspect" } };
}
rpc Cancel(Msg) returns (Msg) { option (google.api.http) = { post: "/v1/x/{name}:cancel" }; }
}
"#;
let pool = common::compile("test/v1/star.proto", STAR_PROTO);
let router = structured_proxy::ProxyServer::from_yaml_str(
"upstream:\n default: \"http://127.0.0.1:1\"\n",
)
.unwrap()
.with_descriptors(pool)
.router();
assert!(
router.is_ok(),
"bindings of different verbs share their path: {router:?}"
);
}