beamup.py

#!/bin/python
import http.server, socketserver, os, urllib, json, mimetypes
from pathlib import Path
import socket,sys,psutil

PORT = 6060
HOST = ""
SHARE_DIR = os.path.expanduser("~")
FRONTEND_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)),"static")
bind = []

def list_if_ips():
    result = {}
    for iface, addrs in psutil.net_if_addrs().items():
        result[iface] = []
        for a in addrs:
            if a.family == socket.AF_INET:
                result[iface].append(("ipv4", a.address, a.netmask))
            elif a.family == socket.AF_INET6:
                result[iface].append(("ipv6", a.address.split('%')[0], a.netmask))
                            
    return result
    
def safe_join(base, *paths):
    final_path = os.path.abspath(os.path.join(base,*paths))
    if not final_path.startswith(base + os.sep):
        return False
    return final_path


class WebServer(http.server.BaseHTTPRequestHandler):
    def _send_json(self, data, code=200):
        self.send_response(code)
        
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())

    def do_OPTIONS(self):
        self.send_response(200)
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def end_headers(self):
        self.send_header("Access-Control-Allow-Origin", "*")
        super().end_headers()

    def do_GET(self):
        if self.path.startswith("/list"):
            query = urllib.parse.urlparse(self.path).query
            params = urllib.parse.parse_qs(query)
            self.list_files(params.get("path",[None])[0])
        elif self.path.startswith("/download/"):
            query = urllib.parse.urlparse(self.path).query
            params = urllib.parse.parse_qs(query)
            self.download_file(params.get("file",[None])[0])
        else:
            self.serve_frontend()

    def list_files(self, path):
        files = []
        dirs  = []
        
        dir_  = path if path != None else SHARE_DIR
        dir_  = os.path.abspath(dir_)
        if not dir_.startswith(SHARE_DIR):
            dir_ = SHARE_DIR
            
        for f in os.listdir(dir_):
            path_ = os.path.join(dir_, f)
            if os.path.isfile(path_):
                files.append({
                    "type":"file",
                    "name": f,
                    "size": os.path.getsize(path_)
                })
            else:
                dirs.append({
                    "type":"dir",
                    "pwd" : path_,
                    "name": f,
                    "size": "",
                })
                
        self._send_json({ "home":dir_ == os.path.abspath(SHARE_DIR), "pwd":dir_, "files": dirs + files})

    def download_file(self, filename):
        filepath = safe_join(SHARE_DIR, "./"+filename)
        if not filepath or not os.path.isfile(filepath):
            self.send_error(404, "File not found")
            return

        self.send_response(200)
        outfile = filepath.split("/")[-1]
        mime, _ = mimetypes.guess_type(filepath)
        self.send_header("Content-Type", mime or "application/octet-stream")
        self.send_header("Content-Disposition", f"attachment; filename={outfile}")
        self.send_header("Content-Length", str(os.path.getsize(filepath)))
        self.end_headers()

        with open(filepath, "rb") as f:
            while chunk := f.read(65536):
                self.wfile.write(chunk)

    def do_POST(self):
        if self.path.startswith("/upload"):
            self.upload_file()
        else:
            self.send_error(404, "Unknown endpoint")

    def upload_file(self):
        query  = urllib.parse.urlparse(self.path).query
        params = urllib.parse.parse_qs(query)
        filename = params.get("name", [None])[0]
        dest = params.get("to", ["./"])[0]
        
        if not filename:
            self.send_error(400, "Missing filename")
            return
            
        dir_ = os.path.abspath(dest)
        print("Dest: ", dir_)
        if not os.path.exists(dir_):
            dir_ = SHARE_DIR
            print("not exists switch to ", dir_)
            
        filepath = safe_join(dir_, os.path.basename(filename))
        print("final res:", filepath )
        
        content_length = int(self.headers.get('Content-Length', 0))
        received = 0
        with open(filepath, "wb") as f:
            while received < content_length:
                chunk_size = min(65536, content_length - received)
                chunk = self.rfile.read(chunk_size)
                if not chunk:
                    break
                f.write(chunk)
                received += len(chunk)
                
        self._send_json({"status": "ok", "message": "Upload complete"})

    def serve_frontend(self):
        path = self.path.strip("/")
        if path == "":
            path = "index.html"
        filepath = safe_join(FRONTEND_DIR, path)

        if not filepath or not os.path.isfile(filepath):
            self.send_error(404, "Not found")
            return

        mime, _ = mimetypes.guess_type(filepath)
        self.send_response(200)
        self.send_header("Content-Type", mime or "text/plain")
        self.end_headers()
        
        with open(filepath, "rb") as f:
            if path == "index.html":
                html = (f.read()).decode("utf-8")
                html = html.replace('"[PATH]"', '"' + SHARE_DIR + '"' )
                self.wfile.write(html.encode())
            else:
                self.wfile.write(f.read())
 
class BeamUp(socketserver.TCPServer):
    allow_reuse_address = True

def usage():
    print("""Usage: beamup [options]
Options:
  -d <directory>    Specify the directory to share. 
                    Defaults to the user’s home directory if not specified.

  -b <host:port>    Set the binding address for the server.
                    Format: "host:port"
                    If omitted:
                      ":port" uses the default host (0.0.0.0)
                      "host:" uses the default port (8080)

Examples:
  beamup -d /var/www -b 127.0.0.1:9000
  beamup -b :8080
  beamup -d ~/Downloads

    """)

if len(sys.argv) > 1:
    if "-h" in sys.argv[1:]:
        usage()
        sys.exit()
    
    if "-d" in sys.argv[1:] and sys.argv.index("-d") + 1 < len(sys.argv):
        SHARE_DIR = sys.argv[sys.argv.index("-d") + 1]
        os.chdir(SHARE_DIR)
    
    if "-b" in sys.argv[1:] and sys.argv.index("-b") + 1 < len(sys.argv):
        tmp  = (sys.argv[sys.argv.index("-b") + 1]).split(":")
        if len(tmp) < 2:
            print("""Error: Invalid binding address
            Use the format 'host:port' e.g: 0.0.0.0:8080
            If ommitted, ":port" uses the defaulf host (localhost), and "host:" uses default port (6060).
            """)
        if tmp[1] != "":
            try:
                PORT = int(tmp[1])
            except:
                print("Invalid port. Must be an integer between 1-65535.")
        
        HOST = tmp[0]
        if PORT <= 0 or PORT > 65535:
            print("Invalid port. Using default port 6060.")
            PORT = 6060
        SHARE_DIR = sys.argv[sys.argv.index("-d") + 1]
        os.chdir(SHARE_DIR)


interface_ip = list_if_ips()
IPS = []

for k in interface_ip.keys():
    for if_ in interface_ip[k]:
        if if_[0] == "ipv4":
            ip_tmp = interface_ip[k][0][1]
            IPS.append(if_[1])
        
bind = [f"• http://{HOST}:{PORT}"]
if HOST == "":
    HOST = "localhost"
    bind = [f"• http://{HOST}:{PORT}"]

bind += [f"• http://{ip}:{PORT}" for ip in IPS]

with BeamUp(("", PORT), WebServer) as httpd:
    print(f"Directory: {os.path.abspath(SHARE_DIR)}")
    print("Visit one of these links from another device on the same network:")
    print("\n".join(bind), "\nPress Ctrl + C to stop the server.")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        httpd.shutdown()
        httpd.server_close()