sender.c

#include "sender.h"
#include <linux/if_packet.h>

/* ── work distribution ───────────────────────────────────────── */
void ip_per_thread(ip_range_t *ip_ranges, int num_ip_ranges,
                   port_range_t *port_ranges, int num_port_ranges,
                   thread_context_t *contexts, int num_threads,
                   uint64_t total_packets) {
    for (int t = 0; t < num_threads; t++) {
        uint64_t per  = total_packets / (uint64_t)num_threads;
        uint64_t st   = (uint64_t)t * per;
        uint64_t en   = (t == num_threads - 1)
                        ? total_packets : st + per;

        contexts[t].work.global_start_idx   = st;
        contexts[t].work.global_end_idx     = en;
        contexts[t].work.current_global_idx = st;
        contexts[t].work.total_packets      = total_packets;
        contexts[t].work.all_ip_ranges      = ip_ranges;
        contexts[t].work.total_ip_ranges    = num_ip_ranges;
        contexts[t].work.port_ranges        = port_ranges;
        contexts[t].work.num_port_ranges    = num_port_ranges;
        contexts[t].work.total_ips =
            calculate_total_ips(ip_ranges, num_ip_ranges);
    }
}

/* ── rate limiter (batch-aware, hybrid spin/sleep) ───────────── */
void rate_limit_batch(thread_context_t *ctx, int batch_size) {
    if (ctx->config->rate_limit == 0) return;

    /* check every 4 calls for finer-grained rate limiting */
    static __thread int calls = 0;
    if (++calls < 4) return;
    calls = 0;

    struct timeval now;
    gettimeofday(&now, NULL);

    double elapsed = (now.tv_sec  - ctx->last_send_time.tv_sec) +
                     (now.tv_usec - ctx->last_send_time.tv_usec) / 1e6;
    double target  = (double)(batch_size * 4) /
                     (double)ctx->config->rate_limit;

    if (elapsed < target) {
        double us = (target - elapsed) * 1e6;
        if (us > 1500.0)
            usleep((useconds_t)(us - 200));
        else
            while (elapsed < target) {
                gettimeofday(&now, NULL);
                elapsed = (now.tv_sec  - ctx->last_send_time.tv_sec) +
                          (now.tv_usec - ctx->last_send_time.tv_usec) / 1e6;
            }
    }
    gettimeofday(&ctx->last_send_time, NULL);
}

/* ══════════════════════════════════════════════════════════════
 * SENDER THREAD — PF_PACKET + TPACKET_V2 TX ring
 * ══════════════════════════════════════════════════════════════ */
