// blast: abre una conexión TCP, envía bytes y la cierra.
//
//   blast <ip> <puerto> <bytes> [--chunk N] [--gap-ms N] [--nodelay] [--reply] [--abort]
//
// Con --chunk y --gap-ms, envía de a N bytes con una pausa entre cada write. Con --reply,
// cierra solo su sentido con shutdown y se queda leyendo lo que conteste el servidor. Con
// --abort, cierra con SO_LINGER en cero: el kernel envía RST en vez de FIN.
#define _POSIX_C_SOURCE 200809L
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

#define BUF_SIZE 65536

static void pause_ms(long ms) {
    struct timespec ts = {ms / 1000, (ms % 1000) * 1000000L};
    while (nanosleep(&ts, &ts) < 0) {
        if (errno != EINTR) {
            perror("nanosleep");
            exit(1);
        }
    }
}

// Lo que el servidor conteste después de nuestro FIN: el otro sentido sigue abierto.
static void read_reply(int fd) {
    char reply[256];
    ssize_t got;
    while ((got = read(fd, reply, sizeof reply - 1)) > 0) {
        reply[got] = '\0';
        printf("el servidor contesta: %s", reply);
    }
    if (got < 0) {
        perror("read");
        exit(1);
    }
}

int main(int argc, char **argv) {
    if (argc < 4) {
        fprintf(stderr, "uso: %s <ip> <puerto> <bytes> [--chunk N] [--gap-ms N] [--nodelay] [--reply] [--abort]\n",
                argv[0]);
        return 2;
    }
    long long total = atoll(argv[3]);
    size_t chunk = BUF_SIZE;
    long gap_ms = 0;
    int nodelay = 0, want_reply = 0, abort_close = 0;
    for (int i = 4; i < argc; i++) {
        if (strcmp(argv[i], "--chunk") == 0 && i + 1 < argc) {
            chunk = (size_t)atol(argv[++i]);
        } else if (strcmp(argv[i], "--gap-ms") == 0 && i + 1 < argc) {
            gap_ms = atol(argv[++i]);
        } else if (strcmp(argv[i], "--nodelay") == 0) {
            nodelay = 1;
        } else if (strcmp(argv[i], "--reply") == 0) {
            want_reply = 1;
        } else if (strcmp(argv[i], "--abort") == 0) {
            abort_close = 1;
        } else {
            fprintf(stderr, "opción desconocida: %s\n", argv[i]);
            return 2;
        }
    }
    if (chunk == 0 || chunk > BUF_SIZE) {
        fprintf(stderr, "--chunk debe estar entre 1 y %d\n", BUF_SIZE);
        return 2;
    }

    struct sockaddr_in addr = {.sin_family = AF_INET, .sin_port = htons((uint16_t)atoi(argv[2]))};
    if (inet_pton(AF_INET, argv[1], &addr.sin_addr) != 1) {
        fprintf(stderr, "dirección inválida: %s\n", argv[1]);
        return 2;
    }
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) {
        perror("socket");
        return 1;
    }
    if (nodelay) {
        int one = 1;
        if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one) < 0) {
            perror("setsockopt TCP_NODELAY");
            return 1;
        }
    }
    if (connect(fd, (struct sockaddr *)&addr, sizeof addr) < 0) {
        perror("connect");
        return 1;
    }

    static char buf[BUF_SIZE];
    memset(buf, 'x', sizeof buf);
    long long sent = 0;
    while (sent < total) {
        size_t want = total - sent < (long long)chunk ? (size_t)(total - sent) : chunk;
        ssize_t done = write(fd, buf, want);
        if (done < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("write");
            return 1;
        }
        sent += done;
        if (gap_ms > 0 && sent < total) {
            pause_ms(gap_ms);
        }
    }
    if (want_reply) {
        if (shutdown(fd, SHUT_WR) < 0) {
            perror("shutdown");
            return 1;
        }
        read_reply(fd);
    }
    if (abort_close) {
        struct linger hard = {.l_onoff = 1, .l_linger = 0};
        if (setsockopt(fd, SOL_SOCKET, SO_LINGER, &hard, sizeof hard) < 0) {
            perror("setsockopt SO_LINGER");
            return 1;
        }
    }
    if (close(fd) < 0) {
        perror("close");
        return 1;
    }
    printf("enviados %lld bytes\n", sent);
    return 0;
}

// connect_probe: intenta una conexión y dice qué devolvió connect y cuánto demoró.
//
//   connect_probe <ip> <puerto> [reintentos_de_syn [espera_ms]]
//
// El tercer argumento fija TCP_SYNCNT, para no esperar los dos minutos que Linux tarda por
// defecto en rendirse cuando nadie contesta. El cuarto mantiene abierta la conexión un rato
// antes de cerrarla, para que el otro extremo pueda cerrar primero.
#define _POSIX_C_SOURCE 200809L
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

static double now_seconds(void) {
    struct timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0) {
        perror("clock_gettime");
        exit(1);
    }
    return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9;
}

static const char *errno_name(int number) {
    switch (number) {
    case ECONNREFUSED: return "ECONNREFUSED";
    case ETIMEDOUT: return "ETIMEDOUT";
    case EHOSTUNREACH: return "EHOSTUNREACH";
    case ENETUNREACH: return "ENETUNREACH";
    default: return "otro";
    }
}

