Last active
July 23, 2024 17:19
-
-
Save BoQsc/69985a2c1cf4ac157e9255f66496498f to your computer and use it in GitHub Desktop.
A simple 7 Days To Die TCP HTTP Steam Link redirecter. Once you visit the webpage, it opens up steam://connect/127.0.0.1:26900 link that auto joins player to the gameserver. This is only a proof of concept, replace 127.0.0.1 with your Public IP.
This file contains 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 http.server | |
import socketserver | |
import logging | |
PORT = 26900 | |
REDIRECT_URL = 'steam://connect/127.0.0.1:26900' | |
# Configure logging to log to a file | |
logging.basicConfig(filename='server.log', level=logging.INFO, format='%(asctime)s - %(message)s') | |
logger = logging.getLogger() | |
class RedirectHandler(http.server.SimpleHTTPRequestHandler): | |
def do_GET(self): | |
logger.info(f"Received request from {self.client_address}") | |
try: | |
self.send_response(302) | |
self.send_header('Location', REDIRECT_URL) | |
self.end_headers() | |
logger.info(f"Redirected to {REDIRECT_URL}") | |
except Exception as e: | |
logger.error(f"Failed to handle request: {e}") | |
def log_message(self, format, *args): | |
# Override to prevent logging to stderr | |
logger.info(f"{self.client_address} - {format % args}") | |
with socketserver.TCPServer(("", PORT), RedirectHandler) as httpd: | |
logger.info(f"Serving on port {PORT}") | |
try: | |
httpd.serve_forever() | |
except Exception as e: | |
logger.error(f"Server stopped due to an error: {e}") |
Self-restarting version | PID termination
import http.server
import socketserver
import os
import sys
import psutil
import time
import logging
from logging.handlers import RotatingFileHandler
PORT = 26900
REDIRECT_URL = 'steam://connect/127.0.0.1:26900'
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
PID_FILE = os.path.join(SCRIPT_DIR, 'redirect_server.pid')
LOG_FILE = os.path.join(SCRIPT_DIR, 'redirect_server.log')
# Setup logging
logger = logging.getLogger('RedirectServer')
logger.setLevel(logging.DEBUG)
handler = RotatingFileHandler(LOG_FILE, maxBytes=1024*1024, backupCount=5)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
class RedirectHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
try:
self.send_response(302)
self.send_header('Location', REDIRECT_URL)
self.end_headers()
logger.info(f"Redirected request from {self.client_address[0]} to {REDIRECT_URL}")
except Exception as e:
logger.error(f"Error handling GET request: {e}", exc_info=True)
def log_message(self, format, *args):
logger.info("%s - - [%s] %s" %
(self.address_string(),
self.log_date_time_string(),
format % args))
def terminate_existing_process():
if os.path.exists(PID_FILE):
with open(PID_FILE, 'r') as pid_file:
pid = int(pid_file.read().strip())
try:
process = psutil.Process(pid)
process.terminate()
process.wait(timeout=5) # Wait for up to 5 seconds for the process to terminate
logger.info(f"Terminated existing process with PID {pid}")
except psutil.NoSuchProcess:
logger.info(f"No process found with PID {pid}")
except psutil.TimeoutExpired:
logger.warning(f"Process with PID {pid} did not terminate within the timeout period")
except Exception as e:
logger.error(f"Error terminating process: {e}", exc_info=True)
finally:
os.remove(PID_FILE)
logger.info(f"Removed PID file: {PID_FILE}")
def main():
logger.info("Starting redirect server")
terminate_existing_process()
try:
with socketserver.TCPServer(("", PORT), RedirectHandler) as httpd:
with open(PID_FILE, 'w') as pid_file:
pid = os.getpid()
pid_file.write(str(pid))
logger.info(f"Written PID {pid} to {PID_FILE}")
logger.info(f"Server running on port {PORT}")
httpd.serve_forever()
except KeyboardInterrupt:
logger.info("Server stopped by user interrupt")
except Exception as e:
logger.error(f"An error occurred: {e}", exc_info=True)
finally:
if os.path.exists(PID_FILE):
os.remove(PID_FILE)
logger.info(f"Removed PID file: {PID_FILE}")
if __name__ == '__main__':
try:
main()
except Exception as e:
logger.critical(f"Unhandled exception in main: {e}", exc_info=true)
Threaded, Fixes the CTRL+C Not working when someone keeps webpage as opened.
import http.server
import socketserver
import os
import sys
import psutil
import time
import logging
from logging.handlers import RotatingFileHandler
import threading
PORT = 26900
REDIRECT_URL = 'steam://connect/127.0.0.1:26900'
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
PID_FILE = os.path.join(SCRIPT_DIR, 'redirect_server.pid')
LOG_FILE = os.path.join(SCRIPT_DIR, 'redirect_server.log')
# Setup logging
logger = logging.getLogger('RedirectServer')
logger.setLevel(logging.DEBUG)
handler = RotatingFileHandler(LOG_FILE, maxBytes=1024*1024, backupCount=5)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
class RedirectHandler(http.server.SimpleHTTPRequestHandler):
def do_GET(self):
try:
self.send_response(302)
self.send_header('Location', REDIRECT_URL)
self.end_headers()
logger.info(f"Redirected request from {self.client_address[0]} to {REDIRECT_URL}")
except Exception as e:
logger.error(f"Error handling GET request: {e}", exc_info=True)
def log_message(self, format, *args):
logger.info("%s - - [%s] %s" %
(self.address_string(),
self.log_date_time_string(),
format % args))
class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
allow_reuse_address = True
def __init__(self, server_address, RequestHandlerClass):
self.allow_reuse_address = True
self.daemon_threads = True
socketserver.TCPServer.__init__(self, server_address, RequestHandlerClass)
def terminate_existing_process():
if os.path.exists(PID_FILE):
with open(PID_FILE, 'r') as pid_file:
pid = int(pid_file.read().strip())
try:
process = psutil.Process(pid)
process.terminate()
process.wait(timeout=5) # Wait for up to 5 seconds for the process to terminate
logger.info(f"Terminated existing process with PID {pid}")
except psutil.NoSuchProcess:
logger.info(f"No process found with PID {pid}")
except psutil.TimeoutExpired:
logger.warning(f"Process with PID {pid} did not terminate within the timeout period")
except Exception as e:
logger.error(f"Error terminating process: {e}", exc_info=True)
finally:
os.remove(PID_FILE)
logger.info(f"Removed PID file: {PID_FILE}")
def run_server(server):
try:
server.serve_forever()
except Exception as e:
logger.error(f"Server error: {e}", exc_info=True)
def main():
logger.info("Starting redirect server")
terminate_existing_process()
try:
server = ThreadedTCPServer(("", PORT), RedirectHandler)
server_thread = threading.Thread(target=run_server, args=(server,))
server_thread.daemon = True
server_thread.start()
with open(PID_FILE, 'w') as pid_file:
pid = os.getpid()
pid_file.write(str(pid))
logger.info(f"Written PID {pid} to {PID_FILE}")
logger.info(f"Server running on port {PORT}")
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("Received keyboard interrupt, shutting down...")
server.shutdown()
server.server_close()
logger.info("Server stopped")
except Exception as e:
logger.error(f"An error occurred: {e}", exc_info=True)
finally:
if os.path.exists(PID_FILE):
os.remove(PID_FILE)
logger.info(f"Removed PID file: {PID_FILE}")
if __name__ == '__main__':
try:
main()
except Exception as e:
logger.critical(f"Unhandled exception in main: {e}", exc_info=True)
Example html web page of usage demonstration.
index.html
<a href="http://localhost:26900/">localhost:26900</a>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Shortened version