server.py

"""Educational HTTP/1.1 server built directly on Python sockets."""
import socket, sys, threading, os, json, random, string, queue
from datetime import datetime, timezone
from typing import Optional, Tuple, Dict

# ----------------------
# Server Configuration
# ----------------------
# I keep basic knobs up top so it's obvious how to tune the server.
DEFAULT_HOST = '127.0.0.1'
DEFAULT_PORT = 8080
MAX_THREADS = 10
MAX_QUEUE_SIZE = 100  # Max pending connections in queue before responding 503
RESOURCES_DIR = 'resources'
UPLOADS_DIR = os.path.join(RESOURCES_DIR, 'uploads')

KEEPALIVE_TIMEOUT = 30
KEEPALIVE_MAX = 100

CURRENT_HOST = DEFAULT_HOST
CURRENT_PORT = DEFAULT_PORT

def get_rfc7231_date():
    # I'm using RFC7231 format so clients recognize it in Date headers.
    now = datetime.now(timezone.utc)
    return now.strftime('%a, %d %b %Y %H:%M:%S GMT')

def log_message(message):
    # Simple print-based logging keeps this project dependency-free.
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {message}")

def main():
    # Parse CLI: port, host, max_threads (keep defaults when missing/invalid)
    # I allow quick overrides for demos and concurrency tests.
    port = DEFAULT_PORT
    host = DEFAULT_HOST
    max_threads = MAX_THREADS

    if len(sys.argv) > 1:
        try:
            port = int(sys.argv[1])
        except ValueError:
            log_message(f"Invalid port number: {sys.argv[1]}. Using default {DEFAULT_PORT}.")
    if len(sys.argv) > 2:
        host = sys.argv[2]
    if len(sys.argv) > 3:
        try:
            max_threads = int(sys.argv[3])
        except ValueError:
            log_message(f"Invalid thread pool size: {sys.argv[3]}. Using default {MAX_THREADS}.")

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

    # Set globals for Host validation
    global CURRENT_HOST, CURRENT_PORT
    CURRENT_HOST = host
    CURRENT_PORT = port

    # Here I create the listening socket and bind to HOST:PORT.
    # I also enable SO_REUSEADDR so quick restarts don't error out.
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    try:
        server_socket.bind((host, port))
        server_socket.listen(50)
    except OSError as e:
        log_message(f"Error binding or listening: {e}")
        sys.exit(1)

    # Now I bring up the thread pool. If the queue fills up, I'll return 503.
    pool = ThreadPool(max_threads, on_dequeued=lambda tn, addr: log_message(f"[{tn}] Connection dequeued, assigned to {tn}"))

    try:
        while True:
            client_socket, client_address = server_socket.accept()
            active, total = pool.status()
            log_message(f"Thread pool status: {active}/{total} active")
            log_message(f"Connection from {client_address[0]}:{client_address[1]}")
            # I try to enqueue every accepted socket. If the queue is full,
            # I am explicit: I answer with 503 Service Unavailable and close.
            if pool.submit(client_socket, client_address):
                log_message("Client connection enqueued for processing")
            else:
                log_message("Warning: Thread pool saturated, queuing connection failed. Responding 503.")
                try:
                    body = b"Service Unavailable"
                    headers = {
                        "Content-Type": "text/plain; charset=utf-8",
                        "Retry-After": "5",
                    }
                    # Send minimal 503
                    send_headers(client_socket, 503, headers, content_length=len(body), connection="close")
                    client_socket.sendall(body)
                except Exception as e:
                    log_message(f"Error sending 503: {e}")
                finally:
                    client_socket.close()
    except KeyboardInterrupt:
        log_message("Server is shutting down.")
    finally:
        server_socket.close()

class ThreadPool:
    """Very small fixed-size thread pool with a bounded task queue."""
    def __init__(self, max_workers: int, on_dequeued=None, queue_size: int = MAX_QUEUE_SIZE):
        # I'm prestarting worker threads so incoming connections get handled immediately.
        self.max_workers = max_workers
        self.on_dequeued = on_dequeued  # callback(thread_name, client_address)
        self.tasks = queue.Queue(maxsize=queue_size)
        self.threads = []
        self._active = 0
        self._lock = threading.Lock()
        for i in range(max_workers):
            t = threading.Thread(target=self._worker, name=f"Thread-{i+1}", daemon=True)
            t.start()
            self.threads.append(t)

    def _worker(self):
        """Workers pull sockets from the queue and handle each client."""
        # Each worker lives forever and processes one client at a time.
        while True:
            client_socket, client_address = self.tasks.get()
            with self._lock:
                self._active += 1
            try:
                if self.on_dequeued:
                    self.on_dequeued(threading.current_thread().name, client_address)
                handle_client(client_socket, client_address)
            except Exception as e:
                log_message(f"[{threading.current_thread().name}] Error processing client {client_address}: {e}")
            finally:
                with self._lock:
                    self._active -= 1
                self.tasks.task_done()

    def submit(self, client_socket, client_address) -> bool:
        """Queue a client connection if space is available."""
        # Non-blocking put: if the queue is full, I signal failure to the caller.
        try:
            self.tasks.put_nowait((client_socket, client_address))
            return True
        except queue.Full:
            return False

    def status(self):
        with self._lock:
            return self._active, self.max_workers