int main(int argc, char **argv) {
    if (argc < 3) {
        fprintf(stderr, "uso: %s <ip> <puerto> [reintentos_de_syn [espera_ms]]\n", argv[0]);
        return 2;
    }
    struct sockaddr_in addr = {.sin_family = AF_INET, .sin_port = htons((uint16_t)atoi(argv[2]))};
    if (inet_pton(AF_INET, argv[1], &addr.sin_addr) != 1) {
        fprintf(stderr, "dirección inválida: %s\n", argv[1]);
        return 2;
    }
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) {
        perror("socket");
        return 1;
    }
    if (argc > 3) {
        int retries = atoi(argv[3]);
        if (setsockopt(fd, IPPROTO_TCP, TCP_SYNCNT, &retries, sizeof retries) < 0) {
            perror("setsockopt TCP_SYNCNT");
            return 1;
        }
    }
    double start = now_seconds();
    int result = connect(fd, (struct sockaddr *)&addr, sizeof addr);
    int saved = errno;
    double seconds = now_seconds() - start;
    if (result == 0) {
        printf("connect devolvió 0 en %.3f s\n", seconds);
    } else {
        printf("connect devolvió -1 en %.3f s: %s (%s)\n", seconds, errno_name(saved),
               strerror(saved));
    }
    if (result == 0 && argc > 4) {
        long ms = atol(argv[4]);
        struct timespec ts = {ms / 1000, (ms % 1000) * 1000000L};
        if (nanosleep(&ts, NULL) < 0) {
            perror("nanosleep");
            return 1;
        }
    }
    if (close(fd) < 0) {
        perror("close");
        return 1;
    }
    return 0;
}

// half_open: deja conexiones a medio abrir. Envía SYN desde una dirección propia y nunca
// termina el handshake, porque en este laboratorio el router descarta los SYN-ACK que
// vuelven a esa dirección. Es la forma de ver una cola de SYN llena sin falsificar nada.
//
//   half_open <ip_propia> <ip_destino> <puerto> <cantidad> <segundos>
#define _POSIX_C_SOURCE 200809L
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <unistd.h>

int main(int argc, char **argv) {
    if (argc != 6) {
        fprintf(stderr, "uso: %s <ip_propia> <ip_destino> <puerto> <cantidad> <segundos>\n", argv[0]);
        return 2;
    }
    struct sockaddr_in local = {.sin_family = AF_INET};
    struct sockaddr_in peer = {.sin_family = AF_INET, .sin_port = htons((uint16_t)atoi(argv[3]))};
    if (inet_pton(AF_INET, argv[1], &local.sin_addr) != 1 ||
        inet_pton(AF_INET, argv[2], &peer.sin_addr) != 1) {
        fprintf(stderr, "dirección inválida\n");
        return 2;
    }
    int count = atoi(argv[4]);
    for (int i = 0; i < count; i++) {
        int fd = socket(AF_INET, SOCK_STREAM, 0);
        if (fd < 0) {
            perror("socket");
            return 1;
        }
        if (bind(fd, (struct sockaddr *)&local, sizeof local) < 0) {
            perror("bind");
            return 1;
        }
        // No bloqueante: connect vuelve de inmediato con EINPROGRESS, y el SYN ya salió.
        int flags = fcntl(fd, F_GETFL);
        if (flags < 0 || fcntl(fd, F_SETFL, flags | O_NONBLOCK) < 0) {
            perror("fcntl");
            return 1;
        }
        if (connect(fd, (struct sockaddr *)&peer, sizeof peer) < 0 && errno != EINPROGRESS) {
            perror("connect");
            return 1;
        }
    }
    printf("%d conexiones a medio abrir\n", count);
    if (fflush(stdout) != 0) {
        perror("fflush");
        return 1;
    }
    sleep((unsigned)atoi(argv[5]));
    return 0;
}

"""Ayudas de la semana 10: levantar la topología y correr comandos en sus nodos."""
import os
import subprocess
import time
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
CAPTURES = os.path.join(HERE, "captures")
# La solución no se publica. Sin ella, las pruebas usan lo que ustedes escriben en lab/.
ANSWERS = os.path.join(HERE, "solution")
if not os.path.isdir(ANSWERS):
    ANSWERS = os.path.join(HERE, "lab")


def answer_path(name):
    """Devuelve la ruta de un archivo del laboratorio, o salta si sigue con el TODO."""
    path = os.path.join(ANSWERS, name)
    with open(path, encoding="utf-8") as file:
        if "TODO" in file.read():
            raise unittest.SkipTest(f"falta completar lab/{name}")
    return path


