From 770252b476bc342ea08da2bc5b83de713463d14a Mon Sep 17 00:00:00 2001

From: Ivan Devat <idevat@redhat.com>

Date: Thu, 12 Mar 2020 15:32:31 +0100

Subject: [PATCH 1/2] send request from python to ruby more directly

Rack protection middleware is launched before

TornadoCommunicationMiddleware. When request parts are unpacked in

TornadoCommunicationMiddleware they are not checked by rack protection.

This commit changes communication between python and ruby - request is

sent to ruby more directly (without need to unpack request in sinatra

middleware).

---

 pcs/daemon/ruby_pcsd.py                   | 217 ++++++++++++++--------

 pcs_test/tier0/daemon/app/fixtures_app.py |   7 +-

 pcs_test/tier0/daemon/test_ruby_pcsd.py   |  61 ++----

 pcsd/rserver.rb                           |  39 ++--

 4 files changed, 175 insertions(+), 149 deletions(-)

diff --git a/pcs/daemon/ruby_pcsd.py b/pcs/daemon/ruby_pcsd.py

index e612f8da..53c53eaf 100644

--- a/pcs/daemon/ruby_pcsd.py

+++ b/pcs/daemon/ruby_pcsd.py

@@ -7,8 +7,8 @@ from time import time as now

 import pycurl

 from tornado.gen import convert_yielded

 from tornado.web import HTTPError

-from tornado.httputil import split_host_and_port, HTTPServerRequest

-from tornado.httpclient import AsyncHTTPClient

+from tornado.httputil import HTTPServerRequest, HTTPHeaders

+from tornado.httpclient import AsyncHTTPClient, HTTPClientError

 from tornado.curl_httpclient import CurlError

@@ -29,6 +29,11 @@ RUBY_LOG_LEVEL_MAP = {

     "DEBUG": logging.DEBUG,

 }

+__id_dict = {"id": 0}

+def get_request_id():

+    __id_dict["id"] += 1

+    return __id_dict["id"]

+

 class SinatraResult(namedtuple("SinatraResult", "headers, status, body")):

     @classmethod

     def from_response(cls, response):

@@ -60,6 +65,59 @@ def process_response_logs(rb_log_list):

             group_id=group_id

         )

+class RubyDaemonRequest(namedtuple(

+    "RubyDaemonRequest",

+    "request_type, path, query, headers, method, body"

+)):

+    def __new__(

+        cls,

+        request_type,

+        http_request: HTTPServerRequest = None,

+        payload=None,

+    ):

+        headers = http_request.headers if http_request else HTTPHeaders()

+        headers.add("X-Pcsd-Type", request_type)

+        if payload:

+            headers.add(

+                "X-Pcsd-Payload",

+                b64encode(json.dumps(payload).encode()).decode()

+            )

+        return super(RubyDaemonRequest, cls).__new__(

+            cls,

+            request_type,

+            http_request.path if http_request else "",

+            http_request.query if http_request else "",

+            headers,

+            http_request.method if http_request else "GET",

+            http_request.body if http_request else None,

+        )

+

+    @property

+    def url(self):

+        # We do not need location for communication with ruby itself since we

+        # communicate via unix socket. But it is required by AsyncHTTPClient so

+        # "localhost" is used.

+        query = f"?{self.query}" if self.query else ""

+        return f"localhost/{self.path}{query}"

+

+    @property

+    def is_get(self):

+        return self.method.upper() == "GET"

+

+    @property

+    def has_http_request_detail(self):

+        return self.path or self.query or self.method != "GET" or self.body

+

+def log_ruby_daemon_request(label, request: RubyDaemonRequest):

+    log.pcsd.debug("%s type: '%s'", label, request.request_type)

+    if request.has_http_request_detail:

+        log.pcsd.debug("%s path: '%s'", label, request.path)

+        if request.query:

+            log.pcsd.debug("%s query: '%s'", label, request.query)

+        log.pcsd.debug("%s method: '%s'", label, request.method)

+        if request.body:

+            log.pcsd.debug("%s body: '%s'", label, request.body)

