path: root/src/source-napatech.c

/* Copyright (C) 2012-2020 Open Information Security Foundation
 *
 * You can copy, redistribute or modify this Program under the terms of
 * the GNU General Public License version 2 as published by the Free
 * Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */

/**
 * \file
 *
 - * \author nPulse Technologies, LLC.
 - * \author Matt Keeler <mk@npulsetech.com>
 *  *
 * Support for NAPATECH adapter with the 3GD Driver/API.
 * Requires libntapi from Napatech A/S.
 *
 */
#include "suricata-common.h"
#include "action-globals.h"
#include "decode.h"
#include "packet.h"
#include "suricata.h"
#include "threadvars.h"
#include "util-datalink.h"
#include "util-optimize.h"
#include "tm-queuehandlers.h"
#include "tm-threads.h"
#include "tm-modules.h"
#include "util-privs.h"
#include "tmqh-packetpool.h"
#include "util-napatech.h"
#include "source-napatech.h"
#include "runmode-napatech.h"

#ifndef HAVE_NAPATECH

TmEcode NoNapatechSupportExit(ThreadVars*, const void*, void**);

void TmModuleNapatechStreamRegister(void)
{
    tmm_modules[TMM_RECEIVENAPATECH].name = "NapatechStream";
    tmm_modules[TMM_RECEIVENAPATECH].ThreadInit = NoNapatechSupportExit;
    tmm_modules[TMM_RECEIVENAPATECH].Func = NULL;
    tmm_modules[TMM_RECEIVENAPATECH].ThreadExitPrintStats = NULL;
    tmm_modules[TMM_RECEIVENAPATECH].ThreadDeinit = NULL;
    tmm_modules[TMM_RECEIVENAPATECH].cap_flags = SC_CAP_NET_ADMIN;
}

void TmModuleNapatechDecodeRegister(void)
{
    tmm_modules[TMM_DECODENAPATECH].name = "NapatechDecode";
    tmm_modules[TMM_DECODENAPATECH].ThreadInit = NoNapatechSupportExit;
    tmm_modules[TMM_DECODENAPATECH].Func = NULL;
    tmm_modules[TMM_DECODENAPATECH].ThreadExitPrintStats = NULL;
    tmm_modules[TMM_DECODENAPATECH].ThreadDeinit = NULL;
    tmm_modules[TMM_DECODENAPATECH].cap_flags = 0;
    tmm_modules[TMM_DECODENAPATECH].flags = TM_FLAG_DECODE_TM;
}

TmEcode NoNapatechSupportExit(ThreadVars *tv, const void *initdata, void **data)
{
    SCLogError("Error creating thread %s: you do not have support for Napatech adapter "
               "enabled please recompile with --enable-napatech",
            tv->name);
    exit(EXIT_FAILURE);
}

#else /* Implied we do have NAPATECH support */


#include <numa.h>
#include <nt.h>

extern uint16_t max_pending_packets;

typedef struct NapatechThreadVars_
{
    ThreadVars *tv;
    NtNetStreamRx_t rx_stream;
    uint16_t stream_id;
    int hba;
    TmSlot *slot;
} NapatechThreadVars;

#ifdef NAPATECH_ENABLE_BYPASS
static int NapatechBypassCallback(Packet *p);
#endif

TmEcode NapatechStreamThreadInit(ThreadVars *, const void *, void **);
void NapatechStreamThreadExitStats(ThreadVars *, void *);
TmEcode NapatechPacketLoop(ThreadVars *tv, void *data, void *slot);

TmEcode NapatechDecodeThreadInit(ThreadVars *, const void *, void **);
TmEcode NapatechDecodeThreadDeinit(ThreadVars *tv, void *data);
TmEcode NapatechDecode(ThreadVars *, Packet *, void *);

/* These are used as the threads are exiting to get a comprehensive count of
 * all the packets received and dropped.
 */
SC_ATOMIC_DECLARE(uint64_t, total_packets);
SC_ATOMIC_DECLARE(uint64_t, total_drops);
SC_ATOMIC_DECLARE(uint16_t, total_tallied);

/* Streams are counted as they are instantiated in order to know when all threads
 * are running*/
SC_ATOMIC_DECLARE(uint16_t, stream_count);

typedef struct NapatechNumaDetect_ {
    SC_ATOMIC_DECLARE(uint16_t, count);
} NapatechNumaDetect;

NapatechNumaDetect *numa_detect = NULL;

SC_ATOMIC_DECLARE(uint64_t, flow_callback_cnt);
SC_ATOMIC_DECLARE(uint64_t, flow_callback_handled_pkts);
SC_ATOMIC_DECLARE(uint64_t, flow_callback_udp_pkts);
SC_ATOMIC_DECLARE(uint64_t, flow_callback_tcp_pkts);
SC_ATOMIC_DECLARE(uint64_t, flow_callback_unhandled_pkts);

/**
 * \brief Initialize the Napatech receiver (reader) module for globals.
 */
static TmEcode NapatechStreamInit(void)
{
    int i;

    SC_ATOMIC_INIT(total_packets);
    SC_ATOMIC_INIT(total_drops);
    SC_ATOMIC_INIT(total_tallied);
    SC_ATOMIC_INIT(stream_count);

    numa_detect = SCMalloc(sizeof(*numa_detect) * (numa_max_node() + 1));
    if (numa_detect == NULL) {
        FatalError("Failed to allocate memory for numa detection array: %s", strerror(errno));
    }

    for (i = 0; i <= numa_max_node(); ++i) {
        SC_ATOMIC_INIT(numa_detect[i].count);
    }

    SC_ATOMIC_INIT(flow_callback_cnt);
    SC_ATOMIC_INIT(flow_callback_handled_pkts);
    SC_ATOMIC_INIT(flow_callback_udp_pkts);
    SC_ATOMIC_INIT(flow_callback_tcp_pkts);
    SC_ATOMIC_INIT(flow_callback_unhandled_pkts);

    return TM_ECODE_OK;
}

