gfwleak/stellar-stellar-2022
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[PACKET_IO]fix read pcap directory bug

Browse Source
This commit is contained in:
liuwentan
2022-08-26 15:59:12 +08:00
parent df25dd5826
commit 9cdc57eaa7
7 changed files with 269 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

375
src/packet_io/pcap_file_mode/pio_pcap_file.cpp
View File

@@ -15,6 +15,7 @@
#include <pthread.h>
#include <sys/stat.h>
#include <sys/syscall.h>
#include <pcap/pcap.h>
#include "utils.h"
#include "util_errors.h"
@@ -25,6 +26,8 @@
#include "packet_io_util.h"
#include "packet_io_internal.h"
#define MAX_RECV_BURST 64
struct safe_pending_file_queue g_pending_file_queue;
/**
@@ -63,26 +66,6 @@ static ssize_t validate_directory_or_file(const char *path, DIR **dir)
return ret;
}
/**
* @brief get the timestamp of the first packet and rewind
*
* @retval true(success), false(error)
*/
static bool peek_first_packet_timestamp(struct pcap_plain_file_info *pfile_info)
{
int ret = pcap_next_ex(pfile_info->pcap_handle, &pfile_info->first_pkt_hdr,
&pfile_info->first_pkt_data);
if (ret <= 0 || (nullptr == pfile_info->first_pkt_hdr)) {
log_error(ST_ERR_PCAP_OPEN_OFFLINE, "failed to get first packet timestamp");
return false;
}
pfile_info->first_pkt_ts.tv_sec = pfile_info->first_pkt_hdr->ts.tv_sec;
pfile_info->first_pkt_ts.tv_usec = pfile_info->first_pkt_hdr->ts.tv_usec;
return true;
}
static ssize_t init_pcap_file(struct pcap_plain_file_info *pfile_info)
{
char errbuf[PCAP_ERRBUF_SIZE] = "";
@@ -118,9 +101,6 @@ static ssize_t init_pcap_file(struct pcap_plain_file_info *pfile_info)
}
#endif
pfile_info->data_link = pcap_datalink(pfile_info->pcap_handle);
if (!peek_first_packet_timestamp(pfile_info)) {
return -1;
}
return 0;
}
@@ -344,7 +324,8 @@ void pcap_file_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_hdr, u_c
p->pkt_hdr = p;
p->pkt_payload = (uint8_t *)p + CUSTOM_ZONE_LEN;
p->pkt_len = pkt_hdr->caplen;
p->data_link = pfile_dev_ctx->entity.file->data_link;
//p->data_link = pfile_dev_ctx->entity.file->data_link;
p->data_link = 1;
if (packet_copy_data((uint8_t *)p->pkt_payload, (uint8_t *)pkt, pkt_hdr->caplen)) {
FREE(p);
@@ -361,80 +342,55 @@ void pcap_file_pkt_callback_oneshot(char *user, struct pcap_pkthdr *pkt_hdr, u_c
pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
}
static ssize_t pcap_directory_file_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx, uint32_t rxq_id,
struct stellar_packet **pkts, size_t nr_pkts)
static ssize_t pcap_directory_file_pkts_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx, uint32_t rxq_id)
{
if (pfile_dev_ctx->entity.dir->current_file[rxq_id]->first_pkt_hdr != nullptr) {
pthread_mutex_lock(&pfile_dev_ctx->entity.dir->current_file[rxq_id]->handle_mutex);
if (pfile_dev_ctx->entity.dir->current_file[rxq_id]->first_pkt_hdr != nullptr) {
pcap_file_pkt_callback_oneshot((char *)pfile_dev_ctx, pfile_dev_ctx->entity.dir->current_file[rxq_id]->first_pkt_hdr,
(u_char *)pfile_dev_ctx->entity.dir->current_file[rxq_id]->first_pkt_data);
pfile_dev_ctx->entity.dir->current_file[rxq_id]->first_pkt_hdr = nullptr;
pfile_dev_ctx->entity.dir->current_file[rxq_id]->first_pkt_data = nullptr;
}
pthread_mutex_unlock(&pfile_dev_ctx->entity.dir->current_file[rxq_id]->handle_mutex);
}
int packet_q_len = nr_pkts;
int packet_q_len = MAX_RECV_BURST;
ssize_t res = -1;
pthread_mutex_lock(&pfile_dev_ctx->entity.dir->current_file[rxq_id]->handle_mutex);
res = pcap_dispatch(pfile_dev_ctx->entity.dir->current_file[rxq_id]->pcap_handle, packet_q_len,
(pcap_handler)pcap_file_pkt_callback_oneshot, (u_char *)pfile_dev_ctx);
pthread_mutex_unlock(&pfile_dev_ctx->entity.dir->current_file[rxq_id]->handle_mutex);
if (res < 0) {
log_error(ST_ERR_PCAP_DISPATCH, "error code %ld %s for %s", res,
pcap_geterr(pfile_dev_ctx->entity.dir->current_file[rxq_id]->pcap_handle),
pfile_dev_ctx->entity.dir->current_file[rxq_id]->file_name);
} else if (res == 0) {
log_info("reach end of pcap file %s (error code %d)", pfile_dev_ctx->entity.dir->current_file[rxq_id]->file_name, res);
//TODO: close pcap file
} else {
// success
struct pio_packet *p = nullptr;
size_t i = 0;
uint32_t q_len = 0;
pthread_mutex_lock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
/* if pkt queue is empty */
if (pfile_dev_ctx->pkt_queues[rxq_id].len == 0) {
pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
return 0;
}
do {
p = pio_packet_dequeue(&pfile_dev_ctx->pkt_queues[rxq_id]);
pkts[i] = (struct stellar_packet *)p;
q_len = pfile_dev_ctx->pkt_queues[rxq_id].len;
i++;
} while ((q_len != 0) && (i < nr_pkts));
pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
if (q_len == 0) {
res = i;
} else {
res = nr_pkts;
}
}
return res;
}
static ssize_t pcap_file_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx, uint32_t rxq_id,
struct stellar_packet **pkts, size_t nr_pkts)
static ssize_t get_pcap_file_pkts(struct pio_pcap_file_device_context *pfile_dev_ctx, uint32_t rxq_id,
struct stellar_packet **pkts, size_t nr_pkts)
{
if (pfile_dev_ctx->entity.file->first_pkt_hdr != nullptr) {
pthread_mutex_lock(&pfile_dev_ctx->entity.file->handle_mutex);
if (pfile_dev_ctx->entity.file->first_pkt_hdr != nullptr) {
pcap_file_pkt_callback_oneshot((char *)pfile_dev_ctx, pfile_dev_ctx->entity.file->first_pkt_hdr,
(u_char *)pfile_dev_ctx->entity.file->first_pkt_data);
pfile_dev_ctx->entity.file->first_pkt_hdr = nullptr;
pfile_dev_ctx->entity.file->first_pkt_data = nullptr;
}
pthread_mutex_unlock(&pfile_dev_ctx->entity.file->handle_mutex);
ssize_t res = -1;
size_t i = 0;
uint32_t q_len = 0;
struct pio_packet *p = nullptr;
pthread_mutex_lock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
if (pfile_dev_ctx->pkt_queues[rxq_id].len > 0) {
do {
p = pio_packet_dequeue(&pfile_dev_ctx->pkt_queues[rxq_id]);
q_len = pfile_dev_ctx->pkt_queues[rxq_id].len;
pkts[i] = (struct stellar_packet *)p;
i++;
} while ((q_len != 0) && (i < nr_pkts));
}
pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
if (q_len == 0) {
res = i;
} else {
res = nr_pkts;
}
int packet_q_len = nr_pkts;
return res;
}
static ssize_t pcap_file_pkts_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx)
{
int packet_q_len = MAX_RECV_BURST;
ssize_t res = -1;
pthread_mutex_lock(&pfile_dev_ctx->entity.file->handle_mutex);
@@ -444,31 +400,8 @@ static ssize_t pcap_file_dispatch(struct pio_pcap_file_device_context *pfile_dev
if (res < 0) {
log_error(ST_ERR_PCAP_DISPATCH, "error code %ld %s for %s", res,
pcap_geterr(pfile_dev_ctx->entity.file->pcap_handle), pfile_dev_ctx->entity.file->file_name);
} else {
if (res == 0) {
log_info("reach end of pcap file %s (error code %d)", pfile_dev_ctx->entity.file->file_name, res);
}
// success
struct pio_packet *p = nullptr;
size_t i = 0;
uint32_t q_len = 0;
pthread_mutex_lock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
if (pfile_dev_ctx->pkt_queues[rxq_id].len != 0) {
do {
p = pio_packet_dequeue(&pfile_dev_ctx->pkt_queues[rxq_id]);
q_len = pfile_dev_ctx->pkt_queues[rxq_id].len;
pkts[i] = (struct stellar_packet *)p;
i++;
} while ((q_len != 0) && (i < nr_pkts));
}
pthread_mutex_unlock(&pfile_dev_ctx->pkt_queues[rxq_id].mutex_q);
if (q_len == 0) {
res = i;
} else {
res = nr_pkts;
}
} else if (res == 0) {
log_info("reach end of pcap file %s (error code %d)", pfile_dev_ctx->entity.file->file_name, res);
}
return res;
@@ -569,6 +502,197 @@ static ssize_t pcap_directory_insert_file(struct pending_file *file_to_add)
return 0;
}
#define SWAPLONG(y) \
(((((u_int)(y))&0xff)<<24) | \
((((u_int)(y))&0xff00)<<8) | \
((((u_int)(y))&0xff0000)>>8) | \
((((u_int)(y))>>24)&0xff))
#define SWAPSHORT(y) \
((u_short)(((((u_int)(y))&0xff)<<8) | \
((((u_int)(y))&0xff00)>>8)))
/* Standard libpcap format. */
#define TCPDUMP_MAGIC 0xa1b2c3d4
/* Alexey Kuznetzov's modified libpcap format. */
#define KUZNETZOV_TCPDUMP_MAGIC 0xa1b2cd34
/*
* Normal libpcap format, except for seconds/nanoseconds timestamps,
* as per a request by Ulf Lamping <ulf.lamping@web.de>
*/
#define NSEC_TCPDUMP_MAGIC 0xa1b23c4d
#define LT_RESERVED1(x) ((x) & 0x03FF0000)
static ssize_t pcap_header_check(const uint8_t *magic, FILE *fp)
{
int swapped = 0;
size_t amt_read;
uint32_t magic_int;
struct pcap_file_header hdr;
/*
* Check whether the first 4 bytes of the file are the magic
* number for a pcap savefile, or for a byte-swapped pcap
* savefile.
*/
memcpy(&magic_int, magic, sizeof(magic_int));
if (magic_int != TCPDUMP_MAGIC && magic_int != KUZNETZOV_TCPDUMP_MAGIC && magic_int != NSEC_TCPDUMP_MAGIC) {
magic_int = SWAPLONG(magic_int);
if (magic_int != TCPDUMP_MAGIC && magic_int != KUZNETZOV_TCPDUMP_MAGIC && magic_int != NSEC_TCPDUMP_MAGIC) {
return -1;
}
swapped = 1;
}
/*
* They are. Put the magic number in the header, and read
* the rest of the header.
*/
hdr.magic = magic_int;
amt_read = fread(((char *)&hdr) + sizeof hdr.magic, 1,
sizeof(hdr) - sizeof(hdr.magic), fp);
if (amt_read != sizeof(hdr) - sizeof(hdr.magic)) {
return -1;
}
/*
* If it's a byte-swapped capture file, byte-swap the header.
*/
if (swapped) {
hdr.version_major = SWAPSHORT(hdr.version_major);
hdr.version_minor = SWAPSHORT(hdr.version_minor);
hdr.thiszone = SWAPLONG(hdr.thiszone);
hdr.sigfigs = SWAPLONG(hdr.sigfigs);
hdr.snaplen = SWAPLONG(hdr.snaplen);
hdr.linktype = SWAPLONG(hdr.linktype);
}
if (hdr.version_major < PCAP_VERSION_MAJOR) {
return -1;
}
/*
* currently only versions 2.[0-4] are supported with
* the exception of 543.0 for DG/UX tcpdump.
*/
if (! ((hdr.version_major == PCAP_VERSION_MAJOR &&
hdr.version_minor <= PCAP_VERSION_MINOR) ||
(hdr.version_major == 543 && hdr.version_minor == 0))) {
return -1;
}
/*
* Check the main reserved field.
*/
if (LT_RESERVED1(hdr.linktype) != 0) {
return -1;
}
return 0;
}
#define BT_SHB 0x0A0D0D0A
#define BT_SHB_INSANE_MAX 1024U*1024U*1U /* 1MB should be enough */
#define BYTE_ORDER_MAGIC 0x1A2B3C4D
struct section_header_block {
uint32_t byte_order_magic;
u_short major_version;
u_short minor_version;
uint64_t section_length;
/* followed by options and trailer */
};
struct block_header {
uint32_t block_type;
uint32_t total_length;
};
static ssize_t pcapng_header_check(const uint8_t *magic, FILE *fp)
{
uint32_t magic_int;
uint32_t total_length;
uint32_t byte_order_magic;
size_t amt_read;
struct block_header *bhdrp;
struct section_header_block *shbp;
/*
* Check whether the first 4 bytes of the file are the block
* type for a pcapng savefile.
*/
memcpy(&magic_int, magic, sizeof(magic_int));
if (magic_int != BT_SHB) {
return -1;
}
/*
* OK, they are. However, that's just \n\r\r\n, so it could,
* conceivably, be an ordinary text file.
*
* It could not, however, conceivably be any other type of
* capture file, so we can read the rest of the putative
* Section Header Block; put the block type in the common
* header, read the rest of the common header and the
* fixed-length portion of the SHB, and look for the byte-order
* magic value.
*/
amt_read = fread(&total_length, 1, sizeof(total_length), fp);
if (amt_read < sizeof(total_length)) {
return -1;
}
amt_read = fread(&byte_order_magic, 1, sizeof(byte_order_magic), fp);
if (amt_read < sizeof(byte_order_magic)) {
return -1;
}
if (byte_order_magic != BYTE_ORDER_MAGIC) {
byte_order_magic = SWAPLONG(byte_order_magic);
if (byte_order_magic != BYTE_ORDER_MAGIC) {
return -1;
}
total_length = SWAPLONG(total_length);
}
/*
* Check the sanity of the total length.
*/
if (total_length < sizeof(*bhdrp) + sizeof(*shbp) + sizeof(total_length) ||
(total_length > BT_SHB_INSANE_MAX)) {
return -1;
}
return 0;
}
static ssize_t validate_pcap_file(struct pio_pcap_file_device_context *pfile_dev_ctx, const char *file_name)
{
uint8_t magic[4];
char abs_path[PATH_MAX] = {0};
snprintf(abs_path, sizeof(abs_path), "%s/%s", pfile_dev_ctx->entity.dir->dir_name, file_name);
FILE *fp = fopen(abs_path, "r");
if (nullptr == fp) {
return -1;
}
size_t ret = fread(&magic, 1, sizeof(magic), fp);
if (ret != sizeof(magic)) {
return -1;
}
if (pcap_header_check(magic, fp) == 0) {
return 0;
} else if (pcapng_header_check(magic, fp) == 0) {
return 0;
}
return -1;
}
static ssize_t pcap_directory_collect_pending_files(struct pio_pcap_file_device_context *pfile_dev_ctx, struct timespec *deadline)
{
if (nullptr == pfile_dev_ctx) {
@@ -594,6 +718,11 @@ static ssize_t pcap_directory_collect_pending_files(struct pio_pcap_file_device_
continue;
}
/* ignore non pcap file */
if (validate_pcap_file(pfile_dev_ctx, dir->d_name) < 0) {
continue;
}
file_to_add = find_pending_file_to_add(pfile_dev_ctx, dir, deadline);
if (nullptr == file_to_add) {
continue;
@@ -609,14 +738,14 @@ static ssize_t pcap_directory_collect_pending_files(struct pio_pcap_file_device_
return 0;
}
static ssize_t pcap_directory_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx, uint32_t rxq_id,
struct stellar_packet **pkts, size_t nr_pkts)
static ssize_t pcap_directory_files_dispatch(struct pio_pcap_file_device_context *pfile_dev_ctx, uint32_t rxq_id)
{
ssize_t res = -1;
struct timespec deadline;
memset(&deadline, 0, sizeof(struct timespec));
get_current_timespec(&deadline);
/* the newest file which can be processed */
deadline.tv_sec = deadline.tv_sec - pfile_dev_ctx->entity.dir->delay;
@@ -632,6 +761,7 @@ static ssize_t pcap_directory_dispatch(struct pio_pcap_file_device_context *pfil
/* not open file yet */
if (nullptr == pfile_dev_ctx->entity.dir->current_file[rxq_id]) {
pthread_mutex_lock(&g_pending_file_queue.queue_mutex);
/* file_queue is empty */
if (TAILQ_EMPTY(&g_pending_file_queue.file_queue_head)) {
pthread_mutex_unlock(&g_pending_file_queue.queue_mutex);
@@ -654,24 +784,24 @@ static ssize_t pcap_directory_dispatch(struct pio_pcap_file_device_context *pfil
res = strncpy_safe(pfile_info->file_name, current_file->file_name, sizeof(pfile_info->file_name));
if (res < 0) {
log_error(ST_ERR_STR_COPY, "pfile_info file name copy failed.");
FREE(current_file);
FREE(pfile_info);
FREE(pfile_dev_ctx->entity.dir->pending_file[rxq_id]);
log_error(ST_ERR_STR_COPY, "pfile_info file name copy failed.");
return -1;
}
pfile_info->shared = &pfile_dev_ctx->shared;
if (init_pcap_file(pfile_info) < 0) {
log_error(ST_ERR_PIO_PCAP_FILE_DEVICE, "init_pcap_file failed.");
FREE(current_file);
FREE(pfile_info);
FREE(pfile_dev_ctx->entity.dir->pending_file[rxq_id]);
log_error(ST_ERR_PIO_PCAP_FILE_DEVICE, "init_pcap_file failed.");
return -1;
} else {
} else {
pfile_dev_ctx->entity.dir->current_file[rxq_id] = pfile_info;
res = pcap_directory_file_dispatch(pfile_dev_ctx, rxq_id, pkts, nr_pkts);
res = pcap_directory_file_pkts_dispatch(pfile_dev_ctx, rxq_id);
if (res < 0) {
FREE(current_file);
FREE(pfile_dev_ctx->entity.dir->pending_file[rxq_id]);
return -1;
}
log_info("processed file %s, processed up to %" PRIuMAX,
@@ -684,13 +814,14 @@ static ssize_t pcap_directory_dispatch(struct pio_pcap_file_device_context *pfil
if (res == 0) { // reach the end of the file
cleanup_pcap_plain_file_info(pfile_dev_ctx->entity.dir->current_file[rxq_id]);
FREE(pfile_dev_ctx->entity.dir->current_file[rxq_id]);
FREE(current_file);
FREE(pfile_dev_ctx->entity.dir->pending_file[rxq_id]);
}
}
} else {
/* file has been opened */
res = pcap_directory_file_dispatch(pfile_dev_ctx, rxq_id, pkts, nr_pkts);
res = pcap_directory_file_pkts_dispatch(pfile_dev_ctx, rxq_id);
if (res < 0) {
FREE(pfile_dev_ctx->entity.dir->pending_file[rxq_id]);
return -1;
} else if (res == 0) {
cleanup_pcap_plain_file_info(pfile_dev_ctx->entity.dir->current_file[rxq_id]);
@@ -718,10 +849,20 @@ ssize_t pio_pcap_file_device_receive(struct packet_io_device *pdev, uint32_t rxq
ssize_t res = -1;
if (pfile_dev_ctx->is_dir == 0) {
log_info("Start reading file:%s", pfile_dev_ctx->entity.file->file_name);
res = pcap_file_dispatch(pfile_dev_ctx, rxq_id, pkts, nr_pkts);
/* read pcap file and dispatch pkts to pkt_queue */
pcap_file_pkts_dispatch(pfile_dev_ctx);
/* get pkts from pkt_queue */
res = get_pcap_file_pkts(pfile_dev_ctx, rxq_id, pkts, nr_pkts);
} else {
log_info("Start reading directory:%s", pfile_dev_ctx->entity.dir->dir_name);
res = pcap_directory_dispatch(pfile_dev_ctx, rxq_id, pkts, nr_pkts);
/* read directory pcap files and dispatch pkts to pkt_queue */
pcap_directory_files_dispatch(pfile_dev_ctx, rxq_id);
/* get pkts from pkt_queue[rxq_id] */
res = get_pcap_file_pkts(pfile_dev_ctx, rxq_id, pkts, nr_pkts);
}
return res;