Skip to content

Instantly share code, notes, and snippets.

@BoQsc
Last active July 23, 2024 17:19
Show Gist options
  • Save BoQsc/69985a2c1cf4ac157e9255f66496498f to your computer and use it in GitHub Desktop.
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.
@BoQsc
Copy link
Author

BoQsc commented Jul 23, 2024

Shortened version

import http.server
import socketserver

PORT = 26900
REDIRECT_URL = 'steam://connect/127.0.0.1:26900'

class RedirectHandler(http.server.SimpleHTTPRequestHandler):
    def do_GET(self):
        try:
            self.send_response(302)
            self.send_header('Location', REDIRECT_URL)
            self.end_headers()
        except Exception:
            pass

with socketserver.TCPServer(("", PORT), RedirectHandler) as httpd:
    try:
        httpd.serve_forever()
    except Exception:
        pass


@BoQsc
Copy link
Author

BoQsc commented Jul 23, 2024

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)

@BoQsc
Copy link
Author

BoQsc commented Jul 23, 2024

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)

@BoQsc
Copy link
Author

BoQsc commented Jul 23, 2024

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