pyhttpserv.py

#!/usr/bin/env python

'''
    This program is free software; you can redistribute it and/or modify
    it under the terms of the Revised BSD License.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    Revised BSD License for more details.

    Copyright 2016-2023 Game Maker 2k - https://github.com/GameMaker2k
    Copyright 2016-2023 Kazuki Przyborowski - https://github.com/KazukiPrzyborowski

    $FileInfo: pyhttpserv.py - Last Update: 10/5/2023 Ver. 2.0.2 RC 1 - Author: cooldude2k $
'''

import os
import argparse
import gzip
import zlib
import bz2

havebrotli = False
try:
    import brotli
    havebrotli = True
except ImportError:
    havebrotli = False
havezstd = False
try:
    import zstandard
    havezstd = True
except ImportError:
    havezstd = False
havelzma = False
try:
    import lzma
    havelzma = True
except ImportError:
    havelzma = False

__program_name__ = "PyHTTPServer"
__program_alt_name__ = "PyHTTPServer"
__program_small_name__ = "httpserver"
__project__ = __program_name__
__project_url__ = "https://github.com/GameMaker2k/PyWWW-Get"
__version_info__ = (2, 0, 2, "RC 1", 1)
__version_date_info__ = (2023, 10, 5, "RC 1", 1)
__version_date__ = str(__version_date_info__[0])+"."+str(__version_date_info__[
    1]).zfill(2)+"."+str(__version_date_info__[2]).zfill(2)
__revision__ = __version_info__[3]
__revision_id__ = "$Id$"
if(__version_info__[4] is not None):
    __version_date_plusrc__ = __version_date__ + \
        "-"+str(__version_date_info__[4])
if(__version_info__[4] is None):
    __version_date_plusrc__ = __version_date__
if(__version_info__[3] is not None):
    __version__ = str(__version_info__[0])+"."+str(__version_info__[1])+"."+str(
        __version_info__[2])+" "+str(__version_info__[3])
if(__version_info__[3] is None):
    __version__ = str(
        __version_info__[0])+"."+str(__version_info__[1])+"."+str(__version_info__[2])

parser = argparse.ArgumentParser(
    description="Simple HTTP Server in Python.", conflict_handler="resolve", add_help=True)
parser.add_argument("-V", "--version", action="version",
                    version=__program_name__+" "+__version__)
parser.add_argument("-e", "--enablessl",
                    action="store_true", help="Enable SSL")
parser.add_argument("-k", "--sslkeypem", default=None,
                    help="specify a custom user agent")
parser.add_argument("-c", "--sslcertpem", default=None,
                    help="specify a custom referer, use if the video access")
parser.add_argument("-p", "--servport", type=int, default=8080,
                    help="specify a file name for output")
getargs = parser.parse_args()

enablessl = getargs.enablessl
sslkeypem = getargs.sslkeypem
sslcertpem = getargs.sslcertpem
servport = int(getargs.servport)
if(isinstance(servport, int)):
    if(servport < 1 or servport > 65535):
        servport = 8080
elif(isinstance(servport, str)):
    if(servport.isnumeric()):
        servport = int(servport)
        if(servport < 1 or servport > 65535):
            servport = 8080
    else:
        servport = 8080
else:
    servport = 8080
if(enablessl):
    if(sslkeypem is not None and
       (not os.path.exists(sslkeypem) or not os.path.isfile(sslkeypem))):
        sslkeypem = None
        enablessl = False
    if(sslcertpem is not None and
       (not os.path.exists(sslkeypem) or not os.path.isfile(sslkeypem))):
        sslcertpem = None
        enablessl = False
pyoldver = True
try:
    from BaseHTTPServer import HTTPServer
    from SimpleHTTPServer import SimpleHTTPRequestHandler
    from urlparse import parse_qs
    from Cookie import SimpleCookie
except ImportError:
    from http.server import SimpleHTTPRequestHandler, HTTPServer
    from urllib.parse import parse_qs
    from http.cookies import SimpleCookie
    pyoldver = False
if(enablessl and
   (sslkeypem is not None and (os.path.exists(sslkeypem) and os.path.isfile(sslkeypem))) and
   (sslcertpem is not None and (os.path.exists(sslkeypem) and os.path.isfile(sslkeypem)))):
    import ssl
# HTTP/HTTPS Server Class


