http1.1 protocol downgrade

Version
hyper 0.14.18

Platform
eulerosv2r7.x86_64

Description

The client sends an HTTP1.1 request to the hyper server, and the state.version of the hyper server is HTTP1.1. The hyper server sends the request to the hyper client, and the state.version of the hyper client is HTTP1.1. Finally, the hyper client sends the request to the server. If the HTTP protocol of the response returned by the server is HTTP1.0 and the response is in the keepalive state, the state.version of the hyper client is HTTP1.0. If HTTP1.1 requests are received later, the HyperClient sends the requests to the server using HTTP1.0.

Related Code：

    // If we know the remote speaks an older version, we try to fix up any messages
    // to work with our older peer.
    fn enforce_version(&mut self, head: &mut MessageHead<T::Outgoing>) {
        if let Version::HTTP_10 = self.state.version {
            // Fixes response or connection when keep-alive header is not present
            self.fix_keep_alive(head);
            // If the remote only knows HTTP/1.0, we should force ourselves
            // to do only speak HTTP/1.0 as well.
            head.version = Version::HTTP_10;
        }
        // If the remote speaks HTTP/1.1, then it *should* be fine with
        // both HTTP/1.0 and HTTP/1.1 from us. So again, we just let
        // the user's headers be.
    }

pub(super) fn poll_read_head() {
        ...
        self.state.keep_alive &= msg.keep_alive;
        self.state.version = msg.head.version;
        ...
}

demo:

demo code:
client microservice

use std::time::Duration;
use hyper::{Body, Request, Version};
use hyper::header::CONNECTION;

#[tokio::main]
async fn main() {
    let mut i = 0;
    loop {
        let client = hyper::client::Client::new();
        i += 1;
        let mut builder = Request::builder().header(CONNECTION, "keep-alive");
        let req = if i % 2 == 0 {
            builder.uri("http://127.0.0.1:8081/yyyyyyyyyyyyyyyyyyyyy").version(Version::HTTP_10).body(Body::empty()).unwrap()
        } else {
            builder.uri("http://127.0.0.1:8081/xxxxxxxxxxxxxxxxxxxxx").version(Version::HTTP_11).body(Body::empty()).unwrap()
        };
        println!("req: {:?}", req);
        let response = client.request(req).await.unwrap();
        println!("response: {:?}", response);
        println!("------------------------------");
        tokio::time::sleep(Duration::from_secs(5)).await;
    }
}

hyper server

use cbb::proxy;

#[tokio::main]
async fn main() {
    println!("BE");
    let listen = "0.0.0.0:8082".parse().unwrap();
    let client = "127.0.0.1:8083".parse().unwrap();
    proxy(listen, client).await;
}

hyper client

use cbb::proxy;

#[tokio::main]
async fn main() {
    println!("BE");
    let listen = "0.0.0.0:8082".parse().unwrap();
    let client = "127.0.0.1:8083".parse().unwrap();
    proxy(listen, client).await;
}

server microservice

use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use cbb::simple_listener;

#[tokio::main]
async fn main() {
    println!("B");
    let listen: SocketAddr = "127.0.0.1:8083".parse().unwrap();
    // simple_listener(listen).await;
    let listener = TcpListener::bind(listen).await.unwrap();
    loop {
        let (mut stream, _) = listener.accept().await.unwrap();
        println!("new stream");
        tokio::spawn(async move {
            let mut data = [0u8; 1024 * 32];
            while let Ok(size) = stream.read(&mut data).await {
                if size == 0 {
                    continue
                }
                let data = String::from_utf8_lossy(&data[..size]);
                println!("[{}]", data);
                if data.contains("1_0") {
                    stream.write_all(b"HTTP/1.0 200 OK\r\nconnection: keep-alive\r\nContent-Length: 0\r\n\r\n").await.unwrap();
                } else {
                    stream.write_all(b"HTTP/1.1 200 OK\r\nconnection: keep-alive\r\nContent-Length: 0\r\n\r\n").await.unwrap();
                }
            }
        });
    }
}

lib

use std::net::SocketAddr;
use hyper::{Body, Response};
use hyper::server::conn::AddrStream;

pub async fn proxy(listen: SocketAddr, c_addr: SocketAddr) {
    use hyper::Error;
    use hyper::service::{make_service_fn, service_fn};
// And a MakeService to handle each connection...
    let client = hyper::client::Client::new();
    let make_svc = make_service_fn(|io: &AddrStream| {
        println!("new conn: {:?}", io.remote_addr());
        let client = client.clone();
        async move {
            let client = client.clone();
            Ok::<_, Error>(service_fn(move |mut req| {
                let client = client.clone();
                async move {
                    println!("request: {:?}", req);
                    let url = format!("http://{}{}", c_addr, req.uri().path());
                    *req.uri_mut() = url.parse().unwrap();
                    let result = client.request(req).await;
                    println!("response: {:?}", result);
                    println!();
                    result
                }
            }))
        }
    });
    hyper::server::Server::bind(&listen).serve(make_svc).await.unwrap();
}


pub async fn simple_listener(listen: SocketAddr) {
    use hyper::Error;
    use hyper::service::{make_service_fn, service_fn};
// And a MakeService to handle each connection...
    let make_svc = make_service_fn(|_| {
        async move {
            Ok::<_, Error>(service_fn(move | req| {
                async move {
                    println!("{:?}",req);
                    Ok::<_, Error>(Response::new(Body::from("hello")))
                }
            }))
        }
    });
    hyper::server::Server::bind(&listen).serve(make_svc).await.unwrap();
}

When the client microservice sends HTTP1.0 and HTTP1.1 requests in turn, the Hyper Server sends the requests to the Hyper Client. The HTTP protocol of the Hyper Client is gradually set to HTTP1.0. As a result, the HTTP protocol of subsequent requests is changed to HTTP1.0
That doesn't seem to be the outcome we want.

Start by tracing the state updates in poll_read_head and the version handling in enforce_version, using the supplied alternating HTTP/1.0 and HTTP/1.1 proxy reproduction. Check how a response version affects later requests on the same connection. Done means alternating requests retain their intended protocol versions rather than subsequent HTTP/1.1 requests being downgraded.