import argparse
import ssl
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class Handler(BaseHTTPRequestHandler):
response_body = b"childflow-demo\n"
def do_GET(self):
self.send_response(200)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(self.response_body)))
self.end_headers()
self.wfile.write(self.response_body)
def do_HEAD(self):
self.send_response(200)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(self.response_body)))
self.end_headers()
def log_message(self, fmt, *args):
return
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, required=True)
parser.add_argument("--body", default="childflow-demo")
parser.add_argument("--tls-cert")
parser.add_argument("--tls-key")
args = parser.parse_args()
handler = type("ConfiguredHandler", (Handler,), {"response_body": (args.body + "\n").encode()})
server = ThreadingHTTPServer((args.host, args.port), handler)
if args.tls_cert and args.tls_key:
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(args.tls_cert, args.tls_key)
server.socket = context.wrap_socket(server.socket, server_side=True)
server.serve_forever()
if __name__ == "__main__":
main()