+

 class Wrapper:

     def __init__(self, pcsd_ruby_socket, debug=False):

         self.__debug = debug

@@ -67,74 +125,87 @@ class Wrapper:

         self.__client = AsyncHTTPClient()

         self.__pcsd_ruby_socket = pcsd_ruby_socket

-    @staticmethod

-    def get_sinatra_request(request: HTTPServerRequest):

-        host, port = split_host_and_port(request.host)

-        return {"env": {

-            "PATH_INFO": request.path,

-            "QUERY_STRING": request.query,

-            "REMOTE_ADDR": request.remote_ip,

-            "REMOTE_HOST": request.host,

-            "REQUEST_METHOD": request.method,

-            "REQUEST_URI": f"{request.protocol}://{request.host}{request.uri}",

-            "SCRIPT_NAME": "",

-            "SERVER_NAME": host,

-            "SERVER_PORT": port,

-            "SERVER_PROTOCOL": request.version,

-            "HTTP_HOST": request.host,

-            "HTTP_ACCEPT": "*/*",

-            "HTTP_COOKIE": ";".join([

-                v.OutputString() for v in request.cookies.values()

-            ]),

-            "HTTPS": "on" if request.protocol == "https" else "off",

-            "HTTP_VERSION": request.version,

-            "REQUEST_PATH": request.path,

-            "rack.input": request.body.decode("utf8"),

-        }}

-

     def prepare_curl_callback(self, curl):

         curl.setopt(pycurl.UNIX_SOCKET_PATH, self.__pcsd_ruby_socket)

         curl.setopt(pycurl.TIMEOUT, 70)

-    async def send_to_ruby(self, request_json):

-        # We do not need location for communication with ruby itself since we

-        # communicate via unix socket. But it is required by AsyncHTTPClient so

-        # "localhost" is used.

-        tornado_request = b64encode(request_json.encode()).decode()

-        return (await self.__client.fetch(

-            "localhost",

-            method="POST",

-            body=f"TORNADO_REQUEST={tornado_request}",

-            prepare_curl_callback=self.prepare_curl_callback,

-        )).body

-

-    async def run_ruby(self, request_type, request=None):

-        """

-        request_type: SINATRA_GUI|SINATRA_REMOTE|SYNC_CONFIGS

-        request: result of get_sinatra_request|None

-            i.e. it has structure returned by get_sinatra_request if the request

-            is not None - so we can get SERVER_NAME and  SERVER_PORT

-        """

-        request = request or {}

-        request.update({"type": request_type})

-        request_json = json.dumps(request)

-

-        if self.__debug:

-            log.pcsd.debug("Ruby daemon request: '%s'", request_json)

+    async def send_to_ruby(self, request: RubyDaemonRequest):

         try:

-            ruby_response = await self.send_to_ruby(request_json)

+            return (await self.__client.fetch(

+                request.url,

+                headers=request.headers,

+                method=request.method,

+                # Tornado enforces body=None for GET method:

+                # Even with `allow_nonstandard_methods` we disallow GET with a

+                # body (because libcurl doesn't allow it unless we use

+                # CUSTOMREQUEST).  While the spec doesn't forbid clients from

+                # sending a body, it arguably disallows the server from doing

+                # anything with them.

+                body=(request.body if not request.is_get else None),

+                prepare_curl_callback=self.prepare_curl_callback,

+            )).body

         except CurlError as e:

+            # This error we can get e.g. when ruby daemon is down.

             log.pcsd.error(

                 "Cannot connect to ruby daemon (message: '%s'). Is it running?",

                 e

             )

             raise HTTPError(500)

+        except HTTPClientError as e:

+            # This error we can get e.g. when rack protection raises exception.

+            log.pcsd.error(

+                (

+                    "Got error from ruby daemon (message: '%s')."

+                    " Try checking system logs (e.g. journal, systemctl status"

+                    " pcsd.service) for more information.."

+                ),

+                e

+            )

+            raise HTTPError(500)

+

+    async def run_ruby(

+        self,

+        request_type,

+        http_request: HTTPServerRequest = None,

+        payload=None,

+    ):

