server.py

import socket
import threading
import queue
import sys
import os
import json
import random
import string
from datetime import datetime, timezone
from email.utils import formatdate

# --- Server Configuration ---
DEFAULT_HOST = '127.0.0.1'
DEFAULT_PORT = 8080
DEFAULT_MAX_THREADS = 10
MAX_QUEUE_SIZE = 50
RESOURCES_DIR = 'resources'
UPLOADS_DIR = os.path.join(RESOURCES_DIR, 'uploads')
REQUEST_BUFFER_SIZE = 8192
KEEP_ALIVE_TIMEOUT = 30
MAX_REQUESTS_PER_CONN = 100
SERVER_NAME = 'Python-MultiThreaded-HTTP-Server/1.0'

# --- MIME Types ---
MIME_TYPES = {
    '.html': 'text/html; charset=utf-8',
    '.txt': 'application/octet-stream',
    '.png': 'application/octet-stream',
    '.jpg': 'application/octet-stream',
    '.jpeg': 'application/octet-stream',
    '.json': 'application/json',
}

# --- HTTP Status Codes ---
STATUS_CODES = {
    200: 'OK',
    201: 'Created',
    400: 'Bad Request',
    403: 'Forbidden',
    404: 'Not Found',
    405: 'Method Not Allowed',
    415: 'Unsupported Media Type',
    500: 'Internal Server Error',
    503: 'Service Unavailable',
}


