Debugging https://github.com/micahflee/onionshare/issues/812

onion-bug.py

#!/usr/bin/env python3
import threading
import queue
import time
import os
import sys
import base64
import hashlib
from http.server import BaseHTTPRequestHandler, HTTPServer
from stem.control import Controller

q = queue.Queue()


class RequestHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # Send response status code
        self.send_response(200)

        # Send headers
        self.send_header('Content-type','plain/text')
        self.end_headers()

        # Send 1mb of A characters
        message = 'A'*1024*1024
        self.wfile.write(bytes(message, "utf8"))

        # Launch a thread that will shutdown the http server after a moment
        # (This is in a new thread, because this method needs to return to
        # end the http request)
        def shutdown_httpd(server):
            time.sleep(0.1)
            print('shutting down http server')
            q.put('shutdown')
            server.shutdown()
        t = threading.Thread(target=shutdown_httpd, args=(self.server,))
        t.start()

        return


def start_http():
    print("http service: http://127.0.0.1:8080/")
    print("")
    httpd = HTTPServer(('127.0.0.1', 8080), RequestHandler)
    httpd.serve_forever()

def usage():
    print("Usage: {} [v2,v3]".format(sys.argv[0]))

def main():
    # v2 or v3 onion service?
    if len(sys.argv) == 2:
        if sys.argv[1] == "v2":
            key_type = "NEW"
            key_content = "RSA1024"
        elif sys.argv[1] == "v3":
            key_type = "NEW"
            key_content = "ED25519-V3"
        else:
            usage()
            sys.exit()
    else:
        usage()
        sys.exit()

    # Start the http server
    t = threading.Thread(target=start_http)
    t.start()

    # Connect to tor controller provided by Tor Browser
    c = Controller.from_port(port=9151)
    c.authenticate()

    print("starting onion service with: key_type='{}', key_content='{}'".format(key_type, key_content))
    res = c.create_ephemeral_hidden_service(
        { 80: 8080 },
        await_publication=True,
        key_type=key_type,
        key_content=key_content
    )
    print("http://{}.onion/".format(res.service_id))
    print("")

    # Wait until the 1mb download completes
    while True:
        if q.qsize() > 0:
            break
        time.sleep(0.1)

    # Stop the onion service
    #time.sleep(5) # let the download finish
    for onion in c.list_ephemeral_hidden_services():
        c.remove_ephemeral_hidden_service(onion)

if __name__ == '__main__':
    main()