SERVER_NAME = "Multi-threaded HTTP Server"
ALLOWED_BINARY_EXTS = {".txt", ".png", ".jpg", ".jpeg"}

STATUS_MESSAGES = {
    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",
}

def status_line(code: int) -> str:
    """HTTP status line for a code."""
    return f"HTTP/1.1 {code} {STATUS_MESSAGES.get(code, '')}"

def send_headers(sock: socket.socket, code: int, extra_headers: dict, content_length: int = 0, connection: str = "close"):
    """Send status line + merged headers + required keep-alive details."""
    # I'm centralizing header formatting here so all responses look consistent.
    headers = {
        "Date": get_rfc7231_date(),
        "Server": SERVER_NAME,
        "Connection": connection,
    }
    if connection.lower() == "keep-alive":
        headers["Keep-Alive"] = f"timeout={KEEPALIVE_TIMEOUT}, max={KEEPALIVE_MAX}"
    if content_length is not None:
        headers["Content-Length"] = str(content_length)
    headers.update(extra_headers or {})
    lines = [status_line(code)] + [f"{k}: {v}" for k, v in headers.items()] + ["", ""]
    sock.sendall("\r\n".join(lines).encode("utf-8"))

def send_plain_error(sock: socket.socket, code: int, message: Optional[str] = None, connection: str = "close"):
    """Send a minimal text/plain error with correct length and connection."""
    body = (message or STATUS_MESSAGES.get(code, "")).encode("utf-8")
    send_headers(sock, code, {"Content-Type": "text/plain; charset=utf-8"}, content_length=len(body), connection=connection)
    sock.sendall(body)

RESOURCE_ROOT_ABS = os.path.realpath(RESOURCES_DIR)

def is_safe_path(candidate_path: str) -> bool:
    """Reject path traversal, absolute paths, and home shortcuts; allow only under resources."""
    # I normalize path separators and block obvious traversal markers.
    lowered = candidate_path.replace('\\', '/').lower()
    if ".." in lowered or lowered.startswith('/') or './' in lowered or lowered.startswith('~/'):
        return False
    # Then I compute the real path and ensure it stays under the resources root.
    abs_path = os.path.realpath(os.path.join(RESOURCES_DIR, candidate_path))
    return abs_path.startswith(RESOURCE_ROOT_ABS + os.sep) or abs_path == RESOURCE_ROOT_ABS

def resolve_request_path(path: str) -> Tuple[Optional[str], Optional[int]]:
    """Resolve request URL to an absolute file path under resources, or return (None, code)."""
    # If the client asks for '/', I intentionally serve index.html.
    rel = "index.html" if path == "/" else (path[1:] if path.startswith("/") else path)
    if not is_safe_path(rel):
        return None, 403
    full = os.path.realpath(os.path.join(RESOURCES_DIR, rel))
    if not os.path.exists(full) or not os.path.isfile(full):
        return None, 404
    return full, None

def get_content_headers_for_path(full_path: str) -> Tuple[Dict[str, str], Optional[int]]:
    """Return appropriate Content-* headers for known types or 415 for others."""
    # I keep MIME mapping tight on purpose—images are forced to download so
    # we can test binary transfers clearly.
    ext = os.path.splitext(full_path)[1].lower()
    text_map = {
        ".html": "text/html; charset=utf-8",
        ".css": "text/css; charset=utf-8",
        ".js": "text/javascript; charset=utf-8",
    }
    if ext in text_map:
        return {"Content-Type": text_map[ext]}, None
    if ext in ALLOWED_BINARY_EXTS:
        filename = os.path.basename(full_path)
        return {
            "Content-Type": "application/octet-stream",
            "Content-Disposition": f"attachment; filename=\"{filename}\"",
        }, None
    return {}, 415

