receiver.c

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

/* global target queue (defined + initialized in main.c) */
tqueue_t g_target_queue;

/* ══════════════════════════════════════════════════════════════
 * RECEIVER THREAD
 *
 * Opens AF_PACKET SOCK_RAW socket, captures all incoming TCP
 * packets, filters SYN+ACK responses, validates the stateless
 * cookie (seq derived from src_ip ^ dst_ip ^ dst_port in sender),
 * and pushes valid targets into g_target_queue.
 * ══════════════════════════════════════════════════════════════ */
void *receiver_thread(void *arg) {
    thread_context_t *ctx = (thread_context_t *)arg;

    /* raw socket to capture incoming packets */
    int rfd = socket(AF_PACKET, SOCK_RAW, htons(ETH_P_IP));
    if (rfd < 0) {
        perror("[receiver] socket");
        return NULL;
    }

    /* bind to interface */
    struct sockaddr_ll sll;
    memset(&sll, 0, sizeof(sll));
    sll.sll_family   = AF_PACKET;
    sll.sll_protocol = htons(ETH_P_IP);
    sll.sll_ifindex  = ctx->config->ifindex;
    if (bind(rfd, (struct sockaddr *)&sll, sizeof(sll)) < 0) {
        perror("[receiver] bind");
        close(rfd);
        return NULL;
    }

    /* set non-blocking */
    int flags = fcntl(rfd, F_GETFL, 0);
    fcntl(rfd, F_SETFL, flags | O_NONBLOCK);

    uint8_t  buf[2048];
    uint32_t src_ip = ctx->src_ip;  /* host byte order */

    while (ctx->running && !g_stop) {
        ssize_t n = recv(rfd, buf, sizeof(buf), 0);
        if (n <= 0) {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                usleep(100);
                continue;
            }
            break;
        }

        /* minimum: ethernet + IP + TCP */
        if ((size_t)n < sizeof(struct ethhdr) +
                         sizeof(struct iphdr)  +
                         sizeof(struct tcphdr))
            continue;

        struct iphdr  *iph  = (struct iphdr  *)(buf + sizeof(struct ethhdr));
        struct tcphdr *tcph = (struct tcphdr *)((uint8_t *)iph + iph->ihl * 4);

        /* only TCP */
        if (iph->protocol != IPPROTO_TCP) continue;

        /* must be addressed to us */
        if (ntohl(iph->daddr) != src_ip) continue;

        /* must be SYN+ACK, no RST/FIN */
        if (!tcph->syn || !tcph->ack) continue;
        if (tcph->rst  || tcph->fin)  continue;

        uint32_t dst_ip_hbo  = ntohl(iph->saddr);
        uint16_t dst_port    = ntohs(tcph->source);

        /* validate stateless cookie:
         * sender set seq = src_ip ^ dst_ip ^ dst_port
         * receiver should see ack_seq = seq + 1  */
        uint32_t expected_seq = src_ip ^ dst_ip_hbo ^ (uint32_t)dst_port;
        uint32_t ack_seq      = ntohl(tcph->ack_seq);
        if (ack_seq - 1 != expected_seq) continue;

        /* skip honeypot / blacklisted IPs */
        if (honeypot_cidr_check(dst_ip_hbo)) continue;
        if (honeypot_check(dst_ip_hbo))      continue;
        if (is_blacklisted(dst_ip_hbo))      continue;

        /* de-duplicate using seen bitmap */
        if (g_seen_bitmap) {
            if (test_bitmap(g_seen_bitmap, dst_ip_hbo)) continue;
            set_bitmap(g_seen_bitmap, dst_ip_hbo);
        }

        atomic_fetch_add(&ctx->stats->packets_recv, 1);
        atomic_fetch_add(&ctx->stats->ports_open, 1);

        /* enqueue for exploit workers */
        target_t t;
        t.ip   = iph->saddr;   /* network byte order */
        t.port = dst_port;

        /* spin-push: drop if queue full rather than block the receiver */
        int retries = 0;
        while (tqueue_push(&g_target_queue, t) < 0 && retries < 100) {
            usleep(500);
            retries++;
        }
        atomic_fetch_add(&ctx->stats->enqueued, 1);
    }

    close(rfd);
    return NULL;
}

#ifdef USE_PFRING_ZC
void *pfring_zc_receiver_thread(void *arg) {
    (void)arg;
    fprintf(stderr, "[pfring] ZC receiver thread started (stub)\n");
    return NULL;
}
#endif