void *sender_thread(void *arg) {
    thread_context_t *ctx = (thread_context_t *)arg;

    /* pin to CPU core */
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(ctx->thread_id % (int)sysconf(_SC_NPROCESSORS_ONLN), &cpuset);
    pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);

    uint32_t xor_state = ctx->current_state;

    /* ── set TPACKET_V2 ─────────────────────────────────────── */
    int version = TPACKET_V2;
    setsockopt(ctx->socket_fd, SOL_PACKET, PACKET_VERSION,
               &version, sizeof(version));

    /* ── QDISC bypass (kernel 3.14+) ────────────────────────── */
    /* OFF by default: bypassing qdisc removes the kernel's TX backpressure
     * and can saturate the NIC driver queue, making SSH unreachable.
     * Enable explicitly with --qdisc-bypass only if you know your NIC
     * can handle the load and you have out-of-band access.             */
    if (ctx->config->qdisc_bypass) {
        int qbypass = 1;
        if (setsockopt(ctx->socket_fd, SOL_PACKET, PACKET_QDISC_BYPASS,
                       &qbypass, sizeof(qbypass)) < 0 && errno != ENOPROTOOPT)
            perror("PACKET_QDISC_BYPASS");
    }

    /* ── TX socket send buffer — cap at 4 MiB to avoid driver starvation */
    {
        int sndbuf = 4 * 1024 * 1024;
        setsockopt(ctx->socket_fd, SOL_SOCKET, SO_SNDBUF,
                   &sndbuf, sizeof(sndbuf));
    }

    /* ── TX ring ─────────────────────────────────────────────── */
    struct tpacket_req req;
    memset(&req, 0, sizeof(req));
    req.tp_block_size = TPACKET_BLOCK_SIZE;
    req.tp_block_nr   = TPACKET_BLOCK_NR;
    req.tp_frame_size = TPACKET_FRAME_SIZE;
    req.tp_frame_nr   = (req.tp_block_size * req.tp_block_nr) /
                         req.tp_frame_size;

    if (setsockopt(ctx->socket_fd, SOL_PACKET, PACKET_TX_RING,
                   &req, sizeof(req)) < 0) {
        perror("PACKET_TX_RING");
        return NULL;
    }

    size_t ring_sz = (size_t)req.tp_block_size * req.tp_block_nr;
    unsigned char *ring = mmap(NULL, ring_sz,
                               PROT_READ | PROT_WRITE,
                               MAP_SHARED, ctx->socket_fd, 0);
    if (ring == MAP_FAILED) { perror("mmap TX ring"); return NULL; }

    /* ── pre-build base packet ──────────────────────────────── */
    packet_t base_pkt;
    if (ctx->config->scan_method == SCAN_METHOD_UDP)
        create_udp_packet(&base_pkt, ctx->src_ip, 0,
                          ctx->src_port,
                          ctx->work.port_ranges[0].start,
                          ctx->config->src_mac, ctx->config->dst_mac,
                          ctx->config->probe_payload,
                          ctx->config->probe_payload_len);
    else
        create_syn_packet(&base_pkt, ctx->src_ip, 0,
                          ctx->src_port,
                          ctx->work.port_ranges[0].start,
                          ctx->config->src_mac, ctx->config->dst_mac);

    /* pre-fill all ring frames with base packet */
    uint32_t tp_mac = TPACKET_ALIGN(sizeof(struct tpacket2_hdr));
    for (unsigned int i = 0; i < req.tp_frame_nr; i++) {
        struct tpacket2_hdr *th =
            (struct tpacket2_hdr *)(ring + i * req.tp_frame_size);
        th->tp_mac = tp_mac;
        th->tp_net = tp_mac + (uint32_t)sizeof(struct ethhdr);
        memcpy((uint8_t *)th + tp_mac,
               base_pkt.buffer, (size_t)base_pkt.length);
    }

    unsigned int frame_idx = 0;
    gettimeofday(&ctx->last_send_time, NULL);
    ctx->packets_sent = 0;

    /* ── main send loop ──────────────────────────────────────── */
    while (ctx->running && !g_stop &&
           ctx->work.current_global_idx < ctx->work.global_end_idx) {

        int batch_count = 0;
        int ring_batch  = 0;

        while (ring_batch < BATCH_SIZE &&
               ctx->work.current_global_idx < ctx->work.global_end_idx &&
               !g_stop) {

            struct tpacket2_hdr *th =
                (struct tpacket2_hdr *)(ring + frame_idx * req.tp_frame_size);

            if (th->tp_status != TP_STATUS_AVAILABLE) {
                if (batch_count > 0) break;
                send(ctx->socket_fd, NULL, 0, MSG_DONTWAIT);
                continue;
            }

            /* BlackRock-shuffled index */
            uint64_t idx = blackrock_shuffle(
                &ctx->config->blackrock,
                ctx->work.current_global_idx);

            ctx->work.current_global_idx++;

            uint64_t ip_idx       = idx % ctx->work.total_ips;
            uint64_t port_tot_idx = idx / ctx->work.total_ips;

            uint32_t ip_hbo = get_ip_from_index(
                ip_idx,
                ctx->work.all_ip_ranges,
                ctx->work.total_ip_ranges);

            /* skip blacklisted / honeypot */
            if (is_blacklisted(ip_hbo))           continue;
            if (honeypot_cidr_check(ip_hbo))      continue;
            if (honeypot_check(ip_hbo))           continue;
            if (!is_whitelisted(ip_hbo))          continue;

            /* select port */
            uint16_t port = 0;
            uint64_t p_acc = 0;
            for (int p = 0; p < ctx->work.num_port_ranges; p++) {
                uint64_t pc = (uint64_t)(ctx->work.port_ranges[p].end
                            - ctx->work.port_ranges[p].start) + 1;
                if (port_tot_idx < p_acc + pc) {
                    port = (uint16_t)(ctx->work.port_ranges[p].start
                           + (port_tot_idx - p_acc));
                    break;
                }
                p_acc += pc;
            }
            if (port == 0) continue;

            uint32_t ip_nbo = htonl(ip_hbo);

            /* patch the ring frame */
            th->tp_len = (unsigned int)base_pkt.length;
            uint8_t *pkt_ptr = (uint8_t *)th + th->tp_mac;

            struct iphdr *iph =
                (struct iphdr *)(pkt_ptr + sizeof(struct ethhdr));
            iph->daddr = ip_nbo;
            iph->id    = (uint16_t)xorshift32(&xor_state);
            iph->check = 0;
            iph->check = checksum_ip(iph);

            if (ctx->config->scan_method == SCAN_METHOD_UDP) {
                struct udphdr *udph =
                    (struct udphdr *)((uint8_t *)iph + sizeof(struct iphdr));
                udph->dest  = htons(port);
                udph->check = 0;
            } else {
                struct tcphdr *tcph =
                    (struct tcphdr *)((uint8_t *)iph + sizeof(struct iphdr));
                tcph->dest  = htons(port);
                /* stateless cookie: seq = f(src_ip, dst_ip, dst_port) */
                tcph->seq   = htonl(ctx->src_ip ^ ip_hbo ^ (uint32_t)port);
                tcph->check = 0;
                tcph->check = checksum_tcp(tcph, ctx->src_ip, ip_hbo);
            }

            th->tp_status = TP_STATUS_SEND_REQUEST;
            frame_idx     = (frame_idx + 1) % req.tp_frame_nr;
            ring_batch++;
            batch_count++;
        }

        if (batch_count > 0) {
            int r = (int)send(ctx->socket_fd, NULL, 0, MSG_DONTWAIT);
            if (r >= 0 || errno == ENOBUFS || errno == EAGAIN) {
                atomic_fetch_add(&ctx->stats->packets_sent,
                                 (unsigned long long)batch_count);
                ctx->packets_sent += batch_count;
            }
        }

        rate_limit_batch(ctx, batch_count);
    }

    munmap(ring, ring_sz);
    return NULL;
}

/* ════════════════════════════════════════════════════════════════
 * PFRING ZC sender (optional)
 * ════════════════════════════════════════════════════════════════ */
#ifdef USE_PFRING_ZC
void *pfring_zc_sender_thread(void *arg) {
    /* Placeholder: pfring ZC API wraps around the same logic above.
     * Requires pfring.h and libpfring at link time.
     * Compile with: make pfring=1
     */
    (void)arg;
    fprintf(stderr, "[pfring] ZC sender thread started (stub)\n");
    return NULL;
}
#endif