server.py

import socket
import argparse
import os
import json
import datetime
import random
import string
import threading
from concurrent.futures import ThreadPoolExecutor
from email.utils import formatdate

# default settings for server
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8080
DEFAULT_MAX_THREADS = 10
LISTEN_BACKLOG = 50
CONNECTION_QUEUE_MAX = 50
RESOURCE_DIRNAME = "resources"
UPLOADS_DIRNAME = "uploads"
KEEP_ALIVE_TIMEOUT = 30
MAX_PERSISTENT_REQUESTS = 100
SUPPORTED_HTML = {".html"}
SUPPORTED_BINARY = {".txt", ".png", ".jpg", ".jpeg"}

# folder setup for resources and uploads
BASE_DIR = os.getcwd()
RESOURCE_DIR = os.path.join(BASE_DIR, RESOURCE_DIRNAME)
UPLOAD_DIR = os.path.join(RESOURCE_DIR, UPLOADS_DIRNAME)
os.makedirs(UPLOAD_DIR, exist_ok=True)

# current time in log format
def timestamp_now():
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# log function to print messages with time and thread name
def log(message):
    thread_name = threading.current_thread().name
    print(f"[{timestamp_now()}] [{thread_name}] {message}")

# generate random id for uploaded files
def generate_id(length=4):
    return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))

# to make sure file path requested is safe and inside resource folder
def safe_path(base, path):
    base_abs = os.path.abspath(base)
    joined = os.path.abspath(os.path.join(base_abs, path))
    if joined.startswith(base_abs + os.sep) or joined == base_abs:
        return joined
    return None

# date format used in HTTP header (standard format)
def rfc7231_date():
    return formatdate(timeval=None, localtime=False, usegmt=True)

# this function creates proper HTTP response with headers and body
def build_response(status_line, headers=None, body=b""):
    headers = headers or {}
    lines = [f"HTTP/1.1 {status_line}", f"Date: {rfc7231_date()}", "Server: Multi-threaded HTTP Server"]
    for k, v in headers.items():
        lines.append(f"{k}: {v}")
    header_block = "\r\n".join(lines) + "\r\n\r\n"
    if isinstance(body, str):
        body = body.encode("utf-8")
    return header_block.encode("utf-8") + body

# reads and separates HTTP request (method, path, headers, body)
def parse_request_bytes(data_bytes):
    try:
        text = data_bytes.decode("utf-8", errors="ignore")
        header_end = text.find("\r\n\r\n")
        if header_end == -1:
            return None, None, None, {}, b""
        header_text = text[:header_end]
        body_start_index = header_end + 4
        lines = header_text.split("\r\n")
        method, path, version = lines[0].split()
        headers = {}
        for line in lines[1:]:
            if ": " in line:
                k, v = line.split(": ", 1)
                headers[k.strip()] = v.strip()
        body_bytes = data_bytes[body_start_index:]
        return method, path, version, headers, body_bytes
    except Exception:
        return None, None, None, {}, b""

# function to handle reading files for GET requests
def read_file_for_path(path):
    if path == "/":
        requested = "index.html"
    else:
        requested = path.lstrip("/")
    if requested.startswith("/") or requested.startswith(".."):
        return 403, {}, b""
    safe = safe_path(RESOURCE_DIR, requested)
    if not safe:
        return 403, {}, b""
    if not os.path.exists(safe) or not os.path.isfile(safe):
        return 404, {}, b""
    _, ext = os.path.splitext(safe)
    ext = ext.lower()

    # if html file -> open and send normally
    if ext in SUPPORTED_HTML:
        with open(safe, "rb") as f:
            content = f.read()
        headers = {"Content-Type": "text/html; charset=utf-8", "Content-Length": str(len(content))}
        return 200, headers, content

    # if binary or text file -> send as downloadable file
    if ext in SUPPORTED_BINARY:
        with open(safe, "rb") as f:
            content = f.read()
        disposition = f'attachment; filename="{os.path.basename(safe)}"'
        headers = {
            "Content-Type": "application/octet-stream",
            "Content-Length": str(len(content)),
            "Content-Disposition": disposition
        }
        return 200, headers, content

    return 415, {}, b""

