diff options
| author | Louis Burda <dev@sinitax.com> | 2026-01-24 04:55:48 +0100 |
|---|---|---|
| committer | Louis Burda <dev@sinitax.com> | 2026-01-24 04:57:14 +0100 |
| commit | 608e959ef71b3d717f55de092d8301d4703a631a (patch) | |
| tree | 27a116a9dfceab58224f30afe84c9da23767426f /src | |
| download | uploadserver-main.tar.gz uploadserver-main.zip | |
Diffstat (limited to 'src')
| -rw-r--r-- | src/uploadserver/__init__.py | 133 |
1 files changed, 133 insertions, 0 deletions
diff --git a/src/uploadserver/__init__.py b/src/uploadserver/__init__.py new file mode 100644 index 0000000..9d64919 --- /dev/null +++ b/src/uploadserver/__init__.py @@ -0,0 +1,133 @@ +import argparse +import os +import http.server +import socketserver +from email.parser import BytesParser +import re +import sys + +def parse_multipart(headers, body): + content_type = headers.get('Content-Type', '') + if not content_type.startswith('multipart/form-data'): + return {} + + header_bytes = f"Content-Type: {content_type}\r\n\r\n".encode() + msg = BytesParser().parsebytes(header_bytes + body) + + parts = {} + for part in msg.walk(): + if part.get_content_maintype() == 'multipart': + continue + disp = part.get('Content-Disposition', '') + name_match = re.search(r'name="([^"]*)"', disp) + filename_match = re.search(r'filename="([^"]*)"', disp) + if name_match: + name = name_match.group(1) + parts[name] = { + 'data': part.get_payload(decode=True), + 'filename': filename_match.group(1) if filename_match else None + } + return parts + +def log_request(handler, body=None): + print(f"--- {handler.command} {handler.path} {handler.request_version}", file=sys.stderr) + for name, value in handler.headers.items(): + print(f"{name}: {value}", file=sys.stderr) + if body is not None: + print(f"--- Body: {len(body)} bytes", file=sys.stderr) + print(file=sys.stderr) + +def make_handler(upload_dir, file_key, debug=False): + class UploadHandler(http.server.SimpleHTTPRequestHandler): + def do_GET(self): + if debug: + log_request(self) + if self.path == '/': + self.send_response(200) + self.send_header('Content-type', 'text/html') + self.end_headers() + files = os.listdir(upload_dir) if os.path.exists(upload_dir) else [] + files_list = ''.join(f'<li><a href="/uploads/{f}">{f}</a></li>' for f in files) + html = f'''<!DOCTYPE html> +<html> +<head><title>File Upload</title></head> +<body> +<h1>Upload File</h1> +<form method="POST" enctype="multipart/form-data"> +<input type="file" name="{file_key}" required> +<input type="submit" value="Upload"> +</form> +<h2>Uploaded Files</h2> +<ul>{files_list}</ul> +</body> +</html>''' + self.wfile.write(html.encode()) + else: + super().do_GET() + + def do_POST(self): + content_length = int(self.headers.get('Content-Length', 0)) + body = self.rfile.read(content_length) + if debug: + log_request(self, body) + if self.path == '/': + parts = parse_multipart(self.headers, body) + if file_key in parts and parts[file_key]['filename']: + filename = os.path.basename(parts[file_key]['filename']) + filepath = os.path.join(upload_dir, filename) + with open(filepath, 'wb') as f: + f.write(parts[file_key]['data']) + print(f"Received file: {filepath}", file=sys.stderr) + self.send_response(303) + self.send_header('Location', '/') + self.end_headers() + return + self.send_response(400) + self.end_headers() + else: + self.send_response(404) + self.end_headers() + return UploadHandler + +def make_raw_handler(output_path, debug=False): + class RawHandler(http.server.BaseHTTPRequestHandler): + def do_POST(self): + content_length = int(self.headers.get('Content-Length', 0)) + body = self.rfile.read(content_length) + if debug: + log_request(self, body) + with open(output_path, 'wb') as f: + f.write(body) + print(f"Received file: {output_path}", file=sys.stderr) + self.send_response(200) + self.send_header('Content-type', 'text/plain') + self.end_headers() + self.wfile.write(f"Saved {len(body)} bytes to {output_path}\n".encode()) + return RawHandler + +def main(): + parser = argparse.ArgumentParser(description='File upload server') + parser.add_argument('-k', '--key', default='file', help='Body parameter name for file uploads') + parser.add_argument('-p', '--port', type=int, default=8000, help='Port to listen on') + parser.add_argument('--raw', action='store_true', help='Accept raw body as file') + parser.add_argument('-o', '--output', help='Output path for raw mode') + parser.add_argument('-d', '--debug', action='store_true', help='Print incoming requests') + args = parser.parse_args() + + if args.raw: + if not args.output: + parser.error('--raw requires -o/--output PATH') + handler = make_raw_handler(args.output, args.debug) + print(f"Server running on http://localhost:{args.port} (raw mode, output: {args.output})") + else: + upload_dir = "uploads" + os.makedirs(upload_dir, exist_ok=True) + handler = make_handler(upload_dir, args.key, args.debug) + print(f"Server running on http://localhost:{args.port} (file key: {args.key})") + + socketserver.TCPServer.allow_reuse_address = True + with socketserver.TCPServer(("", args.port), handler) as httpd: + httpd.serve_forever() + +if __name__ == '__main__': + main() |