test(poc): Secure Control API using cross-pinning mTLS

Note that we are leveraging nix-shell to provide portable dependency
handling.

Co-authored-by: Roberto Abdelkader Martínez Pérez <robertomartinezp@gmail.com>

testutils/poc/README.rst

@@ -17,7 +17,7 @@ at any moment.
             _,-._
            ; ___ :           ,------------------------------.
        ,--' (. .) '--.__    |                                |
-     _;      |||        \   |     Arrr!! Be ye warned!       |
+     _;      |||        \   |     Arrr!! Ye be warned!       |
     '._,-----''';=.____,"   |                                |
       /// < o>   |##|       |                                |
       (o        \`--'       //`-----------------------------'

testutils/poc/kapow

@@ -1,4 +1,7 @@
-#!/usr/bin/env python
+#! /usr/bin/env nix-shell
+#! nix-shell -i python3.7 -p python37 python37Packages.aiohttp python37Packages.requests python37Packages.click
+#
+# TODO: maybe add an option (cli) to supply the external address
 
 #
 # Copyright 2019 Banco Bilbao Vizcaya Argentaria, S.A.
@@ -20,16 +23,28 @@ from collections import namedtuple
 from urllib.parse import urlparse
 from uuid import uuid4
 import asyncio
+import binascii
+import contextlib
+import datetime
 import io
+import ipaddress
 import json
 import logging
 import os
 import shlex
 import ssl
 import sys
+import tempfile
+import uuid
 
 from aiohttp import web, StreamReader
 from aiohttp.web_urldispatcher import UrlDispatcher
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives import serialization
+from cryptography import x509
+from cryptography.x509.oid import NameOID
 import click
 import requests
 
@@ -38,6 +53,79 @@ log = logging.getLogger('kapow')
 loop = asyncio.new_event_loop()
 asyncio.set_event_loop(loop)
 
+KAPOW_CONTROL_URL="https://localhost:8081"
+KAPOW_DATA_URL="http://localhost:8082"
+
+########################################################################
+#                           HTTPS Management                           #
+########################################################################
+
+def generate_ssl_cert(name, alt=None):
+    # Generate our key
+    key = rsa.generate_private_key(
+        public_exponent=65537,
+        key_size=2048,
+    )
+    # Various details about who we are. For a self-signed certificate the
+    # subject and issuer are always the same.
+    subject = issuer = x509.Name([
+        x509.NameAttribute(NameOID.COMMON_NAME, name),
+    ])
+
+    cert = x509.CertificateBuilder().subject_name(
+        subject
+    ).issuer_name(
+        issuer
+    ).public_key(
+        key.public_key()
+    ).serial_number(
+        x509.random_serial_number()
+    ).not_valid_before(
+        datetime.datetime.utcnow()
+    ).not_valid_after(
+        datetime.datetime.utcnow() + datetime.timedelta(days=3650)
+    )
+
+    if alt is not None:
+        try:
+            ip = ipaddress.ip_address(alt)
+        except:
+            cert = cert.add_extension(
+                x509.SubjectAlternativeName([x509.DNSName(alt)]),
+                critical=True,
+            )
+        else:
+            cert = cert.add_extension(
+                x509.SubjectAlternativeName([x509.IPAddress(ip)]),
+                critical=True,
+            )
+        finally:
+            cert = cert.add_extension(
+                x509.ExtendedKeyUsage(
+                    [x509.oid.ExtendedKeyUsageOID.SERVER_AUTH],
+                ),
+                critical=True
+            )
+    else:
+        cert=cert.add_extension(
+            x509.ExtendedKeyUsage(
+                [x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH],
+            ),
+            critical=True
+        )
+
+    cert = cert.sign(key, hashes.SHA256())
+
+    key_bytes = key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            encryption_algorithm=serialization.NoEncryption()
+        )
+    crt_bytes = cert.public_bytes(serialization.Encoding.PEM)
+
+    return (key_bytes, crt_bytes)
+
+
 ########################################################################
 #                         Resource Management                          #
 ########################################################################
@@ -255,8 +343,7 @@ def handle_route(entrypoint, command):
             shell_task = await asyncio.create_subprocess_shell(
                 args,
                 env={**os.environ,
-                     "KAPOW_DATA_URL": "http://localhost:8081",
+                     "KAPOW_DATA_URL": KAPOW_DATA_URL,
-                     "KAPOW_CONTROL_URL": "http://localhost:8081",
                      "KAPOW_HANDLER_ID": id
                      },
                 stdin=asyncio.subprocess.DEVNULL)
