PythonでSSL対応ローカルサーバーをつくる


ソフトの説明

PythonでローカルむけSSL対応のローカルサーバーをたてます。
Pythonで作成した自己SSL証明書を読み込み指定のフォルダ以下をROOTとしたローカルサーバーをたてます。


プログラムの説明

参考にしたPythonのプログラムがTkinterによるGUI付きIP指定ローカルサーバーでしたのでこれを改造し127.0.0.1専用のローカルSSLサーバーとします。
QUEUEで起動する仕様でしたのでSSL証明書があれば自動で読み込みIcon化してポート443のSSLサーバーをたてる仕様としました。 SSL証明書がなければポート80のサーバーをたてます。

SSL証明書作成アプリ



sslserver

プログラムのカレントフォルダの証明書を呼び込みますのでSSL証明書作成アプリの横に配置します。
ここに「root」フォルダを作成してください。起動時このフォルダをデフォルトのルートとして起動します。
アイコンが必要になりますのでこちらからどうぞ


sslフォルダ

この状態にて起動しブラウザのアドレス「https://127.0.0.1:443」を指定すると無事ルートに置いたJSONファイルが確認できました。
証明書がないと「http://127.0.0.1:80」で接続可能です。


SSLTEST

ssl_server.py

#!/usr/bin/env python

import os
import sys
import threading
import tkinter as tk
from tkinter import ttk
from tkinter import messagebox
from tkinter import filedialog
from http.server import HTTPServer, BaseHTTPRequestHandler
from http.server import SimpleHTTPRequestHandler
from socketserver import ThreadingMixIn
import queue as Queue
#import ssl
#from ipaddress import ip_address
#from urllib import request
#import tempfile



USE_HTTPS_PEM = False
USE_HTTPS_KEY = False
PORT = 80
PEM =""
PEM_KEY=""
public_key=""
Fpath = ""

class Pem:
    def isthere(self):
        cpath = os.getcwd()
        #print(os.listdir(cpath))
        global USE_HTTPS_PEM
        global USE_HTTPS_KEY
        global PORT
        global PEM
        global PEM_KEY


        if not os.listdir(cpath):
            USE_HTTPS_PEM = False
            USE_HTTPS_KEY = False
            PORT = 80

        pathf = './127.0.0.1.pem'

        if os.path.exists(pathf):
            USE_HTTPS_PEM = True
            USE_HTTPS_KEY = False
            PORT = 443
            PEM = os.getcwd() + '\\127.0.0.1.pem'
            PEM_KEY = os.getcwd() + '\\127.0.0.1_key.pem'

        #pathf = './127.0.0.1.crt'

        #if os.path.exists(pathf):

        #    USE_HTTPS_PEM = False
        #    USE_HTTPS_KEY = True
        #    PORT = 443
        #    PEM = os.getcwd()+'\\127.0.0.1.crt'
        #    PEM_KEY = os.getcwd()+'\\127.0.0.1.key'



class CORSHTTPRequestHandler(SimpleHTTPRequestHandler,BaseHTTPRequestHandler):
    #consle off for nuitka
    def log_message(self, format, *args):
        pass
    def send_head(self):
        """Common code for GET and HEAD commands.
        This sends the response code and MIME headers.
        Return value is either a file object (which has to be copied
        to the outputfile by the caller unless the command was HEAD,
        and must be closed by the caller under all circumstances), or
        None, in which case the caller has nothing further to do.
        """
        path = self.translate_path(self.path)
        #path = os.getcwd()
        #print(path)

        f = None
        if os.path.isdir(path):
            if not self.path.endswith('/'):
                # redirect browser - doing basically what apache does
                self.send_response(301)
                self.send_header("Location", self.path + "/")
                self.end_headers()
                return None
            for index in "index.html", "index.htm":
                index = os.path.join(path, index)
                if os.path.exists(index):
                    path = index
                    break
            else:
                return self.list_directory(path)
        ctype = self.guess_type(path)
        try:
            # Always read in binary mode. Opening files in text mode may cause
            # newline translations, making the actual size of the content
            # transmitted *less* than the content-length!
            f = open(path, 'rb')
        except IOError:
            self.send_error(404, "File not found")
            return None
        self.send_response(200)
        self.send_header("Content-type", ctype)
        fs = os.fstat(f.fileno())
        self.send_header("Content-Length", str(fs[6]))
        self.send_header("Last-Modified", self.date_time_string(fs.st_mtime))
        #self.send_header("Access-Control-Allow-Origin", "https://booleaneffect.info")
        self.send_header("Access-Control-Allow-Origin", "*")

        self.end_headers()
        return f



class ThreadingSimpleServer(ThreadingMixIn, HTTPServer):
    pass

class ServerThread(threading.Thread):

    def __init__(self):

        threading.Thread.__init__(self)
        self.queue = server_queue
        self.is_server_running = False
        self.daemon = True


    def create_server(self, port=PORT):

        self.server = ThreadingSimpleServer(("127.0.0.1", port), CORSHTTPRequestHandler, HTTPServer)
        if USE_HTTPS_PEM:
            import ssl
            self.server.socket = ssl.wrap_socket(self.server.socket,
                                                 server_side=True,
                                                 do_handshake_on_connect=True,
                                                 certfile=PEM,
                                                 keyfile=PEM_KEY)
        if USE_HTTPS_KEY:
            import ssl
            self.server.socket = ssl.wrap_socket(self.server.socket,
                                                server_side=True,
                                                do_handshake_on_connect=True,
                                                certfile=PEM,
                                                keyfile=PEM_KEY)

        self.serv_info = self.server.socket.getsockname()

    def run(self):
        while True:
            if not self.queue.empty():
                i = self.queue.get()
                if i[0] == 'start':
                    self.create_server(int(i[1]))
                    self.is_server_running = True
                    #print('SERVING @: {}:{}'.format(self.serv_info[0], self.serv_info[1]))
                    self.server.serve_forever()
                elif i[0] == 'stop':
                    self.is_server_running = False
                    #print('SERVER SHUTDOWN')
                    self.server = None



