Merging upstream version 1.2.1.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

13 files changed, 869 insertions, 188 deletions

diff --git a/src/input/fpcap.c b/src/input/fpcap.c
index de054ec..6804608 100644
--- a/src/input/fpcap.c
+++ b/src/input/fpcap.c
@@ -24,6 +24,7 @@
 #include "core/assert.h"
 #include "core/object/pcap.h"
 
+#include <fcntl.h>
 #include <stdio.h>
 #ifdef HAVE_ENDIAN_H
 #include <endian.h>
@@ -60,7 +61,7 @@ static input_fpcap_t _defaults = {
     0, 0,
     0, 0, 0,
     CORE_OBJECT_PCAP_INIT(0),
-    0, 0, 0, 0, 0,
+    0, 0, 0, 0, 0, 0,
     0, 0, 0, 0, 0, 0, 0,
     0
 };
@@ -91,6 +92,16 @@ static int _open(input_fpcap_t* self)
 {
     mlassert_self();
 
+#if _POSIX_C_SOURCE >= 200112L || defined(__FreeBSD__)
+    if (self->use_fadvise) {
+        int err = posix_fadvise(fileno((FILE*)self->file), 0, 0, POSIX_FADV_SEQUENTIAL);
+        if (err) {
+            lcritical("posix_fadvise() failed: %s", core_log_errstr(err));
+            return -2;
+        }
+    }
+#endif
+
     if (fread(&self->magic_number, 1, 4, self->file) != 4
         || fread(&self->version_major, 1, 2, self->file) != 2
         || fread(&self->version_minor, 1, 2, self->file) != 2
diff --git a/src/input/fpcap.h b/src/input/fpcap.h
index da3596c..4e1e5ee 100644
--- a/src/input/fpcap.h
+++ b/src/input/fpcap.h
@@ -18,14 +18,14 @@
  * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "core/log.h"
-#include "core/receiver.h"
-#include "core/producer.h"
-#include "core/object/pcap.h"
+#include <dnsjit/core/log.h>
+#include <dnsjit/core/receiver.h>
+#include <dnsjit/core/producer.h>
+#include <dnsjit/core/object/pcap.h>
 
 #ifndef __dnsjit_input_fpcap_h
 #define __dnsjit_input_fpcap_h
 
-#include "input/fpcap.hh"
+#include <dnsjit/input/fpcap.hh>
 
 #endif
diff --git a/src/input/fpcap.hh b/src/input/fpcap.hh
index 0ced83c..ab06a71 100644
--- a/src/input/fpcap.hh
+++ b/src/input/fpcap.hh
@@ -35,7 +35,7 @@ typedef struct input_fpcap {
     core_object_pcap_t prod_pkt;
 
     void*    file;
-    int      extern_file;
+    int      extern_file, use_fadvise;
     size_t   pkts;
     uint8_t* buf;
     size_t   buf_size;
diff --git a/src/input/fpcap.lua b/src/input/fpcap.lua
index 2fc2906..b50fc11 100644
--- a/src/input/fpcap.lua
+++ b/src/input/fpcap.lua
@@ -100,6 +100,15 @@ function Fpcap:produce()
     return C.input_fpcap_producer(self.obj), self.obj
 end
 
+-- Use
+-- .B posix_fadvise()
+-- to indicate sequential reading (if supported), may increase performance.
+-- MUST be called before
+-- .BR open() .
+function Fpcap:fadvise_sequential()
+    self.obj.use_fadvise = 1
+end
+
 -- Open a PCAP file for processing and read the PCAP header.
 -- Returns 0 on success.
 function Fpcap:open(file)
diff --git a/src/input/mmpcap.h b/src/input/mmpcap.h
index 0873858..b4ae09a 100644
--- a/src/input/mmpcap.h
+++ b/src/input/mmpcap.h
@@ -18,14 +18,14 @@
  * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "core/log.h"
-#include "core/receiver.h"
-#include "core/producer.h"
-#include "core/object/pcap.h"
+#include <dnsjit/core/log.h>
+#include <dnsjit/core/receiver.h>
+#include <dnsjit/core/producer.h>
+#include <dnsjit/core/object/pcap.h>
 
 #ifndef __dnsjit_input_mmpcap_h
 #define __dnsjit_input_mmpcap_h
 
-#include "input/mmpcap.hh"
+#include <dnsjit/input/mmpcap.hh>
 
 #endif
diff --git a/src/input/pcap.h b/src/input/pcap.h
index 0a53de5..cba5227 100644
--- a/src/input/pcap.h
+++ b/src/input/pcap.h
@@ -18,16 +18,16 @@
  * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "core/log.h"
-#include "core/receiver.h"
-#include "core/producer.h"
-#include "core/object/pcap.h"
+#include <dnsjit/core/log.h>
+#include <dnsjit/core/receiver.h>
+#include <dnsjit/core/producer.h>
+#include <dnsjit/core/object/pcap.h>
 
 #ifndef __dnsjit_input_pcap_h
 #define __dnsjit_input_pcap_h
 
 #include <pcap/pcap.h>
 
-#include "input/pcap.hh"
+#include <dnsjit/input/pcap.hh>
 
 #endif
diff --git a/src/input/zero.c b/src/input/zero.c
deleted file mode 100644
index 8bb1cf6..0000000
--- a/src/input/zero.c
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * Copyright (c) 2018-2021, OARC, Inc.
- * All rights reserved.
- *
- * This file is part of dnsjit.
- *
- * dnsjit is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * dnsjit 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
- * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-#include "config.h"
-
-#include "input/zero.h"
-#include "core/assert.h"
-#include "core/object/null.h"
-
-#include <time.h>
-
-static core_log_t   _log      = LOG_T_INIT("input.zero");
-static input_zero_t _defaults = {
-    LOG_T_INIT_OBJ("input.zero"),
-    0,
-    0,
-};
-
-static core_object_null_t _null = CORE_OBJECT_NULL_INIT(0);
-
-core_log_t* input_zero_log()
-{
-    return &_log;
-}
-
-void input_zero_init(input_zero_t* self)
-{
-    mlassert_self();
-
-    *self = _defaults;
-}
-
-void input_zero_destroy(input_zero_t* self)
-{
-    mlassert_self();
-}
-
-void input_zero_run(input_zero_t* self, uint64_t num)
-{
-    mlassert_self();
-    if (!self->recv) {
-        lfatal("no receiver set");
-    }
-
-    while (num--) {
-        self->recv(self->ctx, (core_object_t*)&_null);
-    }
-}
-
-static const core_object_t* _produce(void* ctx)
-{
-    return (core_object_t*)&_null;
-}
-
-core_producer_t input_zero_producer()
-{
-    return _produce;
-}
diff --git a/src/input/zero.hh b/src/input/zero.hh
deleted file mode 100644
index 9373c90..0000000
--- a/src/input/zero.hh
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Copyright (c) 2018-2021, OARC, Inc.
- * All rights reserved.
- *
- * This file is part of dnsjit.
- *
- * dnsjit is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * dnsjit 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
- * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-//lua:require("dnsjit.core.log")
-//lua:require("dnsjit.core.receiver_h")
-//lua:require("dnsjit.core.producer_h")
-
-typedef struct input_zero {
-    core_log_t      _log;
-    core_receiver_t recv;
-    void*           ctx;
-} input_zero_t;
-
-core_log_t* input_zero_log();
-
-void input_zero_init(input_zero_t* self);
-void input_zero_destroy(input_zero_t* self);
-void input_zero_run(input_zero_t* self, uint64_t num);
-
-core_producer_t input_zero_producer();
diff --git a/src/input/zero.lua b/src/input/zero.lua
index 1907dce..c6bf5a4 100644
--- a/src/input/zero.lua
+++ b/src/input/zero.lua
@@ -17,59 +17,15 @@
 -- along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
 
 -- dnsjit.input.zero
--- Generate empty objects (/dev/zero)
---   local input = require("dnsjit.input.zero").new()
---   input:receiver(filter_or_output)
---   input:run(1e6)
+-- Dummy layer to example.input.zero
 --
--- Input module for generating empty
--- .I core.object.null
--- objects, mostly used for testing.
+-- This module has moved to example.input.zero, see examples/modules/input-example in
+-- dnsjit source repository.
 module(...,package.seeall)
 
-require("dnsjit.input.zero_h")
-local ffi = require("ffi")
-local C = ffi.C
-
-local t_name = "input_zero_t"
-local input_zero_t = ffi.typeof(t_name)
-local Zero = {}
-
--- Create a new Zero input.
-function Zero.new()
-    local self = {
-        _receiver = nil,
-        obj = input_zero_t(),
-    }
-    C.input_zero_init(self.obj)
-    ffi.gc(self.obj, C.input_zero_destroy)
-    return setmetatable(self, { __index = Zero })
-end
-
--- Return the Log object to control logging of this instance or module.
-function Zero:log()
-    if self == nil then
-        return C.input_zero_log()
-    end
-    return self.obj._log
-end
-
--- Set the receiver to pass objects to.
-function Zero:receiver(o)
-    self.obj.recv, self.obj.ctx = o:receive()
-    self._receiver = o
-end
-
--- Return the C functions and context for producing objects.
-function Zero:produce()
-    return C.input_zero_producer(), self.obj
-end
-
--- Generate
--- .I num
--- empty objects and send them to the receiver.
-function Zero:run(num)
-    C.input_zero_run(self.obj, num)
+ok, cls = pcall(require, "example.input.zero")
+if not ok then
+    error("You need to install the example module input-example\n" .. cls)
 end
 
-return Zero
+return cls
diff --git a/src/input/zpcap.c b/src/input/zpcap.c
new file mode 100644
index 0000000..11eab9a
--- /dev/null
+++ b/src/input/zpcap.c
@@ -0,0 +1,582 @@
+/*
+ * Copyright (c) 2018-2021, OARC, Inc.
+ * All rights reserved.
+ *
+ * This file is part of dnsjit.
+ *
+ * dnsjit is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * dnsjit 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
+ * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "config.h"
+
+#include "input/zpcap.h"
+#include "core/assert.h"
+#include "core/object/pcap.h"
+
+#include <fcntl.h>
+#include <stdio.h>
+#ifdef HAVE_ENDIAN_H
+#include <endian.h>
+#else
+#ifdef HAVE_SYS_ENDIAN_H
+#include <sys/endian.h>
+#else
+#ifdef HAVE_MACHINE_ENDIAN_H
+#include <machine/endian.h>
+#endif
+#endif
+#endif
+#ifdef HAVE_BYTESWAP_H
+#include <byteswap.h>
+#endif
+#ifndef bswap_16
+#ifndef bswap16
+#define bswap_16(x) swap16(x)
+#define bswap_32(x) swap32(x)
+#define bswap_64(x) swap64(x)
+#else
+#define bswap_16(x) bswap16(x)
+#define bswap_32(x) bswap32(x)
+#define bswap_64(x) bswap64(x)
+#endif
+#endif
+#include <pcap/pcap.h>
+#include <string.h>
+
+#ifdef HAVE_LZ4
+#include <lz4frame.h>
+struct _lz4_ctx {
+    LZ4F_dctx*               ctx;
+    LZ4F_decompressOptions_t opts;
+};
+#define lz4 ((struct _lz4_ctx*)self->comp_ctx)
+#endif
+#ifdef HAVE_ZSTD
+#include <zstd.h>
+struct _zstd_ctx {
+    ZSTD_DCtx*     ctx;
+    ZSTD_inBuffer  in;
+    ZSTD_outBuffer out;
+};
+#define zstd ((struct _zstd_ctx*)self->comp_ctx)
+#endif
+
+#define MAX_SNAPLEN 0x40000
+
+static core_log_t    _log      = LOG_T_INIT("input.zpcap");
+static input_zpcap_t _defaults = {
+    LOG_T_INIT_OBJ("input.zpcap"),
+    0, 0,
+    0, 0, 0,
+    CORE_OBJECT_PCAP_INIT(0),
+    input_zpcap_type_none, 0,
+    0, 0, 0, 0, 0, 0, 0, 0,
+    0, 0, 0, 0, 0, 0,
+    0, 0, 0, 0, 0, 0, 0,
+    0
+};
+
+core_log_t* input_zpcap_log()
+{
+    return &_log;
+}
+
+void input_zpcap_init(input_zpcap_t* self)
+{
+    mlassert_self();
+
+    *self = _defaults;
+}
+
+void input_zpcap_destroy(input_zpcap_t* self)
+{
+    mlassert_self();
+
+    switch (self->compression) {
+#ifdef HAVE_LZ4
+    case input_zpcap_type_lz4: {
+        LZ4F_errorCode_t code;
+
+        if (lz4) {
+            if (lz4->ctx && (code = LZ4F_freeDecompressionContext(lz4->ctx))) {
+                lfatal("LZ4F_freeDecompressionContext() failed: %s", LZ4F_getErrorName(code));
+            }
+        }
+        free(lz4);
+        free(self->in);
+        free(self->out);
+        break;
+    }
+#endif
+#ifdef HAVE_ZSTD
+    case input_zpcap_type_zstd:
+        if (zstd) {
+            if (zstd->ctx) {
+                ZSTD_freeDCtx(zstd->ctx);
+            }
+        }
+        free(zstd);
+        free(self->in);
+        free(self->out);
+        break;
+#endif
+    default:
+        break;
+    }
+
+    if (!self->extern_file && self->file) {
+        fclose(self->file);
+    }
+    free(self->buf);
+}
+
+static ssize_t _read(input_zpcap_t* self, void* dst, size_t len, void** dstp)
+{
+    switch (self->compression) {
+#ifdef HAVE_LZ4
+    case input_zpcap_type_lz4: {
+        size_t need = len;
+
+        if (dstp && self->out_have > need) {
+            *dstp = self->out + self->out_at;
+            self->out_have -= need;
+            self->out_at += need;
+            return len;
+        }
+
+        for (;;) {
+            if (self->out_have > need) {
+                memcpy(dst, self->out + self->out_at, need);
+                self->out_have -= need;
+                self->out_at += need;
+                return len;
+            }
+
+            memcpy(dst, self->out + self->out_at, self->out_have);
+            need -= self->out_have;
+            dst += self->out_have;
+
+            ssize_t n = fread(self->in + self->in_at, 1, self->in_size - self->in_have, self->file);
+            if (n < 0) {
+                return n;
+            }
+            self->in_at += n;
+            self->in_have += n;
+            if (!self->in_have) {
+                return 0;
+            }
+
+            size_t dst_size = self->out_size, src_size = self->in_have;
+            size_t code = LZ4F_decompress(lz4->ctx, self->out, &dst_size, self->in, &src_size, &lz4->opts);
+            if (LZ4F_isError(code)) {
+                lfatal("LZ4F_decompress() failed: %s", LZ4F_getErrorName(code));
+            }
+
+            if (src_size < self->in_have) {
+                self->in_have -= src_size;
+                memmove(self->in, self->in + src_size, self->in_have);
+                self->in_at = self->in_have;
+            } else {
+                self->in_at   = 0;
+                self->in_have = 0;
+            }
+
+            self->out_at   = 0;
+            self->out_have = dst_size;
+        }
+    }
+#endif
+#ifdef HAVE_ZSTD
+    case input_zpcap_type_zstd: {
+        size_t need = len;
+
+        if (dstp && self->out_have > need) {
+            *dstp = self->out + self->out_at;
+            self->out_have -= need;
+            self->out_at += need;
+            return len;
+        }
+
+        for (;;) {
+            if (self->out_have > need) {
+                memcpy(dst, self->out + self->out_at, need);
+                self->out_have -= need;
+                self->out_at += need;
+                return len;
+            }
+
+            memcpy(dst, self->out + self->out_at, self->out_have);
+            need -= self->out_have;
+            dst += self->out_have;
+
+            if (zstd->in.pos >= zstd->in.size) {
+                ssize_t n = fread(self->in, 1, self->in_size, self->file);
+                if (n < 1) {
+                    return n;
+                }
+                zstd->in.size = n;
+                zstd->in.pos  = 0;
+            }
+
+            zstd->out.size = self->out_size;
+            zstd->out.pos  = 0;
+            size_t code    = ZSTD_decompressStream(zstd->ctx, &zstd->out, &zstd->in);
+            if (ZSTD_isError(code)) {
+                lfatal("ZSTD_decompressStream() failed: %s", ZSTD_getErrorName(code));
+            }
+
+            self->out_have = zstd->out.pos;
+            self->out_at   = 0;
+        }
+    }
+#endif
+    default:
+        return 0;
+    }
+}
+
+static int _open(input_zpcap_t* self)
+{
+    mlassert_self();
+
+#if _POSIX_C_SOURCE >= 200112L || defined(__FreeBSD__)
+    if (self->use_fadvise) {
+        int err = posix_fadvise(fileno((FILE*)self->file), 0, 0, POSIX_FADV_SEQUENTIAL);
+        if (err) {
+            lcritical("posix_fadvise() failed: %s", core_log_errstr(err));
+            return -2;
+        }
+    }
+#endif
+
+    switch (self->compression) {
+#ifdef HAVE_LZ4
+    case input_zpcap_type_lz4: {
+        LZ4F_errorCode_t code;
+
+        if (lz4) {
+            if (lz4->ctx && (code = LZ4F_freeDecompressionContext(lz4->ctx))) {
+                lfatal("LZ4F_freeDecompressionContext() failed: %s", LZ4F_getErrorName(code));
+            }
+        }
+        free(lz4);
+        free(self->in);
+        free(self->out);
+
+        lfatal_oom(self->comp_ctx = calloc(1, sizeof(struct _lz4_ctx)));
+        if ((code = LZ4F_createDecompressionContext(&lz4->ctx, LZ4F_VERSION))) {
+            lfatal("LZ4F_createDecompressionContext() failed: %s", LZ4F_getErrorName(code));
+        }
+        lz4->opts.stableDst = 1;
+
+        self->in_size = 256 * 1024;
+        lfatal_oom(self->in = malloc(self->in_size));
+        self->out_size = 256 * 1024;
+        lfatal_oom(self->out = malloc(self->out_size));
+
+        break;
+    }
+#endif
+#ifdef HAVE_ZSTD
+    case input_zpcap_type_zstd:
+        if (zstd) {
+            if (zstd->ctx) {
+                ZSTD_freeDCtx(zstd->ctx);
+            }
+        }
+        free(zstd);
+        free(self->in);
+        free(self->out);
+
+        lfatal_oom(self->comp_ctx = calloc(1, sizeof(struct _zstd_ctx)));
+        lfatal_oom(zstd->ctx = ZSTD_createDCtx());
+        self->in_size = ZSTD_DStreamInSize();
+        lfatal_oom(self->in = malloc(self->in_size));
+        self->out_size = ZSTD_DStreamOutSize();
+        lfatal_oom(self->out = malloc(self->out_size));
+
+        zstd->in.src   = self->in;
+        zstd->out.dst  = self->out;
+        zstd->out.size = self->out_size;
+        break;
+#endif
+    default:
+        lcritical("no support for selected compression");
+        return -2;
+    }
+
+    if (_read(self, &self->magic_number, 4, 0) != 4
+        || _read(self, &self->version_major, 2, 0) != 2
+        || _read(self, &self->version_minor, 2, 0) != 2
+        || _read(self, &self->thiszone, 4, 0) != 4
+        || _read(self, &self->sigfigs, 4, 0) != 4
+        || _read(self, &self->snaplen, 4, 0) != 4
+        || _read(self, &self->network, 4, 0) != 4) {
+        lcritical("could not read full PCAP header");
+        return -2;
+    }
+    switch (self->magic_number) {
+    case 0x4d3cb2a1:
+        self->is_nanosec = 1;
+    case 0xd4c3b2a1:
+        self->is_swapped    = 1;
+        self->version_major = bswap_16(self->version_major);
+        self->version_minor = bswap_16(self->version_minor);
+        self->thiszone      = (int32_t)bswap_32((uint32_t)self->thiszone);
+        self->sigfigs       = bswap_32(self->sigfigs);
+        self->snaplen       = bswap_32(self->snaplen);
+        self->network       = bswap_32(self->network);
+        break;
+    case 0xa1b2c3d4:
+    case 0xa1b23c4d:
+        break;
+    default:
+        lcritical("invalid PCAP header");
+        return -2;
+    }
+
+    if (self->snaplen > MAX_SNAPLEN) {
+        lcritical("too large snaplen (%u)", self->snaplen);
+        return -2;
+    }
+
+    if (self->version_major != 2 || self->version_minor != 4) {
+        lcritical("unsupported PCAP version v%u.%u", self->version_major, self->version_minor);
+        return -2;
+    }
+
+    /*
+     * Translation taken from https://github.com/the-tcpdump-group/libpcap/blob/90543970fd5fbed261d3637f5ec4811d7dde4e49/pcap-common.c#L1212 .
+     */
+    switch (self->network) {
+    case 101:
+        self->linktype = DLT_RAW;
+        break;
+#ifdef DLT_FR
+    case 107: /* LINKTYPE_FRELAY */
+        self->linktype = DLT_FR;
+        break;
+#endif
+    case 100: /* LINKTYPE_ATM_RFC1483 */
+        self->linktype = DLT_ATM_RFC1483;
+        break;
+    case 102: /* LINKTYPE_SLIP_BSDOS */
+        self->linktype = DLT_SLIP_BSDOS;
+        break;
+    case 103: /* LINKTYPE_PPP_BSDOS */
+        self->linktype = DLT_PPP_BSDOS;
+        break;
+    case 104: /* LINKTYPE_C_HDLC */
+        self->linktype = DLT_C_HDLC;
+        break;
+    case 106: /* LINKTYPE_ATM_CLIP */
+        self->linktype = DLT_ATM_CLIP;
+        break;
+    case 50: /* LINKTYPE_PPP_HDLC */
+        self->linktype = DLT_PPP_SERIAL;
+        break;
+    case 51: /* LINKTYPE_PPP_ETHER */
+        self->linktype = DLT_PPP_ETHER;
+        break;
+    default:
+        self->linktype = self->network;
+    }
+
+    lfatal_oom(self->buf = malloc(self->snaplen));
+    self->prod_pkt.snaplen    = self->snaplen;
+    self->prod_pkt.linktype   = self->linktype;
+    self->prod_pkt.is_swapped = self->is_swapped;
+
+    ldebug("pcap v%u.%u snaplen:%lu %s", self->version_major, self->version_minor, self->snaplen, self->is_swapped ? " swapped" : "");
+
+    return 0;
+}
+
+int input_zpcap_open(input_zpcap_t* self, const char* file)
+{
+    mlassert_self();
+    lassert(file, "file is nil");
+
+    if (self->file) {
+        lfatal("already opened");
+    }
+
+    if (!(self->file = fopen(file, "rb"))) {
+        lcritical("fopen(%s) error: %s", file, core_log_errstr(errno));
+        return -1;
+    }
+
+    return _open(self);
+}
+
+int input_zpcap_openfp(input_zpcap_t* self, void* fp)
+{
+    mlassert_self();
+
+    if (self->file) {
+        lfatal("already opened");
+    }
+
+    self->file        = fp;
+    self->extern_file = 1;
+
+    return _open(self);
+}
+
+int input_zpcap_run(input_zpcap_t* self)
+{
+    struct {
+        uint32_t ts_sec;
+        uint32_t ts_usec;
+        uint32_t incl_len;
+        uint32_t orig_len;
+    } hdr;
+    core_object_pcap_t pkt = CORE_OBJECT_PCAP_INIT(0);
+    int                ret;
+    mlassert_self();
+
+    if (!self->file) {
+        lfatal("no PCAP opened");
+    }
+    if (!self->recv) {
+        lfatal("no receiver set");
+    }
+
+    pkt.snaplen    = self->snaplen;
+    pkt.linktype   = self->linktype;
+    pkt.is_swapped = self->is_swapped;
+
+    while ((ret = _read(self, &hdr, 16, 0)) == 16) {
+        if (self->is_swapped) {
+            hdr.ts_sec   = bswap_32(hdr.ts_sec);
+            hdr.ts_usec  = bswap_32(hdr.ts_usec);
+            hdr.incl_len = bswap_32(hdr.incl_len);
+            hdr.orig_len = bswap_32(hdr.orig_len);
+        }
+        if (hdr.incl_len > self->snaplen) {
+            lwarning("invalid packet length, larger then snaplen");
+            return -1;
+        }
+        pkt.bytes = (unsigned char*)self->buf;
+        if (_read(self, self->buf, hdr.incl_len, (void**)&pkt.bytes) != hdr.incl_len) {
+            lwarning("could not read all of packet, aborting");
+            return -1;
+        }
+
+        self->pkts++;
+
+        pkt.ts.sec = hdr.ts_sec;
+        if (self->is_nanosec) {
+            pkt.ts.nsec = hdr.ts_usec;
+        } else {
+            pkt.ts.nsec = hdr.ts_usec * 1000;
+        }
+        pkt.caplen = hdr.incl_len;
+        pkt.len    = hdr.orig_len;
+
+        self->recv(self->ctx, (core_object_t*)&pkt);
+    }
+    if (ret) {
+        lwarning("could not read next PCAP header, aborting");
+        return -1;
+    }
+
+    return 0;
+}
+
+int input_zpcap_have_support(input_zpcap_t* self)
+{
+    mlassert_self();
+
+    switch (self->compression) {
+#ifdef HAVE_LZ4
+    case input_zpcap_type_lz4:
+        return 1;
+#endif
+#ifdef HAVE_ZSTD
+    case input_zpcap_type_zstd:
+        return 1;
+#endif
+    default:
+        break;
+    }
+    return 0;
+}
+
+static const core_object_t* _produce(input_zpcap_t* self)
+{
+    struct {
+        uint32_t ts_sec;
+        uint32_t ts_usec;
+        uint32_t incl_len;
+        uint32_t orig_len;
+    } hdr;
+    int ret;
+    mlassert_self();
+
+    if (self->is_broken) {
+        lwarning("PCAP is broken, will not read next packet");
+        return 0;
+    }
+
+    if ((ret = _read(self, &hdr, 16, 0)) != 16) {
+        if (ret) {
+            lwarning("could not read next PCAP header, aborting");
+            self->is_broken = 1;
+        }
+        return 0;
+    }
+
+    if (self->is_swapped) {
+        hdr.ts_sec   = bswap_32(hdr.ts_sec);
+        hdr.ts_usec  = bswap_32(hdr.ts_usec);
+        hdr.incl_len = bswap_32(hdr.incl_len);
+        hdr.orig_len = bswap_32(hdr.orig_len);
+    }
+    if (hdr.incl_len > self->snaplen) {
+        lwarning("invalid packet length, larger then snaplen");
+        self->is_broken = 1;
+        return 0;
+    }
+    self->prod_pkt.bytes = (unsigned char*)self->buf;
+    if (_read(self, self->buf, hdr.incl_len, (void**)&self->prod_pkt.bytes) != hdr.incl_len) {
+        lwarning("could not read all of packet, aborting");
+        self->is_broken = 1;
+        return 0;
+    }
+
+    self->pkts++;
+
+    self->prod_pkt.ts.sec = hdr.ts_sec;
+    if (self->is_nanosec) {
+        self->prod_pkt.ts.nsec = hdr.ts_usec;
+    } else {
+        self->prod_pkt.ts.nsec = hdr.ts_usec * 1000;
+    }
+    self->prod_pkt.caplen = hdr.incl_len;
+    self->prod_pkt.len    = hdr.orig_len;
+
+    return (core_object_t*)&self->prod_pkt;
+}
+
+core_producer_t input_zpcap_producer(input_zpcap_t* self)
+{
+    mlassert_self();
+
+    if (!self->file) {
+        lfatal("no PCAP opened");
+    }
+
+    return (core_producer_t)_produce;
+}
diff --git a/src/input/zero.h b/src/input/zpcap.h
index 882525a..7f6f304 100644
--- a/src/input/zero.h
+++ b/src/input/zpcap.h
@@ -18,15 +18,14 @@
  * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "core/log.h"
-#include "core/receiver.h"
-#include "core/producer.h"
+#include <dnsjit/core/log.h>
+#include <dnsjit/core/receiver.h>
+#include <dnsjit/core/producer.h>
+#include <dnsjit/core/object/pcap.h>
 
-#ifndef __dnsjit_input_zero_h
-#define __dnsjit_input_zero_h
+#ifndef __dnsjit_input_zpcap_h
+#define __dnsjit_input_zpcap_h
 
-#include <stdint.h>
-
-#include "input/zero.hh"
+#include <dnsjit/input/zpcap.hh>
 
 #endif
diff --git a/src/input/zpcap.hh b/src/input/zpcap.hh
new file mode 100644
index 0000000..bc013fa
--- /dev/null
+++ b/src/input/zpcap.hh
@@ -0,0 +1,77 @@
+/*
+ * Copyright (c) 2018-2021, OARC, Inc.
+ * All rights reserved.
+ *
+ * This file is part of dnsjit.
+ *
+ * dnsjit is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * dnsjit 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
+ * along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+//lua:require("dnsjit.core.log")
+//lua:require("dnsjit.core.receiver_h")
+//lua:require("dnsjit.core.producer_h")
+//lua:require("dnsjit.core.object.pcap_h")
+
+typedef enum input_zpcap_type {
+    input_zpcap_type_none,
+    input_zpcap_type_lz4,
+    input_zpcap_type_zstd
+} input_zpcap_type_t;
+
+typedef struct input_zpcap {
+    core_log_t      _log;
+    core_receiver_t recv;
+    void*           ctx;
+
+    uint8_t is_swapped;
+    uint8_t is_nanosec;
+    uint8_t is_broken;
+
+    core_object_pcap_t prod_pkt;
+
+    input_zpcap_type_t compression;
+    void*              comp_ctx;
+
+    void * in, *out;
+    size_t in_size, out_size;
+    size_t in_have, out_have;
+    size_t in_at, out_at;
+
+    void*    file;
+    int      extern_file, use_fadvise;
+    size_t   pkts;
+    uint8_t* buf;
+    size_t   buf_size;
+
+    uint32_t magic_number;
+    uint16_t version_major;
+    uint16_t version_minor;
+    int32_t  thiszone;
+    uint32_t sigfigs;
+    uint32_t snaplen;
+    uint32_t network;
+
+    uint32_t linktype;
+} input_zpcap_t;
+
+core_log_t* input_zpcap_log();
+
+void input_zpcap_init(input_zpcap_t* self);
+void input_zpcap_destroy(input_zpcap_t* self);
+int  input_zpcap_open(input_zpcap_t* self, const char* file);
+int  input_zpcap_openfp(input_zpcap_t* self, void* fp);
+int  input_zpcap_run(input_zpcap_t* self);
+int  input_zpcap_have_support(input_zpcap_t* self);
+
+core_producer_t input_zpcap_producer(input_zpcap_t* self);
diff --git a/src/input/zpcap.lua b/src/input/zpcap.lua
new file mode 100644
index 0000000..9a0230f
--- /dev/null
+++ b/src/input/zpcap.lua
@@ -0,0 +1,159 @@
+-- Copyright (c) 2018-2021, OARC, Inc.
+-- All rights reserved.
+--
+-- This file is part of dnsjit.
+--
+-- dnsjit is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU General Public License as published by
+-- the Free Software Foundation, either version 3 of the License, or
+-- (at your option) any later version.
+--
+-- dnsjit 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
+-- along with dnsjit.  If not, see <http://www.gnu.org/licenses/>.
+
+-- dnsjit.input.zpcap
+-- Read input from a PCAP file that is compressed
+--   local input = require("dnsjit.input.zpcap").new()
+--   input:zstd()
+--   input:open("file.pcap.zst")
+--   input:receiver(filter_or_output)
+--   input:run()
+--
+-- Read input from a PCAP file that is compressed and parse the PCAP without
+-- libpcap.
+-- After opening a file and reading the PCAP header, the attributes are
+-- populated.
+-- .SS Attributes
+-- .TP
+-- is_swapped
+-- Indicate if the byte order in the PCAP is in reverse order of the host.
+-- .TP
+-- is_nanosec
+-- Indicate if the time stamps are in nanoseconds or not.
+-- .TP
+-- magic_number
+-- Magic number.
+-- .TP
+-- version_major
+-- Major version number.
+-- .TP
+-- version_minor
+-- Minor version number.
+-- .TP
+-- thiszone
+-- GMT to local correction.
+-- .TP
+-- sigfigs
+-- Accuracy of timestamps.
+-- .TP
+-- snaplen
+-- Max length of captured packets, in octets.
+-- .TP
+-- network
+-- The link type found in the PCAP header, see https://www.tcpdump.org/linktypes.html .
+-- .TP
+-- linktype
+-- The data link type, mapped from
+-- .IR network .
+module(...,package.seeall)
+
+require("dnsjit.input.zpcap_h")
+local ffi = require("ffi")
+local C = ffi.C
+
+local t_name = "input_zpcap_t"
+local input_zpcap_t = ffi.typeof(t_name)
+local Zpcap = {}
+
+-- Create a new Zpcap input.
+function Zpcap.new()
+    local self = {
+        _receiver = nil,
+        obj = input_zpcap_t(),
+    }
+    C.input_zpcap_init(self.obj)
+    ffi.gc(self.obj, C.input_zpcap_destroy)
+    return setmetatable(self, { __index = Zpcap })
+end
+
+-- Return the Log object to control logging of this instance or module.
+function Zpcap:log()
+    if self == nil then
+        return C.input_zpcap_log()
+    end
+    return self.obj._log
+end
+
+-- Set the receiver to pass objects to.
+function Zpcap:receiver(o)
+    self.obj.recv, self.obj.ctx = o:receive()
+    self._receiver = o
+end
+
+-- Return the C functions and context for producing objects.
+function Zpcap:produce()
+    return C.input_zpcap_producer(self.obj), self.obj
+end
+
+-- Use
+-- .B posix_fadvise()
+-- to indicate sequential reading (if supported), may increase performance.
+-- MUST be called before
+-- .BR open() .
+function Zpcap:fadvise_sequential()
+    self.obj.use_fadvise = 1
+end
+
+-- Use liblz4 to decompress the input file/data.
+function Zpcap:lz4()
+    self.obj.compression = "input_zpcap_type_lz4"
+end
+
+-- Use libzstd to decompress the input file/data.
+function Zpcap:zstd()
+    self.obj.compression = "input_zpcap_type_zstd"
+end
+
+-- Return true if support for selected compression library is built in.
+function Zpcap:have_support()
+    if C.input_zpcap_have_support(self.obj) == 1 then
+        return true
+    end
+    return false
+end
+
+-- Open a PCAP file for processing and read the PCAP header.
+-- Returns 0 on success.
+function Zpcap:open(file)
+    return C.input_zpcap_open(self.obj, file)
+end
+
+-- Open a PCAP file for processing and read the PCAP header using a
+-- file descriptor, for example
+-- .B io.stdin
+-- or with
+-- .BR io.open() .
+-- Will not take ownership of the file descriptor.
+-- Returns 0 on success.
+function Zpcap:openfp(fp)
+    return C.input_zpcap_openfp(self.obj, fp)
+end
+
+-- Start processing packets and send each packet read to the receiver.
+-- Returns 0 if all packets was read successfully.
+function Zpcap:run()
+    return C.input_zpcap_run(self.obj)
+end
+
+-- Return the number of packets seen.
+function Zpcap:packets()
+    return tonumber(self.obj.pkts)
+end
+
+-- dnsjit.input.fpcap (3)
+return Zpcap