quickjs-tart

quickjs-based runtime for wallet-core logic
Log | Files | Refs | README | LICENSE

test_https.py (4217B)

      1 #!/usr/bin/env python3
      2 
      3 import http.server
      4 import os
      5 from pathlib import Path
      6 import shutil
      7 import ssl
      8 import subprocess
      9 import sys
     10 import tempfile
     11 import threading
     12 
     13 
     14 class TestHandler(http.server.BaseHTTPRequestHandler):
     15     def do_GET(self):
     16         body = b"ok"
     17         self.send_response(200)
     18         self.send_header("Content-Length", str(len(body)))
     19         self.end_headers()
     20         self.wfile.write(body)
     21 
     22     def log_message(self, format, *args):
     23         pass
     24 
     25 
     26 def run_checked(argv):
     27     subprocess.run(
     28         argv,
     29         check=True,
     30         stdout=subprocess.DEVNULL,
     31         stderr=subprocess.PIPE,
     32         text=True,
     33     )
     34 
     35 
     36 def make_certificates(openssl, directory):
     37     ca_key = directory / "ca.key"
     38     ca_cert = directory / "ca.crt"
     39     server_key = directory / "server.key"
     40     server_csr = directory / "server.csr"
     41     server_cert = directory / "server.crt"
     42     server_extensions = directory / "server.ext"
     43 
     44     run_checked([
     45         openssl, "req", "-x509", "-newkey", "rsa:2048", "-sha256",
     46         "-nodes", "-days", "1", "-subj", "/CN=quickjs-tart test CA",
     47         "-addext", "basicConstraints=critical,CA:TRUE",
     48         "-addext", "keyUsage=critical,keyCertSign,cRLSign",
     49         "-keyout", str(ca_key), "-out", str(ca_cert),
     50     ])
     51     run_checked([
     52         openssl, "req", "-new", "-newkey", "rsa:2048", "-sha256",
     53         "-nodes", "-subj", "/CN=127.0.0.1", "-keyout", str(server_key),
     54         "-out", str(server_csr),
     55     ])
     56     server_extensions.write_text(
     57         "basicConstraints=critical,CA:FALSE\n"
     58         "keyUsage=critical,digitalSignature,keyEncipherment\n"
     59         "extendedKeyUsage=serverAuth\n"
     60         "subjectAltName=IP:127.0.0.1\n",
     61         encoding="ascii",
     62     )
     63     run_checked([
     64         openssl, "x509", "-req", "-in", str(server_csr), "-CA",
     65         str(ca_cert), "-CAkey", str(ca_key), "-CAcreateserial", "-days",
     66         "1", "-sha256", "-extfile", str(server_extensions), "-out",
     67         str(server_cert),
     68     ])
     69     return ca_cert, server_cert, server_key
     70 
     71 
     72 def run_https_probe(qtart, ca_cert, port, ca_environment_variable):
     73     marker = f"https-ok-{ca_environment_variable}"
     74     expression = (
     75         f'os.fetchHttp("https://127.0.0.1:{port}/", {{}}).promise.then('
     76         f'(r) => print(r.status === 200 ? "{marker}" : "bad-status"), '
     77         '(e) => print("https-error", e));'
     78     )
     79     environment = os.environ.copy()
     80     environment.pop("CURL_CA_BUNDLE", None)
     81     environment.pop("SSL_CERT_DIR", None)
     82     environment.pop("SSL_CERT_FILE", None)
     83     environment[ca_environment_variable] = str(ca_cert)
     84     result = subprocess.run(
     85         [qtart, "--std", "-e", expression],
     86         env=environment,
     87         capture_output=True,
     88         text=True,
     89         timeout=20,
     90     )
     91     if result.returncode != 0 or marker not in result.stdout:
     92         sys.stderr.write(result.stdout)
     93         sys.stderr.write(result.stderr)
     94         return False
     95     return True
     96 
     97 
     98 def main():
     99     if len(sys.argv) != 2:
    100         print("usage: test_https.py QTART", file=sys.stderr)
    101         return 2
    102 
    103     qtart = sys.argv[1]
    104     openssl = shutil.which("openssl")
    105     if openssl is None:
    106         print("openssl is required", file=sys.stderr)
    107         return 2
    108     with tempfile.TemporaryDirectory(prefix="quickjs-tart-https-") as tmp:
    109         ca_cert, server_cert, server_key = make_certificates(
    110             openssl,
    111             Path(tmp),
    112         )
    113         server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), TestHandler)
    114         tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    115         tls_context.load_cert_chain(server_cert, server_key)
    116         server.socket = tls_context.wrap_socket(server.socket, server_side=True)
    117         server_thread = threading.Thread(target=server.serve_forever)
    118         server_thread.start()
    119         try:
    120             port = server.server_address[1]
    121             if not run_https_probe(qtart, ca_cert, port, "SSL_CERT_FILE"):
    122                 return 1
    123             if not run_https_probe(qtart, ca_cert, port, "CURL_CA_BUNDLE"):
    124                 return 1
    125         finally:
    126             server.shutdown()
    127             server.server_close()
    128             server_thread.join()
    129     return 0
    130 
    131 
    132 if __name__ == "__main__":
    133     raise SystemExit(main())