Masking: Handle larger than 16MB payloads

Larger than 16MB payloads can either be ignored or then they may
cause the session to be aborted.

server/modules/filter/masking/maskingfilterconfig.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "masking"
 #include "maskingfilterconfig.hh"
 
 namespace

server/modules/filter/masking/maskingfiltersession.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "masking"
 #include "maskingfiltersession.hh"
 #include <sstream>
 #include <maxscale/buffer.hh>
@@ -98,10 +99,25 @@ int MaskingFilterSession::clientReply(GWBUF* pPacket)
         case EXPECTING_ROW_EOF:
             handle_eof(pPacket);
             break;
+
+        case SUPPRESSING_RESPONSE:
+            break;
         }
     }
 
-    return FilterSession::clientReply(pPacket);
+    // The state may change by the code above, so need to check it again.
+    int rv;
+    if (m_state != SUPPRESSING_RESPONSE)
+    {
+        rv = FilterSession::clientReply(pPacket);
+    }
+    else
+    {
+        // TODO: The return value should mean something.
+        rv = 0;
+    }
+
+    return rv;
 }
 
 void MaskingFilterSession::handle_response(GWBUF* pPacket)
@@ -130,6 +146,12 @@ void MaskingFilterSession::handle_field(GWBUF* pPacket)
 {
     ComQueryResponse::ColumnDef column_def(pPacket);
 
+    if (column_def.payload_len() >= ComPacket::MAX_PAYLOAD_LEN) // Not particularly likely...
+    {
+        handle_large_payload();
+    }
+    else
+    {
         const char *zUser = session_get_user(m_pSession);
         const char *zHost = session_get_remote(m_pSession);
 
@@ -151,6 +173,7 @@ void MaskingFilterSession::handle_field(GWBUF* pPacket)
             m_state = EXPECTING_FIELD_EOF;
         }
     }
+}
 
 void MaskingFilterSession::handle_eof(GWBUF* pPacket)
 {
@@ -215,6 +238,37 @@ void MaskingFilterSession::handle_row(GWBUF* pPacket)
         break;
 
     default:
+        if (m_res.some_rule_matches())
+        {
+            if (response.payload_len() >= ComPacket::MAX_PAYLOAD_LEN)
+            {
+                handle_large_payload();
+            }
+            else
+            {
+                mask_values(response);
+            }
+        }
+    }
+}
+
+void MaskingFilterSession::handle_large_payload()
+{
+    if (m_filter.config().large_payload() == Config::LARGE_ABORT)
+    {
+        MXS_WARNING("Payload > 16MB, closing the connection.");
+        poll_fake_hangup_event(m_pSession->client_dcb);
+        m_state = SUPPRESSING_RESPONSE;
+    }
+    else
+    {
+        MXS_WARNING("Payload > 16MB, no masking is performed.");
+        m_state = IGNORING_RESPONSE;
+    }
+}
+
+void MaskingFilterSession::mask_values(ComResponse& response)
+{
     switch (m_res.command())
     {
     case MYSQL_COM_QUERY:
@@ -278,4 +332,3 @@ void MaskingFilterSession::handle_row(GWBUF* pPacket)
         ss_dassert(!true);
     }
 }
-}

server/modules/filter/masking/maskingfiltersession.hh

@@ -49,13 +49,17 @@ private:
         EXPECTING_FIELD_EOF,
         EXPECTING_ROW,
         EXPECTING_ROW_EOF,
-        IGNORING_RESPONSE
+        IGNORING_RESPONSE,
+        SUPPRESSING_RESPONSE
     };
 
     void handle_response(GWBUF* pPacket);
     void handle_field(GWBUF* pPacket);
     void handle_row(GWBUF* pPacket);
     void handle_eof(GWBUF* pPacket);
+    void handle_large_payload();
+
+    void mask_values(ComResponse& response);
 
 private:
     typedef std::tr1::shared_ptr<MaskingRules> SMaskingRules;
@@ -68,14 +72,17 @@ private:
             , m_nTotal_fields(0)
             , m_index(0)
             , m_multi_result(false)
+            , m_some_rule_matches(false)
         {}
 
         void reset(uint8_t command, const SMaskingRules& sRules)
         {
+            reset_multi();
+
             m_command = command;
             m_sRules = sRules;
-
+            m_multi_result = false;
-            reset_multi();
+            m_some_rule_matches = false;
         }
 
         void reset_multi()
@@ -97,6 +104,11 @@ private:
             return m_sRules;
         }
 
+        bool some_rule_matches() const
+        {
+            return m_some_rule_matches;
+        }
+
         bool is_multi_result() const
         {
             return m_multi_result;
@@ -117,6 +129,11 @@ private:
             m_types.push_back(type);
             m_rules.push_back(pRule);
 
+            if (pRule)
+            {
+                m_some_rule_matches = true;
+            }
+
             return m_rules.size() == m_nTotal_fields;
         }
 
@@ -139,11 +156,12 @@ private:
     private:
         uint8_t                                m_command;           /*<! What command. */
         SMaskingRules                          m_sRules;            /*<! The rules that are used. */
-        uint32_t                               m_nTotal_fields; /*<! The total number of fields/columns. */
+        uint32_t                               m_nTotal_fields;     /*<! The total number of fields. */
         std::vector<enum_field_types>          m_types;             /*<! The column types. */
         std::vector<const MaskingRules::Rule*> m_rules;             /*<! The rules applied for columns. */
         size_t                                 m_index;             /*<! Index to the current rule.*/
         bool                                   m_multi_result;      /*<! Are we processing multi-results. */
+        bool                                   m_some_rule_matches; /*<! At least one rule matches. */
     };
 
     const MaskingFilter& m_filter;

server/modules/filter/masking/maskingrules.cc

@@ -11,6 +11,7 @@
  * Public License.
  */
 
+#define MXS_MODULE_NAME "masking"
 #include "maskingrules.hh"
 #include <algorithm>
 #include <errno.h>

server/modules/filter/masking/mysql.hh

@@ -488,9 +488,14 @@ public:
         ERR_PACKET = 0xff,
     };
 
-    uint32_t packet_len() const
+    enum
     {
-        return m_packet_len;
+        MAX_PAYLOAD_LEN = 0xffffff
+    };
+
+    uint32_t payload_len() const
+    {
+        return m_payload_len;
     }
     uint8_t packet_no() const
     {
@@ -501,7 +506,7 @@ protected:
     ComPacket(GWBUF* pPacket)
         : m_pPacket(pPacket)
         , m_pData(GWBUF_DATA(pPacket))
-        , m_packet_len(MYSQL_GET_PAYLOAD_LEN(m_pData))
+        , m_payload_len(MYSQL_GET_PAYLOAD_LEN(m_pData))
         , m_packet_no(MYSQL_GET_PACKET_NO(m_pData))
     {
         m_pData += MYSQL_HEADER_LEN;
@@ -510,7 +515,7 @@ protected:
     ComPacket(const ComPacket& packet)
         : m_pPacket(packet.m_pPacket)
         , m_pData(GWBUF_DATA(m_pPacket))
-        , m_packet_len(packet.m_packet_len)
+        , m_payload_len(packet.m_payload_len)
         , m_packet_no(packet.m_packet_no)
     {
         m_pData += MYSQL_HEADER_LEN;
@@ -520,7 +525,7 @@ protected:
     uint8_t* m_pData;
 
 private:
-    uint32_t m_packet_len;
+    uint32_t m_payload_len;
     uint8_t  m_packet_no;
 };