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()