Last active
September 10, 2023 04:14
-
-
Save hoangnd-monkey/d6dc3592c4c0de643959b4971101f0d5 to your computer and use it in GitHub Desktop.
Python Forwarder Server
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import functools | |
| from http.server import BaseHTTPRequestHandler, HTTPServer | |
| from typing import Callable | |
| import requests | |
| import sys | |
| hostName = "0.0.0.0" | |
| serverPort = 8080 | |
| host = "https://jsonplaceholder.typicode.com" | |
| class ForwardServer(BaseHTTPRequestHandler): | |
| target = None | |
| header_handle: Callable = None | |
| def do_GET(self): | |
| self._handle_request("get", requests.get) | |
| def do_DELETE(self): | |
| self._handle_request("delete", requests.delete) | |
| def do_POST(self): | |
| self._handle_request("post", requests.post) | |
| def do_PUT(self): | |
| self._handle_request("put", requests.put) | |
| def do_PATCH(self): | |
| self._handle_request("patch", requests.patch) | |
| def _handle_request(self, method, requests_func): | |
| full_url = f"{self.target}{self.path}" | |
| if self.header_handle and isinstance(self.header_handle, Callable): | |
| headers = self.header_handle(method, self.headers) | |
| else: | |
| headers = dict(self.headers) | |
| headers.pop("Host") | |
| body = b"" | |
| if "Content-Length" in headers: | |
| body = self.rfile.read(int(headers["Content-Length"])) | |
| resp = requests_func(url=full_url, data=body, headers=headers) | |
| self.send_response(resp.status_code) | |
| self.send_header("Content-Type", resp.headers["Content-Type"]) | |
| self.end_headers() | |
| self.wfile.write(resp.content) | |
| if __name__ == "__main__": | |
| if len(sys.argv) > 1: | |
| host = sys.argv[1] | |
| if not host.startswith("http"): | |
| host = f"https://{host}" | |
| ForwardServer.target = host | |
| listener_on = (hostName, serverPort) | |
| webServer = HTTPServer(listener_on, ForwardServer) | |
| print("Server started http://%s:%s" % (hostName, serverPort)) | |
| try: | |
| webServer.serve_forever() | |
| except KeyboardInterrupt: | |
| pass | |
| webServer.server_close() | |
| print("Server stopped.") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment