grpc-argument-validator 0.2.2

gRPC argument validator utility.

gRPC argument validator

gRPC argument validator is a library that provides decorators to automatically validate arguments in requests to rpc methods.

Getting Started

This is an example of how you may give instructions on setting up your project locally. To get a local copy up and running follow these simple example steps.

Installation

From PyPI

pip install grpc-argument-validator

From source

• Install poetry

• Clone repo

git clone https://github.com/messagebird/python-grpc-argument-validator.git

• Install packages

cd python-grpc-argument-validator && poetry install

• Run the tests

cd src/tests
poetry run python -m unittest

Quick Example

from google.protobuf.descriptor import FieldDescriptor
from grpc_argument_validator import validate_args
from grpc_argument_validator import AbstractArgumentValidator, ValidationResult, ValidationContext

class PathValidator(AbstractArgumentValidator):

    def check(self, name: str, value: Path, field_descriptor: FieldDescriptor, validation_context: ValidationContext) -> ValidationResult:
        if len(value.points) > 5:
            return ValidationResult(valid=True)
        return ValidationResult(False, f"path for '{name}' should be at least five points long")

class RouteService(RouteCheckerServicer):
    @validate_args(
        non_empty=["tags", "tags[]", "path.points"],
        validators={"path": PathValidator()},
    )
    def Create(self, request: Route, context: grpc.ServicerContext):
        return BoolValue(value=True)

Documentation

We host the full API reference on GitHub pages.

Argument field syntax

To specify which argument field should be validated, grpc-argument-validator expects strings that match the field names as defined in the protobufs. To access nested fields, use a dot (.).

Consider the following protobuf definition:

syntax = "proto3";

package routeguide;

import "google/protobuf/empty.proto";
import "google/protobuf/wrappers.proto";

message Point {
  int32 x = 1;
  int32 y = 2;
  google.protobuf.StringValue name = 3;
}

message Rectangle {
  Point lo = 1;
  Point hi = 2;
}

message Area {
  Rectangle rectangle = 1;
  google.protobuf.StringValue message = 2;
  google.protobuf.BytesValue uuid = 3;
}

message Path {
  repeated Point points = 1;
}

enum Planet {
  PLANET_INVALID = 0;
  PLANET_EARTH = 1;
  PLANET_MARS = 2;
}

message PlanetValue {
  Planet value = 1;
}

message Route {
  Path path = 1;
  google.protobuf.StringValue name = 2;
  PlanetValue planet = 3;
  repeated string tags = 4;
}

service RouteService {
  rpc CreateRoute(Route) returns (google.protobuf.Empty);
  rpc CreateArea(Area) returns (google.protobuf.Empty);
}

• If you want to validate the field planet in a Route proto, simply specify "planet" or equivalently ".planet".
• If you want to validate the value field within the name field of a Route proto, use "name.value" or equivalently ".name.value".
• If you want to apply a check to each element of a repeated field, append [] to the name of the field.
• If you want to apply a check to the 'root proto' (i.e. the request itself), use "." as the field path.

To clarify this, let's say that we know that both planet and name.value should have non-default values. We can then decorate a method in our gRPC server as follows:

import grpc
from google.protobuf.empty_pb2 import Empty
from grpc_argument_validator import validate_args
from tests.route_guide_protos.route_guide_pb2 import Route
from tests.route_guide_protos.route_guide_pb2_grpc import RouteServiceServicer


class RouteServiceImpl(RouteServiceServicer):
    @validate_args(non_empty=["planet", "name.value"])
    def CreateRoute(self, request: Route, context: grpc.ServicerContext):
        return Empty()

Calling the service with a default value for either planet or name.value will yield an INVALID_ARGUMENT status code with further details on which fields violate the validation.

Validators

There are two kinds of validators you might consider:

• There are predefined validators which we will cover shortly
• Another option is to define your own validators

In the examples below, we have used exactly one validator + field path per validate_args decorator for clarity. Fortunately, our API allows you to use multiple validators and fields!

'Has' validator

