server.py

import argparse
import math
import random
import time
import socket
import threading
from RUDP.rudp_protocol import RUDPPacket
from client_dns import register_domain


DEBUG = True

_real_print = print

def print(*args, **kwargs):
    if DEBUG:
        _real_print(*args, **kwargs)

#------------------- controllers ------------------- #
from Controller.HomeController import HomeController
from Controller.WatchController import WatchController
from Controller.MediaController import MediaController
from Controller.JsController import JsController
from Controller.CssController import CssController

# ------------------- Server Setup ------------------- #
controllers = {"home": HomeController(),\
                "watch": WatchController(),\
                "media": MediaController(),\
                "js": JsController(),\
                "css":CssController() }

RUDP_clients={}
packet_loss_probabilty = 0.01

MAX_RETRIES = 5

stop_event = threading.Event()

def httpResponse(body, status=200, content_type="text/plain; charset=utf-8", connection="close", headers=None):
    if headers is None:
        headers = {}

    if isinstance(body, str):
        body_bytes = body.encode("utf-8")
    else:
        body_bytes = body

    statuses = {
    200: "OK",
    201: "Created",
    202: "Accepted",
    206: "Partial Content",
    400: "Bad Request",
    404: "Not Found",
    416: "Range Not Satisfiable",
    }
    status_reply = statuses.get(status, "OK")

    header_str = (
        f"HTTP/1.1 {status} {status_reply}\r\n"
        f"Content-Type: {content_type}\r\n"
        f"Content-Length: {len(body_bytes)}\r\n"
        f"Connection: {connection}\r\n"
    )

    for k, v in headers.items():
        header_str += f"{k}: {v}\r\n"

    print(f"[server] response headers:\n{header_str}\n")

    header_str += "\r\n"

    return header_str.encode("utf-8") + body_bytes

def handle_http_request(request):
    global packet_loss_probabilty

    headers_end = request.find(b"\r\n\r\n")
    if headers_end == -1:
        return httpResponse("Bad Request", status=400, connection="close"), False

    headers_part = request[:headers_end]
    body = request[headers_end + 4:]
    lines = headers_part.split(b"\r\n")
    hdrs = get_headers(lines)

    if "chunked" in hdrs.get("transfer-encoding", "").lower():
        return httpResponse("Chunked uploads not supported", status=400, connection="close"), False

    method, target, version = lines[0].split(b" ", 2)
    method = method.decode()
    target = target.decode()

    path = target.split("?", 1)[0].strip("/")
    controller = path.split("/", 1)[0] if "/" in path else path
    route = path.split("/", 1)[-1] if "/" in path else ""

    content_length = int(hdrs.get("content-length", "0"))
    if len(body) < content_length:
        print(f"[server] incomplete POST body received, expected {content_length} bytes, got {len(body)} bytes")
        return httpResponse("Incomplete POST body", status=400, connection="close"), False

    req_conn = hdrs.get("connection", "").lower()
    connection = "keep-alive" if req_conn == "keep-alive" else "close"

    content_type_header = hdrs.get("content-type", "").lower()
    if method == "POST" and "multipart/form-data" in content_type_header:
        connection = "close"

    if controller in controllers:
        body, status, type, addedHeaders = controllers[controller].control(route, body, method, headers=hdrs)
    elif controller == "loss_probability" and method == "POST":
        value = route
        try:
            value = float(value)
            if 0 <= value <= 1:
                packet_loss_probabilty = value
                print(f"[server] packet_loss_probabilty set to {packet_loss_probabilty:.3f}")
                return httpResponse(
                    f"packet_loss_probabilty set to {packet_loss_probabilty:.3f}",
                    status=200,
                    content_type="text/plain; charset=utf-8",
                    connection="close"
                ), False
            return httpResponse("Invalid float value", status=400, connection="close"), False
        except:
            return httpResponse("Invalid float value", status=400, connection="close"), False
    else:
        body, status, type, addedHeaders = controllers["home"].control(path, body, method, headers=hdrs)

    res = httpResponse(body, status=status, content_type=type, headers=addedHeaders, connection=connection)
    resHeaders = res.split(b"\r\n\r\n", 1)[0].split(b"\r\n")[1:]
    resHeadersDict = get_headers(resHeaders)
    keepalive = resHeadersDict.get("connection", "close").lower() == "keep-alive"
    return res, keepalive

def get_headers(lines):
    headers={}
    for line in lines:
        if b":" in line:
            key,value = line.split(b":",1)
            headers[key.decode().lower()] = value.decode().strip()
    return headers


def handle_TCP_client(conn, addr):
    print(f"[server] connection from {addr}")

    with conn:
        buffer = b""
        while True:
            try:
                data = conn.recv(65536)
            except (ConnectionResetError, ConnectionAbortedError, OSError) as e:
                print(f"[server] error handling request from {addr}: {e}")
                break

            if not data:
                print(f"[server] no data from {addr}, closing connection")
                break

            buffer += data

            while True:
                headers_end = buffer.find(b"\r\n\r\n")
                if headers_end == -1:
                    break

                headers_content = buffer[:headers_end]
                lines = headers_content.split(b"\r\n")
                headersDict = get_headers(lines)
                content_length = int(headersDict.get("content-length", "0"))
                packet_length = headers_end + 4 + content_length

                if len(buffer) < packet_length:
                    break

                full_request = buffer[:packet_length]
                buffer = buffer[packet_length:]

                response, keepalive = handle_http_request(full_request)
                conn.sendall(response)

                if not keepalive:
                    return

def handle_RUDP_client(soc:socket.socket,data:bytes,addr,host):
    packet = RUDPPacket.from_bytes(data)
    
    if not packet:
        return
    
    if packet.flags == 0 and random.random()<packet_loss_probabilty:
                    print(f"[server] Simulating packet lost, lost packet sequence {packet.seq_num}")
                    return
    
    if addr not in RUDP_clients:
        add_RUDP_client(addr=addr,packet=packet)
    client = RUDP_clients[addr]

    #SYN
    if (packet.flags & RUDPPacket.SYN) and not (packet.flags & RUDPPacket.ACK):
        server_isn = random.randint(0, 65535)
        client["state"] = "SYN_RCVD"
        client["client_isn"] = packet.seq_num
        client["req_expected_seq"] = packet.seq_num + 1
        client["server_isn"] = server_isn
        if random.random() > packet_loss_probabilty:
            syn_ack_packet = RUDPPacket(
                version=1,
                flags=RUDPPacket.SYN | RUDPPacket.ACK,
                ack_num=packet.seq_num + 1,
                seq_num=server_isn,
                src_ip=host
            )
            soc.sendto(syn_ack_packet.to_bytes(), addr)
            return
        else:
            print(f"[server] simulating ACK loss, ACK {packet.seq_num+1} lost")
            return

    #ACK to SYN/ACK
    server_isn = client.get("server_isn")
    if (packet.flags & RUDPPacket.ACK) and server_isn is not None and packet.ack_num == server_isn + 1 and client.get("state") == "SYN_RCVD":
        client["state"] = "ESTABLISHED"
        client["req_buff"].clear()
        client["req_bytes"].clear()
        return
    
    #ACK 
    elif (packet.flags & RUDPPacket.ACK) and client["state"] == "ESTABLISHED":
        ack = packet.ack_num  # next expected seq
        print(f"ack number {ack} arrived")
        # cumulative ACK: everything < ack is acknowledged
        if ack > client["send_base"]:
            for seq in range(client["send_base"], ack):
                client["unacked"].pop(seq, None)
                client["retries"].pop(seq, None)

            client["send_base"] = ack
            if client["window_size"] < client["ssthresh"]:
                client["window_size"] += 1 
            else:
                client["ack_count"] +=1
                if client["ack_count"] >= client["window_size"]:
                    client["window_size"] +=1
                    client["ack_count"] = 0


        if client["segments"] and client["send_base"] >= len(client["segments"]):
            print("[RUDP] Response fully acknowledged. Resetting sender state.")

            client["segments"] = []
            client["unacked"].clear()
            client["send_base"] = 0
            client["next_seq"] = 0

        send_RUDP_resp(soc,client,addr,None,host)
        return

    #FIN
    if packet.flags & RUDPPacket.FIN:
        client["state"] = "FIN_WAIT"
        server_isn = client.get("server_isn")
        fin_packet = RUDPPacket(
            version=1,
            flags=RUDPPacket.FIN | RUDPPacket.ACK,
            ack_num=packet.seq_num + 1,
            seq_num=(server_isn if server_isn is not None else 0) + 1,
            src_ip=host
        )
        soc.sendto(fin_packet.to_bytes(), addr)
        remove_client(addr=addr)
        return
    
    #DATA
    if not packet.flags and client.get("state")=="ESTABLISHED":
        seq = packet.seq_num
        if seq== client.get("req_expected_seq"):
            client["req_bytes"].extend(packet.payload)
            client["req_expected_seq"] +=1
            while client.get("req_expected_seq") in client.get("req_buff"):
                client["req_bytes"].extend(client.get("req_buff").pop(client["req_expected_seq"]))
                client["req_expected_seq"] +=1

        elif seq > client.get("req_expected_seq"):
            client.get("req_buff")[seq] = packet.payload

        if random.random() > packet_loss_probabilty:
            req_ack_packet = RUDPPacket(version=1,flags=RUDPPacket.ACK,seq_num=0,ack_num=client.get("req_expected_seq"),src_ip=host)
            soc.sendto(req_ack_packet.to_bytes(),addr)
        else:
            print(f"[server] simulating ACK loss, ACK {client.get('req_expected_seq')} lost")

    #read HTTP requst from payload
    payload = client.get("req_bytes")
    if not payload:
        return
    while True:
        headers_end = payload.find(b"\r\n\r\n")
        if headers_end == -1:
            break
        headers_content = bytes(payload[:headers_end])
        lines = headers_content.split(b"\r\n")
        headersDict = get_headers(lines[1:])
        content_length = int(headersDict.get("content-length", "0"))
        packet_length = headers_end + 4 + content_length
        if len(payload)<packet_length:
            break
        full_request = bytes(payload[:packet_length])
        payload = (payload[packet_length:])
        response, keepalive = handle_http_request(full_request)
        send_RUDP_resp(soc=soc,client=client,addr=addr,response=response,host=host)
        if not keepalive:
            break
    client["req_bytes"] = payload
    return

