rpyc+https双向认证

https://rpyc.readthedocs.io/en/latest/docs/secure-connection.html

server

import rpyc
from rpyc.utils.server import ThreadedServer  # or ForkingServer
from rpyc.utils.authenticators import SSLAuthenticator
import ssl

if __name__ == "__main__":
    server_key = "/server.key"
    server_cert = "/server.crt"
    ca_cert = "/ca.crt"
	
    # 如果是单向认证的话只需要传server.key和server.crt即可，
    # 如果需要双向认证的话还需要传ca证书，ca_certs或者指定cert_reqs=ssl.CERT_REQUIRED
    authenticator = SSLAuthenticator(server_key, server_cert, ca_certs=ca_cert, cert_reqs=ssl.CERT_REQUIRED)
    server = ThreadedServer(rpyc.SlaveService, port=12345, authenticator=authenticator)
    server.start()

client

import rpyc

ssl_client_key = "/client.key"
ssl_client_cert = "/client.crt"

ssl_client_pair = dict(keyfile=ssl_client_key, certfile=ssl_client_cert)

conn = rpyc.utils.factory.ssl_connect('localhost', 12345,
                    service=rpyc.SlaveService,
                    config={"allow_pickle":True},
                    keepalive=10,
                    **ssl_client_pair)

conn.close()