[Proof Of Concept] Low-level HTTP requests handling with python sockets.

POC_http_server_socket.py

import datetime
import json

HTTP_VERSION = '1.0'
HTTP_STATUS_CODES = {
    200: 'OK',
    201: 'CREATED',
    202: 'ACCEPTED',
    204: 'NO CONTENT',
    400: 'BAD REQUEST',
    401: 'UNAUTHORIZED',
    403: 'FORBIDDEN',
    404: 'NOT FOUND',
}
SERVER_NAME = 'fherbine/0.1'


def is_json(string):
    try:
        json.loads(string)
        return True
    except json.decoder.JSONDecodeError:
        return False
    except:
        raise

def http_dateformat(dt):
    return dt.strftime('%a, %d %b %Y %H:%M:%S GMT')


class HttpRequest:
    headers = dict()

    def __init__(self, headers, content):
        self.headers = headers
        self.content = content

    @classmethod
    def from_raw_request(cls, request):
        headers, content = HttpParser().parse(request)
        return cls(headers, content)

    def force_json_content(self):
        if not is_json(self._content):
            raise ValueError('Request content cannot be converted into JSON')

        self.headers['Content-Type'] = 'application/json'
        self._content = json.loads(self._content)


    @property
    def content(self):
        return self._content

    @content.setter
    def content(self, new_content):
        if not self.headers:
            self._content = new_content
            self.headers['Content-Type'] = 'text/plain'
            return

        ct = self.headers.get('Content-Type')

        if not ct or ct == 'text/plain':
            self._content = new_content
            self.headers['Content-Type'] = 'text/plain'
        elif ct == 'application/json':
            self._content = json.loads(new_content)
        else:
            self._content = new_content
            self.headers['Content-Type'] = 'text/plain'


class HttpParser:
    def parse(self, content):
        content = content.decode()
        headers, raw_content = content.split('\r\n\r\n')
        headers = headers.split('\r\n')

        http_head = headers.pop(0)
        headers = {
            header.split(
                ': '
            )[0]: header.split(': ')[1] for header in headers
        }
        method, route, version = http_head.split(' ')
        headers['method'] = method
        headers['route'] = route
        headers['version'] = version

        return headers, raw_content

    def make_response(self, content, status_code):
        now = datetime.datetime.now()
        expires = http_dateformat(now + datetime.timedelta(hours=1))
        now = http_dateformat(now)
        content_type = 'application/json' if is_json(content) else 'text/plain'

        if status_code in (200, 201, 202) and not content:
            status_code = 204

        content = HTTP_STATUS_CODES[status_code] if status_code in (
            400,
            401,
            403,
            404,
        ) else content

        content = content.replace('\n', '\r\n')

        http_code = 'HTTP/{v} {code} {msg}\r\n'.format(
            v=HTTP_VERSION,
            code=status_code,
            msg=HTTP_STATUS_CODES[status_code],
        )
        now = 'Date: %s\r\n' % now
        server = 'Server: %s\r\n' % SERVER_NAME
        content_type = 'Content-Type: %s\r\n' % content_type
        content_len = 'Content-Length: %s\r\n' % len(content)
        expires = 'Expires: %s\r\n' % expires

        header = '{}{}{}{}{}{}\r\n'.format(
            http_code,
            now,
            server,
            content_type,
            content_len,
            expires,
        )

        return (header + content).encode()

    def make_json_response(self, content):
        content = json.dumps(content)
        return self.make_response(content, 200)

if __name__ == '__main__':
    import socket
    import sys
    import signal
    import time

    main_connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    main_connection.bind(('', int(sys.argv[-1])))
    main_connection.listen(5)

    client_socket, infos = main_connection.accept()
    receive = client_socket.recv(8192)
    req = HttpRequest.from_raw_request(receive)
    req.force_json_content()
    print(req.content, req.headers, sep='\n')
    client_socket.send(HttpParser().make_json_response({'hello': 'world'}))

    client_socket.shutdown(socket.SHUT_RDWR)
    client_socket.close()

    main_connection.shutdown(socket.SHUT_RDWR)
    main_connection.close()

Pas mal, je me suis amusé à faire la même chose en plus poussé en Go. Ça permet de bien le fonctionnement du protocole HTTP et d'un server web. C'est sur https://github.com/vomnes/ImpetusResel si tu veux voir.

https://gist.github.com/vomnes/be42868583db5812b7266b2f45262dca