/**
 * \brief Deinitialize the Napatech receiver (reader) module for globals.
 */
static TmEcode NapatechStreamDeInit(void)
{
    if (numa_detect != NULL) {
        SCFree(numa_detect);
    }

    return TM_ECODE_OK;
}

/**
 * \brief Register the Napatech  receiver (reader) module.
 */
void TmModuleNapatechStreamRegister(void)
{
    tmm_modules[TMM_RECEIVENAPATECH].name = "NapatechStream";
    tmm_modules[TMM_RECEIVENAPATECH].ThreadInit = NapatechStreamThreadInit;
    tmm_modules[TMM_RECEIVENAPATECH].Func = NULL;
    tmm_modules[TMM_RECEIVENAPATECH].PktAcqLoop = NapatechPacketLoop;
    tmm_modules[TMM_RECEIVENAPATECH].PktAcqBreakLoop = NULL;
    tmm_modules[TMM_RECEIVENAPATECH].ThreadExitPrintStats = NapatechStreamThreadExitStats;
    tmm_modules[TMM_RECEIVENAPATECH].ThreadDeinit = NapatechStreamThreadDeinit;
    tmm_modules[TMM_RECEIVENAPATECH].cap_flags = SC_CAP_NET_RAW;
    tmm_modules[TMM_RECEIVENAPATECH].flags = TM_FLAG_RECEIVE_TM;
    tmm_modules[TMM_RECEIVENAPATECH].Init = NapatechStreamInit;
    tmm_modules[TMM_RECEIVENAPATECH].DeInit = NapatechStreamDeInit;
}

/**
 * \brief Register the Napatech decoder module.
 */
void TmModuleNapatechDecodeRegister(void)
{
    tmm_modules[TMM_DECODENAPATECH].name = "NapatechDecode";
    tmm_modules[TMM_DECODENAPATECH].ThreadInit = NapatechDecodeThreadInit;
    tmm_modules[TMM_DECODENAPATECH].Func = NapatechDecode;
    tmm_modules[TMM_DECODENAPATECH].ThreadExitPrintStats = NULL;
    tmm_modules[TMM_DECODENAPATECH].ThreadDeinit = NapatechDecodeThreadDeinit;
    tmm_modules[TMM_DECODENAPATECH].cap_flags = 0;
    tmm_modules[TMM_DECODENAPATECH].flags = TM_FLAG_DECODE_TM;
}

#ifdef NAPATECH_ENABLE_BYPASS
/**
 * \brief template of IPv4 header
 */
struct ipv4_hdr
{
    uint8_t version_ihl; /**< version and header length */
    uint8_t type_of_service; /**< type of service */
    uint16_t total_length; /**< length of packet */
    uint16_t packet_id; /**< packet ID */
    uint16_t fragment_offset; /**< fragmentation offset */
    uint8_t time_to_live; /**< time to live */
    uint8_t next_proto_id; /**< protocol ID */
    uint16_t hdr_checksum; /**< header checksum */
    uint32_t src_addr; /**< source address */
    uint32_t dst_addr; /**< destination address */
} __attribute__ ((__packed__));

/**
 * \brief template of IPv6 header
 */
struct ipv6_hdr
{
    uint32_t vtc_flow; /**< IP version, traffic class & flow label. */
    uint16_t payload_len; /**< IP packet length - includes sizeof(ip_header). */
    uint8_t proto; /**< Protocol, next header. */
    uint8_t hop_limits; /**< Hop limits. */
    uint8_t src_addr[16]; /**< IP address of source host. */
    uint8_t dst_addr[16]; /**< IP address of destination host(s). */
} __attribute__ ((__packed__));

/**
 * \brief template of UDP header
 */
struct udp_hdr
{
    uint16_t src_port; /**< UDP source port. */
    uint16_t dst_port; /**< UDP destination port. */
    uint16_t dgram_len; /**< UDP datagram length */
    uint16_t dgram_cksum; /**< UDP datagram checksum */
} __attribute__ ((__packed__));

/**
 * \brief template of TCP header
 */
struct tcp_hdr
{
    uint16_t src_port; /**< TCP source port. */
    uint16_t dst_port; /**< TCP destination port. */
    uint32_t sent_seq; /**< TX data sequence number. */
    uint32_t recv_ack; /**< RX data acknowledgement sequence number. */
    uint8_t data_off; /**< Data offset. */
    uint8_t tcp_flags; /**< TCP flags */
    uint16_t rx_win; /**< RX flow control window. */
    uint16_t cksum; /**< TCP checksum. */
    uint16_t tcp_urp; /**< TCP urgent pointer, if any. */
} __attribute__ ((__packed__));


/*  The hardware will assign a "color" value indicating what filters are matched
 * by a given packet.  These constants indicate what bits are set in the color
 * field for different protocols
 *
 */
#define RTE_PTYPE_L2_ETHER                  0x10000000
#define RTE_PTYPE_L3_IPV4                   0x01000000
#define RTE_PTYPE_L3_IPV6                   0x04000000
#define RTE_PTYPE_L4_TCP                    0x00100000
#define RTE_PTYPE_L4_UDP                    0x00200000

/* These masks are used to extract layer 3 and layer 4 protocol
 * values from the color field in the packet descriptor.
 */