class CustomHTTPRequestHandler(SimpleHTTPRequestHandler):
    def compress_content(self, content):
        """Compress content using gzip or deflate depending on Accept-Encoding header."""
        accept_encoding = self.headers.get('Accept-Encoding', '')
        accept_encoding = None
        if 'gzip' in accept_encoding:
            self.send_header('Content-Encoding', 'gzip')
            compressed_content = gzip.compress(content.encode('utf-8'))
            self.send_header('Content-Length', len(compressed_content))
            return compressed_content
        elif 'deflate' in accept_encoding:
            self.send_header('Content-Encoding', 'deflate')
            compressed_content = zlib.compress(content.encode('utf-8'))
            self.send_header('Content-Length', len(compressed_content))
            return compressed_content
        elif 'br' in accept_encoding:
            self.send_header('Content-Encoding', 'br')
            compressed_content = brotli.compress(content.encode('utf-8'))
            self.send_header('Content-Length', len(compressed_content))
            return compressed_content
        elif 'zstd' in accept_encoding:
            self.send_header('Content-Encoding', 'zstd')
            compressed_content = zstandard.compress(content.encode('utf-8'))
            self.send_header('Content-Length', len(compressed_content))
            return compressed_content
        elif 'lzma' in accept_encoding:
            self.send_header('Content-Encoding', 'lzma')
            compressed_content = lzma.compress(content.encode('utf-8'))
            self.send_header('Content-Length', len(compressed_content))
            return compressed_content
        elif 'xz' in accept_encoding:
            self.send_header('Content-Encoding', 'xz')
            compressed_content = lzma.compress(content.encode('utf-8'))
            self.send_header('Content-Length', len(compressed_content))
            return compressed_content
        else:
            self.send_header('Content-Length', len(content))
            return content.encode('utf-8')
        return content.encode()

    def display_info(self):
        # Setting headers for the response
        self.send_response(200)
        self.send_header('Content-type', 'text/plain')
        # Set a sample cookie in the response;
        self.send_header('Set-Cookie', 'sample_cookie=sample_value; Path=/;')
        self.end_headers()
        # Displaying request method
        response = 'Method: {}\n'.format(self.command)
        response += 'Path: {}\n'.format(self.path)
        # Displaying all headers
        headers_list = ["{}: {}".format(
            key.title(), self.headers[key]) for key in self.headers]
        response += '\nHeaders:\n' + '\n'.join(headers_list) + '\n'
        # Extract and display cookies from headers
        if 'Cookie' in self.headers:
            response += '\nCookies:\n'
            cookies = SimpleCookie(self.headers['Cookie'])
            for key, morsel in cookies.items():
                response += '{}: {}\n'.format(key, morsel.value)
        # Displaying GET parameters (if any)
        if self.command == 'GET':
            query = self.path.split('?', 1)[-1]
            params = parse_qs(query)
            if params:
                response += '\nGET Parameters:\n'
                for key, values in params.items():
                    response += '{}: {}\n'.format(key, ', '.join(values))
        # Sending the response
        self.wfile.write(self.compress_content(response))
    # Get Method

    def do_GET(self):
        self.display_info()
    # Post Method

    def do_POST(self):
        if 'Transfer-Encoding' in self.headers and self.headers['Transfer-Encoding'] == 'chunked':
            post_data = ''
            while True:
                chunk_size_line = self.rfile.readline().decode('utf-8')
                chunk_size = int(chunk_size_line, 16)
                if chunk_size == 0:
                    self.rfile.readline()
                    break
                chunk_data = self.rfile.read(chunk_size).decode('utf-8')
                post_data += chunk_data
                self.rfile.readline()
        else:
            content_length = int(self.headers['Content-Length'])
            post_data = self.rfile.read(content_length).decode('utf-8')
        params = parse_qs(post_data)
        response = 'POST Parameters:\n'
        for key, values in params.items():
            response += '{}: {}\n'.format(key, ', '.join(values))
        self.send_response(200)
        self.send_header('Content-type', 'text/plain')
        self.send_header('Set-Cookie', 'sample_cookie=sample_value; Path=/;')
        self.end_headers()
        self.wfile.write(self.compress_content(response))


# Start Server Forever
if __name__ == "__main__":
    server_address = ('', int(servport))
    httpd = HTTPServer(server_address, CustomHTTPRequestHandler)
    if(enablessl and sslkeypem is not None and sslcertpem is not None):
        httpd.socket = ssl.wrap_socket(httpd.socket,
                                       keyfile=sslkeypem, certfile=sslcertpem, server_side=True)
    if(enablessl):
        print("Server started at https://localhost:"+str(servport))
    else:
        print("Server started at http://localhost:"+str(servport))
    httpd.serve_forever()