class Gui(tk.Tk):
    def __init__(self):
        tk.Tk.__init__(self)
        self.server_running = False

        #f0 = ttk.Frame(self)
        #f1 = ttk.Frame(self)
        f2 = ttk.Frame(self)
        f3 = ttk.Frame(self)
        f4 = ttk.Frame(self)
        f5 = ttk.Frame(self)

        #ttk.Label(f1, text='Your IP address:  {}'.format(ip_address), font='Helvetica 10 bold').pack()
        ttk.Label(f2, text='ポート: ', font='Helvetica 10 bold').pack(side='left')
        self.port = ttk.Entry(f2, width=6)
        self.port.insert('end', PORT)
        self.port.pack(side='left', padx=0, pady=10)
        ttk.Label(f3, text='フォルダパス: ', font='Helvetica 10 bold').pack(side='left')
        self.path_entry = ttk.Entry(f3, width=55)
        self.path_entry.insert('end', OWD)
        self.path_entry.pack(side='left')
        ttk.Button(f3, text='Browse', command=self.browse).pack(side='left', padx=10)
        ttk.Label(f4, text='スタート/ストップ: ', font='Helvetica 10 bold').pack(side='left')
        self.start_btn = ttk.Button(f4, text='Start server', command=self.start_clicked)
        self.start_btn.pack(side='left', padx=10)
        self.stop_btn = ttk.Button(f4, text='Stop server', command=self.stop_clicked, state='disabled')
        self.stop_btn.pack(side='left', padx=10)
        ttk.Label(f5, text='Server state: ', font='Helvetica 10 bold').pack(side='left')
        self.serverstate_lbl = ttk.Label(f5, text='Stopped', font='Helvetica 15 bold', foreground='red')
        self.serverstate_lbl.pack(side='left')

        #f0.grid(column=1, row=0, columnspan=3, padx=10, pady=10, sticky='w')
        #f1.grid(column=1, row=1, padx=10, pady=10, sticky='w')
        f2.grid(column=1, row=2, padx=10, pady=10, sticky='w')
        f3.grid(column=1, row=3, columnspan=3, padx=10, pady=10, sticky='w')
        f4.grid(column=1, row=4, padx=10, pady=10, sticky='w')
        f5.grid(column=2, row=4, padx=10, pady=10, sticky='w')

    def browse(self):
        path = filedialog.askdirectory()
        if os.path.isdir(path):
            self.path_entry.delete('0', tk.END)
            self.path_entry.insert('end', path)


    def start_clicked(self):
        # if server_thread.is_server_running:
        # messagebox.showerror('Already running','The server is already running. Please stop the server first!')
        # return
        if switch_dir(self.path_entry.get()):

            #port = check_port(self.port.get().strip())
            port = PORT
            if not port:
                #messagebox.showerror('Error', 'Please enter a valid port number between 1-65535')
                return
            server_queue.put(('start', port))

            # wait for the server_thread to switch is_server_running = True
            while not server_thread.is_server_running:
                pass
            self.toggle_state_widgets()
            #messagebox.showinfo('NOW SERVING', 'Now serving {} @{}:{}'.format(os.getcwd(), ip_address, port))
        else:
            messagebox.showerror('Invalid path', 'The path you have provided does not exist!')

    def stop_clicked(self, notify=True):
        if server_thread.is_server_running:
            server_queue.put(('stop', ''))
            server_thread.server.shutdown()

            # wait for the server thread to switch isruuning = False
            while server_thread.is_server_running:
                pass
            self.toggle_state_widgets()
            #if notify:
            #    messagebox.showinfo('SERVER SHUTDOWN', 'The server has been shutdown.')

    def toggle_state_widgets(self):
        if server_thread.is_server_running:
            self.start_btn.configure(state='disabled')
            self.stop_btn.configure(state='normal')
            txt = 'Serving'
            color = 'green'
        else:
            self.start_btn.configure(state='normal')
            self.stop_btn.configure(state='disabled')
            txt = 'Stopped'
            color = 'red'
        self.serverstate_lbl.configure(text=txt, foreground=color)




def switch_dir(path):
    if os.path.isdir(path):
        os.chdir(path)
        return True
    if path == '':
        os.chdir(Fpath)
        return True
    else:
        return


def check_port(port):
    try:
        p = int(port)
        if p > 65535:
            return
        else:
            return p
    except Exception as e:
        print(e)
        return


if __name__ == '__main__':


    OWD = os.path.dirname(sys.argv[0])
    OWD = os.path.join(OWD, "root").replace('/', os.sep)
    Fpath = OWD



    pem = Pem()
    pem.isthere()

    # 起動時にlocalhostを立てる。最小化
    root = Gui()
    root.title('LocalServer')
    root.iconbitmap(default='sk.ico')
    root.iconify()
    root.after(1, root.start_clicked)

    server_queue = Queue.Queue()
    server_thread = ServerThread()
    server_thread.start()
    root.mainloop()

参考リンク Spcial thanks dojafoja / GUI-python-server
logo