動機
業務で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