+        request = RubyDaemonRequest(request_type, http_request, payload)

+        request_id = get_request_id()

+

+        def log_request():

+            log_ruby_daemon_request(

+                f"Ruby daemon request (id: {request_id})",

+                request,

+            )

+

+        if self.__debug:

+            log_request()

+

+        return self.process_ruby_response(

+            f"Ruby daemon response (id: {request_id})",

+            log_request,

+            await self.send_to_ruby(request),

+        )

+

+    def process_ruby_response(self, label, log_request, ruby_response):

+        """

+        Return relevant part of unpacked ruby response. As a side effect

+        relevant logs are writen.

+        string label -- is used as a log prefix

+        callable log_request -- is used to log request when some errors happen;

+            we want to log request before error even if there is not debug mode

+        string ruby_response -- body of response from ruby; it should contain

+            json with dictionary with response specific keys

+        """

         try:

             response = json.loads(ruby_response)

             if "error" in response:

+                if not self.__debug:

+                    log_request()

                 log.pcsd.error(

-                    "Ruby daemon response contains an error: '%s'",

+                    "%s contains an error: '%s'",

+                    label,

                     json.dumps(response)

                 )

                 raise HTTPError(500)

@@ -144,56 +215,52 @@ class Wrapper:

                 body = b64decode(response.pop("body"))

                 if self.__debug:

                     log.pcsd.debug(

-                        "Ruby daemon response (without logs and body): '%s'",

+                        "%s (without logs and body): '%s'",

+                        label,

                         json.dumps(response)

                     )

-                    log.pcsd.debug("Ruby daemon response body: '%s'", body)

+                    log.pcsd.debug("%s body: '%s'", label, body)

                 response["body"] = body

             elif self.__debug:

                 log.pcsd.debug(

-                    "Ruby daemon response (without logs): '%s'",

+                    "%s (without logs): '%s'",

+                    label,

                     json.dumps(response)

                 )

             process_response_logs(logs)

             return response

         except (json.JSONDecodeError, binascii.Error) as e:

             if self.__debug:

-                log.pcsd.debug("Ruby daemon response: '%s'", ruby_response)

+                log.pcsd.debug("%s: '%s'", label, ruby_response)

+            else:

+                log_request()

+

             log.pcsd.error("Cannot decode json from ruby pcsd wrapper: '%s'", e)

             raise HTTPError(500)

     async def request_gui(

         self, request: HTTPServerRequest, user, groups, is_authenticated

     ) -> SinatraResult:

-        sinatra_request = self.get_sinatra_request(request)

         # Sessions handling was removed from ruby. However, some session

         # information is needed for ruby code (e.g. rendering some parts of

         # templates). So this information must be sent to ruby by another way.

-        sinatra_request.update({

-            "session": {

+        return SinatraResult.from_response(

+            await convert_yielded(self.run_ruby(SINATRA_GUI, request, {

                 "username": user,

                 "groups": groups,

                 "is_authenticated": is_authenticated,

-            }

-        })

-        response = await convert_yielded(self.run_ruby(

-            SINATRA_GUI,

-            sinatra_request

-        ))

-        return SinatraResult.from_response(response)

+            }))

+        )

     async def request_remote(self, request: HTTPServerRequest) -> SinatraResult:

-        response = await convert_yielded(self.run_ruby(

-            SINATRA_REMOTE,

-            self.get_sinatra_request(request)

-        ))

-        return SinatraResult.from_response(response)

+        return SinatraResult.from_response(

+            await convert_yielded(self.run_ruby(SINATRA_REMOTE, request))

+        )

     async def sync_configs(self):

         try:

-            response = await convert_yielded(self.run_ruby(SYNC_CONFIGS))

-            return response["next"]

+            return (await convert_yielded(self.run_ruby(SYNC_CONFIGS)))["next"]

         except HTTPError:

             log.pcsd.error("Config synchronization failed")

             return int(now()) + DEFAULT_SYNC_CONFIG_DELAY

diff --git a/pcs_test/tier0/daemon/app/fixtures_app.py b/pcs_test/tier0/daemon/app/fixtures_app.py

