Forwarding IP for submit_tx works

core/bin/zksync_api/Cargo.toml

@@ -44,7 +44,7 @@ jsonrpc-core-client = {version= "18", features=["ws", "http"]}
 jsonrpc-pubsub = "18.0.0"
 jsonrpc-ws-server = "18"
 jsonrpc-http-server = "18"
-hyper = "0.14.14"
+hyper = {version="0.14.14", features=["stream"]}
 jsonrpc-derive = "18"
 qstring = "0.7.2"
 

core/bin/zksync_api/src/api_server/rpc_server/mod.rs

@@ -1,10 +1,10 @@
 // Built-in uses
-use std::time::Instant;
+use std::{convert::TryFrom, time::Instant};
 
 // External uses
 use futures::{
     channel::{mpsc, oneshot},
-    SinkExt,
+    FutureExt, SinkExt, StreamExt,
 };
 use jsonrpc_core::{Error, IoHandler, MetaIoHandler, Metadata, Middleware, Result};
 use jsonrpc_http_server::{RequestMiddleware, RequestMiddlewareAction, ServerBuilder};
@@ -91,16 +91,104 @@ impl RpcApp {
 
 struct IpInsertMiddleWare {}
 
+async fn insert_ip(body: hyper::Body, ip: String) -> hyper::Result<Vec<u8>> {
+    let body_stream: Vec<_> = body.collect().await;
+
+    let body_bytes: hyper::Result<Vec<_>> = body_stream.into_iter().collect();
+    let body_bytes = (body_bytes)?;
+
+    let mut body_bytes: Vec<u8> = body_bytes
+        .into_iter()
+        .map(|b| {
+            let d: Vec<u8> = b.into_iter().collect();
+            d
+        })
+        .flatten()
+        .collect();
+
+    let tmp_str = String::from_utf8(body_bytes.clone());
+    if let Ok(s) = tmp_str {
+        dbg!("HAHA");
+        dbg!(s.clone());
+
+        let call: std::result::Result<jsonrpc_core::MethodCall, _> = serde_json::from_str(&s); //.map_err(|e| hyper::Error)
+
+        let new_call: Option<jsonrpc_core::MethodCall> = if let Ok(call) = call {
+            if call.method == "tx_submit" {
+                dbg!("TXXXXXXXX SUBMITTTTTTT");
+
+                let mut new_call = call.clone();
+                match call.params {
+                    jsonrpc_core::Params::Array(params) => {
+                        dbg!("ARRAY");
+                        dbg!(params.len());
+                        let mut new_params = params.clone();
+                        // The IP param can only be set by the middleware
+                        if params.len() > 3 {
+                            dbg!("MORE THAN 3");
+                            return Ok(vec![]);
+                        }
+
+                        if params.len() == 2 {
+                            // fast processing
+                            dbg!("EXACTLY 2");
+                            new_params.push(serde_json::Value::Null);
+                        }
+
+                        if params.len() == 3 {
+                            dbg!("DOING FOR 3");
+                            new_params.push(serde_json::Value::String(ip));
+                        }
+
+                        new_call.params = jsonrpc_core::Params::Array(new_params);
+                    }
+                    _ => {
+                        dbg!("OMG");
+                    }
+                };
+                Some(new_call)
+            } else {
+                None
+            }
+        } else {
+            None
+        };
+
+        if let Some(call) = new_call {
+            let new_body_bytes = serde_json::to_string(&call);
+            if let Ok(s) = new_body_bytes {
+                dbg!("HERE IS THE NEW CALL");
+                dbg!(s.clone());
+                body_bytes = s.as_bytes().to_owned();
+            }
+        }
+    } else {
+        dbg!("CANT UTF-8...");
+    }
+
+    Ok(body_bytes)
+}
+
 impl RequestMiddleware for IpInsertMiddleWare {
     fn on_request(&self, request: hyper::Request<hyper::Body>) -> RequestMiddlewareAction {
         let (parts, body) = request.into_parts();
+        let cloudflare_sent_ip = "CF-Connecting-IP";
+
+        let remote_ip = match parts.headers.get(cloudflare_sent_ip) {
+            Some(ip) => ip.to_str(),
+            None => {
+                dbg!("NO CF IP HEADER, DEFAULTING TO NULL");
+                Ok("")
+            }
+        };
+        let remote_ip = "NEW_IP".to_owned();
+
+        let body_bytes = insert_ip(body, remote_ip).into_stream();
+        let body = hyper::Body::wrap_stream(body_bytes);
 
         RequestMiddlewareAction::Proceed {
-            /// Should the request be processed even if invalid CORS headers are detected?
-            /// This allows for side effects to take place.
-            should_continue_on_invalid_cors: true,
-            /// The request object returned
-            request,
+            should_continue_on_invalid_cors: false,
+            request: hyper::Request::from_parts(parts, body),
         }
     }
 }

core/bin/zksync_api/src/api_server/rpc_server/rpc_impl.rs

@@ -128,6 +128,12 @@ impl RpcApp {
         ip: Option<String>,
     ) -> Result<TxHash> {
         let start = Instant::now();
+
+        if let Some(ip) = ip {
+            dbg!("IPPPPPPPPPPPPPPP");
+            dbg!(ip);
+        }
+
         let result = self
             .tx_sender
             .submit_tx_with_separate_fp(*tx, *signature, fast_processing)