#define RTE_PTYPE_L3_MASK                   0x0f000000
#define RTE_PTYPE_L4_MASK                   0x00f00000

#define COLOR_IS_SPAN                       0x00001000

static int is_inline = 0;
static int inline_port_map[MAX_PORTS] = { -1 };

/**
 * \brief Binds two ports together for inline operation.
 *
 * Get the ID of an adapter on which a given port resides.
 *
 * \param port one of the ports in a pairing.
 * \param peer the other port in a pairing.
 * \return ID of the adapter.
 *
 */
int NapatechSetPortmap(int port, int peer)
{
    if ((inline_port_map[port] == -1) && (inline_port_map[peer] == -1)) {
        inline_port_map[port] = peer;
        inline_port_map[peer] = port;
    } else {
        SCLogError("Port pairing is already configured.");
        return 0;
    }
    return 1;
}

/**
 * \brief Returns the ID of the adapter
 *
 * Get the ID of an adapter on which a given port resides.
 *
 * \param port for which adapter ID is requested.
 * \return ID of the adapter.
 *
 */
int NapatechGetAdapter(uint8_t port)
{
    static int port_adapter_map[MAX_PORTS] = { -1 };
    int status;
    NtInfo_t h_info; /* Info handle */
    NtInfoStream_t h_info_stream; /* Info stream handle */

    if (unlikely(port_adapter_map[port] == -1)) {
        if ((status = NT_InfoOpen(&h_info_stream, "ExampleInfo")) != NT_SUCCESS) {
            NAPATECH_ERROR(status);
            return -1;
        }
        /* Read the system info */
        h_info.cmd = NT_INFO_CMD_READ_PORT_V9;
        h_info.u.port_v9.portNo = (uint8_t) port;
        if ((status = NT_InfoRead(h_info_stream, &h_info)) != NT_SUCCESS) {
            /* Get the status code as text */
            NAPATECH_ERROR(status);
            NT_InfoClose(h_info_stream);
            return -1;
        }
        port_adapter_map[port] = h_info.u.port_v9.data.adapterNo;
    }
    return port_adapter_map[port];
}

/**
 * \brief IPv4 4-tuple convenience structure
 */
struct IPv4Tuple4
{
    uint32_t sa; /*!< Source address */
    uint32_t da; /*!< Destination address */
    uint16_t sp; /*!< Source port */
    uint16_t dp; /*!< Destination port */
};

/**
 * \brief IPv6 4-tuple convenience structure
 */
struct IPv6Tuple4
{
    uint8_t sa[16]; /*!< Source address */
    uint8_t da[16]; /*!< Destination address */
    uint16_t sp;    /*!< Source port */
    uint16_t dp;    /*!< Destination port */
};

/**
 * \brief Compares the byte order value of two IPv6 addresses.
 *
 *
 * \param addr_a The first address to compare
 * \param addr_b The second address to compare
 *
 * \return -1 if addr_a < addr_b
 *          1 if addr_a > addr_b
 *          0 if addr_a == addr_b
 */
static int CompareIPv6Addr(uint8_t addr_a[16], uint8_t addr_b[16]) {
    uint16_t pos;
    for (pos = 0; pos < 16; ++pos) {
        if (addr_a[pos] < addr_b[pos]) {
            return -1;
        } else if (addr_a[pos] > addr_b[pos]) {
            return 1;
        } /* else they are equal - check next position*/
    }

    /* if we get here the addresses are equal */
    return 0;
}

/**
 * \brief  Initializes the FlowStreams used to program flow data.
 *
 * Opens a FlowStream on the adapter associated with the rx port.  This
 * FlowStream is subsequently used to program the adapter with
 * flows to bypass.
 *
 * \return the flow stream handle, NULL if failure.
 */
static NtFlowStream_t InitFlowStream(int adapter, int stream_id)
{
    int status;
    NtFlowStream_t hFlowStream;

    NtFlowAttr_t attr;
    char flow_name[80];

    NT_FlowOpenAttrInit(&attr);
    NT_FlowOpenAttrSetAdapterNo(&attr, adapter);

    snprintf(flow_name, sizeof(flow_name), "Flow_stream_%d", stream_id );
    SCLogDebug("Opening flow programming stream:  %s", flow_name);
    if ((status = NT_FlowOpen_Attr(&hFlowStream, flow_name, &attr)) != NT_SUCCESS) {
        SCLogWarning("Napatech bypass functionality not supported by the FPGA version on adapter "
                     "%d - disabling support.",
                adapter);
        return NULL;
    }
    return hFlowStream;
}

/**
 * \brief Callback function to process Bypass events on Napatech Adapter.
 *
 * Callback function that sets up the Flow tables on the Napatech card
 * so that subsequent packets from this flow are bypassed on the hardware.
 *
 * \param p packet containing information about the flow to be bypassed
 * \param is_inline indicates if Suricata is being run in inline mode.
 *
 * \return Error code indicating success (1) or failure (0).
 *
 */