def send_file(sock: socket.socket, full_path: str, headers: Dict[str, str], connection: str = "close"):
    """Stream a file in 8KB chunks after sending headers."""
    # Here I stream the file in chunks to keep memory usage steady for big files.
    size = os.path.getsize(full_path)
    send_headers(sock, 200, headers, content_length=size, connection=connection)
    with open(full_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            sock.sendall(chunk)
    return size

READ_MAX = 8192

def read_http_request(client_socket: socket.socket, max_bytes: int = READ_MAX, timeout: float = 5.0) -> bytes:
    """Read raw HTTP bytes up to max_bytes, including body if Content-Length fits."""
    # I read up to 8192 bytes for the request (headers + body). That’s enough
    # to demo HTTP parsing without risking huge memory usage from a single client.
    client_socket.settimeout(timeout)
    data = bytearray()
    sep = b"\r\n\r\n"
    try:
        # Read until headers end or max_bytes reached
        while sep not in data and len(data) < max_bytes:
            chunk = client_socket.recv(min(4096, max_bytes - len(data)))
            if not chunk:
                break
            data.extend(chunk)
        # If headers present, try to read body per Content-Length (bounded by max_bytes)
        idx = data.find(sep)
        if idx != -1:
            head = bytes(data[:idx])
            rest = bytes(data[idx + len(sep):])
            # Parse Content-Length if any
            try:
                header_text = head.decode('iso-8859-1')
            except UnicodeDecodeError:
                header_text = head.decode('utf-8', errors='replace')
            content_length = 0
            for line in header_text.split('\r\n')[1:]:
                if not line:
                    break
                if line.lower().startswith('content-length:'):
                    try:
                        content_length = int(line.split(':', 1)[1].strip())
                    except Exception:
                        content_length = 0
                    break
            needed = max(0, content_length - len(rest))
            while needed > 0 and len(data) < max_bytes:
                chunk = client_socket.recv(min(4096, max_bytes - len(data)))
                if not chunk:
                    break
                data.extend(chunk)
                needed -= len(chunk)
        return bytes(data)
    except socket.timeout:
        return bytes(data)
    finally:
        try:
            client_socket.settimeout(None)
        except Exception:
            pass

# (kept minimal) JSON responses are written inline where needed

def choose_connection(http_version: str, headers: Dict[str, str], req_count: int) -> str:
    """Pick 'keep-alive' or 'close' honoring HTTP version, header, and max-requests."""
    # I cap the number of requests per connection to keep things predictable.
    if req_count >= KEEPALIVE_MAX:
        return "close"
    ver = http_version.upper()
    conn = (headers.get("Connection") or headers.get("connection", "")).lower()
    if ver == "HTTP/1.1":
        return "close" if "close" in conn else "keep-alive"
    if ver == "HTTP/1.0":
        return "keep-alive" if "keep-alive" in conn else "close"
    return "close"

def handle_client(client_socket, client_address):
    """Serve one TCP client; supports multiple requests over the same connection."""
    # This is the heart of the server: I accept bytes, parse HTTP, and send responses.
    # I intentionally keep this loop small and rely on helpers to stay readable.
    thread_name = threading.current_thread().name
    req_count = 0
    try:
        while True:
            # Idle timeout for keep-alive connections
            try:
                client_socket.settimeout(KEEPALIVE_TIMEOUT)
            except Exception:
                pass

            request_data = read_http_request(client_socket)
            if not request_data:
                # Peer went idle or closed
                break

            method, path, http_version, headers, body = parse_request(request_data)
            req_count += 1
            log_message(f"[{thread_name}] Request: {method} {path} {http_version}")

            # Host header validation for all requests
            # I'm strict here so random Host injection won't route outside localhost.
            ok, code, seen = validate_host_header(headers)
            if not ok:
                log_message(f"[{thread_name}] Host validation: {seen} ✗")
                send_plain_error(client_socket, code)
                break
            else:
                log_message(f"[{thread_name}] Host validation: {seen} ✓")

            # Determine connection policy for this response
            connection = choose_connection(http_version, headers, req_count)

            method_upper = method.upper()
            if method_upper == "GET":
                # Strip any query string from the path
                clean_path = path.split('?', 1)[0]

                full, err = resolve_request_path(clean_path)
                if err:
                    send_plain_error(client_socket, err)
                    if connection == "close":
                        break
                    else:
                        continue
                content_headers, err2 = get_content_headers_for_path(full)
                if err2:
                    send_plain_error(client_socket, err2)
                    if connection == "close":
                        break
                    else:
                        continue
                size = os.path.getsize(full)
                base = os.path.basename(full)
                kind = "binary" if content_headers.get("Content-Type", "").startswith("application/octet-stream") else "HTML/CSS"
                log_message(f"[{thread_name}] Sending {kind} file: {base} ({size} bytes)")
                transferred = send_file(client_socket, full, content_headers, connection=connection)
                log_message(f"[{thread_name}] Response: 200 OK ({transferred} bytes transferred)")
            elif method_upper == "POST":
                if path != "/upload":
                    send_plain_error(client_socket, 404)
                    if connection == "close":
                        break
                    else:
                        continue
                ctype = headers.get("Content-Type", "")
                if ctype.split(";")[0].strip().lower() != "application/json":
                    # I only accept JSON here to keep the server predictable.
                    send_plain_error(client_socket, 415, "Unsupported Media Type")
                    if connection == "close":
                        break
                    else:
                        continue
                try:
                    decoded = body.decode('utf-8')
                except UnicodeDecodeError:
                    # If the client sends non-UTF-8, I make it explicit.
                    send_plain_error(client_socket, 400, "Bad Request: body must be UTF-8")
                    if connection == "close":
                        break
                    else:
                        continue
                try:
                    _ = json.loads(decoded)
                except json.JSONDecodeError:
                    # Here I validate the JSON structure to avoid writing bad data.
                    send_plain_error(client_socket, 400, "Bad Request: invalid JSON")
                    if connection == "close":
                        break
                    else:
                        continue
                os.makedirs(UPLOADS_DIR, exist_ok=True)
                ts = datetime.now().strftime('%Y%m%d_%H%M%S')
                rand = ''.join(random.choices(string.hexdigits.lower(), k=4))
                filename = f"upload_{ts}_{rand}.json"
                fullpath = os.path.join(UPLOADS_DIR, filename)
                with open(fullpath, 'wb') as f:
                    # I write the exact bytes the client sent (UTF-8) into uploads/.
                    f.write(decoded.encode('utf-8'))
                relpath = f"/uploads/{filename}"
                log_message(f"[{thread_name}] Created upload file: {filename}")
                # Inline JSON response (no helper needed beyond this point)
                resp_bytes = json.dumps({
                    "status": "success",
                    "message": "File created successfully",
                    "filepath": relpath,
                }).encode('utf-8')
                send_headers(client_socket, 201, {"Content-Type": "application/json"}, content_length=len(resp_bytes), connection=connection)
                client_socket.sendall(resp_bytes)
            else:
                # Anything else (PUT/DELETE/etc.) is out of scope for this assignment.
                send_plain_error(client_socket, 405)

            # Log connection policy and apply
            log_message(f"[{thread_name}] Connection: {connection}")
            if connection == "close":
                break
            if req_count >= KEEPALIVE_MAX:
                break

    except ValueError as ve:
        send_plain_error(client_socket, 400, str(ve))
    except FileNotFoundError:
        send_plain_error(client_socket, 404)
    except socket.timeout:
        # Idle timeout
        pass
    except Exception as e:
        log_message(f"[{thread_name}] Error handling client {client_address}: {e}")
        try:
            send_plain_error(client_socket, 500)
        except Exception:
            pass
    finally:
        try:
            client_socket.shutdown(socket.SHUT_RDWR)
        except Exception:
            pass
        client_socket.close()


def parse_request(request_bytes: bytes):
    """Parse raw HTTP request bytes into (method, path, version, headers, body).
    Raises ValueError on malformed structures.
    """
    # I parse just the essentials: request line, headers, body (if Content-Length says so).
    # Find end of headers (CRLF CRLF)
    sep = b"\r\n\r\n"
    idx = request_bytes.find(sep)
    if idx == -1:
        # No header terminator found within 8192
        raise ValueError("Malformed HTTP request: missing header terminator")

    head = request_bytes[:idx]
    body = request_bytes[idx + len(sep):]

    try:
        header_text = head.decode('iso-8859-1')  # HTTP headers are ISO-8859-1 per RFC
    except UnicodeDecodeError:
        header_text = head.decode('utf-8', errors='replace')

    lines = header_text.split('\r\n')
    if not lines:
        raise ValueError("Empty request")

    # Parse request line
    request_line = lines[0]
    try:
        method, path, http_version = request_line.split(' ', 2)
    except ValueError:
        raise ValueError(f"Malformed request line: {request_line}")

    # Parse headers (skip malformed ones but keep going)
    headers = {}
    for line in lines[1:]:
        if not line:
            break
        if ':' not in line:
            # Skip malformed header lines but continue parsing
            log_message(f"Skipping malformed header: {line}")
            continue
        key, value = line.split(':', 1)
        headers[key.strip()] = value.lstrip()

    return method, path, http_version, headers, body

def validate_host_header(headers: Dict[str, str]) -> Tuple[bool, int, str]:
    # Why I’m strict: I want to ensure requests are intended for this local server
    # and not accidentally routed from other hosts/ports.
    host_hdr = headers.get("Host")
    if not host_hdr:
        return False, 400, "(missing)"
    host_val = host_hdr.strip().lower()
    allowed_hosts = {"localhost", "127.0.0.1", CURRENT_HOST.lower()}
    if ':' in host_val:
        h, p = host_val.rsplit(':', 1)
        try:
            port_val = int(p)
        except ValueError:
            return False, 403, host_hdr
    else:
        h, port_val = host_val, CURRENT_PORT
    if h not in allowed_hosts or port_val != CURRENT_PORT:
        return False, 403, host_hdr
    return True, 200, host_hdr

if __name__ == "__main__":
    main()