index 8d5b8f4c..590203b4 100644

--- a/pcs_test/tier0/daemon/app/fixtures_app.py

+++ b/pcs_test/tier0/daemon/app/fixtures_app.py

@@ -20,7 +20,12 @@ class RubyPcsdWrapper(ruby_pcsd.Wrapper):

         self.headers = {"Some": "value"}

         self.body = b"Success action"

-    async def run_ruby(self, request_type, request=None):

+    async def run_ruby(

+        self,

+        request_type,

+        http_request=None,

+        payload=None,

+    ):

         if request_type != self.request_type:

             raise AssertionError(

                 f"Wrong request type: expected '{self.request_type}'"

diff --git a/pcs_test/tier0/daemon/test_ruby_pcsd.py b/pcs_test/tier0/daemon/test_ruby_pcsd.py

index 28f14c87..32eb74cc 100644

--- a/pcs_test/tier0/daemon/test_ruby_pcsd.py

+++ b/pcs_test/tier0/daemon/test_ruby_pcsd.py

@@ -4,7 +4,7 @@ from base64 import b64encode

 from unittest import TestCase, mock

 from urllib.parse import urlencode

-from tornado.httputil import HTTPServerRequest

+from tornado.httputil import HTTPServerRequest, HTTPHeaders

 from tornado.testing import AsyncTestCase, gen_test

 from tornado.web import HTTPError

@@ -22,46 +22,17 @@ def create_http_request():

     return HTTPServerRequest(

         method="POST",

         uri="/pcsd/uri",

-        headers={"Cookie": "cookie1=first;cookie2=second"},

+        headers=HTTPHeaders({"Cookie": "cookie1=first;cookie2=second"}),

         body=str.encode(urlencode({"post-key": "post-value"})),

         host="pcsd-host:2224"

     )

-class GetSinatraRequest(TestCase):

-    def test_translate_request(self):

-        # pylint: disable=invalid-name

-        self.maxDiff = None

-        self.assertEqual(

-            create_wrapper().get_sinatra_request(create_http_request()),

-            {

-                'env': {

-                    'HTTPS': 'off',

-                    'HTTP_ACCEPT': '*/*',

-                    'HTTP_COOKIE': 'cookie1=first;cookie2=second',

-                    'HTTP_HOST': 'pcsd-host:2224',

-                    'HTTP_VERSION': 'HTTP/1.0',

-                    'PATH_INFO': '/pcsd/uri',

-                    'QUERY_STRING': '',

-                    'REMOTE_ADDR': None, # It requires complicated request args

-                    'REMOTE_HOST': 'pcsd-host:2224',

-                    'REQUEST_METHOD': 'POST',

-                    'REQUEST_PATH': '/pcsd/uri',

-                    'REQUEST_URI': 'http://pcsd-host:2224/pcsd/uri',

-                    'SCRIPT_NAME': '',

-                    'SERVER_NAME': 'pcsd-host',

-                    'SERVER_PORT': 2224,

-                    'SERVER_PROTOCOL': 'HTTP/1.0',

-                    'rack.input': 'post-key=post-value'

-                }

-            }

-        )

-

 patch_ruby_pcsd = create_patcher(ruby_pcsd)

 class RunRuby(AsyncTestCase):

     def setUp(self):

         self.ruby_response = ""

-        self.request = self.create_request()

+        self.request = ruby_pcsd.RubyDaemonRequest(ruby_pcsd.SYNC_CONFIGS)

         self.wrapper = create_wrapper()

         patcher = mock.patch.object(

             self.wrapper,

@@ -72,14 +43,10 @@ class RunRuby(AsyncTestCase):

         patcher.start()

         super().setUp()

-    async def send_to_ruby(self, request_json):

-        self.assertEqual(json.loads(request_json), self.request)

+    async def send_to_ruby(self, ruby_request):

+        self.assertEqual(ruby_request, self.request)

         return self.ruby_response

-    @staticmethod

-    def create_request(_type=ruby_pcsd.SYNC_CONFIGS):

-        return {"type": _type}

-

     def set_run_result(self, run_result):

         self.ruby_response = json.dumps({**run_result, "logs": []})

@@ -125,10 +92,10 @@ class RunRuby(AsyncTestCase):

             "body": b64encode(str.encode(body)).decode(),

         })

         http_request = create_http_request()

