okhttp: fix header scheme does not match transport type. (#6260)

okhttp: fix header scheme does not match transport type.

okhttp/src/main/java/io/grpc/okhttp/Headers.java

@@ -34,7 +34,8 @@ import okio.ByteString;
  */
 class Headers {
 
-  public static final Header SCHEME_HEADER = new Header(Header.TARGET_SCHEME, "https");
+  public static final Header HTTPS_SCHEME_HEADER = new Header(Header.TARGET_SCHEME, "https");
+  public static final Header HTTP_SCHEME_HEADER = new Header(Header.TARGET_SCHEME, "http");
   public static final Header METHOD_HEADER = new Header(Header.TARGET_METHOD, GrpcUtil.HTTP_METHOD);
   public static final Header METHOD_GET_HEADER = new Header(Header.TARGET_METHOD, "GET");
   public static final Header CONTENT_TYPE_HEADER =
@@ -47,7 +48,12 @@ class Headers {
    * application thread context.
    */
   public static List<Header> createRequestHeaders(
-      Metadata headers, String defaultPath, String authority, String userAgent, boolean useGet) {
+      Metadata headers,
+      String defaultPath,
+      String authority,
+      String userAgent,
+      boolean useGet,
+      boolean usePlaintext) {
     Preconditions.checkNotNull(headers, "headers");
     Preconditions.checkNotNull(defaultPath, "defaultPath");
     Preconditions.checkNotNull(authority, "authority");
@@ -61,7 +67,11 @@ class Headers {
     List<Header> okhttpHeaders = new ArrayList<>(7 + InternalMetadata.headerCount(headers));
 
     // Set GRPC-specific headers.
-    okhttpHeaders.add(SCHEME_HEADER);
+    if (usePlaintext) {
+      okhttpHeaders.add(HTTP_SCHEME_HEADER);
+    } else {
+      okhttpHeaders.add(HTTPS_SCHEME_HEADER);
+    }
     if (useGet) {
       okhttpHeaders.add(METHOD_GET_HEADER);
     } else {

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java

@@ -408,7 +408,14 @@ class OkHttpClientStream extends AbstractClientStream {
 
     @GuardedBy("lock")
     private void streamReady(Metadata metadata, String path) {
-      requestHeaders = Headers.createRequestHeaders(metadata, path, authority, userAgent, useGet);
+      requestHeaders =
+          Headers.createRequestHeaders(
+              metadata,
+              path,
+              authority,
+              userAgent,
+              useGet,
+              transport.isUsingPlaintext());
       transport.streamReadyToStart(OkHttpClientStream.this);
     }
 

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -315,6 +315,11 @@ class OkHttpClientTransport implements ConnectionClientTransport, TransportExcep
     initTransportTracer();
   }
 
+  // sslSocketFactory is set to null when use plaintext.
+  boolean isUsingPlaintext() {
+    return sslSocketFactory == null;
+  }
+
   private void initTransportTracer() {
     synchronized (lock) { // to make @GuardedBy linter happy
       transportTracer.setFlowControlWindowReader(new TransportTracer.FlowControlReader() {

okhttp/src/test/java/io/grpc/okhttp/HeadersTest.java

@@ -53,6 +53,7 @@ public class HeadersTest {
         path,
         authority,
         userAgent,
+        false,
         false);
 
     // 7 reserved headers, 1 user header

okhttp/src/test/java/io/grpc/okhttp/OkHttpClientStreamTest.java

@@ -25,6 +25,7 @@ import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 
 import com.google.common.io.BaseEncoding;
 import io.grpc.CallOptions;
@@ -182,7 +183,31 @@ public class OkHttpClientStreamTest {
     verify(mockedFrameWriter)
         .synStream(eq(false), eq(false), eq(3), eq(0), headersCaptor.capture());
     assertThat(headersCaptor.getValue()).containsExactly(
-        Headers.SCHEME_HEADER,
+        Headers.HTTPS_SCHEME_HEADER,
+        Headers.METHOD_HEADER,
+        new Header(Header.TARGET_AUTHORITY, "localhost"),
+        new Header(Header.TARGET_PATH, "/" + methodDescriptor.getFullMethodName()),
+        new Header(GrpcUtil.USER_AGENT_KEY.name(), "good-application"),
+        Headers.CONTENT_TYPE_HEADER,
+        Headers.TE_HEADER)
+            .inOrder();
+  }
+
+  @Test
+  public void start_headerPlaintext() throws IOException {
+    Metadata metaData = new Metadata();
+    metaData.put(GrpcUtil.USER_AGENT_KEY, "misbehaving-application");
+    when(transport.isUsingPlaintext()).thenReturn(true);
+    stream = new OkHttpClientStream(methodDescriptor, metaData, frameWriter, transport,
+        flowController, lock, MAX_MESSAGE_SIZE, INITIAL_WINDOW_SIZE, "localhost",
+        "good-application", StatsTraceContext.NOOP, transportTracer, CallOptions.DEFAULT, false);
+    stream.start(new BaseClientStreamListener());
+    stream.transportState().start(3);
+
+    verify(mockedFrameWriter)
+        .synStream(eq(false), eq(false), eq(3), eq(0), headersCaptor.capture());
+    assertThat(headersCaptor.getValue()).containsExactly(
+        Headers.HTTP_SCHEME_HEADER,
         Headers.METHOD_HEADER,
         new Header(Header.TARGET_AUTHORITY, "localhost"),
         new Header(Header.TARGET_PATH, "/" + methodDescriptor.getFullMethodName()),

okhttp/src/test/java/io/grpc/okhttp/OkHttpClientTransportTest.java

@@ -22,8 +22,8 @@ import static io.grpc.internal.ClientStreamListener.RpcProgress.PROCESSED;
 import static io.grpc.internal.ClientStreamListener.RpcProgress.REFUSED;
 import static io.grpc.internal.GrpcUtil.DEFAULT_MAX_MESSAGE_SIZE;
 import static io.grpc.okhttp.Headers.CONTENT_TYPE_HEADER;
+import static io.grpc.okhttp.Headers.HTTP_SCHEME_HEADER;
 import static io.grpc.okhttp.Headers.METHOD_HEADER;
-import static io.grpc.okhttp.Headers.SCHEME_HEADER;
 import static io.grpc.okhttp.Headers.TE_HEADER;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -650,7 +650,7 @@ public class OkHttpClientTransportTest {
     stream.start(listener);
     Header userAgentHeader = new Header(GrpcUtil.USER_AGENT_KEY.name(),
             GrpcUtil.getGrpcUserAgent("okhttp", null));
-    List<Header> expectedHeaders = Arrays.asList(SCHEME_HEADER, METHOD_HEADER,
+    List<Header> expectedHeaders = Arrays.asList(HTTP_SCHEME_HEADER, METHOD_HEADER,
             new Header(Header.TARGET_AUTHORITY, "notarealauthority:80"),
             new Header(Header.TARGET_PATH, "/" + method.getFullMethodName()),
             userAgentHeader, CONTENT_TYPE_HEADER, TE_HEADER);
@@ -667,7 +667,7 @@ public class OkHttpClientTransportTest {
     OkHttpClientStream stream =
         clientTransport.newStream(method, new Metadata(), CallOptions.DEFAULT);
     stream.start(listener);
-    List<Header> expectedHeaders = Arrays.asList(SCHEME_HEADER, METHOD_HEADER,
+    List<Header> expectedHeaders = Arrays.asList(HTTP_SCHEME_HEADER, METHOD_HEADER,
         new Header(Header.TARGET_AUTHORITY, "notarealauthority:80"),
         new Header(Header.TARGET_PATH, "/" + method.getFullMethodName()),
         new Header(GrpcUtil.USER_AGENT_KEY.name(),