static int ProgramFlow(Packet *p, int inline_mode)
{
    NtFlow_t flow_match;
    memset(&flow_match, 0, sizeof(flow_match));

    NapatechPacketVars *ntpv = &(p->ntpv);

    /*
     * The hardware decoder will "color" the packets according to the protocols
     * in the packet and the port the packet arrived on.  packet_type gets
     * these bits and we mask out layer3, layer4, and is_span to determine
     * the protocols and if the packet is coming in from a SPAN port.
     */
    uint32_t packet_type = ((ntpv->dyn3->color_hi << 14) & 0xFFFFC000) | ntpv->dyn3->color_lo;
    uint8_t *packet = (uint8_t *) ntpv->dyn3 + ntpv->dyn3->descrLength;

    uint32_t layer3 = packet_type & RTE_PTYPE_L3_MASK;
    uint32_t layer4 = packet_type & RTE_PTYPE_L4_MASK;
    uint32_t is_span = packet_type & COLOR_IS_SPAN;

    /*
     * When we're programming the flows to arrive on a span port,
     * where upstream and downstream packets arrive on the same port,
     * the hardware is configured to swap the source and dest
     * fields if the src addr > dest addr.  We need to program the
     * flow tables to match.  We'll compare addresses and set
     * do_swap accordingly.
     */

    uint32_t do_swap = 0;

    SC_ATOMIC_ADD(flow_callback_cnt, 1);

    /* Only bypass TCP and UDP */
    if (PKT_IS_TCP(p)) {
        SC_ATOMIC_ADD(flow_callback_tcp_pkts, 1);
    } else if PKT_IS_UDP(p) {
        SC_ATOMIC_ADD(flow_callback_udp_pkts, 1);
    } else {
        SC_ATOMIC_ADD(flow_callback_unhandled_pkts, 1);
    }

    struct IPv4Tuple4 v4Tuple;
    struct IPv6Tuple4 v6Tuple;
    struct ipv4_hdr *pIPv4_hdr = NULL;
    struct ipv6_hdr *pIPv6_hdr = NULL;

    switch (layer3) {
        case RTE_PTYPE_L3_IPV4:
        {
            pIPv4_hdr = (struct ipv4_hdr *) (packet + ntpv->dyn3->offset0);
            if (!is_span) {
                v4Tuple.sa = pIPv4_hdr->src_addr;
                v4Tuple.da = pIPv4_hdr->dst_addr;
            } else {
                do_swap = (htonl(pIPv4_hdr->src_addr) > htonl(pIPv4_hdr->dst_addr));
                if (!do_swap) {
                    /* already in order */
                    v4Tuple.sa = pIPv4_hdr->src_addr;
                    v4Tuple.da = pIPv4_hdr->dst_addr;
                } else { /* swap */
                    v4Tuple.sa = pIPv4_hdr->dst_addr;
                    v4Tuple.da = pIPv4_hdr->src_addr;
                }
            }
            break;
        }
        case RTE_PTYPE_L3_IPV6:
        {
            pIPv6_hdr = (struct ipv6_hdr *) (packet + ntpv->dyn3->offset0);
            do_swap = (CompareIPv6Addr(pIPv6_hdr->src_addr, pIPv6_hdr->dst_addr) > 0);

            if (!is_span) {
                memcpy(&(v6Tuple.sa), pIPv6_hdr->src_addr, 16);
                memcpy(&(v6Tuple.da), pIPv6_hdr->dst_addr, 16);
            } else {
                /* sort src/dest address before programming */
                if (!do_swap) {
                    /* already in order */
                    memcpy(&(v6Tuple.sa), pIPv6_hdr->src_addr, 16);
                    memcpy(&(v6Tuple.da), pIPv6_hdr->dst_addr, 16);
                } else { /* swap the addresses */
                    memcpy(&(v6Tuple.sa), pIPv6_hdr->dst_addr, 16);
                    memcpy(&(v6Tuple.da), pIPv6_hdr->src_addr, 16);
                }
            }
            break;
        }
        default:
        {
            return 0;
        }
    }

    switch (layer4) {
        case RTE_PTYPE_L4_TCP:
        {
            struct tcp_hdr *tcp_hdr = (struct tcp_hdr *) (packet + ntpv->dyn3->offset1);
            if (layer3 == RTE_PTYPE_L3_IPV4) {
                if (!is_span) {
                    v4Tuple.dp = tcp_hdr->dst_port;
                    v4Tuple.sp = tcp_hdr->src_port;
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV4;
                } else {
                    if (!do_swap) {
                        v4Tuple.sp = tcp_hdr->src_port;
                        v4Tuple.dp = tcp_hdr->dst_port;
                    } else {
                        v4Tuple.sp = tcp_hdr->dst_port;
                        v4Tuple.dp = tcp_hdr->src_port;
                    }
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV4_SPAN;
                }
                memcpy(&(flow_match.keyData), &v4Tuple, sizeof(v4Tuple));
            } else {
                if (!is_span) {
                    v6Tuple.dp = tcp_hdr->dst_port;
                    v6Tuple.sp = tcp_hdr->src_port;
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV6;
                } else {
                    if (!do_swap) {
                        v6Tuple.sp = tcp_hdr->src_port;
                        v6Tuple.dp = tcp_hdr->dst_port;
                    } else {
                        v6Tuple.dp = tcp_hdr->src_port;
                        v6Tuple.sp = tcp_hdr->dst_port;
                    }
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV6_SPAN;
                }
                memcpy(&(flow_match.keyData), &v6Tuple, sizeof(v6Tuple));
            }
            flow_match.ipProtocolField = 6;
            break;
        }
        case RTE_PTYPE_L4_UDP:
        {
            struct udp_hdr *udp_hdr = (struct udp_hdr *) (packet + ntpv->dyn3->offset1);
            if (layer3 == RTE_PTYPE_L3_IPV4) {
                if (!is_span) {
                    v4Tuple.dp = udp_hdr->dst_port;
                    v4Tuple.sp = udp_hdr->src_port;
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV4;
                } else {
                    if (!do_swap) {
                        v4Tuple.sp = udp_hdr->src_port;
                        v4Tuple.dp = udp_hdr->dst_port;
                    } else {
                        v4Tuple.dp = udp_hdr->src_port;
                        v4Tuple.sp = udp_hdr->dst_port;
                    }
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV4_SPAN;
                }
                memcpy(&(flow_match.keyData), &v4Tuple, sizeof(v4Tuple));
            } else { /* layer3 is IPV6 */
                if (!is_span) {
                    v6Tuple.dp = udp_hdr->dst_port;
                    v6Tuple.sp = udp_hdr->src_port;
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV6;
                } else {
                    if (!do_swap) {
                        v6Tuple.sp = udp_hdr->src_port;
                        v6Tuple.dp = udp_hdr->dst_port;
                    } else {
                        v6Tuple.dp = udp_hdr->src_port;
                        v6Tuple.sp = udp_hdr->dst_port;
                    }
                    flow_match.keyId = NAPATECH_KEYTYPE_IPV6_SPAN;
                }
                memcpy(&(flow_match.keyData), &v6Tuple, sizeof(v6Tuple));
            }
            flow_match.ipProtocolField = 17;
            break;
        }
        default:
        {
            return 0;
        }
    }

    flow_match.op = 1;  /* program flow */
    flow_match.gfi = 1; /* Generate FlowInfo records */
    flow_match.tau = 1; /* tcp automatic unlearn */

    if (PacketCheckAction(p, ACTION_DROP)) {
        flow_match.keySetId = NAPATECH_FLOWTYPE_DROP;
    } else {
        if (inline_mode) {
            flow_match.keySetId = NAPATECH_FLOWTYPE_PASS;
        } else {
            flow_match.keySetId = NAPATECH_FLOWTYPE_DROP;
        }
    }

    if (NT_FlowWrite(ntpv->flow_stream, &flow_match, -1) != NT_SUCCESS) {
        if (!(suricata_ctl_flags & SURICATA_STOP)) {
            SCLogError("NT_FlowWrite failed!.");
            exit(EXIT_FAILURE);
        }
    }

    return 1;
}