class Lab:
    """La topología de Containerlab, que se levanta y se destruye desde las pruebas."""

    def __init__(self, folder="tcp", name="w10-tcp", network="redes-w10-tcp"):
        self.folder = os.path.join(HERE, folder)
        self.topology = f"{folder}.clab.yml"
        self.prefix = f"clab-{name}-"
        self.network = network

    def deploy(self):
        done = subprocess.run(
            ["containerlab", "deploy", "-t", self.topology, "--reconfigure"],
            cwd=self.folder, capture_output=True, text=True, timeout=300)
        if done.returncode != 0:
            raise RuntimeError(done.stdout + done.stderr)

    def destroy(self):
        subprocess.run(["containerlab", "destroy", "-t", self.topology, "--cleanup"],
                       cwd=self.folder, capture_output=True, text=True, timeout=300)
        subprocess.run(["docker", "network", "rm", self.network], capture_output=True, text=True)

    def sh(self, node, *command, timeout=90):
        """Corre un comando en un nodo y devuelve el resultado completo."""
        return subprocess.run(["docker", "exec", self.prefix + node, *command],
                              capture_output=True, text=True, timeout=timeout)

    def out(self, node, *command, timeout=90):
        """Corre un comando en un nodo y devuelve stdout y stderr juntos."""
        done = self.sh(node, *command, timeout=timeout)
        return done.stdout + done.stderr

    def daemon(self, node, command):
        """Deja un comando corriendo en segundo plano dentro de un nodo."""
        subprocess.run(["docker", "exec", "-d", self.prefix + node, "sh", "-c", command],
                       capture_output=True, text=True, check=True)

    def counters(self, node):
        """Los contadores TCP del kernel de un nodo, como diccionario."""
        table = {}
        for line in self.out(node, "nstat", "-azs").splitlines():
            parts = line.split()
            if len(parts) >= 2 and parts[1].isdigit():
                table[parts[0]] = int(parts[1])
        return table


def wait_until(predicate, seconds=30, step=0.25):
    """Espera a que una condición se cumpla; devuelve si se cumplió."""
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        if predicate():
            return True
        time.sleep(step)
    return predicate()

"""Lee una captura con tshark y la devuelve como una lista de segmentos."""
import subprocess

FIELDS = {
    "frame": "frame.number",
    "time": "frame.time_relative",
    "src": "ip.src",
    "dst": "ip.dst",
    "sport": "tcp.srcport",
    "dport": "tcp.dstport",
    "seq": "tcp.seq",
    "seq_raw": "tcp.seq_raw",
    "ack": "tcp.ack",
    "len": "tcp.len",
    "flags": "tcp.flags.str",
    "win_field": "tcp.window_size_value",
    "win": "tcp.window_size",
    "mss": "tcp.options.mss_val",
    "wscale": "tcp.options.wscale.shift",
    "sack_perm": "tcp.options.sack_perm",
    "ts": "tcp.options.timestamp.tsval",
    "sack_le": "tcp.options.sack_le",
    "sack_re": "tcp.options.sack_re",
    "dup_ack": "tcp.analysis.duplicate_ack_num",
    "retrans": "tcp.analysis.retransmission",
    "fast": "tcp.analysis.fast_retransmission",
    "spurious": "tcp.analysis.spurious_retransmission",
    "out_of_order": "tcp.analysis.out_of_order",
    "zero_window": "tcp.analysis.zero_window",
    "probe": "tcp.analysis.keep_alive",
    "window_update": "tcp.analysis.window_update",
    "ack_rtt": "tcp.analysis.ack_rtt",
    "rto": "tcp.analysis.rto",
}
NUMBERS = {"frame", "sport", "dport", "seq", "seq_raw", "ack", "len", "win_field", "win", "mss",
           "wscale", "dup_ack"}
FLOATS = {"time", "ack_rtt", "rto"}
MARKS = {"sack_perm", "ts", "retrans", "fast", "spurious", "out_of_order", "zero_window", "probe",
         "window_update"}


class Segment:
    """Un segmento de la captura. Los campos de FIELDS se leen como atributos."""

    def __init__(self, values):
        for name, text in values.items():
            if name in MARKS:
                value = text != ""
            elif name in NUMBERS:
                value = int(text) if text else None
            elif name in FLOATS:
                value = float(text) if text else None
            else:
                value = text
            setattr(self, name, value)
        # tshark escribe las banderas con un carácter por bit: ··A·S· y similares.
        self.syn = "S" in self.flags
        self.has_ack = "A" in self.flags
        self.fin = "F" in self.flags
        self.rst = "R" in self.flags
        self.resent = self.retrans or self.fast or self.spurious


def read_capture(path):
    """Devuelve los segmentos TCP de la captura, en orden."""
    command = ["tshark", "-r", path, "-Y", "tcp", "-T", "fields", "-E", "separator=\t"]
    for field in FIELDS.values():
        command += ["-e", field]
    done = subprocess.run(command, capture_output=True, text=True, timeout=120)
    if done.returncode != 0:
        raise RuntimeError(done.stderr)
    segments = []
    for line in done.stdout.splitlines():
        segments.append(Segment(dict(zip(FIELDS, line.split("\t")))))
    return segments

"""Calcula EstimatedRTT, DevRTT y el RTO para una lista de muestras, como el RFC 6298.

    python3 rto_table.py 100 90 120 95             muestras en ms; la primera inicializa
    python3 rto_table.py --start 100 5 120         parte de EstimatedRTT = 100 y DevRTT = 5
    python3 rto_table.py --pcap captures/x-a.pcap  toma las muestras de una captura

Solo hace la cuenta. No tiene temporizador ni retroceso: no es un emisor.
"""
import sys

from pcaptools import read_capture

ALPHA = 1 / 8
BETA = 1 / 4


def first_sample(sample):
    """Regla 2.2 del RFC 6298: la primera muestra inicializa los dos valores."""
    return sample, sample / 2