# saves uploaded JSON files into uploads folder
def save_upload_json(json_payload):
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    file_id = generate_id(4)
    filename = f"upload_{timestamp}_{file_id}.json"
    filepath = os.path.join(UPLOAD_DIR, filename)
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(json_payload, f, indent=2, ensure_ascii=False)
    return f"/{UPLOADS_DIRNAME}/{filename}"

# checks if Host header in request is valid for our server
def validate_host_header(host_header, server_host, server_port):
    if not host_header:
        return False
    host_header = host_header.strip()
    if host_header == f"{server_host}:{server_port}" or host_header == server_host:
        return True
    if server_host in ("127.0.0.1", "localhost") and (
        host_header == "localhost" or host_header == f"localhost:{server_port}"
    ):
        return True
    return False

# handles GET request logic
def handle_get(path, conn, headers, keep_alive):
    status, resp_headers, body = read_file_for_path(path)
    if status == 403:
        conn.sendall(build_response("403 Forbidden", {"Content-Length": "0", "Connection": "close"}))
        log(f"GET {path} -> 403 Forbidden")
        return False
    if status == 404:
        conn.sendall(build_response("404 Not Found", {"Content-Length": "0", "Connection": "close"}))
        log(f"GET {path} -> 404 Not Found")
        return False
    if status == 415:
        conn.sendall(build_response("415 Unsupported Media Type", {"Content-Length": "0", "Connection": "close"}))
        log(f"GET {path} -> 415 Unsupported Media Type")
        return False
    conn_header = "keep-alive" if keep_alive else "close"
    resp_headers["Connection"] = conn_header
    if keep_alive:
        resp_headers["Keep-Alive"] = f"timeout={KEEP_ALIVE_TIMEOUT}, max={MAX_PERSISTENT_REQUESTS}"
    conn.sendall(build_response("200 OK", resp_headers, body))
    log(f"GET {path} -> 200 OK ({resp_headers.get('Content-Length','?')} bytes)")
    return keep_alive

# handles POST request logic (JSON upload)
def handle_post(headers, body_bytes, conn, keep_alive):
    content_type = headers.get("Content-Type", "")
    if "application/json" not in content_type.lower():
        conn.sendall(build_response("415 Unsupported Media Type", {"Content-Length": "0", "Connection": "close"}))
        log("POST -> 415 Unsupported Media Type")
        return False
    try:
        payload = json.loads(body_bytes.decode("utf-8"))
    except Exception:
        conn.sendall(build_response("400 Bad Request", {"Content-Length": "0", "Connection": "close"}))
        log("POST -> 400 Bad Request")
        return False
    try:
        relpath = save_upload_json(payload)
    except Exception:
        conn.sendall(build_response("500 Internal Server Error", {"Content-Length": "0", "Connection": "close"}))
        log("POST -> 500 Internal Server Error")
        return False

    # preparing success response
    response_body = json.dumps({
        "status": "success",
        "message": "File created successfully",
        "filepath": relpath
    }).encode("utf-8")
    resp_headers = {
        "Content-Type": "application/json; charset=utf-8",
        "Content-Length": str(len(response_body)),
        "Connection": "keep-alive" if keep_alive else "close"
    }
    if keep_alive:
        resp_headers["Keep-Alive"] = f"timeout={KEEP_ALIVE_TIMEOUT}, max={MAX_PERSISTENT_REQUESTS}"
    conn.sendall(build_response("201 Created", resp_headers, response_body))
    log(f"POST -> 201 Created {relpath}")
    return keep_alive