/**
 * \brief     Callback from Suricata when a flow that should be bypassed
 *            is identified.
 */

static int NapatechBypassCallback(Packet *p)
{
    NapatechPacketVars *ntpv = &(p->ntpv);

    /*
     *  Since, at this point, we don't know what action to take,
     *  simply mark this packet as one that should be
     *  bypassed when the packet is returned by suricata with a
     *  pass/drop verdict.
     */
    ntpv->bypass = 1;

    return 1;
}

#endif

/**
 * \brief   Initialize the Napatech receiver thread, generate a single
 *          NapatechThreadVar structure for each thread, this will
 *          contain a NtNetStreamRx_t stream handle which is used when the
 *          thread executes to acquire the packets.
 *
 * \param tv        Thread variable to ThreadVars
 * \param initdata  Initial data to the adapter passed from the user,
 *                  this is processed by the user.
 *
 *                  For now, we assume that we have only a single name for the NAPATECH
 *                  adapter.
 *
 * \param data      data pointer gets populated with
 *
 */
TmEcode NapatechStreamThreadInit(ThreadVars *tv, const void *initdata, void **data)
{
    SCEnter();
    struct NapatechStreamDevConf *conf = (struct NapatechStreamDevConf *) initdata;
    uint16_t stream_id = conf->stream_id;
    *data = NULL;

    NapatechThreadVars *ntv = SCCalloc(1, sizeof (NapatechThreadVars));
    if (unlikely(ntv == NULL)) {
        FatalError("Failed to allocate memory for NAPATECH  thread vars.");
    }

    memset(ntv, 0, sizeof (NapatechThreadVars));
    ntv->stream_id = stream_id;
    ntv->tv = tv;
    ntv->hba = conf->hba;

    DatalinkSetGlobalType(LINKTYPE_ETHERNET);

    SCLogDebug("Started processing packets from NAPATECH  Stream: %u", ntv->stream_id);

    *data = (void *) ntv;
    SCReturnInt(TM_ECODE_OK);
}

/**
 * \brief Callback to indicate that the packet buffer can be returned to the hardware.
 *
 *  Called when Suricata is done processing the packet.  Before the packet is released
 *  this also checks the action to see if the packet should be dropped and programs the
 *  flow hardware if the flow is to be bypassed and the Napatech packet buffer is released.
 *
 *
 * \param p Packet to return to the system.
 *
 */
static void NapatechReleasePacket(struct Packet_ *p)
{
    /*
     * If the packet is to be dropped we need to set the wirelength
     * before releasing the Napatech buffer back to NTService.
     */
#ifdef NAPATECH_ENABLE_BYPASS
    if (is_inline && PacketCheckAction(p, ACTION_DROP)) {
        p->ntpv.dyn3->wireLength = 0;
    }

    /*
     *  If this flow is to be programmed for hardware bypass we do it now.  This is done
     *  here because the action is not available in the packet structure at the time of the
     *  bypass callback and it needs to be done before we release the packet structure.
     */
    if (p->ntpv.bypass == 1) {
        ProgramFlow(p, is_inline);
    }
#endif

    NT_NetRxRelease(p->ntpv.rx_stream, p->ntpv.nt_packet_buf);
    PacketFreeOrRelease(p);
}

/**
 * \brief Returns the NUMA node associated with the currently running thread.
 *
 * \return ID of the NUMA node.
 *
 */
static int GetNumaNode(void)
{
    int cpu = 0;
    int node = 0;

#if defined(__linux__)
    cpu = sched_getcpu();
    node = numa_node_of_cpu(cpu);
#else
    SCLogWarning("Auto configuration of NUMA node is not supported on this OS.");
#endif

    return node;
}

/**
 * \brief Outputs hints on the optimal host-buffer configuration to aid tuning.
 *
 * \param log_level of the currently running instance.
 *
 */