def next_sample(estimated, deviation, sample):
    """Regla 2.3 del RFC 6298. DevRTT se calcula antes y usa el EstimatedRTT anterior."""
    deviation = (1 - BETA) * deviation + BETA * abs(estimated - sample)
    estimated = (1 - ALPHA) * estimated + ALPHA * sample
    return estimated, deviation


def rto(estimated, deviation):
    """El RTO sin el piso de 1 s, para poder compararlo con las muestras."""
    return estimated + 4 * deviation


def table(samples, start=None):
    """Devuelve filas (muestra, EstimatedRTT, DevRTT, RTO)."""
    rows = []
    state = start
    for sample in samples:
        state = next_sample(*state, sample) if state else first_sample(sample)
        rows.append((sample, state[0], state[1], rto(*state)))
    return rows


def samples_from_capture(path):
    """Las muestras de RTT que Wireshark deduce al emparejar cada dato con su ACK, en ms."""
    segments = read_capture(path)
    client = (segments[0].src, segments[0].sport)
    return [s.ack_rtt * 1000 for s in segments
            if s.ack_rtt is not None and (s.src, s.sport) != client and not s.syn]


def main():
    args = sys.argv[1:]
    start = None
    if args[:1] == ["--pcap"] and len(args) == 2:
        samples = samples_from_capture(args[1])
        samples = samples[:: max(1, len(samples) // 12)]
    elif args[:1] == ["--start"] and len(args) >= 4:
        start = (float(args[1]), float(args[2]))
        samples = [float(x) for x in args[3:]]
    elif args and not args[0].startswith("-"):
        samples = [float(x) for x in args]
    else:
        print(__doc__)
        return 2
    print(f"{'muestra':>9} {'EstimatedRTT':>13} {'DevRTT':>8} {'RTO':>8}")
    for row in table(samples, start):
        print(f"{row[0]:9.2f} {row[1]:13.2f} {row[2]:8.2f} {row[3]:8.2f}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Los experimentos de la semana: cada uno prepara el enlace, transfiere y captura.

    python3 scenarios.py             lista los escenarios
    python3 scenarios.py clean       corre uno; las capturas quedan en captures/

Cada escenario imprime los comandos que corre: léanlos, porque son lo que se hace a mano en
el laboratorio. Hace falta la topología levantada (make up).
"""
import os
import sys
import time

from netlab import CAPTURES, Lab, wait_until

LAB = Lab()
SERVER = "10.210.2.10"
ALIAS = "10.210.1.20"  # la segunda dirección de a: la de las conexiones a medio abrir
PORT = 5001
LINK = "delay 20ms rate 20mbit limit 200"  # 40 ms de RTT y 20 Mbit/s, de a hacia b
# Con una cola de 200, el arranque de una transferencia a veces la desborda. Una cola de 2000
# es mayor que todos los segmentos de 2 MB juntos: aquí nada se pierde si no lo pedimos.
DEEP_LINK = "delay 20ms rate 20mbit limit 2000"
CLASSIC = {"tcp_sack": 0, "tcp_recovery": 0, "tcp_early_retrans": 0}
MODERN = {"tcp_sack": 1, "tcp_recovery": 1, "tcp_early_retrans": 3}
VERBOSE = True


def say(node, command):
    if VERBOSE:
        print(f"  {node}# {command}")


def report(text):
    """Muestra un resultado. Las pruebas no lo imprimen: lo revisan."""
    if VERBOSE:
        print(text.rstrip())


def run(node, command):
    """Corre un comando visible en un nodo; falla si el comando falla."""
    say(node, command)
    done = LAB.sh(node, "sh", "-c", command)
    if done.returncode != 0:
        raise RuntimeError(f"{node}: {command}\n{done.stdout}{done.stderr}")
    return done.stdout + done.stderr


def quiet(node, command):
    """Limpieza: corre un comando sin mostrarlo y sin importar si falla."""
    return LAB.sh(node, "sh", "-c", command)


def set_link(toward_b=LINK, toward_a="delay 20ms"):
    # Se borra y se vuelve a crear: con replace, netem conserva el rate anterior si no se le da
    # uno nuevo, y con rate no desordena.
    quiet("m", "tc qdisc del dev eth2 root; tc qdisc del dev eth1 root")
    run("m", f"tc qdisc add dev eth2 root netem {toward_b}")
    run("m", f"tc qdisc add dev eth1 root netem {toward_a}")


def set_tcp(values):
    """Fija sysctls de TCP en a y en b. Cada nodo tiene su propio network namespace."""
    settings = " ".join(f"net.ipv4.{name}={value}" for name, value in values.items())
    for node in "ab":
        run(node, f"sysctl -qw {settings}")


def nft_drop(rule):
    run("m", f"nft 'add rule ip w10 forward {rule} drop'")


def reset():
    """Deja todo como recién levantado: enlace bueno, sin filtros, sin programas colgados."""
    for node in "ab":
        quiet(node, "pkill sink; pkill slow_reader; pkill half_open; pkill tcpdump; "
                    "pkill blast; true")
        quiet(node, "sysctl -qw net.ipv4.tcp_sack=1 net.ipv4.tcp_recovery=1 "
                    "net.ipv4.tcp_early_retrans=3 net.ipv4.tcp_syncookies=1 "
                    "net.ipv4.tcp_syn_linear_timeouts=4")
    quiet("a", "ip link set eth1 mtu 1500; ip route change default via 10.210.1.1; "
               "ip route flush cache")
    quiet("m", f"tc qdisc del dev eth2 root; tc qdisc del dev eth1 root; "
               f"tc qdisc add dev eth2 root netem {LINK}; "
               "tc qdisc add dev eth1 root netem delay 20ms; "
               "nft delete table ip w10; nft add table ip w10; "
               "nft add chain ip w10 forward '{ type filter hook forward priority 0; }'")
    time.sleep(0.2)


class Capture:
    """Captura en a y en b mientras dura el bloque with. Quedan name-a.pcap y name-b.pcap."""

    def __init__(self, name, ports=f"tcp port {PORT}", nodes="ab"):
        self.name = name
        self.ports = ports
        self.nodes = nodes

    def path(self, node):
        return os.path.join(CAPTURES, f"{self.name}-{node}.pcap")

    def __enter__(self):
        for node in self.nodes:
            if os.path.exists(self.path(node)):
                os.remove(self.path(node))
            # -B agranda el buffer del kernel: sin eso, a toda velocidad, tcpdump pierde segmentos.
            # --immediate-mode evita perder lo último al detenerlo con una señal.
            command = (f"tcpdump -i eth1 -n -s 128 -B 16384 --immediate-mode -Z root "
                       f"-w /captures/{self.name}-{node}.pcap {self.ports}")
            say(node, command + " &")
            LAB.daemon(node, f"{command} >/tmp/tcpdump.log 2>&1")
        ready = wait_until(lambda: all("listening" in quiet(n, "cat /tmp/tcpdump.log").stdout
                                       for n in self.nodes), seconds=15)
        if not ready:
            raise RuntimeError("tcpdump no arrancó")
        time.sleep(0.3)
        return self

    def __exit__(self, *_):
        time.sleep(1.0)  # lo último todavía viene en camino por el enlace con retardo
        for node in self.nodes:
            quiet(node, "pkill -INT tcpdump")
        for node in self.nodes:
            wait_until(lambda n=node: quiet(n, "pidof tcpdump").stdout.strip() == "", seconds=10)
        return False


def start_server(args, program="sink"):
    """Arranca el servidor en b. Su salida queda en /tmp/server.out."""
    say("b", f"/w10/{program} {args} &")
    quiet("b", "rm -f /tmp/server.out")
    LAB.daemon("b", f"/w10/{program} {args} >/tmp/server.out 2>&1")
    if not wait_until(lambda: f":{args.split()[0]} " in quiet("b", "ss -ltn").stdout, seconds=10):
        raise RuntimeError("el servidor no arrancó: " + quiet("b", "cat /tmp/server.out").stdout)


def server_output(seconds=60):
    """Espera a que termine el servidor y devuelve lo que escribió."""
    wait_until(lambda: quiet("b", "pidof sink slow_reader").stdout.strip() == "",
               seconds=seconds)
    return quiet("b", "cat /tmp/server.out").stdout


def send(args=f"{SERVER} {PORT} 2000000"):
    return run("a", f"/w10/blast {args}")


def transfer(name, size=2000000, server_args="", blast_args=""):
    """Una transferencia completa dentro de una captura. Devuelve la salida del servidor."""
    with Capture(name):
        start_server(f"{PORT} {server_args}".strip())
        send(f"{SERVER} {PORT} {size} {blast_args}".strip())
        result = server_output()
    return result


def scenario_clean():
    """Transferencia de 2 MB sin pérdidas, con una cola que no se puede desbordar."""
    reset()
    set_link(DEEP_LINK)
    return transfer("clean")


def scenario_mss():
    """Baja el MTU de a a 1400: el MSS que a anuncia cambia, y también el tamaño de los datos."""
    reset()
    run("a", "ip link set eth1 mtu 1400")
    try:
        return transfer("mss", size=100000)
    finally:
        run("a", "ip link set eth1 mtu 1500")


def scenario_drop():
    """El router descarta un solo segmento, el número 40. TCP moderno: con SACK y RACK."""
    reset()
    set_link(DEEP_LINK)
    nft_drop(f"tcp dport {PORT} numgen inc mod 100000 == 40 counter")
    before = LAB.counters("a")
    result = transfer("drop")
    return result, before, LAB.counters("a")


def scenario_classic():
    """El mismo descarte, con SACK y RACK apagados: TCP con los ACK duplicados de los libros."""
    reset()
    set_link(DEEP_LINK)
    set_tcp(CLASSIC)
    nft_drop(f"tcp dport {PORT} numgen inc mod 100000 == 40 counter")
    try:
        return transfer("classic")
    finally:
        set_tcp(MODERN)


def scenario_rto():
    """Un flujo lento y un corte total de 3 s: nadie manda ACK, y queda solo el temporizador."""
    reset()
    before = LAB.counters("a")
    samples = {}
    with Capture("rto"):
        start_server(str(PORT))
        LAB.daemon("a", f"/w10/blast {SERVER} {PORT} 60000 --chunk 1000 --gap-ms 100 --nodelay")
        say("a", f"/w10/blast {SERVER} {PORT} 60000 --chunk 1000 --gap-ms 100 --nodelay &")
        time.sleep(1.5)
        samples["before"] = run("a", f"ss -tin dst {SERVER}")
        report(samples["before"])
        set_link(f"{LINK} loss 100%", "delay 20ms loss 100%")
        time.sleep(2.2)
        samples["during"] = run("a", f"ss -tin dst {SERVER}")
        report(samples["during"])
        time.sleep(0.8)
        set_link()
        server_output()
    after = LAB.counters("a")
    show_counters((before, after))
    return before, after, samples


def scenario_tail():
    """Se pierden el último segmento y el FIN: no llega ningún ACK que avise. ¿Quién lo arregla?"""
    reset()
    # Hacia el servidor van: SYN (0), ACK (1), cinco de datos (2 a 6) y el FIN (7).
    nft_drop(f"tcp dport {PORT} numgen inc mod 100000 {{ 6, 7 }} counter")
    before = LAB.counters("a")
    result = transfer("tail", size=7000)
    return result, before, LAB.counters("a")


def scenario_loss():
    """2 % de pérdida aleatoria en el sentido de los datos."""
    reset()
    set_link(f"{LINK} loss 2%")
    before = LAB.counters("a")
    result = transfer("loss", size=3000000)
    return result, before, LAB.counters("a")


def scenario_reorder():
    """Sin pérdida, sin tope de velocidad y con 3 ms de jitter: los segmentos se desordenan."""
    reset()
    set_link("delay 20ms 3ms limit 2000")
    before = LAB.counters("a")
    result = transfer("reorder", size=3000000)
    return result, before, LAB.counters("a")


def scenario_reorder_classic():
    """El mismo enlace desordenado, con el TCP de los ACK duplicados."""
    reset()
    set_tcp(CLASSIC)
    set_link("delay 20ms 3ms limit 2000")
    before = LAB.counters("a")
    try:
        result = transfer("reorder-classic", size=3000000)
    finally:
        set_tcp(MODERN)
    return result, before, LAB.counters("a")


def scenario_zero():
    """El receptor acepta y se queda 3 s sin leer: su buffer se llena y anuncia ventana cero."""
    reset()
    return transfer("zero", size=300000, server_args="--rcvbuf 16384 --stall-ms 3000")


def scenario_slow():
    """Su slow_reader del laboratorio: 2 KB cada 100 ms, o sea unos 160 kbit/s."""
    reset()
    with Capture("slow"):
        start_server(f"{PORT} 8192 2048 100", program="slow_reader")
        send(f"{SERVER} {PORT} 100000")
        result = server_output()
    return result


def scenario_limit():
    """La misma transferencia con buffers de recepción distintos: ¿quién limita?"""
    reset()
    results = {}
    for rcvbuf in (8192, 32768, 262144):
        name = f"limit-{rcvbuf}"
        results[rcvbuf] = transfer(name, size=3000000, server_args=f"--rcvbuf {rcvbuf}")
        report(f"    rcvbuf {rcvbuf}: {results[rcvbuf].strip()}")
    return results


def scenario_ports():
    """Qué devuelve connect: puerto cerrado, filtrado en silencio, rechazado, y sin ruta."""
    reset()
    results = {}
    results["closed"] = run("a", f"/w10/connect_probe {SERVER} 5999 2")
    nft_drop("tcp dport 5999")
    with Capture("filtered", ports="tcp port 5999", nodes="a"):
        results["filtered"] = run("a", f"/w10/connect_probe {SERVER} 5999 2")
    # Linux reciente reintenta el SYN cada 1 s unas veces antes de empezar a duplicar.
    run("a", "sysctl -qw net.ipv4.tcp_syn_linear_timeouts=0")
    with Capture("filtered-exp", ports="tcp port 5999", nodes="a"):
        results["filtered_exp"] = run("a", f"/w10/connect_probe {SERVER} 5999 2")
    run("a", "sysctl -qw net.ipv4.tcp_syn_linear_timeouts=4")
    reset()
    run("m", "nft 'add rule ip w10 forward tcp dport 5999 reject with icmp type admin-prohibited'")
    results["rejected"] = run("a", f"/w10/connect_probe {SERVER} 5999 2")
    reset()
    results["no_route"] = run("a", "/w10/connect_probe 10.211.0.1 5999 2")
    start_server(str(PORT))
    results["open"] = run("a", f"/w10/connect_probe {SERVER} {PORT} 2")
    server_output()
    for name, text in results.items():
        report(f"    {name}: {text.strip()}")
    return results


def half_open_round(cookies):
    """Llena la cola de SYN de un servidor con backlog 16 y prueba a conectarse."""
    reset()
    run("b", f"sysctl -qw net.ipv4.tcp_syncookies={cookies}")
    # El servidor no acepta nunca: importa la cola del handshake, no lo que pasa después.
    start_server(f"{PORT} --stay --backlog 16")
    nft_drop(f"ip saddr {SERVER} ip daddr {ALIAS} tcp flags & (syn|ack) == syn|ack")
    before = LAB.counters("b")
    LAB.daemon("a", f"/w10/half_open {ALIAS} {SERVER} {PORT} 100 12")
    say("a", f"/w10/half_open {ALIAS} {SERVER} {PORT} 100 12 &")
    time.sleep(2)
    pending = len(run("b", "ss -tn state syn-recv").splitlines()) - 1
    probe = run("a", f"/w10/connect_probe {SERVER} {PORT} 2")
    after = LAB.counters("b")
    quiet("a", "pkill half_open")
    quiet("b", "pkill sink")
    return {"pending": pending, "probe": probe, "before": before, "after": after}


def scenario_syn():
    """Una cola de SYN llena, sin y con SYN cookies."""
    results = {"off": half_open_round(0), "on": half_open_round(1)}
    for name, result in results.items():
        grew = {key: result["after"].get(key, 0) - result["before"].get(key, 0)
                for key in ("TcpExtTCPReqQFullDrop", "TcpExtSyncookiesSent", "TcpExtSyncookiesRecv")}
        report(f"    cookies {name}: {result['pending']} en SYN-RECV, {result['probe'].strip()}, {grew}")
    reset()
    return results


def scenario_delack():
    """Datos chicos cada 100 ms a un receptor que no lee: ¿cuánto espera antes del ACK?"""
    reset()
    return transfer("delack", size=4000, server_args="--stall-ms 6000",
                    blast_args="--chunk 100 --gap-ms 100 --nodelay")


def scenario_halfclose():
    """El cliente cierra su sentido y después todavía recibe la respuesta del servidor."""
    reset()
    return transfer("halfclose", size=3000, server_args="--reply", blast_args="--reply")


def scenario_timewait():
    """Se pierde el último ACK del cierre: ese es el trabajo de quien espera en TIME_WAIT."""
    reset()
    # Hacia el servidor van: SYN (0), ACK (1), FIN (2) y el último ACK (3).
    nft_drop(f"tcp dport {PORT} numgen inc mod 100000 == 3 counter")
    result = transfer("timewait", size=0)
    states = run("a", f"ss -tano state time-wait dst {SERVER}")
    report(states)
    return result, states


def close_first_round(port, flags):
    """Un servidor que cierra primero; devuelve lo que dice uno nuevo en el mismo puerto."""
    start_server(f"{port} --close-first {flags}".strip())
    run("a", f"/w10/connect_probe {SERVER} {port} 2 300")
    server_output()
    time.sleep(0.2)
    states = run("b", f"ss -tano state time-wait sport = :{port}")
    report(states)
    say("b", f"timeout 1 /w10/sink {port} {flags}")
    again = quiet("b", f"timeout 1 /w10/sink {port} {flags}")
    return states, again.stdout + again.stderr


def scenario_bindfail():
    """El servidor cierra primero y queda en TIME_WAIT: ¿puede volver a abrir ese puerto?"""
    reset()
    # Puertos propios: el TIME_WAIT de una ronda no debe estorbar a los otros escenarios.
    without = close_first_round(5003, "--no-reuseaddr")
    report(f"    sin SO_REUSEADDR: {without[1].strip() or 'bind funciona'}")
    with_reuse = close_first_round(5004, "")
    report(f"    con SO_REUSEADDR: {with_reuse[1].strip() or 'bind funciona'}")
    return without, with_reuse


def scenario_abort():
    """El cliente cierra con RST en vez de FIN; el servidor se entera en su read."""
    reset()
    with Capture("abort"):
        start_server(str(PORT))
        send(f"{SERVER} {PORT} 3000 --abort")
        result = server_output()
    return result


def scenario_closed():
    """Conectarse a un puerto cerrado: el RST del kernel de b, capturado."""
    reset()
    with Capture("closed", ports="tcp port 5999"):
        result = run("a", f"/w10/connect_probe {SERVER} 5999 2")
    return result


def scenario_rtomin():
    """Fija rto_min en la ruta de a: se ve qué parte del RTO de Linux es ese piso."""
    reset()
    rows = {}
    for floor in (None, "500ms"):
        if floor:
            run("a", f"ip route change default via 10.210.1.1 rto_min {floor}")
        start_server(str(PORT))
        LAB.daemon("a", f"/w10/blast {SERVER} {PORT} 20000 --chunk 1000 --gap-ms 100 --nodelay")
        say("a", f"/w10/blast {SERVER} {PORT} 20000 --chunk 1000 --gap-ms 100 --nodelay &")
        time.sleep(1.2)
        rows[floor or "default"] = run("a", f"ss -tin dst {SERVER}")
        server_output()
    run("a", "ip route change default via 10.210.1.1")
    for floor, text in rows.items():
        report(f"    {floor}: {' '.join(w for w in text.split() if w.startswith(('rto:', 'rtt:')))}")
    return rows


def scenario_ss():
    """Una conexión activa vista con ss cada segundo: rtt, rto, cwnd."""
    reset()
    start_server(str(PORT))
    LAB.daemon("a", f"/w10/blast {SERVER} {PORT} 8000000")
    say("a", f"/w10/blast {SERVER} {PORT} 8000000 &")
    rows = []
    for _ in range(3):
        time.sleep(0.8)
        rows.append(run("a", f"ss -tin dst {SERVER}"))
        report(rows[-1])
    server_output()
    return rows


def show_counters(result):
    """Muestra cuánto cambiaron los contadores del kernel entre (antes, después)."""
    names = ("TcpRetransSegs", "TcpExtTCPTimeouts", "TcpExtTCPLossProbes", "TcpExtTCPSackRecovery",
             "TcpExtTCPRenoRecovery")
    before, after = result
    report("  ".join(f"{name}={after.get(name, 0) - before.get(name, 0)}" for name in names))


def scenarios():
    module = sys.modules[__name__]
    return {name[len("scenario_"):]: getattr(module, name)
            for name in dir(module) if name.startswith("scenario_")}


def main():
    table = scenarios()
    if len(sys.argv) != 2 or sys.argv[1] not in table:
        width = max(len(name) for name in table)
        for name, function in table.items():
            print(f"{name:<{width}}  {function.__doc__}")
        return 2
    result = table[sys.argv[1]]()
    if isinstance(result, str):
        print(result, end="")
    elif isinstance(result, tuple) and len(result) == 3 and isinstance(result[0], str):
        print(result[0], end="")
        show_counters(result[1:])
    return 0


if __name__ == "__main__":
    sys.exit(main())

// sink: acepta una conexión TCP, lee todo lo que llegue y cuenta los bytes.
//
//   sink <puerto> [--rcvbuf B] [--stall-ms N] [--reply] [--close-first] [--backlog N]
//                 [--no-reuseaddr] [--stay]
//
// --rcvbuf      fija SO_RCVBUF antes de listen, para que la conexión lo herede
// --stall-ms    acepta y se queda N ms sin leer: el buffer de recepción se llena
// --reply       cuando el cliente cierra su sentido, le contesta cuántos bytes recibió
// --close-first apenas acepta, cierra él primero (se queda con TIME_WAIT)
// --backlog     segundo argumento de listen
// --stay        no acepta nunca: sirve para ver la cola de conexiones
#define _POSIX_C_SOURCE 200809L
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <time.h>
#include <unistd.h>

static double now_seconds(void) {
    struct timespec ts;
    if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0) {
        perror("clock_gettime");
        exit(1);
    }
    return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9;
}

static void pause_ms(long ms) {
    struct timespec ts = {ms / 1000, (ms % 1000) * 1000000L};
    while (nanosleep(&ts, &ts) < 0) {
        if (errno != EINTR) {
            perror("nanosleep");
            exit(1);
        }
    }
}

int main(int argc, char **argv) {
    if (argc < 2) {
        fprintf(stderr, "uso: %s <puerto> [opciones]\n", argv[0]);
        return 2;
    }
    int rcvbuf = 0, backlog = 16, reply = 0, close_first = 0, reuse = 1, stay = 0;
    long stall_ms = 0;
    for (int i = 2; i < argc; i++) {
        if (strcmp(argv[i], "--rcvbuf") == 0 && i + 1 < argc) {
            rcvbuf = atoi(argv[++i]);
        } else if (strcmp(argv[i], "--stall-ms") == 0 && i + 1 < argc) {
            stall_ms = atol(argv[++i]);
        } else if (strcmp(argv[i], "--backlog") == 0 && i + 1 < argc) {
            backlog = atoi(argv[++i]);
        } else if (strcmp(argv[i], "--reply") == 0) {
            reply = 1;
        } else if (strcmp(argv[i], "--close-first") == 0) {
            close_first = 1;
        } else if (strcmp(argv[i], "--no-reuseaddr") == 0) {
            reuse = 0;
        } else if (strcmp(argv[i], "--stay") == 0) {
            stay = 1;
        } else {
            fprintf(stderr, "opción desconocida: %s\n", argv[i]);
            return 2;
        }
    }

    int listener = socket(AF_INET, SOCK_STREAM, 0);
    if (listener < 0) {
        perror("socket");
        return 1;
    }
    if (reuse && setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof reuse) < 0) {
        perror("setsockopt SO_REUSEADDR");
        return 1;
    }
    if (rcvbuf > 0 && setsockopt(listener, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof rcvbuf) < 0) {
        perror("setsockopt SO_RCVBUF");
        return 1;
    }
    struct sockaddr_in addr = {.sin_family = AF_INET,
                               .sin_port = htons((uint16_t)atoi(argv[1])),
                               .sin_addr.s_addr = htonl(INADDR_ANY)};
    if (bind(listener, (struct sockaddr *)&addr, sizeof addr) < 0) {
        perror("bind");
        return 1;
    }
    if (listen(listener, backlog) < 0) {
        perror("listen");
        return 1;
    }
    if (stay) {
        for (;;) {
            pause_ms(1000);
        }
    }

    int fd = accept(listener, NULL, NULL);
    if (fd < 0) {
        perror("accept");
        return 1;
    }
    double start = now_seconds();
    if (stall_ms > 0) {
        pause_ms(stall_ms);
    }
    if (close_first) {
        if (close(fd) < 0) {
            perror("close");
            return 1;
        }
        printf("cerré primero\n");
        return 0;
    }

    static char buf[65536];
    long long total = 0;
    ssize_t got;
    while ((got = read(fd, buf, sizeof buf)) != 0) {
        if (got < 0) {
            if (errno == EINTR) {
                continue;
            }
            perror("read");
            return 1;
        }
        total += got;
    }
    if (reply) {
        char text[64];
        int len = snprintf(text, sizeof text, "%lld bytes\n", total);
        if (write(fd, text, (size_t)len) < 0) {
            perror("write");
            return 1;
        }
    }
    double seconds = now_seconds() - start;
    if (close(fd) < 0) {
        perror("close");
        return 1;
    }
    // Se mide aquí y no en el emisor: write termina cuando el kernel acepta los bytes en su
    // buffer, no cuando llegan.
    printf("recibidos %lld bytes en %.2f s (%.1f Mbit/s)\n", total, seconds,
           seconds > 0 ? (double)total * 8 / seconds / 1e6 : 0.0);
    return 0;
}