# main function that serves each client connection (handles keep-alive too)
def serve_client(conn, addr, server_host, server_port):
    log(f"Connection from {addr[0]}:{addr[1]}")
    conn.settimeout(KEEP_ALIVE_TIMEOUT)
    persistent_count = 0
    try:
        while persistent_count < MAX_PERSISTENT_REQUESTS:
            try:
                data = conn.recv(8192)
            except socket.timeout:
                log("Connection timed out.")
                break
            if not data:
                break
            method, path, version, headers, body_bytes = parse_request_bytes(data)
            if not method:
                conn.sendall(build_response("400 Bad Request", {"Content-Length": "0", "Connection": "close"}))
                log("Bad Request")
                break
            log(f"Request: {method} {path}")
            host_header = headers.get("Host", "")
            if not host_header:
                conn.sendall(build_response("400 Bad Request", {"Content-Length": "0", "Connection": "close"}))
                log("Missing Host header")
                break
            if not validate_host_header(host_header, server_host, server_port):
                conn.sendall(build_response("403 Forbidden", {"Content-Length": "0", "Connection": "close"}))
                log(f"Host validation failed: {host_header}")
                break
            
            conn_hdr = headers.get("Connection", "").lower()
            log(f"Host validation: {host_header}")
            if conn_hdr:
                keep_alive = conn_hdr == "keep-alive"
            else:
                keep_alive = (version == "HTTP/1.1")

            # handle GET or POST
            if method == "POST":
                content_length = int(headers.get("Content-Length", "0"))
                while len(body_bytes) < content_length:
                    try:
                        more = conn.recv(8192)
                    except socket.timeout:
                        break
                    if not more:
                        break
                    body_bytes += more
                keep = handle_post(headers, body_bytes, conn, keep_alive)
            elif method == "GET":
                keep = handle_get(path, conn, headers, keep_alive)
            else:
                conn.sendall(build_response("405 Method Not Allowed", {"Content-Length": "0", "Connection": "close"}))
                log(f"{method} -> 405 Method Not Allowed")
                keep = False

            persistent_count += 1
            if not keep:
                break
    except Exception as e:
        log(f"Error while serving client: {e}")
    finally:
        try:
            conn.close()
        except Exception:
            pass
        log(f"Connection closed after {persistent_count} request(s).")

# main server function that runs and accepts clients using thread pool
def run_server(host, port, max_threads):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(LISTEN_BACKLOG)
    server.settimeout(1.0)
    print(f"[{timestamp_now()}] HTTP Server started on http://{host}:{port}")
    print(f"[{timestamp_now()}] Thread pool size: {max_threads}\n")

    pool = ThreadPoolExecutor(max_workers=max_threads)
    connection_queue = []
    active_tasks = 0
    lock = threading.Lock()

    # inner function that handles a single client inside thread
    def client_wrapper(conn, addr):
        nonlocal active_tasks
        threading.current_thread().name = f"Thread-{threading.get_ident() % 1000}"
        with lock:
            active_tasks += 1
            log(f"Thread pool active: {active_tasks}/{max_threads}")
        try:
            serve_client(conn, addr, host, port)
        finally:
            with lock:
                active_tasks -= 1
                log(f"Thread pool active: {active_tasks}/{max_threads}")
                # if queue has pending connections, take next one
                if connection_queue:
                    queued_conn, queued_addr = connection_queue.pop(0)
                    log("Connection dequeued")
                    pool.submit(client_wrapper, queued_conn, queued_addr)

    try:
        while True:
            try:
                conn, addr = server.accept()
                with lock:
                    if active_tasks >= max_threads:
                        # if too many connections, queue or reject
                        if len(connection_queue) >= CONNECTION_QUEUE_MAX:
                            try:
                                retry_headers = {"Content-Length": "0", "Connection": "close", "Retry-After": "5"}
                                conn.sendall(build_response("503 Service Unavailable", retry_headers))
                            except Exception:
                                pass
                            conn.close()
                            log(f"Connection refused (503) from {addr[0]}:{addr[1]}")
                            continue
                        connection_queue.append((conn, addr))
                        log(f"Thread pool full, queued {addr[0]}:{addr[1]}")
                        continue
                pool.submit(client_wrapper, conn, addr)
            except socket.timeout:
                continue
    except KeyboardInterrupt:
        print(f"[{timestamp_now()}] Server shutting down.")
    finally:
        server.close()
        pool.shutdown(wait=True)
        print(f"[{timestamp_now()}] Server stopped.")

# arguments from command line (host, port, threads)
def parse_args():
    parser = argparse.ArgumentParser(description="Multi-threaded HTTP Server")
    parser.add_argument("port", nargs="?", type=int, default=DEFAULT_PORT)
    parser.add_argument("host", nargs="?", type=str, default=DEFAULT_HOST)
    parser.add_argument("max_threads", nargs="?", type=int, default=DEFAULT_MAX_THREADS)
    return parser.parse_args()

# program starts from here
if __name__ == "__main__":
    args = parse_args()
    run_server(args.host, args.port, args.max_threads)