static void RecommendNUMAConfig(void)
{
    char *buffer, *p;
    int set_cpu_affinity = 0;

    p = buffer = SCCalloc(sizeof(char), (32 * (numa_max_node() + 1) + 1));
    if (buffer == NULL) {
        FatalError("Failed to allocate memory for temporary buffer: %s", strerror(errno));
    }

    if (ConfGetBool("threading.set-cpu-affinity", &set_cpu_affinity) != 1) {
        set_cpu_affinity = 0;
    }

    if (set_cpu_affinity) {
        SCLogPerf("Minimum host buffers that should be defined in ntservice.ini:");
        for (int i = 0; i <= numa_max_node(); ++i) {
            SCLogPerf("   NUMA Node %d: %d", i, SC_ATOMIC_GET(numa_detect[i].count));
            p += snprintf(p, 32, "%s[%d, 16, %d]", (i == 0 ? "" : ","),
                    SC_ATOMIC_GET(numa_detect[i].count), i);
        }
        SCLogPerf("E.g.: HostBuffersRx=%s", buffer);
    }

    SCFree(buffer);
}

/**
 * \brief   Main Napatechpacket processing loop
 *
 * \param tv     Thread variable to ThreadVars
 * \param data   Pointer to NapatechThreadVars with data specific to Napatech
 * \param slot   TMSlot where this instance is running.
 *
 */
