Fix fuzz harness blockers (#3257)

* Fix fuzz harness blockers

Harden several code paths with increased error handling. The existing
fuzzing harneses are running into various blockers stopping them from
explore further code. This is an effort to harden the code so the
fuzzers will run better without crashing.

Signed-off-by: David Korczynski <david@adalogics.com>

* Add fatal logging

Signed-off-by: David Korczynski <david@adalogics.com>

---------

Signed-off-by: David Korczynski <david@adalogics.com>

Author: DavidKorczynski

src/brpc/policy/mongo_protocol.cpp

@@ -113,6 +113,13 @@ void SendMongoResponse::Run() {
 ParseResult ParseMongoMessage(butil::IOBuf* source,
                               Socket* socket, bool /*read_eof*/, const void *arg) {
     const Server* server = static_cast<const Server*>(arg);
+    // arg may be NULL when the parser is invoked outside of a full Server
+    // context (e.g. during protocol probing or fuzz testing).  Without this
+    // guard, server->options() dereferences a null pointer and crashes.
+    if (NULL == server) {
+        LOG(FATAL) << "Failed creating server";
+        return MakeParseError(PARSE_ERROR_TRY_OTHERS);
+    }
     const MongoServiceAdaptor* adaptor = server->options().mongo_service_adaptor;
     if (NULL == adaptor) {
         // The server does not enable mongo adaptor.

src/brpc/policy/streaming_rpc_protocol.cpp

@@ -116,7 +116,16 @@ ParseResult ParseStreamingMessage(butil::IOBuf* source,
             break;
         }
         meta_buf.clear();  // to reduce memory resident
-        ((Stream*)ptr->conn())->OnReceived(fm, &payload, socket);
+        // ptr->conn() returns the connection-level context attached to the
+        // socket.  It may be NULL when the socket was found by ID but has no
+        // Stream object associated (e.g. during protocol probing or fuzz
+        // testing).  Calling OnReceived on a null pointer would crash.
+        Stream* stream_conn = (Stream*)ptr->conn();
+        if (stream_conn == NULL) {
+            LOG(FATAL) << "No stream object found";
+            break;
+        }
+        stream_conn->OnReceived(fm, &payload, socket);
     } while (0);
 
     // Hack input messenger

src/brpc/redis_command.cpp

@@ -410,6 +410,12 @@ RedisCommandConsumeState RedisCommandParser::ConsumeImpl(butil::IOBuf& buf,
         }
         const size_t buf_size = buf.size();
         const auto copy_str = static_cast<char *>(arena->allocate(buf_size + 1));
+        // arena->allocate() may return NULL on allocation failure
+        if (copy_str == NULL) {
+            LOG(FATAL) << "Arena failed allocation";
+            *err = PARSE_ERROR_ABSOLUTELY_WRONG;
+            return CONSUME_STATE_ERROR;
+        }
         buf.copy_to(copy_str, buf_size);
         if (*copy_str == ' ') {
             *err = PARSE_ERROR_ABSOLUTELY_WRONG;
@@ -520,6 +526,12 @@ RedisCommandConsumeState RedisCommandParser::ConsumeImpl(butil::IOBuf& buf,
     }
     buf.pop_front(crlf_pos + 2/*CRLF*/);
     char* d = (char*)arena->allocate((len/8 + 1) * 8);
+    // Guard against allocation failure
+    if (d == NULL) {
+        LOG(FATAL) << "Arena failed allocation";
+        *err = PARSE_ERROR_ABSOLUTELY_WRONG;
+        return CONSUME_STATE_ERROR;
+    }
     buf.cutn(d, len);
     d[len] = '\0';
     _args[_index].set(d, len);