Serveur HTTP de base Python – Code Exchange Stack Exchange

C'est un serveur HTTP extrêmement simple en Python utilisant la bibliothèque de sockets, et quelques autres pour obtenir le type MIME, etc …

J'ai aussi évité le ../../ vulnérabilité, bien qu'une partie du code de la envoyer le fichier la fonction semble un peu faible.

Il devrait également être conforme à la norme PEP8, à part peut-être quelques espaces dans les commentaires.

type de fichier d'importation
prise d'importation
importer _fil


classe ServerSocket ():
    '' 'Reçoit les connexions, analyse la demande et l'envoie à un gestionnaire
    '' '
    def __init __ (self, adresse, handler = None, * arguments, ** kwargs):
        '' 'Crée un socket serveur et définit un gestionnaire pour le serveur
        '' '
        self.socket = socket.socket ()
        self.socket.setsockopt (socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.socket.bind (adresse)

        self.handler_args = args
        self.handler_kwargs = kwargs

        si gestionnaire:
            self.handler = handler # Le gestionnaire personnalisé
        autre:
            self.handler = Handler # Le gestionnaire par défaut

    def initialise (self, open_connections = 5):
        '' 'Initialise le socket du serveur et le fait écouter les connexions
        '' '
        self.socket.listen (open_connections)
        auto.listen ()

    def parse (self, data):
        '' 'Divise un paquet en
                la demande,
                les en-têtes (qui incluent la demande),
                et contenu
        '' '
        stringed = str (data, 'utf-8')
        split = stringed.split (' r  n  r  n')
        en-têtes = split[0]
        si len (split)> 1:
            contenu = divisé[1]
        autre:
            contenu = []
        request = headers.split ('')[0]

        demande de retour, en-têtes, contenu

    def handle (auto, client, adresse):
        '' 'Analyse les données et gère la demande. Il ferme ensuite la connexion
        '' '
        essayer:
            data = client.recv (1024)
        sauf ConnectionResetError:
            si self.handler_kwargs["logging"] est vrai:
                print (f '{adresse[0]} quitter de façon inattendue ')
            client.close ()
            revenir
        analysé = self.parse (données)
        gestionnaire = self.handler (self.handler_args, self.handler_kwargs)
        handler.handle (client, analysé, adresse)
        client.close ()

    def écouter (auto):
        '' 'Écoute jusqu'à l'interruption du clavier et gère chaque connexion de manière
                nouveau fil
        '' '
        essayer:
            alors que vrai:
                client_data = self.socket.accept ()
                si self.handler_kwargs['logging'] est vrai:
                    print (f'Connection from {client_data[1][0]} ')
                _thread.start_new_thread (self.handle, client_data)
        sauf KeyboardInterrupt:
            self.socket.close ()


gestionnaire de classe ():
    '' 'Traite les requêtes du socket serveur
    '' '
    def __init __ (self, args, kwargs):
        self.args = args
        self.kwargs = kwargs

    def set_status (self, code, message):
        '' 'Utilisé pour ajouter une ligne d'état:
                - "HTTP / 1.1 200 OK" ou "HTTP / 1.1 404 Introuvable", etc.
        '' '
        self.reply_headers = [f'HTTP/1.0 {code} {message}']

    def set_header (self, en-tête, contenu):
        '' 'Définit un en-tête personnalisé et l'ajoute à la réponse
        '' '
        self.reply_headers + = [f'{header}: {content}']

    réponse def (auto, contenu):
        '' 'Ajoute au contenu de la réponse
        '' '
        si type (contenu) == str:
            self.reply_content + = content.split (' n')
        autre:
            self.reply_content + = [content]

    def Calculate_content_length (auto):
        '' 'Calcule la longueur du contenu et l'ajoute à l'en-tête
        '' '
        longueur = len (self.reply_content) * 2
        longueurs = [len(line) for line in self.reply_content]
        longueur + = somme (longueurs)
        self.set_header ('Content-Length', longueur)

    def get_type (self, nom_fichier):
        return filetype.guess ('./ public /' + nom_fichier)

    def extract_file_name (self, file_name = None):
        si nom_fichier:
            nom_f = nom_fichier[1:]
        autre:
            f_name = self.request_status.split ('')[1][1:]
        
        retourne f_name

    def send_file (self, file_name = None):
        si nom_fichier est Aucun:
            nom_fichier = self.extract_nom_fichier ()

        si nom_fichier == '':
            nom_fichier = 'index.html'
        elif nom_fichier[0] dans './':
            self.set_status (403, "Interdit")
            self.set_header ('Content-Type', 'text / html')
            self.reply_content =['['

Erreur 403: Interdit

']revenir         essayer:             avec open ('./ public /' + nom_fichier, 'rb') en tant que fichier:                 file_contents = file.read ()         sauf FileNotFoundError:             self.set_status (404, 'Introuvable')             self.set_header ('Content-Type', 'text / html')             self.reply_content =['['

Erreur 404: fichier non trouvé

']revenir         type_fichier = self.get_type (nom_fichier)         si type_fichier n'est pas None:             self.set_header ('Content-Type', type_fichier.MIME)         elif nom_fichier.split ('.')[-1] == 'html':             self.set_header ('Content-Type', 'text / html')         autre:             self.set_header ('Content-Type', 'text / txt')         self.response (file_contents)     def get_request_address (self):         retourne à moi.adresse     def parse_headers (auto, en-têtes):         t = {}         pour en-tête dans les en-têtes[1:]:             t[headersplit(':')[headersplit(':')[0]]= header.split (':')[1] retourner t     def reply (auto):         '' 'Assemble la réponse et l'envoie au client         '' '         si self.reply_headers[0][0:4] ! = "HTTP":             self.set_status (200, 'OK')             self.set_header ('Content-Type', 'text / html')             self.reply_content =['['

État de la réponse non spécifié

']self.calculate_content_length ()         message = ' r n'.join (self.reply_headers)         message + = ' r n r n'         essayer:             message + = ' r n'.join (self.reply_content)             message + = ' r n'         sauf TypeError:             message = octets (message, 'utf-8')             message + = b ' r n'.join (self.reply_content)             message + = b ' r n'         essayer:             si type (message) == str:                 self.client.send (octets (message, 'utf-8'))             autre:                 self.client.send (message)         sauf:             passer     def handle (self, client, parsed_data, address):         '' 'Initialise les variables et met en casse le type de requête sur                 déterminer la fonction de gestionnaire         '' '         self.client = client         self.address = adresse         self.reply_headers = [] self.reply_content = [] self.headers = True         self.request_status = données_analysées[1].split (' r n')[0] request = parsed_data[0] headers = self.parse_headers (parsed_data[1].split (' r n'))         contents = données_données[2] si requête == "GET":             func = self.get         demande elif == "POST":             func = self.post         demande elif == "HEAD":             func = self.head         demande elif == "PUT":             func = self.put         demande elif == "DELETE":             func = self.delete         demande elif == "CONNECT":             func = self.connect         demande elif == "OPTIONS":             func = self.options         demande elif == "TRACE":             func = self.trace         demande elif == "PATCH":             func = self.patch         autre:             func = self.default         func (en-têtes, contenu)         auto.reply ()     def default (self, en-têtes, contenu):         '' 'Si la demande n'est pas connue, sa valeur par défaut est         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Type de demande inconnu

'' ')     def get (self, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes GET         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande GET

'' ')     def post (auto, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes POST         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande POST

'' ')     def head (soi, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes HEAD         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande HEAD

'' ')     def put (auto, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes PUT         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande PUT

'' ')     def delete (auto, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes DELETE         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande DELETE

'' ')     def connect (self, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes CONNECT         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande de connexion

'' ')     Options de def (auto, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes OPTIONS         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande d'OPTIONS

'' ')     def trace (soi, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes TRACE         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande TRACE

'' ')     patch correctif (auto, en-têtes, contenu):         '' 'Écraser pour gérer de manière personnalisée les demandes PATCH         '' '         self.set_status (200, 'OK')         self.set_header ('Content-Type', 'text / html')         self.response ('' '

Avec succès une demande de PATCH

'' ') # ============================================== ======================= si __name__ == "__main__":     système d'importation     classe CustomHandler (gestionnaire):         def get (self, en-têtes, contenu):             self.set_status (200, 'OK')             request_address = self.get_request_address ()[0] nom_fichier = self.extract_nom_fichier ()             print (f '{request_address} -> {en-têtes["Host"]}/{nom de fichier}')             self.send_file ()     def run ():         si len (sys.argv) == 2:             port = int (sys.argv[1])         autre:             port = 80         essayer:             print ('Initializing ...', end = '')             http_server = ServerSocket (                 ('0.0.0.0', port),                 CustomHandler,                 journalisation = True             )             print ('Terminé')             http_server.initialise ()         sauf exception comme e:             print (f '{e}')     courir()