class HTTPServer:
    """
    A multi-threaded HTTP server that handles GET and POST requests.
    """

    def __init__(self, host, port, max_threads):
        self.host = host
        self.port = port
        self.max_threads = max_threads
        self.server_socket = None
        self.connection_queue = queue.Queue(maxsize=MAX_QUEUE_SIZE)
        self.thread_pool = []
        self.lock = threading.Lock()
        self.active_threads = 0

    def log(self, message, thread_name=None):
        """Prints a log message with a timestamp and optional thread name."""
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        if thread_name:
            print(f"[{timestamp}] [{thread_name}] {message}")
        else:
            print(f"[{timestamp}] {message}")

    def start(self):
        """Initializes the server, starts the thread pool, and listens for connections."""
        if not os.path.exists(RESOURCES_DIR):
            os.makedirs(RESOURCES_DIR)
        if not os.path.exists(UPLOADS_DIR):
            os.makedirs(UPLOADS_DIR)

        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        try:
            self.server_socket.bind((self.host, self.port))
            self.server_socket.listen(MAX_QUEUE_SIZE)
        except OSError as e:
            self.log(f"Error binding to {self.host}:{self.port} - {e}")
            sys.exit(1)

        self.log(f"HTTP Server started on http://{self.host}:{self.port}")
        self.log(f"Thread pool size: {self.max_threads}")
        self.log(f"Serving files from '{RESOURCES_DIR}' directory")
        self.log("Press Ctrl+C to stop the server")

        for i in range(self.max_threads):
            thread = threading.Thread(target=self.worker_thread, name=f"Thread-{i + 1}")
            thread.daemon = True
            thread.start()
            self.thread_pool.append(thread)

        try:
            while True:
                client_socket, client_address = self.server_socket.accept()
                try:
                    self.connection_queue.put((client_socket, client_address), block=False)
                except queue.Full:
                    self.log("Warning: Thread pool saturated, queuing connection. Queue is full, rejecting connection.")
                    response = self.build_response(503, headers={'Retry-After': '30'})
                    client_socket.sendall(response)
                    client_socket.close()

        except KeyboardInterrupt:
            self.log("Server shutting down.")
        finally:
            if self.server_socket:
                self.server_socket.close()

    def worker_thread(self):
        """The main loop for each worker thread."""
        thread_name = threading.current_thread().name
        while True:
            client_socket, client_address = self.connection_queue.get()

            with self.lock:
                self.active_threads += 1
                self.log(f"Thread pool status: {self.active_threads}/{self.max_threads} active", thread_name)

            self.log(f"Connection from {client_address[0]}:{client_address[1]}", thread_name)
            self.handle_client_connection(client_socket, client_address)

            with self.lock:
                self.active_threads -= 1

            self.connection_queue.task_done()

    def handle_client_connection(self, client_socket, client_address):
        """Manages a persistent connection with a client."""
        thread_name = threading.current_thread().name
        request_count = 0
        keep_alive = True

        client_socket.settimeout(KEEP_ALIVE_TIMEOUT)

        while keep_alive and request_count < MAX_REQUESTS_PER_CONN:
            try:
                request_data = client_socket.recv(REQUEST_BUFFER_SIZE)
                if not request_data:
                    break  # Client closed connection

                response, connection_header = self.handle_request(request_data, client_address)
                client_socket.sendall(response)

                request_count += 1
                if connection_header == 'close':
                    keep_alive = False

            except socket.timeout:
                self.log("Connection timed out.", thread_name)
                break
            except Exception as e:
                self.log(f"Error handling request: {e}", thread_name)
                response = self.build_response(500)
                try:
                    client_socket.sendall(response)
                except:
                    pass  # Client might have already disconnected
                break

        self.log("Closing connection.", thread_name)
        client_socket.close()

    def handle_request(self, request_data, client_address):
        """Parses a request and routes it to the appropriate handler."""
        thread_name = threading.current_thread().name
        try:
            method, path, version, headers, body = self.parse_request(request_data)
            self.log(f"Request: {method} {path} {version}", thread_name)

            # Determine keep-alive status
            connection_header = headers.get('Connection', 'keep-alive' if version == 'HTTP/1.1' else 'close').lower()

            # --- Host Header Validation ---
            host_header = headers.get('Host')
            if not host_header:
                self.log("Security violation: Missing Host header", thread_name)
                return self.build_response(400), 'close'

            expected_host = f"{self.host}:{self.port}"
            if host_header != expected_host:
                self.log(f"Security violation: Host mismatch. Expected '{expected_host}', got '{host_header}'",
                         thread_name)
                return self.build_response(403), 'close'
            self.log(f"Host validation: {host_header} ✓", thread_name)

            # --- Route request ---
            if method == 'GET':
                response = self.handle_get(path)
            elif method == 'POST':
                response = self.handle_post(path, headers, body)
            else:
                response = self.build_response(405, headers={'Allow': 'GET, POST'})

            # --- Add common headers ---
            status_code_str = response.split(b' ')[1].decode()
            final_headers = {
                'Connection': connection_header,
                'Keep-Alive': f'timeout={KEEP_ALIVE_TIMEOUT}, max={MAX_REQUESTS_PER_CONN}'
            }
            if connection_header == 'keep-alive':
                self.log(f"Connection: keep-alive", thread_name)

            final_response = self.add_headers_to_response(response, final_headers)

            content_length = len(final_response.split(b'\r\n\r\n', 1)[1])
            self.log(f"Response: {status_code_str} ({content_length} bytes transferred)", thread_name)

            return final_response, connection_header

        except Exception as e:
            self.log(f"Error parsing request: {e}", thread_name)
            return self.build_response(400), 'close'

    def parse_request(self, request_data):
        """Parses raw HTTP request data into its components."""
        request_line_and_headers, body = request_data.split(b'\r\n\r\n', 1)
        lines = request_line_and_headers.split(b'\r\n')

        method, path, version = lines[0].decode().split()

        headers = {}
        for line in lines[1:]:
            key, value = line.decode().split(': ', 1)
            headers[key] = value

        return method, path, version, headers, body

    def handle_get(self, path):
        """Handles GET requests for serving static files."""
        if path == '/':
            path = '/index.html'

        # --- Path Traversal Protection ---
        base_dir = os.path.abspath(RESOURCES_DIR)
        requested_path = os.path.abspath(os.path.join(base_dir, path.lstrip('/')))

        if path.startswith('/..') or path.startswith('/./') or not requested_path.startswith(base_dir):
            self.log(f"Security violation: Path traversal attempt blocked for path: {path}", threading.current_thread().name)
            return self.build_response(403)

        if not os.path.exists(requested_path) or not os.path.isfile(requested_path):
            return self.build_response(404)

        _, extension = os.path.splitext(requested_path)
        content_type = MIME_TYPES.get(extension.lower())

        if not content_type:
            return self.build_response(415)

        try:
            with open(requested_path, 'rb') as f:
                content = f.read()

            headers = {'Content-Type': content_type}
            if 'application/octet-stream' in content_type:
                filename = os.path.basename(requested_path)
                headers['Content-Disposition'] = f'attachment; filename="{filename}"'
                self.log(f"Sending binary file: {filename} ({len(content)} bytes)", threading.current_thread().name)

            return self.build_response(200, headers=headers, body=content)

        except IOError as e:
            self.log(f"File I/O error for {requested_path}: {e}", threading.current_thread().name)
            return self.build_response(500)

    def handle_post(self, path, headers, body):
        """Handles POST requests for uploading JSON data."""
        thread_name = threading.current_thread().name
        if path != '/upload':
            return self.build_response(404)

        content_type = headers.get('Content-Type', '')
        if 'application/json' not in content_type:
            return self.build_response(415)

        try:
            json_data = json.loads(body.decode('utf-8'))
        except json.JSONDecodeError:
            return self.build_response(400, body=b'{"error": "Invalid JSON format"}')

        try:
            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
            random_id = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
            filename = f"upload_{timestamp}_{random_id}.json"
            filepath = os.path.join(UPLOADS_DIR, filename)

            with open(filepath, 'w') as f:
                json.dump(json_data, f, indent=4)

            response_body = {
                "status": "success",
                "message": "File created successfully",
                "filepath": f"/uploads/{filename}"
            }
            self.log(f"Created file: {filepath}", thread_name)
            return self.build_response(201,
                                       headers={'Content-Type': 'application/json'},
                                       body=json.dumps(response_body).encode('utf-8'))
        except Exception as e:
            self.log(f"Error creating file on POST: {e}", thread_name)
            return self.build_response(500)

    def build_response(self, status_code, headers=None, body=b''):
        """Constructs a complete HTTP response."""
        status_message = STATUS_CODES.get(status_code, 'Unknown Status')
        response_line = f"HTTP/1.1 {status_code} {status_message}\r\n"

        response_headers = {
            'Date': formatdate(timeval=None, localtime=False, usegmt=True),
            'Server': SERVER_NAME,
            'Content-Length': str(len(body))
        }

        if headers:
            response_headers.update(headers)

        if status_code >= 400 and not body:
            body = f"<h1>{status_code} {status_message}</h1>".encode('utf-8')
            response_headers['Content-Type'] = 'text/html; charset=utf-8'
            response_headers['Content-Length'] = str(len(body))

        header_lines = "".join([f"{k}: {v}\r\n" for k, v in response_headers.items()])

        return response_line.encode('utf-8') + header_lines.encode('utf-8') + b'\r\n' + body

    def add_headers_to_response(self, response_bytes, additional_headers):
        """Injects additional headers into an existing response."""
        parts = response_bytes.split(b'\r\n\r\n', 1)
        headers_part, body = parts[0], parts[1] if len(parts) > 1 else b''

        header_lines = headers_part.split(b'\r\n')
        status_line = header_lines[0]

        existing_headers = {}
        for line in header_lines[1:]:
            key, value = line.decode().split(': ', 1)
            existing_headers[key.lower()] = value

        final_headers = {}
        # Update with additional headers first, so they can be overwritten by original if needed
        for k, v in additional_headers.items():
            final_headers[k.lower()] = v
        for k, v in existing_headers.items():
            final_headers[k] = v

        new_headers_str = "".join([f"{k}: {v}\r\n" for k, v in final_headers.items()])

        return status_line + b'\r\n' + new_headers_str.encode('utf-8') + b'\r\n' + body


def main():
    """Parses command-line arguments and starts the server."""
    host = sys.argv[2] if len(sys.argv) > 2 else DEFAULT_HOST
    port = int(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_PORT
    max_threads = int(sys.argv[3]) if len(sys.argv) > 3 else DEFAULT_MAX_THREADS

    server = HTTPServer(host, port, max_threads)
    server.start()


if __name__ == '__main__':
    main()