grpc-py: extend fuzzing suite (#8348)

- Fixes https://bugs.chromium.org/p/oss-fuzz/issues/detail?id=50586
- Ensures coverage can be run by enabling clean server exit in coverage
runs
- Extends to also reach grpc_status code

projects/grpc-py/build.sh

@@ -20,6 +20,10 @@ pip3 install -r ./requirements.txt
 
 GRPC_PYTHON_CFLAGS="${CFLAGS}" GRPC_PYTHON_BUILD_SYSTEM_RE2=true GRPC_PYTHON_BUILD_SYSTEM_OPENSSL=true GRPC_PYTHON_BUILD_SYSTEM_ZLIB=true pip3 install -v . 
 
+# Install grpcio_status
+cd src/python/grpcio_status
+pip3 install .
+
 cd $SRC/grpc/examples/python/helloworld
 for fuzzer in $(find $SRC -name 'fuzz_*.py'); do
   compile_python_fuzzer $fuzzer --add-data helloworld_pb2.py:. --add-data helloworld_pb2_grpc.py:.

projects/grpc-py/fuzz_server.py

@@ -14,12 +14,18 @@
 # limitations under the License.
 """Fuzz grpc server using the Greeter example"""
 
+import os
 import sys
 import time
 import grpc
+from google.protobuf import any_pb2
+from google.rpc import status_pb2
+from grpc_status import rpc_status
+
 import socket
 import atheris
 import threading
+import argparse
 from concurrent.futures import ThreadPoolExecutor
 from google.protobuf.internal import builder as _builder
 
@@ -35,6 +41,8 @@ sys.path.append(app_path)
 import helloworld_pb2
 import helloworld_pb2_grpc
 
+runs_left = None
+server = None
 
 # Simple server
 class FuzzGreeter(helloworld_pb2_grpc.GreeterServicer):
@@ -45,35 +53,75 @@ class FuzzGreeter(helloworld_pb2_grpc.GreeterServicer):
 
 def serve() -> None:
     """Starts fuzz server"""
+    global server
     server = grpc.server(ThreadPoolExecutor(max_workers=1))
     helloworld_pb2_grpc.add_GreeterServicer_to_server(FuzzGreeter(), server)
     server.add_insecure_port('[::]:50051')
     server.start()
-    server.wait_for_termination()
+    #server.wait_for_termination()
+    return
+
+@atheris.instrument_func
+def TestInput(input_bytes):
+    """Send fuzzing input to the server"""
+    global runs_left
+    global server
+    if runs_left != None:
+        runs_left = runs_left - 1
+        if runs_left <= 2:
+            server.stop()
+            return
+
+    time.sleep(0.02)
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.connect(("localhost", 50051))
+            s.sendall(input_bytes)
+            data = s.recv(1024)
+    except OSError:
+        # We don't want to report network errors
+        return
+
+    # Hit the rpc_status too
+    fdp = atheris.FuzzedDataProvider(input_bytes)
+    try:
+        rich_status = status_pb2.Status(
+            code=fdp.ConsumeIntInRange(1,30000),
+            message=fdp.ConsumeUnicodeNoSurrogates(60)
+        )
+        rpc_status.to_status(rich_status)
+    except ValueError:
+        pass
+
     return
 
 
-def TestInput(input_bytes):
-   """Send fuzzing input to the server"""
-   time.sleep(0.02)
-   with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-      s.connect(("localhost", 50051))
-      s.sendall(input_bytes)
-      data = s.recv(1024)
-   return
+def get_run_count_if_there():
+    """Ensure proper exit for coverage builds"""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-atheris_runs", required=False, default=None)
+    args, _ = parser.parse_known_args()
+    if args.atheris_runs is None:
+        print("None args")
+        return None
+    print(f"Got a fixed set of runs {args.atheris_runs}")
+    return args.atheris_runs
 
 
 def main():
-   # Launch a grpc server
-   _thread = threading.Thread(target=serve)
-   _thread.start()
-   time.sleep(0.2)
+    global runs_left
+    max_runs = get_run_count_if_there()
+    if max_runs is not None:
+        runs_left = int(max_runs)
 
-   # Start fuzzing
-   atheris.instrument_all()
-   atheris.Setup(sys.argv, TestInput, enable_python_coverage=True)
-   atheris.Fuzz()
+    # Launch a grpc server
+    serve()
+
+    # Start fuzzing
+    atheris.instrument_all()
+    atheris.Setup(sys.argv, TestInput, enable_python_coverage=True)
+    atheris.Fuzz()
 
 
 if __name__ == "__main__":
-   main()
+    main()