MAX_RESP_SIZE = 1200
RTO = 0.1

def send_RUDP_resp(soc: socket.socket, client: dict, addr, response, host):
    if addr not in RUDP_clients:
        return

    if response is not None and client["segments"]:
        return

    if response is not None:
        client["segments"] = segment_response(response)
        client["send_base"] = 0
        client["next_seq"] = 0
        client["ssthresh"] = 16
        client["ack_count"] = 0
        client["unacked"].clear()
        client["retries"].clear()

    if not client["segments"]:
        return

    while client["next_seq"] < client["send_base"] + client["window_size"] and client["next_seq"] < len(client["segments"]):
        seq = client["next_seq"]
        seq_packet = RUDPPacket(
            version=1,
            flags=0,
            seq_num=seq,
            ack_num=0,
            src_ip=host,
            payload=client["segments"][seq]
        )
        packet_bytes = seq_packet.to_bytes()

        try:
            soc.sendto(packet_bytes, addr)
        except (TimeoutError, OSError) as e:
            print(f"[server] sendto failed for {addr}, seq={seq}: {e}")
            return

        print(f"[server] Packet sequence {seq} sent")
        client["unacked"][seq] = (packet_bytes, time.time())
        client["next_seq"] += 1

        if client["next_seq"] >= len(client["segments"]):
            print("[server] sent all data")

    now = time.time()
    if client["send_base"] in client["unacked"]:
        base_bytes, base_time = client["unacked"][client["send_base"]]
        if now - base_time > RTO:
            for seq in range(client["send_base"], client["next_seq"]):
                if seq in client["unacked"]:
                    client["retries"][seq] = client["retries"].get(seq, 0) + 1
                    tries = client["retries"][seq]

                    if tries > MAX_RETRIES:
                        print(f"[server] GIVING UP seq={seq} after {tries-1} retries. Closing RUDP client {addr}")
                        remove_client(addr)
                        return

                    packet, _ = client["unacked"][seq]
                    print(f"[server] timeout! resending packet sequence {seq} resizing window to 4")
                    client["ssthresh"] = max(4,client["window_size"]//2)
                    client["window_size"] = 4
                    client["ack_count"] = 0


                    try:
                        soc.sendto(packet, addr)
                    except (TimeoutError, OSError) as e:
                        print(f"[server] resend failed for {addr}, seq={seq}: {e}")
                        return

                    client["unacked"][seq] = (packet, now)

def segment_response(response:bytes)->list:
    segments = []
    seq = 0
    packet_size = int(MAX_RESP_SIZE*math.pow(1-packet_loss_probabilty,4))
    for i in range(0,len(response),packet_size):
        segments.append(response[i:i+packet_size])
        seq+=1
    return segments


def add_RUDP_client(addr, packet: RUDPPacket):
    RUDP_clients[addr] = {
        "state": "CLOSED",
        "client_isn": None,
        "server_isn": None,
        "send_base": 0,
        "next_seq": 0,
        "window_size": 4,
        "ssthresh": 16,
        "ack_count": 0,
        "unacked": {},
        "segments": [],
        "expected_seq": 0,
        "req_expected_seq": 0,
        "req_buff": {},
        "req_bytes": bytearray(),
        "retries": {}
    }
    
def remove_client(addr):
    if addr not in RUDP_clients:
        return
    client = RUDP_clients[addr]
    client.get("req_buff").clear()
    client.get("unacked").clear()
    client.get("segments").clear()
    client.get("req_bytes").clear()

    RUDP_clients.pop(addr)
    
    
def serverTCP(host, port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen(5)
        s.settimeout(0.5)  
        print(f"[server] listening on {host}:{port}")
        while not stop_event.is_set():
            try:
                conn, addr = s.accept()
            except socket.timeout:
                continue
            except OSError:
                break
            threading.Thread(
                target=handle_TCP_client,
                args=(conn, addr),
                daemon=True
            ).start()
        print("[server] TCP stopped")

TICK_INTERVAL = 0.005   
SOCKET_TIMEOUT = 0.005  

def serverRUDP(host, port):
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.settimeout(SOCKET_TIMEOUT)
        print(f"[server] RUDP listening on {host}:{port}")
        last_tick = time.time()
        while not stop_event.is_set():
            try:
                data, addr = s.recvfrom(65535)
                print(f"[server] new rudp connection from {addr}")
                handle_RUDP_client(s, data, addr, host)
            except socket.timeout:
                pass
            except ConnectionResetError:
                continue
            except OSError as e:
                if stop_event.is_set():
                    break
                print("[server] UDP socket error:", e)
                continue
            now = time.time()
            if now - last_tick >= TICK_INTERVAL:
                for addr, client in list(RUDP_clients.items()):
                    send_RUDP_resp(soc=s, client=client, addr=addr, response=None, host=host)
                last_tick = now

        print("[server] RUDP stopped")

def serve(host, tcp_port, udp_port):
    tcp_thread = threading.Thread(target=serverTCP, args=(host, tcp_port), daemon=True)
    udp_thread = threading.Thread(target=serverRUDP, args=(host, udp_port), daemon=True)
    tcp_thread.start()
    udp_thread.start()
    try:
        while tcp_thread.is_alive() and udp_thread.is_alive():
            time.sleep(0.2)
    except KeyboardInterrupt:
        print("\n[server] Ctrl+C received, shutting down...")
        stop_event.set()
    tcp_thread.join(timeout=2)
    udp_thread.join(timeout=2)
    print("[server] Shutdown complete")

def get_my_local_ip():
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.connect(("8.8.8.8", 80))  
    local_ip = s.getsockname()[0]
    s.close()
    return local_ip

def register_in_dns(dns:str,domain:str,host:str):
    local_ip = "127.0.0.1"
    if host == "0.0.0.0":
        local_ip = get_my_local_ip()
    register_domain(domain=domain,ip=local_ip,dns_ip=dns)



# ------------------- Main ------------------- #

def main():
    ap = argparse.ArgumentParser(description="DASH Video Server")
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--tcp_port", type=int, default=80)
    ap.add_argument("--udp_port", type=int, default=8080)
    ap.add_argument("--dns",default="127.0.0.1")
    ap.add_argument("--domain",default="my-app.home")

    args = ap.parse_args()
    register_in_dns(args.dns,args.domain,args.host)
    serve(args.host, args.tcp_port,args.udp_port)

if __name__ == "__main__":
    main()