@@ -279,7 +366,7 @@ def handle_route(entrypoint, command):
 
 
 def error_body(reason):
-    return {"reason": reason, "foo": "bar"}
+    return {"reason": reason}
 
 def get_routes(app):
     async def _get_routes(request):
@@ -399,41 +486,31 @@ def delete_route(app):
 #                            aiohttp webapp                            #
 ########################################################################
 
+async def report_result(proc):
+    await proc.communicate()
+    print(f"Process exited with code {proc.returncode}")
+
 
 async def run_init_script(app, scripts, interactive):
     """
     Run the init script if given, then wait for the shell to finish.
 
     """
-    if not scripts:
+    for script in scripts:
-        # No script given
+        try:
-        if not interactive:
+            result = await asyncio.create_subprocess_exec(
-            return
+                script,
+                env={**os.environ,
+                     "KAPOW_CONTROL_CLIENT_CERT": app["client_cert"],
+                     "KAPOW_CONTROL_CLIENT_KEY": app["client_key"],
+                     "KAPOW_CONTROL_SERVER_CERT": app["server_cert"],
+                     "KAPOW_CONTROL_URL": KAPOW_CONTROL_URL,
+                     })
+        except Exception as exc:
+            print(exc)
         else:
-            cmd = "/bin/bash"
+            asyncio.create_task(report_result(result))
-    else:
-        def build_filenames():
-            for filename in scripts:
-                yield shlex.quote(filename)
-                yield "<(echo)"
-        filenames = " ".join(build_filenames())
-        if interactive:
-            cmd = f"/bin/bash --init-file <(cat {filenames})"
-        else:
-            cmd = f"/bin/bash <(cat {filenames})"
 
-    shell_task = await asyncio.create_subprocess_shell(
-        cmd,
-        executable="/bin/bash",
-        env={**os.environ,
-             "KAPOW_DATA_URL": "http://localhost:8081",
-             "KAPOW_CONTROL_URL": "http://localhost:8081"
-             })
-
-    await shell_task.wait()
-    if interactive:
-        await app.cleanup()
-        os._exit(shell_task.returncode)
 
 
 class InvalidRouteError(Exception):
@@ -481,7 +558,28 @@ async def start_background_tasks(app):
     app["debug_tasks"] = loop.create_task(run_init_script(app, app["scripts"], app["interactive"]))
 
 
-async def start_kapow_server(bind, scripts, certfile=None, interactive=False, keyfile=None):
+def reduce_addr(addr):
+    """Drop the port part from an `addr:port` string (IPv6 aware)"""
+    addr, *_ = addr.rsplit(':', 1)
+    if addr.startswith('[') and addr.endswith(']'):
+        return addr[1:-1]
+    else:
+        return addr
+
+
+async def start_kapow_server(user_bind,
+                             control_bind,
+                             data_bind,
+                             scripts,
+                             certfile=None,
+                             interactive=False,
+                             keyfile=None,
+                             control_reachable_addr="localhost:8081"):
+    global KAPOW_CONTROL_URL
+    KAPOW_CONTROL_URL=f"https://{control_reachable_addr}"
+    #
+    # USER
+    #
     user_app = DynamicApplication(client_max_size=1024**3)
     user_app["user_routes"] = list()  # [KapowRoute]
     user_runner = web.AppRunner(user_app)