The simplest of all predefined validators is the 'has' validator which simply checks whether a HasField evaluates to True. This of course works in combination with nested fields.

In the example below, calling the Create endpoint without setting Route.name would result in an INVALID_ARGUMENT status.

class RouteServiceImpl(RouteServiceServicer):
    @validate_args(has=["name"])
    def CreateRoute(self, request: Route, context: grpc.ServicerContext):
        return Empty()

Run this on a local machine and make a request with an invalid argument:

with grpc.insecure_channel("127.0.0.1:50051") as c:
    route_client = RouteServiceStub(channel=c)
    try:
        route_client.CreateRoute(Route(tags=["tag"]))
    except grpc.RpcError as e:
        if isinstance(e, grpc.Call):
            print(e.details())

The following will be printed: must have 'name'

UUID validator

Another common use-case is the validation of UUIDs. You can enlist the fields that should be UUIDs (represented as 16 bytes) with the uuids argument:

class RouteServiceImpl(RouteServiceServicer):
    @validate_args(uuids=["uuid.value"])
    def CreateArea(self, request: Area, context: grpc.ServicerContext):
        return Empty()

The client side might violate the UUID requirement as follows:

with grpc.insecure_channel("127.0.0.1:50051") as c:
    route_client = RouteServiceStub(channel=c)
    try:
        route_client.CreateArea(Area(uuid=BytesValue(value="not a uuid".encode())))
    except grpc.RpcError as e:
        if isinstance(e, grpc.Call):
            print(e.details())

This will print 'uuid.value' must be a valid UUID.

Non-default validator

For fields that should have a non-default value, such as enums, we have provided the non_default argument:

class RouteServiceImpl(RouteServiceServicer):
    @validate_args(non_default=["planet.value"])
    def CreateRoute(self, request: Route, context: grpc.ServicerContext):
        return Empty()

The client side may violate this as follows:

with grpc.insecure_channel("127.0.0.1:50051") as c:
    route_client = RouteServiceStub(channel=c)
    try:
        route_client.CreateRoute(Route(planet=PlanetValue()))
    except grpc.RpcError as e:
        if isinstance(e, grpc.Call):
            print(e.details())

Which will print 'planet.value' must have non-default value.

Non-empty validator

We provide a 'non-'empty validator which can be used to ensure that a repeated field has more than zero elements.

class RouteServiceImpl(RouteServiceServicer):
    @validate_args(non_empty=["tags"])
    def CreateRoute(self, request: Route, context: grpc.ServicerContext):
        return Empty()

Which can be violated as follows:

with grpc.insecure_channel("127.0.0.1:50051") as c:
    route_client = RouteServiceStub(channel=c)
    try:
        route_client.CreateRoute(Route(tags=[]))
    except grpc.RpcError as e:
        if isinstance(e, grpc.Call):
            print(e.details())

Which will print 'tags' must be non-empty.

Regexp validator

Finally, we have the regexp validator that can be used to check whether a string field matches a regular expression.

class RouteServiceImpl(RouteServiceServicer):
    @validate_args(validators={"message.value": RegexpValidator(pattern=r"\d+")})
    def CreateArea(self, request: Area, context: grpc.ServicerContext):
        return Empty()

with grpc.insecure_channel("127.0.0.1:50051") as c:
    route_client = RouteServiceStub(channel=c)
    try:
        route_client.CreateArea(Area(message=StringValue(value="hello world")))
    except grpc.RpcError as e:
        if isinstance(e, grpc.Call):
            print(e.details())

Which will print 'message.value' must match regexp pattern: \d+.

Custom validators

You can also write custom validators to flexibily handle your use-case. You need to derive a class from AbstractArgumentValidator and implement its check method. The example below shows how to implement a simple validator for checking that a path has 5 points. You can provide such custom validators through a dict that maps a field path to a validator:

from grpc_argument_validator import AbstractArgumentValidator
from grpc_argument_validator import ValidationContext
from grpc_argument_validator import ValidationResult
from google.protobuf.descriptor import FieldDescriptor

from examples.route_guide_pb2 import Path