TmEcode NapatechPacketLoop(ThreadVars *tv, void *data, void *slot)
{
    int32_t status;
    char error_buffer[100];
    uint64_t pkt_ts;
    NtNetBuf_t packet_buffer;
    NapatechThreadVars *ntv = (NapatechThreadVars *) data;
    uint64_t hba_pkt_drops = 0;
    uint64_t hba_byte_drops = 0;
    uint16_t hba_pkt = 0;
    int numa_node = -1;
    int set_cpu_affinity = 0;
    int closer = 0;
    int is_autoconfig = 0;

    /* This just keeps the startup output more orderly. */
    usleep(200000 * ntv->stream_id);

#ifdef NAPATECH_ENABLE_BYPASS
    NtFlowStream_t flow_stream[MAX_ADAPTERS] = { 0 };
    if (NapatechUseHWBypass()) {
        /* Get a FlowStream handle for each adapter so we can efficiently find the
         * correct handle corresponding to the port on which a packet is received.
         */
        int adapter = 0;
        for (adapter = 0; adapter < NapatechGetNumAdapters(); ++adapter) {
            flow_stream[adapter] = InitFlowStream(adapter, ntv->stream_id);
        }
    }
#endif

    if (ConfGetBool("napatech.auto-config", &is_autoconfig) == 0) {
        is_autoconfig = 0;
    }

    if (is_autoconfig) {
        numa_node = GetNumaNode();

        if (numa_node <= numa_max_node()) {
            SC_ATOMIC_ADD(numa_detect[numa_node].count, 1);
        }

        if (ConfGetBool("threading.set-cpu-affinity", &set_cpu_affinity) != 1) {
            set_cpu_affinity = 0;
        }

        if (set_cpu_affinity) {
            NapatechSetupNuma(ntv->stream_id, numa_node);
        }

        SC_ATOMIC_ADD(stream_count, 1);
        if (SC_ATOMIC_GET(stream_count) == NapatechGetNumConfiguredStreams()) {
            /* Print the recommended NUMA configuration early because it
             * can fail with "No available hostbuffers" in NapatechSetupTraffic */
            RecommendNUMAConfig();

#ifdef NAPATECH_ENABLE_BYPASS
            if (ConfGetBool("napatech.inline", &is_inline) == 0) {
                is_inline = 0;
            }

            /* Initialize the port map before we setup traffic filters */
            for (int i = 0; i < MAX_PORTS; ++i) {
                inline_port_map[i] = -1;
            }
#endif
            /* The last thread to run sets up and deletes the streams */
            status = NapatechSetupTraffic(NapatechGetNumFirstStream(),
                    NapatechGetNumLastStream());

            closer = 1;

            if (status == 0x20002061) {
                FatalError("Check host buffer configuration in ntservice.ini"
                           " or try running /opt/napatech3/bin/ntpl -e "
                           "\"delete=all\" to clean-up stream NUMA config.");
            } else if (status == 0x20000008) {
                FatalError("Check napatech.ports in the suricata config file.");
            }
            SCLogNotice("Napatech packet input engine started.");
        }
    } // is_autoconfig

    SCLogInfo(
            "Napatech Packet Loop Started - cpu: %3d, cpu_numa: %3d   stream: %3u ",
            sched_getcpu(), numa_node, ntv->stream_id);

    if (ntv->hba > 0) {
        char *s_hbad_pkt = SCCalloc(1, 32);
        if (unlikely(s_hbad_pkt == NULL)) {
            FatalError("Failed to allocate memory for NAPATECH stream counter.");
        }
        snprintf(s_hbad_pkt, 32, "nt%d.hba_drop", ntv->stream_id);
        hba_pkt = StatsRegisterCounter(s_hbad_pkt, tv);
        StatsSetupPrivate(tv);
        StatsSetUI64(tv, hba_pkt, 0);
    }
    SCLogDebug("Opening NAPATECH Stream: %u for processing", ntv->stream_id);

    if ((status = NT_NetRxOpen(&(ntv->rx_stream), "SuricataStream",
            NT_NET_INTERFACE_PACKET, ntv->stream_id, ntv->hba)) != NT_SUCCESS) {

        NAPATECH_ERROR(status);
        SCFree(ntv);
        SCReturnInt(TM_ECODE_FAILED);
    }
    TmSlot *s = (TmSlot *) slot;
    ntv->slot = s->slot_next;

    // Indicate that the thread is actually running its application level code (i.e., it can poll
    // packets)
    TmThreadsSetFlag(tv, THV_RUNNING);

    while (!(suricata_ctl_flags & SURICATA_STOP)) {
        /* make sure we have at least one packet in the packet pool, to prevent
         * us from alloc'ing packets at line rate */
        PacketPoolWait();

        /* Napatech returns packets 1 at a time */
        status = NT_NetRxGet(ntv->rx_stream, &packet_buffer, 1000);
        if (unlikely(
                status == NT_STATUS_TIMEOUT || status == NT_STATUS_TRYAGAIN)) {
            if (status == NT_STATUS_TIMEOUT) {
                TmThreadsCaptureHandleTimeout(tv, NULL);
            }
            continue;
        } else if (unlikely(status != NT_SUCCESS)) {
            NAPATECH_ERROR(status);
            SCLogInfo("Failed to read from Napatech Stream %d: %s",
                    ntv->stream_id, error_buffer);
            break;
        }

        Packet *p = PacketGetFromQueueOrAlloc();
        if (unlikely(p == NULL)) {
            NT_NetRxRelease(ntv->rx_stream, packet_buffer);
            SCReturnInt(TM_ECODE_FAILED);
        }

#ifdef NAPATECH_ENABLE_BYPASS
        p->ntpv.bypass = 0;
#endif
        p->ntpv.rx_stream = ntv->rx_stream;

        pkt_ts = NT_NET_GET_PKT_TIMESTAMP(packet_buffer);

        /*
         * Handle the different timestamp forms that the napatech cards could use
         *   - NT_TIMESTAMP_TYPE_NATIVE is not supported due to having an base
         *     of 0 as opposed to NATIVE_UNIX which has a base of 1/1/1970
         */
        switch (NT_NET_GET_PKT_TIMESTAMP_TYPE(packet_buffer)) {
            case NT_TIMESTAMP_TYPE_NATIVE_UNIX:
                p->ts = SCTIME_ADD_USECS(SCTIME_FROM_SECS(pkt_ts / 100000000),
                        ((pkt_ts % 100000000) / 100) + ((pkt_ts % 100) > 50 ? 1 : 0));
                break;
            case NT_TIMESTAMP_TYPE_PCAP:
                p->ts = SCTIME_ADD_USECS(SCTIME_FROM_SECS(pkt_ts >> 32), pkt_ts & 0xFFFFFFFF);
                break;
            case NT_TIMESTAMP_TYPE_PCAP_NANOTIME:
                p->ts = SCTIME_ADD_USECS(SCTIME_FROM_SECS(pkt_ts >> 32),
                        ((pkt_ts & 0xFFFFFFFF) / 1000) + ((pkt_ts % 1000) > 500 ? 1 : 0));
                break;
            case NT_TIMESTAMP_TYPE_NATIVE_NDIS:
                /* number of seconds between 1/1/1601 and 1/1/1970 */
                p->ts = SCTIME_ADD_USECS(SCTIME_FROM_SECS((pkt_ts / 100000000) - 11644473600),
                        ((pkt_ts % 100000000) / 100) + ((pkt_ts % 100) > 50 ? 1 : 0));
                break;
            default:
                SCLogError("Packet from Napatech Stream: %u does not have a supported timestamp "
                           "format",
                        ntv->stream_id);
                NT_NetRxRelease(ntv->rx_stream, packet_buffer);
                SCReturnInt(TM_ECODE_FAILED);
        }

        if (unlikely(ntv->hba > 0)) {
            NtNetRx_t stat_cmd;
            stat_cmd.cmd = NT_NETRX_READ_CMD_STREAM_DROP;
            /* Update drop counter */
            if (unlikely((status = NT_NetRxRead(ntv->rx_stream, &stat_cmd)) != NT_SUCCESS)) {
                NAPATECH_ERROR(status);
                SCLogInfo("Couldn't retrieve drop statistics from the RX stream: %u",
                        ntv->stream_id);
            } else {
                hba_pkt_drops = stat_cmd.u.streamDrop.pktsDropped;

                StatsSetUI64(tv, hba_pkt, hba_pkt_drops);
            }
            StatsSyncCountersIfSignalled(tv);
        }

#ifdef NAPATECH_ENABLE_BYPASS
        p->ntpv.dyn3 = _NT_NET_GET_PKT_DESCR_PTR_DYN3(packet_buffer);
        p->BypassPacketsFlow = (NapatechIsBypassSupported() ? NapatechBypassCallback : NULL);
        NT_NET_SET_PKT_TXPORT(packet_buffer, inline_port_map[p->ntpv.dyn3->rxPort]);
        p->ntpv.flow_stream = flow_stream[NapatechGetAdapter(p->ntpv.dyn3->rxPort)];

#endif

        p->ReleasePacket = NapatechReleasePacket;
        p->ntpv.nt_packet_buf = packet_buffer;
        p->ntpv.stream_id = ntv->stream_id;
        p->datalink = LINKTYPE_ETHERNET;

        if (unlikely(PacketSetData(p, (uint8_t *)NT_NET_GET_PKT_L2_PTR(packet_buffer), NT_NET_GET_PKT_WIRE_LENGTH(packet_buffer)))) {
            TmqhOutputPacketpool(ntv->tv, p);
            SCReturnInt(TM_ECODE_FAILED);
        }

        if (unlikely(TmThreadsSlotProcessPkt(ntv->tv, ntv->slot, p) != TM_ECODE_OK)) {
            SCReturnInt(TM_ECODE_FAILED);
        }

        /*
         * At this point the packet and the Napatech Packet Buffer have been returned
         * to the system in the NapatechReleasePacket() Callback.
         */

        StatsSyncCountersIfSignalled(tv);
    } // while

    if (closer) {
        NapatechDeleteFilters();
    }

    if (unlikely(ntv->hba > 0)) {
        SCLogInfo("Host Buffer Allowance Drops - pkts: %ld,  bytes: %ld", hba_pkt_drops, hba_byte_drops);
    }

    SCReturnInt(TM_ECODE_OK);
}

