Merge pull request #1 from AkihiroSuda/suda/wip

Avoid golang/go#14210 (Go 1.6 panic: runtime error: cgo argument has Go pointer to Go pointer)

Author: AkihiroSuda

.travis.yml

@@ -1,6 +1,8 @@
 language: go
 
 go:
+  - tip
+  - 1.6
   - 1.5
 
 before_install:

netfilter.go

@@ -32,9 +32,12 @@ package netfilter
 import "C"
 
 import (
+	"fmt"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
-	"fmt"
+	"os"
+	"sync"
+	"time"
 	"unsafe"
 )
 
@@ -61,6 +64,7 @@ type NFQueue struct {
 	qh      *C.struct_nfq_q_handle
 	fd      C.int
 	packets chan NFPacket
+	idx     uint32
 }
 
 //Verdict for a packet
@@ -79,6 +83,9 @@ const (
 	NF_DEFAULT_PACKET_SIZE uint32 = 0xffff
 )
 
+var theTable = make(map[uint32]*chan NFPacket, 0)
+var theTabeLock sync.RWMutex
+
 //Create and bind to queue specified by queueId
 func NewNFQueue(queueId uint16, maxPacketsInQueue uint32, packetSize uint32) (*NFQueue, error) {
 	var nfq = NFQueue{}
@@ -98,7 +105,11 @@ func NewNFQueue(queueId uint16, maxPacketsInQueue uint32, packetSize uint32) (*N
 	}
 
 	nfq.packets = make(chan NFPacket)
-	if nfq.qh, err = C.CreateQueue(nfq.h, C.u_int16_t(queueId), unsafe.Pointer(&nfq.packets)); err != nil || nfq.qh == nil {
+	nfq.idx = uint32(time.Now().UnixNano())
+	theTabeLock.Lock()
+	theTable[nfq.idx] = &nfq.packets
+	theTabeLock.Unlock()
+	if nfq.qh, err = C.CreateQueue(nfq.h, C.u_int16_t(queueId), C.u_int32_t(nfq.idx)); err != nil || nfq.qh == nil {
 		C.nfq_close(nfq.h)
 		return nil, fmt.Errorf("Error binding to queue: %v\n", err)
 	}
@@ -130,6 +141,9 @@ func NewNFQueue(queueId uint16, maxPacketsInQueue uint32, packetSize uint32) (*N
 func (nfq *NFQueue) Close() {
 	C.nfq_destroy_queue(nfq.qh)
 	C.nfq_close(nfq.h)
+	theTabeLock.Lock()
+	delete(theTable, nfq.idx)
+	theTabeLock.Unlock()
 }
 
 //Get the channel for packets
@@ -142,15 +156,23 @@ func (nfq *NFQueue) run() {
 }
 
 //export go_callback
-func go_callback(queueId C.int, data *C.uchar, len C.int, cb *chan NFPacket) Verdict {
+func go_callback(queueId C.int, data *C.uchar, len C.int, idx uint32) Verdict {
 	xdata := C.GoBytes(unsafe.Pointer(data), len)
 	packet := gopacket.NewPacket(xdata, layers.LayerTypeIPv4, gopacket.DecodeOptions{Lazy: true, NoCopy: true})
 	p := NFPacket{verdictChannel: make(chan Verdict), Packet: packet}
+	theTabeLock.RLock()
+	cb, ok := theTable[idx]
+	theTabeLock.RUnlock()
+	if !ok {
+		fmt.Fprintf(os.Stderr, "Dropping, unexpectedly due to bad idx=%d\n", idx)
+		return NF_DROP
+	}
 	select {
 	case (*cb) <- p:
 		v := <-p.verdictChannel
 		return v
 	default:
+		fmt.Fprintf(os.Stderr, "Dropping, unexpectedly due to no recv, idx=%d\n", idx)
 		return NF_DROP
 	}
 }

netfilter.h

@@ -27,27 +27,29 @@
 #include <linux/netfilter.h>
 #include <libnetfilter_queue/libnetfilter_queue.h>
 
-extern uint go_callback(int id, unsigned char* data, int len, void** cb_func);
+extern uint go_callback(int id, unsigned char* data, int len, u_int32_t idx);
 
 static int nf_callback(struct nfq_q_handle *qh, struct nfgenmsg *nfmsg, struct nfq_data *nfa, void *cb_func){
     uint32_t id = -1;
     struct nfqnl_msg_packet_hdr *ph = NULL;
     unsigned char *buffer = NULL;
     int ret = 0;
     int verdict = 0;
+    u_int32_t idx;
 
     ph = nfq_get_msg_packet_hdr(nfa);
     id = ntohl(ph->packet_id);
 
     ret = nfq_get_payload(nfa, &buffer);
-    verdict = go_callback(id, buffer, ret, cb_func);
+    idx = (uint32_t)((uintptr_t)cb_func);
+    verdict = go_callback(id, buffer, ret, idx);
 
     return nfq_set_verdict(qh, id, verdict, 0, NULL);
 }
 
-static inline struct nfq_q_handle* CreateQueue(struct nfq_handle *h, u_int16_t queue, void* cb_func)
+static inline struct nfq_q_handle* CreateQueue(struct nfq_handle *h, u_int16_t queue, u_int32_t idx)
 {
-    return nfq_create_queue(h, queue, &nf_callback, cb_func);
+  return nfq_create_queue(h, queue, &nf_callback, (void*)((uintptr_t)idx));
 }
 
 static inline void Run(struct nfq_handle *h, int fd)

netfilter_test.go

@@ -0,0 +1,39 @@
+package netfilter
+
+import (
+	"testing"
+	"time"
+)
+
+var stopCh = make(chan struct{})
+
+func serve(t *testing.T, queueNum uint16) {
+	nfq, err := NewNFQueue(queueNum, 100, NF_DEFAULT_PACKET_SIZE)
+	if err != nil {
+		t.Skipf("Skipping the test due to %s", err)
+	}
+	defer nfq.Close()
+	packets := nfq.GetPackets()
+
+	t.Logf("Starting (NFQ %d)..", queueNum)
+	for true {
+		select {
+		case p := <-packets:
+			t.Logf("Accepting %s", p.Packet)
+			p.SetVerdict(NF_ACCEPT)
+		case <-stopCh:
+			t.Logf("Exiting..")
+			return
+		}
+	}
+}
+
+// very dumb test, but enough for testing golang/go#14210
+func TestNetfilter(t *testing.T) {
+	queueNum := 42
+	go serve(t, uint16(queueNum))
+	wait := 3 * time.Second
+	t.Logf("Sleeping for %s", wait)
+	time.Sleep(wait)
+	close(stopCh)
+}