class PathValidator(AbstractArgumentValidator):
    def check(
        self, name: str, value: Path, field_descriptor: FieldDescriptor, validation_context: ValidationContext
    ) -> ValidationResult:
        if len(value.points) > 5:
            return ValidationResult(valid=True)
        return ValidationResult(False, f"path for '{name}' should be at least five points long")


class RouteServiceImpl(RouteServiceServicer):

    @validate_args(validators={"path": PathValidator()})
    def CreateRoute(self, request: Route, context: grpc.ServicerContext):
        return Empty()

Optional vs. required validators

For each of the built-in validators (except for the has validator), validate_args has not one but two keyword arguments. One of those is prepended with optional_. This means that apart from uuid, non_default and non_empty we also have optional_uuid, optional_non_default and optional_non_empty. The behavior is slightly different: for any of the optional_* validators, it is OK if the field is not contained by the incoming request. Sometimes fields are simply optional, and you only want to validate them if they are present.

Since it is also common that fields are not optional, we also provide the required validators (without optional_*) for which HasField must evaluate to True for that field and all preceding fields in the protos hierarchy.

The custom validator counterparts are validators and optional_validators. Each takes a dict with a mapping of field paths to validators. These can be used for validators that might be preconfigured such as the RegexpValidator or for customer validators.

Streaming requests

You can also use the validators for streaming requests. Since streaming requests might not all look the same in a single stream (e.g. the first request might have metadata describing the remainder of the stream), we provide a streaming request index in a ValidationContext that is passed to an AbstractArgumentValidator.

Here's an example of how that could be used:

class StreamingPathValidator(AbstractArgumentValidator):
    def __init__(self, first_number_of_points: int, second_number_of_points: int):
        self._first_number_of_points = first_number_of_points
        self._second_number_of_points = second_number_of_points

    def check(
        self, name: str, value: Any, field_descriptor: FieldDescriptor, validation_context: ValidationContext
    ) -> ValidationResult:
        if not validation_context.is_streaming:
            return ValidationResult(False, "request must be a streaming request")

        if validation_context.streaming_message_index == 0:
            if len(value.points) != self._first_number_of_points:
                return ValidationResult(False, f"first path should have {self._first_number_of_points} points")

        if validation_context.streaming_message_index == 1:
            if len(value.points) != self._second_number_of_points:
                return ValidationResult(False, f"second path should have {self._second_number_of_points} points")

        return ValidationResult(True)

Enabling rich error details

To enable richer error responses where each violation is contained in a BadRequest proto, you can use

from grpc_argument_validator import ArgumentValidatorConfig

ArgumentValidatorConfig.set_rich_grpc_errors(enabled=True)

Now, your client-side can parse the error details as follows:

def extract_error_details(err):
    status_proto = status_pb2.Status()

    for metadatum in err.trailing_metadata():
        if isinstance(metadatum, _Metadatum):
            if metadatum.key == "grpc-status-details-bin":
                status_proto.MergeFromString(metadatum.value)

    unpacked = [_unpack_error_detail(det) for det in status_proto.details]
    return unpacked

def _unpack_error_detail(grpc_detail):
    val = error_details_pb2.BadRequest()
    grpc_detail.Unpack(val)
    return val

with grpc.insecure_channel("127.0.0.1:50051") as c:
    route_client = RouteServiceStub(channel=c)
    try:
        route_client.CreateArea(Area(message=StringValue(value="hello world")))
    except grpc.RpcError as e:
        error_details = extract_error_details(e)
        print(error_details)

Contributing

Contributions are what make the open source community such an amazing place to be learn, inspire, and create. Any contributions you make are greatly appreciated.

1. Fork the Project
2. Create your Feature Branch (git checkout -b feature/AmazingFeature)
3. Commit your Changes (git commit -m 'Add some AmazingFeature')
4. Push to the Branch (git push origin feature/AmazingFeature)
5. Open a Pull Request

Generating HTML Documentation

Generate the docs by running:

pdoc --html -o docs src/grpc_argument_validator

License

Distributed under The BSD 3-Clause License. Copyright (c) 2021, MessageBird