-        self.request = {

-            **self.create_request(ruby_pcsd.SINATRA_REMOTE),

-            **self.wrapper.get_sinatra_request(http_request),

-        }

+        self.request = ruby_pcsd.RubyDaemonRequest(

+            ruby_pcsd.SINATRA_REMOTE,

+            http_request,

+        )

         result = yield self.wrapper.request_remote(http_request)

         self.assert_sinatra_result(result, headers, status, body)

@@ -148,15 +115,15 @@ class RunRuby(AsyncTestCase):

             "body": b64encode(str.encode(body)).decode(),

         })

         http_request = create_http_request()

-        self.request = {

-            **self.create_request(ruby_pcsd.SINATRA_GUI),

-            **self.wrapper.get_sinatra_request(http_request),

-            "session": {

+        self.request = ruby_pcsd.RubyDaemonRequest(

+            ruby_pcsd.SINATRA_GUI,

+            http_request,

+            {

                 "username": user,

                 "groups": groups,

                 "is_authenticated": is_authenticated,

             }

-        }

+        )

         result = yield self.wrapper.request_gui(

             http_request,

             user=user,

diff --git a/pcsd/rserver.rb b/pcsd/rserver.rb

index 6002a73c..4b58f252 100644

--- a/pcsd/rserver.rb

+++ b/pcsd/rserver.rb

@@ -11,42 +11,25 @@ def pack_response(response)

   return [200, {}, [response.to_json.to_str]]

 end

-def unpack_request(transport_env)

-  return JSON.parse(Base64.strict_decode64(

-    transport_env["rack.request.form_hash"]["TORNADO_REQUEST"]

-  ))

-end

-

 class TornadoCommunicationMiddleware

   def initialize(app)

     @app = app

   end

-  def call(transport_env)

+  def call(env)

     Thread.current[:pcsd_logger_container] = []

     begin

-      request = unpack_request(transport_env)

+      type = env["HTTP_X_PCSD_TYPE"]

-      if ["sinatra_gui", "sinatra_remote"].include?(request["type"])

-        if request["type"] == "sinatra_gui"

-          session = request["session"]

+      if ["sinatra_gui", "sinatra_remote"].include?(type)

+        if type == "sinatra_gui"

+          session = JSON.parse(Base64.strict_decode64(env["HTTP_X_PCSD_PAYLOAD"]))

           Thread.current[:tornado_username] = session["username"]

           Thread.current[:tornado_groups] = session["groups"]

           Thread.current[:tornado_is_authenticated] = session["is_authenticated"]

         end

-        # Keys rack.input and rack.errors are required. We make sure they are

-        # there.

-        request_env = request["env"]

-        request_env["rack.input"] = StringIO.new(request_env["rack.input"])

-        request_env["rack.errors"] = StringIO.new()

-

-        status, headers, body = @app.call(request_env)

-

-        rack_errors = request_env['rack.errors'].string()

-        if not rack_errors.empty?()

-          $logger.error(rack_errors)

-        end

+        status, headers, body = @app.call(env)

         return pack_response({

           :status => status,

@@ -56,16 +39,20 @@ class TornadoCommunicationMiddleware

         })

       end

-      if request["type"] == "sync_configs"

+      if type == "sync_configs"

         return pack_response({

           :next => Time.now.to_i + run_cfgsync(),

           :logs => Thread.current[:pcsd_logger_container],

         })

       end

-      raise "Unexpected value for key 'type': '#{request['type']}'"

+      return pack_response({

+        :error => "Unexpected value for key 'type': '#{type}'"

+      })

     rescue => e

-      return pack_response({:error => "Processing request error: '#{e}'"})

+      return pack_response({

+        :error => "Processing request error: '#{e}' '#{e.backtrace}'"

+      })

     end

   end

 end

-- 

2.21.1