動機

業務でtls利用したgrpcサーバを建てる必要があったがサンプルを探してもinsecure channelを利用した例しか見つからなかったので調べた。

証明書用意

# 秘密鍵と公開鍵を作成
$ openssl req -x509 -nodes -newkey rsa:2048 -days 365 -keyout privkey.pem -out cert.pem -subj "/CN=127.0.0.1"
# 公開鍵でroot証明書を作る
$ openssl  x509 -in cert.pem -out root.crt

.proto

test.proto

syntax = "proto3";

message EchoRequest {
    string message = 1;
}

message EchoResponse {
    string message = 1;
}

service AVRGateway {
    rpc Echo (EchoRequest) returns (EchoResponse) {}
}

スタブ作成

$ pipenv shell --three
$ pipenv install grpcio-tools
$ python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./test.proto

サーバ作成


from concurrent import futures
import grpc
import test_pb2_grpc
import test_pb2

class Servicer(test_pb2_grpc.AVRGatewayServicer):
  def Echo(self, request, context):
    return test_pb2.EchoResponse(message = "hoge")

def serve():
    pkey=open("privkey.pem", "rb").read()
    chain=open("cert.pem", "rb").read()
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=2))
    test_pb2_grpc.add_AVRGatewayServicer_to_server(
        Servicer(), server)
    cred = grpc.ssl_server_credentials([(pkey, chain)])
    server.add_secure_port('localhost:51011', cred)
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    serve();

クライアント作成

import grpc
from test_pb2_grpc import AVRGatewayStub
from test_pb2 import *

if __name__ == "__main__":
    message = sys.argv[1]
    root = open("root.crt","rb").read()
    cred = grpc.ssl_channel_credentials(root_certificates=root)

    with grpc.secure_channel("localhost:51011", cred) as channel:
        stub = AVRGatewayStub(channel)
        response = stub.Echo(EchoRequest(message=message))
        print(response)

テスト

$ python server.py &
$ $ python client.py hoge
message: "hoge"

参考

リファレンス
 https://qiita.com/qiita_kuru/items/4d51b11f76dc0eafba7f
https://grpc.io/docs/tutorials/basic/python/
https://qiita.com/ike_dai/items/1a28f6a42c4a0d3a49d0