@@ -492,32 +590,73 @@ async def start_kapow_server(bind, scripts, certfile=None, interactive=False, ke
         ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
         ssl_context.load_cert_chain(certfile, keyfile)
 
-    ip, port = bind.split(':')
+    user_ip, user_port = user_bind.rsplit(':', 2)
-    user_site = web.TCPSite(user_runner, ip, int(port), ssl_context=ssl_context)
+    user_site = web.TCPSite(user_runner, user_ip, int(user_port),
+                            ssl_context=ssl_context)
     await user_site.start()
 
+    #
+    # CONTROL
+    #
+    alternate_name = reduce_addr(control_reachable_addr)
+    srv_key_bytes, srv_crt_bytes = generate_ssl_cert("control", alternate_name)
+    cli_key_bytes, cli_crt_bytes = generate_ssl_cert("control")
+
+    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+    with tempfile.NamedTemporaryFile(suffix=".pem", delete=True) as pem_file, \
+         tempfile.NamedTemporaryFile(suffix=".key", delete=True) as key_file, \
+         tempfile.NamedTemporaryFile(suffix=".pem", delete=True) as cli_crt_file:
+        pem_file.write(srv_crt_bytes)
+        pem_file.flush()
+        key_file.write(srv_key_bytes)
+        key_file.flush()
+        cli_crt_file.write(cli_crt_bytes)
+        cli_crt_file.flush()
+
+        context.verify_mode = ssl.CERT_REQUIRED
+        context.load_cert_chain(pem_file.name, key_file.name)
+        context.load_verify_locations(cafile=cli_crt_file.name)
+
     control_app = web.Application(client_max_size=1024**3)
     control_app.add_routes([
-        # Control API
         web.get('/routes', get_routes(user_app)),
         web.get('/routes/{id}', get_route(user_app)),
         web.post('/routes', append_route(user_app)),
         web.put('/routes', insert_route(user_app)),
         web.delete('/routes/{id}', delete_route(user_app)),
-
-        # Data API
-        web.get('/handlers/{id}/{field:.*}', get_field),
-        web.put('/handlers/{id}/{field:.*}', set_field),
     ])
     control_app["scripts"] = scripts
+    control_app["client_cert"] = cli_crt_bytes 
+    control_app["client_key"] = cli_key_bytes 
+    control_app["server_cert"] = srv_crt_bytes 
     control_app["interactive"] = interactive
     control_app.on_startup.append(start_background_tasks)
 
     control_runner = web.AppRunner(control_app)
+
     await control_runner.setup()
-    control_site = web.TCPSite(control_runner, '127.0.0.1', 8081)
+
+    control_ip, control_port = control_bind.rsplit(':', 2)
+    control_site = web.TCPSite(control_runner, control_ip,
+                               int(control_port), ssl_context=context)
     await control_site.start()
 
+    #
+    # DATA
+    #
+    data_app = web.Application(client_max_size=1024**3)
+    data_app.add_routes([
+        # Data API
+        web.get('/handlers/{id}/{field:.*}', get_field),
+        web.put('/handlers/{id}/{field:.*}', set_field),
+    ])
+
+    data_runner = web.AppRunner(data_app)
+
+    await data_runner.setup()
+    data_ip, data_port = data_bind.rsplit(':', 2)
+    data_site = web.TCPSite(data_runner, data_ip, int(data_port))
+    await data_site.start()
 
 
 ########################################################################
@@ -536,13 +675,25 @@ def kapow(ctx):
 @click.option("--certfile", default=None)
 @click.option("--keyfile", default=None)
 @click.option("--bind", default="0.0.0.0:8080")
+@click.option("--control-bind", default="0.0.0.0:8081")
+@click.option("--data-bind", default="0.0.0.0:8082")
+@click.option("--control-reachable-addr", default="localhost:8081")
 @click.option("-i", "--interactive", is_flag=True)
 @click.argument("scripts", nargs=-1)
-def server(certfile, keyfile, bind, interactive, scripts):
+def server(certfile, keyfile, bind, interactive, scripts,
+           control_reachable_addr, control_bind, data_bind):
     if bool(certfile) ^ bool(keyfile):
         print("For SSL both 'certfile' and 'keyfile' should be provided.")
         sys.exit(1)
-    loop.run_until_complete(start_kapow_server(bind, scripts, certfile, interactive, keyfile))
+    loop.run_until_complete(
+        start_kapow_server(bind,
+                           control_bind,
+                           data_bind,
+                           scripts,
+                           certfile,
+                           interactive,
+                           keyfile,
+                           control_reachable_addr))
     loop.run_forever()
 
 @kapow.group(help="Manage current server HTTP routes")
@@ -550,59 +701,79 @@ def route():
     pass
 
 
+@contextlib.contextmanager
+def kapow_control_certs():
+    with tempfile.NamedTemporaryFile(suffix='.crt', encoding='utf-8', mode='w') as srv_cert, \
+         tempfile.NamedTemporaryFile(suffix='.crt', encoding='utf-8', mode='w') as cli_cert, \
+         tempfile.NamedTemporaryFile(suffix='.key', encoding='utf-8', mode='w') as cli_key:
+        srv_cert.write(os.environ["KAPOW_CONTROL_SERVER_CERT"])
+        srv_cert.file.flush()
+        cli_cert.write(os.environ["KAPOW_CONTROL_CLIENT_CERT"])
+        cli_cert.file.flush()
+        cli_key.write(os.environ["KAPOW_CONTROL_CLIENT_KEY"])
+        cli_key.file.flush()
+        session=requests.Session()
+        session.verify=srv_cert.name
+        session.cert=(cli_cert.name, cli_key.name)
+        yield session
+
+
 @route.command("add")
 @click.option("-c", "--command", nargs=1)
 @click.option("-e", "--entrypoint", default="/bin/sh -c")
 @click.option("-X", "--method", default="GET")
-@click.option("--url", envvar='KAPOW_CONTROL_URL')
+@click.option("--url", envvar='KAPOW_CONTROL_URL', default=KAPOW_CONTROL_URL)
 @click.argument("url_pattern", nargs=1)
 @click.argument("command_file", required=False)
 def route_add(url_pattern, entrypoint, command, method, url, command_file):
-    if command:
+    with kapow_control_certs() as requests:
-        # Command is given inline
+        if command:
-        source = command
+            # Command is given inline
-    elif command_file is None:
+            source = command
-        # No command
+        elif command_file is None:
-        source = ""
+            # No command
-    elif command_file == '-':
+            source = ""
-        # Read commands from stdin
+        elif command_file == '-':
-        source = sys.stdin.read()
+            # Read commands from stdin
-    else:
+            source = sys.stdin.read()
-        # Read commands from a file
+        else:
-        with open(command_file, 'r', encoding='utf-8') as handler:
+            # Read commands from a file
-            source = handler.read()
+            with open(command_file, 'r', encoding='utf-8') as handler:
+                source = handler.read()
 
-    response = requests.post(f"{url}/routes",
+        response = requests.post(f"{url}/routes",
-                             json={"method": method,
+                                 json={"method": method,
-                                   "url_pattern": url_pattern,
+                                       "url_pattern": url_pattern,
-                                   "entrypoint": entrypoint,
+                                       "entrypoint": entrypoint,
-                                   "command": source})
+                                       "command": source})
-    response.raise_for_status()
+        response.raise_for_status()
-    print(json.dumps(response.json(), indent=2))
+        print(json.dumps(response.json(), indent=2))
 
 
 @route.command("remove")
-@click.option("--url", envvar='KAPOW_CONTROL_URL')
+@click.option("--url", envvar='KAPOW_CONTROL_URL', default=KAPOW_CONTROL_URL)
 @click.argument("route-id")
 def route_remove(route_id, url):
-    response = requests.delete(f"{url}/routes/{route_id}")
+    with kapow_control_certs() as requests:
-    response.raise_for_status()
+        response = requests.delete(f"{url}/routes/{route_id}")
+        response.raise_for_status()
 
 
 @route.command("list")
-@click.option("--url", envvar='KAPOW_CONTROL_URL')
+@click.option("--url", envvar='KAPOW_CONTROL_URL', default=KAPOW_CONTROL_URL)
 @click.argument("route-id", nargs=1, required=False, default=None)
 def route_list(route_id, url):
-    if route_id is None:
+    with kapow_control_certs() as requests:
-        response = requests.get(f"{url}/routes")
+        if route_id is None:
-    else:
+            response = requests.get(f"{url}/routes")
-        response = requests.get(f"{url}/routes/{route_id}")
+        else:
-    response.raise_for_status()
+            response = requests.get(f"{url}/routes/{route_id}")
-    print(json.dumps(response.json(), indent=2))
+        response.raise_for_status()
+        print(json.dumps(response.json(), indent=2))
 
 
 @kapow.command("set", help="Set data from the current context")
-@click.option("--url", envvar='KAPOW_DATA_URL')
+@click.option("--url", envvar='KAPOW_DATA_URL', default=KAPOW_DATA_URL)
 @click.option("--handler-id", envvar='KAPOW_HANDLER_ID')
 @click.argument("path", nargs=1)
 @click.argument("value", required=False)
@@ -622,7 +793,7 @@ def kapow_set(url, handler_id, path, value):
 
 
 @kapow.command("get", help="Get data from the current context")
-@click.option("--url", envvar='KAPOW_DATA_URL')
+@click.option("--url", envvar='KAPOW_DATA_URL', default=KAPOW_DATA_URL)
 @click.option("--handler-id", envvar='KAPOW_HANDLER_ID')
 @click.argument("path", nargs=1)
 def kapow_get(url, handler_id, path):