/**
 * \brief Print some stats to the log at program exit.
 *
 * \param tv Pointer to ThreadVars.
 * \param data Pointer to data, ErfFileThreadVars.
 */
void NapatechStreamThreadExitStats(ThreadVars *tv, void *data)
{
    NapatechThreadVars *ntv = (NapatechThreadVars *) data;
    NapatechCurrentStats stat = NapatechGetCurrentStats(ntv->stream_id);

    double percent = 0;
    if (stat.current_drop_packets > 0)
        percent = (((double) stat.current_drop_packets)
                  / (stat.current_packets + stat.current_drop_packets)) * 100;

    SCLogInfo("nt%lu - pkts: %lu; drop: %lu (%5.2f%%); bytes: %lu",
                 (uint64_t) ntv->stream_id, stat.current_packets,
                  stat.current_drop_packets, percent, stat.current_bytes);

    SC_ATOMIC_ADD(total_packets, stat.current_packets);
    SC_ATOMIC_ADD(total_drops, stat.current_drop_packets);
    SC_ATOMIC_ADD(total_tallied, 1);

    if (SC_ATOMIC_GET(total_tallied) == NapatechGetNumConfiguredStreams()) {
        if (SC_ATOMIC_GET(total_drops) > 0)
            percent = (((double) SC_ATOMIC_GET(total_drops)) / (SC_ATOMIC_GET(total_packets)
                         + SC_ATOMIC_GET(total_drops))) * 100;

        SCLogInfo(" ");
        SCLogInfo("--- Total Packets: %ld  Total Dropped: %ld (%5.2f%%)",
                SC_ATOMIC_GET(total_packets), SC_ATOMIC_GET(total_drops), percent);

#ifdef NAPATECH_ENABLE_BYPASS
        SCLogInfo("--- BypassCB - Total: %ld,  UDP: %ld,  TCP: %ld,  Unhandled: %ld",
                SC_ATOMIC_GET(flow_callback_cnt),
                SC_ATOMIC_GET(flow_callback_udp_pkts),
                SC_ATOMIC_GET(flow_callback_tcp_pkts),
                SC_ATOMIC_GET(flow_callback_unhandled_pkts));
#endif
    }
}

/**
 * \brief   Deinitializes the NAPATECH card.
 * \param   tv pointer to ThreadVars
 * \param   data pointer that gets cast into PcapThreadVars for ptv
 */
TmEcode NapatechStreamThreadDeinit(ThreadVars *tv, void *data)
{
    SCEnter();
    NapatechThreadVars *ntv = (NapatechThreadVars *) data;

    SCLogDebug("Closing Napatech Stream: %d", ntv->stream_id);
    NT_NetRxClose(ntv->rx_stream);

    SCReturnInt(TM_ECODE_OK);
}

/**
 * \brief   This function passes off to link type decoders.
 *
 * NapatechDecode decodes packets from Napatech and passes
 * them off to the proper link type decoder.
 *
 * \param t pointer to ThreadVars
 * \param p pointer to the current packet
 * \param data pointer that gets cast into PcapThreadVars for ptv
 */
TmEcode NapatechDecode(ThreadVars *tv, Packet *p, void *data)
{
    SCEnter();

    DecodeThreadVars *dtv = (DecodeThreadVars *) data;

    BUG_ON(PKT_IS_PSEUDOPKT(p));

    // update counters
    DecodeUpdatePacketCounters(tv, dtv, p);

    switch (p->datalink) {
        case LINKTYPE_ETHERNET:
            DecodeEthernet(tv, dtv, p, GET_PKT_DATA(p), GET_PKT_LEN(p));
            break;
        default:
            SCLogError("Datalink type %" PRId32 " not yet supported in module NapatechDecode",
                    p->datalink);
            break;
    }

    PacketDecodeFinalize(tv, dtv, p);
    SCReturnInt(TM_ECODE_OK);
}

/**
 * \brief   Initialization of Napatech Thread.
 *
 * \param t pointer to ThreadVars
 * \param initdata - unused.
 * \param data pointer that gets cast into DecoderThreadVars
 */
TmEcode NapatechDecodeThreadInit(ThreadVars *tv, const void *initdata, void **data)
{
    SCEnter();
    DecodeThreadVars *dtv = NULL;
    dtv = DecodeThreadVarsAlloc(tv);
    if (dtv == NULL) {
        SCReturnInt(TM_ECODE_FAILED);
    }

    DecodeRegisterPerfCounters(dtv, tv);
    *data = (void *) dtv;
    SCReturnInt(TM_ECODE_OK);
}

/**
 * \brief   Deinitialization of Napatech Thread.
 *
 * \param tv pointer to ThreadVars
 * \param data pointer that gets cast into DecoderThreadVars
 */
TmEcode NapatechDecodeThreadDeinit(ThreadVars *tv, void *data)
{
    if (data != NULL) {
        DecodeThreadVarsFree(tv, data);
    }
    SCReturnInt(TM_ECODE_OK);
}

#endif /* HAVE_NAPATECH */