From 6532ca571d8c1a21c1d754cd495fe0aaadd604ed Mon Sep 17 00:00:00 2001 From: Jacob Alexander Date: Wed, 10 Jun 2020 13:54:14 -0700 Subject: [PATCH] Adding SSL version of the calculator example - Used to test asyncio SSL on Windows mainly - Skipping asyncio thread tests on Windows (due to getTimer wrapper bug on Windows) --- examples/async_ssl_calculator_client.py | 353 ++++++++++++++++++++++++ examples/async_ssl_calculator_server.py | 239 ++++++++++++++++ examples/async_ssl_client.py | 6 +- examples/async_ssl_server.py | 2 +- test/test_examples.py | 10 + 5 files changed, 606 insertions(+), 4 deletions(-) create mode 100755 examples/async_ssl_calculator_client.py create mode 100755 examples/async_ssl_calculator_server.py diff --git a/examples/async_ssl_calculator_client.py b/examples/async_ssl_calculator_client.py new file mode 100755 index 0000000..f77a940 --- /dev/null +++ b/examples/async_ssl_calculator_client.py @@ -0,0 +1,353 @@ +#!/usr/bin/env python3 + +import argparse +import asyncio +import os +import socket +import ssl + +import capnp +import calculator_capnp + + +this_dir = os.path.dirname(os.path.abspath(__file__)) + + +class PowerFunction(calculator_capnp.Calculator.Function.Server): + + '''An implementation of the Function interface wrapping pow(). Note that + we're implementing this on the client side and will pass a reference to + the server. The server will then be able to make calls back to the client.''' + + def call(self, params, **kwargs): + '''Note the **kwargs. This is very necessary to include, since + protocols can add parameters over time. Also, by default, a _context + variable is passed to all server methods, but you can also return + results directly as python objects, and they'll be added to the + results struct in the correct order''' + + return pow(params[0], params[1]) + + +async def myreader(client, reader): + while True: + data = await reader.read(4096) + client.write(data) + + +async def mywriter(client, writer): + while True: + data = await client.read(4096) + writer.write(data.tobytes()) + await writer.drain() + + +def parse_args(): + parser = argparse.ArgumentParser(usage='Connects to the Calculator server \ +at the given address and does some RPCs') + parser.add_argument("host", help="HOST:PORT") + + return parser.parse_args() + + +async def main(host): + host = host.split(':') + addr = host[0] + port = host[1] + + # Setup SSL context + ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, 'selfsigned.cert')) + + # Handle both IPv4 and IPv6 cases + try: + print("Try IPv4") + reader, writer = await asyncio.open_connection( + addr, port, + ssl=ctx, + family=socket.AF_INET + ) + except Exception: + print("Try IPv6") + reader, writer = await asyncio.open_connection( + addr, port, + ssl=ctx, + family=socket.AF_INET6 + ) + + # Start TwoPartyClient using TwoWayPipe (takes no arguments in this mode) + client = capnp.TwoPartyClient() + + # Assemble reader and writer tasks, run in the background + coroutines = [myreader(client, reader), mywriter(client, writer)] + asyncio.gather(*coroutines, return_exceptions=True) + + # Bootstrap the Calculator interface + calculator = client.bootstrap().cast_as(calculator_capnp.Calculator) + + '''Make a request that just evaluates the literal value 123. + + What's interesting here is that evaluate() returns a "Value", which is + another interface and therefore points back to an object living on the + server. We then have to call read() on that object to read it. + However, even though we are making two RPC's, this block executes in + *one* network round trip because of promise pipelining: we do not wait + for the first call to complete before we send the second call to the + server.''' + + print('Evaluating a literal... ', end="") + + # Make the request. Note we are using the shorter function form (instead + # of evaluate_request), and we are passing a dictionary that represents a + # struct and its member to evaluate + eval_promise = calculator.evaluate({"literal": 123}) + + # This is equivalent to: + ''' + request = calculator.evaluate_request() + request.expression.literal = 123 + + # Send it, which returns a promise for the result (without blocking). + eval_promise = request.send() + ''' + + # Using the promise, create a pipelined request to call read() on the + # returned object. Note that here we are using the shortened method call + # syntax read(), which is mostly just sugar for read_request().send() + read_promise = eval_promise.value.read() + + # Now that we've sent all the requests, wait for the response. Until this + # point, we haven't waited at all! + response = await read_promise.a_wait() + assert response.value == 123 + + print("PASS") + + '''Make a request to evaluate 123 + 45 - 67. + + The Calculator interface requires that we first call getOperator() to + get the addition and subtraction functions, then call evaluate() to use + them. But, once again, we can get both functions, call evaluate(), and + then read() the result -- four RPCs -- in the time of *one* network + round trip, because of promise pipelining.''' + + print("Using add and subtract... ", end='') + + # Get the "add" function from the server. + add = calculator.getOperator(op='add').func + # Get the "subtract" function from the server. + subtract = calculator.getOperator(op='subtract').func + + # Build the request to evaluate 123 + 45 - 67. Note the form is 'evaluate' + # + '_request', where 'evaluate' is the name of the method we want to call + request = calculator.evaluate_request() + subtract_call = request.expression.init('call') + subtract_call.function = subtract + subtract_params = subtract_call.init('params', 2) + subtract_params[1].literal = 67.0 + + add_call = subtract_params[0].init('call') + add_call.function = add + add_params = add_call.init('params', 2) + add_params[0].literal = 123 + add_params[1].literal = 45 + + # Send the evaluate() request, read() the result, and wait for read() to finish. + eval_promise = request.send() + read_promise = eval_promise.value.read() + + response = await read_promise.a_wait() + assert response.value == 101 + + print("PASS") + + ''' + Note: a one liner version of building the previous request (I highly + recommend not doing it this way for such a complicated structure, but I + just wanted to demonstrate it is possible to set all of the fields with a + dictionary): + + eval_promise = calculator.evaluate( +{'call': {'function': subtract, + 'params': [{'call': {'function': add, + 'params': [{'literal': 123}, + {'literal': 45}]}}, + {'literal': 67.0}]}}) + ''' + + '''Make a request to evaluate 4 * 6, then use the result in two more + requests that add 3 and 5. + + Since evaluate() returns its result wrapped in a `Value`, we can pass + that `Value` back to the server in subsequent requests before the first + `evaluate()` has actually returned. Thus, this example again does only + one network round trip.''' + + print("Pipelining eval() calls... ", end="") + + # Get the "add" function from the server. + add = calculator.getOperator(op='add').func + # Get the "multiply" function from the server. + multiply = calculator.getOperator(op='multiply').func + + # Build the request to evaluate 4 * 6 + request = calculator.evaluate_request() + + multiply_call = request.expression.init("call") + multiply_call.function = multiply + multiply_params = multiply_call.init("params", 2) + multiply_params[0].literal = 4 + multiply_params[1].literal = 6 + + multiply_result = request.send().value + + # Use the result in two calls that add 3 and add 5. + + add_3_request = calculator.evaluate_request() + add_3_call = add_3_request.expression.init("call") + add_3_call.function = add + add_3_params = add_3_call.init("params", 2) + add_3_params[0].previousResult = multiply_result + add_3_params[1].literal = 3 + add_3_promise = add_3_request.send().value.read() + + add_5_request = calculator.evaluate_request() + add_5_call = add_5_request.expression.init("call") + add_5_call.function = add + add_5_params = add_5_call.init("params", 2) + add_5_params[0].previousResult = multiply_result + add_5_params[1].literal = 5 + add_5_promise = add_5_request.send().value.read() + + # Now wait for the results. + assert (await add_3_promise.a_wait()).value == 27 + assert (await add_5_promise.a_wait()).value == 29 + + print("PASS") + + '''Our calculator interface supports defining functions. Here we use it + to define two functions and then make calls to them as follows: + + f(x, y) = x * 100 + y + g(x) = f(x, x + 1) * 2; + f(12, 34) + g(21) + + Once again, the whole thing takes only one network round trip.''' + + print("Defining functions... ", end="") + + # Get the "add" function from the server. + add = calculator.getOperator(op='add').func + # Get the "multiply" function from the server. + multiply = calculator.getOperator(op='multiply').func + + # Define f. + request = calculator.defFunction_request() + request.paramCount = 2 + + # Build the function body. + add_call = request.body.init("call") + add_call.function = add + add_params = add_call.init("params", 2) + add_params[1].parameter = 1 # y + + multiply_call = add_params[0].init("call") + multiply_call.function = multiply + multiply_params = multiply_call.init("params", 2) + multiply_params[0].parameter = 0 # x + multiply_params[1].literal = 100 + + f = request.send().func + + # Define g. + request = calculator.defFunction_request() + request.paramCount = 1 + + # Build the function body. + multiply_call = request.body.init("call") + multiply_call.function = multiply + multiply_params = multiply_call.init("params", 2) + multiply_params[1].literal = 2 + + f_call = multiply_params[0].init("call") + f_call.function = f + f_params = f_call.init("params", 2) + f_params[0].parameter = 0 + + add_call = f_params[1].init("call") + add_call.function = add + add_params = add_call.init("params", 2) + add_params[0].parameter = 0 + add_params[1].literal = 1 + + g = request.send().func + + # OK, we've defined all our functions. Now create our eval requests. + + # f(12, 34) + f_eval_request = calculator.evaluate_request() + f_call = f_eval_request.expression.init("call") + f_call.function = f + f_params = f_call.init("params", 2) + f_params[0].literal = 12 + f_params[1].literal = 34 + f_eval_promise = f_eval_request.send().value.read() + + # g(21) + g_eval_request = calculator.evaluate_request() + g_call = g_eval_request.expression.init("call") + g_call.function = g + g_call.init('params', 1)[0].literal = 21 + g_eval_promise = g_eval_request.send().value.read() + + # Wait for the results. + assert (await f_eval_promise.a_wait()).value == 1234 + assert (await g_eval_promise.a_wait()).value == 4244 + + print("PASS") + + '''Make a request that will call back to a function defined locally. + + Specifically, we will compute 2^(4 + 5). However, exponent is not + defined by the Calculator server. So, we'll implement the Function + interface locally and pass it to the server for it to use when + evaluating the expression. + + This example requires two network round trips to complete, because the + server calls back to the client once before finishing. In this + particular case, this could potentially be optimized by using a tail + call on the server side -- see CallContext::tailCall(). However, to + keep the example simpler, we haven't implemented this optimization in + the sample server.''' + + print("Using a callback... ", end="") + + # Get the "add" function from the server. + add = calculator.getOperator(op='add').func + + # Build the eval request for 2^(4+5). + request = calculator.evaluate_request() + + pow_call = request.expression.init("call") + pow_call.function = PowerFunction() + pow_params = pow_call.init("params", 2) + pow_params[0].literal = 2 + + add_call = pow_params[1].init("call") + add_call.function = add + add_params = add_call.init("params", 2) + add_params[0].literal = 4 + add_params[1].literal = 5 + + # Send the request and wait. + response = await request.send().value.read().a_wait() + assert response.value == 512 + + print("PASS") + +if __name__ == '__main__': + # Using asyncio.run hits an asyncio ssl bug + # https://bugs.python.org/issue36709 + # asyncio.run(main(parse_args().host), loop=loop, debug=True) + loop = asyncio.get_event_loop() + loop.run_until_complete(main(parse_args().host)) diff --git a/examples/async_ssl_calculator_server.py b/examples/async_ssl_calculator_server.py new file mode 100755 index 0000000..e9b276b --- /dev/null +++ b/examples/async_ssl_calculator_server.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 + +import argparse +import asyncio +import logging +import os +import socket +import ssl + +import capnp +import calculator_capnp + + +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + +this_dir = os.path.dirname(os.path.abspath(__file__)) + + +class Server: + async def myreader(self): + while self.retry: + try: + # Must be a wait_for so we don't block on read() + data = await asyncio.wait_for( + self.reader.read(4096), + timeout=0.1 + ) + except asyncio.TimeoutError: + logger.debug("myreader timeout.") + continue + except Exception as err: + logger.error("Unknown myreader err: %s", err) + return False + await self.server.write(data) + logger.debug("myreader done.") + return True + + + async def mywriter(self): + while self.retry: + try: + # Must be a wait_for so we don't block on read() + data = await asyncio.wait_for( + self.server.read(4096), + timeout=0.1 + ) + self.writer.write(data.tobytes()) + except asyncio.TimeoutError: + logger.debug("mywriter timeout.") + continue + except Exception as err: + logger.error("Unknown mywriter err: %s", err) + return False + logger.debug("mywriter done.") + return True + + + async def myserver(self, reader, writer): + # Start TwoPartyServer using TwoWayPipe (only requires bootstrap) + self.server = capnp.TwoPartyServer(bootstrap=CalculatorImpl()) + self.reader = reader + self.writer = writer + self.retry = True + + # Assemble reader and writer tasks, run in the background + coroutines = [self.myreader(), self.mywriter()] + tasks = asyncio.gather(*coroutines, return_exceptions=True) + + while True: + self.server.poll_once() + # Check to see if reader has been sent an eof (disconnect) + if self.reader.at_eof(): + self.retry = False + break + await asyncio.sleep(0.01) + + # Make wait for reader/writer to finish (prevent possible resource leaks) + await tasks + + +def read_value(value): + '''Helper function to asynchronously call read() on a Calculator::Value and + return a promise for the result. (In the future, the generated code might + include something like this automatically.)''' + + return value.read().then(lambda result: result.value) + + +def evaluate_impl(expression, params=None): + '''Implementation of CalculatorImpl::evaluate(), also shared by + FunctionImpl::call(). In the latter case, `params` are the parameter + values passed to the function; in the former case, `params` is just an + empty list.''' + + which = expression.which() + + if which == 'literal': + return capnp.Promise(expression.literal) + elif which == 'previousResult': + return read_value(expression.previousResult) + elif which == 'parameter': + assert expression.parameter < len(params) + return capnp.Promise(params[expression.parameter]) + elif which == 'call': + call = expression.call + func = call.function + + # Evaluate each parameter. + paramPromises = [evaluate_impl(param, params) for param in call.params] + + joinedParams = capnp.join_promises(paramPromises) + # When the parameters are complete, call the function. + ret = (joinedParams + .then(lambda vals: func.call(vals)) + .then(lambda result: result.value)) + + return ret + else: + raise ValueError("Unknown expression type: " + which) + + +class ValueImpl(calculator_capnp.Calculator.Value.Server): + + "Simple implementation of the Calculator.Value Cap'n Proto interface." + + def __init__(self, value): + self.value = value + + def read(self, **kwargs): + return self.value + + +class FunctionImpl(calculator_capnp.Calculator.Function.Server): + + '''Implementation of the Calculator.Function Cap'n Proto interface, where the + function is defined by a Calculator.Expression.''' + + def __init__(self, paramCount, body): + self.paramCount = paramCount + self.body = body.as_builder() + + def call(self, params, _context, **kwargs): + '''Note that we're returning a Promise object here, and bypassing the + helper functionality that normally sets the results struct from the + returned object. Instead, we set _context.results directly inside of + another promise''' + + assert len(params) == self.paramCount + # using setattr because '=' is not allowed inside of lambdas + return evaluate_impl(self.body, params).then(lambda value: setattr(_context.results, 'value', value)) + + +class OperatorImpl(calculator_capnp.Calculator.Function.Server): + + '''Implementation of the Calculator.Function Cap'n Proto interface, wrapping + basic binary arithmetic operators.''' + + def __init__(self, op): + self.op = op + + def call(self, params, **kwargs): + assert len(params) == 2 + + op = self.op + + if op == 'add': + return params[0] + params[1] + elif op == 'subtract': + return params[0] - params[1] + elif op == 'multiply': + return params[0] * params[1] + elif op == 'divide': + return params[0] / params[1] + else: + raise ValueError('Unknown operator') + + +class CalculatorImpl(calculator_capnp.Calculator.Server): + + "Implementation of the Calculator Cap'n Proto interface." + + def evaluate(self, expression, _context, **kwargs): + return evaluate_impl(expression).then(lambda value: setattr(_context.results, 'value', ValueImpl(value))) + + def defFunction(self, paramCount, body, _context, **kwargs): + return FunctionImpl(paramCount, body) + + def getOperator(self, op, **kwargs): + return OperatorImpl(op) + + +def parse_args(): + parser = argparse.ArgumentParser(usage='''Runs the server bound to the\ +given address/port ADDRESS. ''') + + parser.add_argument("address", help="ADDRESS:PORT") + + return parser.parse_args() + + +async def new_connection(reader, writer): + server = Server() + await server.myserver(reader, writer) + + +async def main(): + address = parse_args().address + host = address.split(':') + addr = host[0] + port = host[1] + + # Setup SSL context + ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) + ctx.load_cert_chain(os.path.join(this_dir, 'selfsigned.cert'), os.path.join(this_dir, 'selfsigned.key')) + + # Handle both IPv4 and IPv6 cases + try: + print("Try IPv4") + server = await asyncio.start_server( + new_connection, + addr, port, + ssl=ctx, + family=socket.AF_INET + ) + except Exception: + print("Try IPv6") + server = await asyncio.start_server( + new_connection, + addr, port, + ssl=ctx, + family=socket.AF_INET6 + ) + + async with server: + await server.serve_forever() + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/examples/async_ssl_client.py b/examples/async_ssl_client.py index 548cac1..d69e196 100755 --- a/examples/async_ssl_client.py +++ b/examples/async_ssl_client.py @@ -1,13 +1,13 @@ #!/usr/bin/env python3 -import asyncio import argparse +import asyncio import os -import time -import capnp import socket import ssl +import time +import capnp import thread_capnp this_dir = os.path.dirname(os.path.abspath(__file__)) diff --git a/examples/async_ssl_server.py b/examples/async_ssl_server.py index ad0f6b3..575d482 100755 --- a/examples/async_ssl_server.py +++ b/examples/async_ssl_server.py @@ -11,7 +11,7 @@ import capnp import thread_capnp -logger = logging.getLogger(__) +logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) this_dir = os.path.dirname(os.path.abspath(__file__)) diff --git a/test/test_examples.py b/test/test_examples.py index dfd711a..448c67e 100644 --- a/test/test_examples.py +++ b/test/test_examples.py @@ -126,6 +126,7 @@ def test_addressbook_example(cleanup): assert ret == 0 +@pytest.mark.skipif(sys.platform == "win32", reason="Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. See https://github.com/capnproto/pycapnp/issues/196") def test_async_example(cleanup): address = '{}:36434'.format(hostname) server = 'async_server.py' @@ -133,6 +134,7 @@ def test_async_example(cleanup): run_subprocesses(address, server, client) +@pytest.mark.skipif(sys.platform == "win32", reason="Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. See https://github.com/capnproto/pycapnp/issues/196") def test_ssl_async_example(cleanup): address = '{}:36435'.format(hostname) server = 'async_ssl_server.py' @@ -140,8 +142,16 @@ def test_ssl_async_example(cleanup): run_subprocesses(address, server, client) +@pytest.mark.skipif(sys.platform == "win32", reason="Asyncio bug with libcapnp timer, likely due to asyncio starving some event loop. See https://github.com/capnproto/pycapnp/issues/196") def test_ssl_reconnecting_async_example(cleanup): address = '{}:36436'.format(hostname) server = 'async_ssl_server.py' client = 'async_reconnecting_ssl_client.py' run_subprocesses(address, server, client) + + +def test_async_ssl_calculator_example(cleanup): + address = '{}:36437'.format(hostname) + server = 'async_ssl_calculator_server.py' + client = 'async_ssl_calculator_client.py' + run_subprocesses(address, server, client)