tun_darwin_test.go 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637
  1. //go:build !ios && !e2e_testing
  2. package overlay
  3. import (
  4. "bytes"
  5. "os"
  6. "sync/atomic"
  7. "syscall"
  8. "testing"
  9. "time"
  10. "github.com/slackhq/nebula/internal/msgx"
  11. "github.com/slackhq/nebula/test"
  12. "github.com/stretchr/testify/assert"
  13. "github.com/stretchr/testify/require"
  14. "golang.org/x/sys/unix"
  15. )
  16. // newSocketpairTun returns a tun writing to one end of a connected AF_UNIX datagram socketpair, which
  17. // takes sendmsg_x the way a utun control socket does, and the other end's fd for reading back.
  18. // sndbuf is the writer's send buffer, which caps the largest datagram it accepts.
  19. func newSocketpairTun(t *testing.T, sndbuf int) (*tun, int) {
  20. t.Helper()
  21. fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_DGRAM, 0)
  22. require.NoError(t, err)
  23. require.NoError(t, unix.SetsockoptInt(fds[0], unix.SOL_SOCKET, unix.SO_SNDBUF, sndbuf))
  24. require.NoError(t, unix.SetsockoptInt(fds[1], unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20))
  25. // A dropped datagram fails the read-back instead of hanging it.
  26. require.NoError(t, unix.SetsockoptTimeval(fds[1], unix.SOL_SOCKET, unix.SO_RCVTIMEO, &unix.Timeval{Sec: 2}))
  27. require.NoError(t, unix.SetNonblock(fds[0], true))
  28. f := os.NewFile(uintptr(fds[0]), "socketpair")
  29. t.Cleanup(func() {
  30. _ = f.Close()
  31. _ = unix.Close(fds[1])
  32. })
  33. return &tun{f: f, l: test.NewLogger()}, fds[1]
  34. }
  35. // testTunPkt returns packet i of size bytes, alternating IPv4 and IPv6, and the datagram the utun
  36. // device should receive for it.
  37. func testTunPkt(i, size int) (pkt, wire []byte) {
  38. pkt = make([]byte, size)
  39. pkt[0] = 0x45
  40. af := byte(syscall.AF_INET)
  41. if i%2 == 1 {
  42. pkt[0] = 0x60
  43. af = syscall.AF_INET6
  44. }
  45. pkt[1] = byte(i)
  46. pkt[2] = byte(i >> 8)
  47. return pkt, append([]byte{0, 0, 0, af}, pkt...)
  48. }
  49. // readTunBack asserts r receives exactly want, in order.
  50. func readTunBack(t *testing.T, r int, want [][]byte) {
  51. t.Helper()
  52. buf := make([]byte, 4096)
  53. for i, w := range want {
  54. n, _, err := unix.Recvfrom(r, buf, 0)
  55. require.NoError(t, err, "datagram %d", i)
  56. if !bytes.Equal(w, buf[:n]) {
  57. t.Fatalf("datagram %d: got % x, want % x", i, buf[:n], w)
  58. }
  59. }
  60. _, _, err := unix.Recvfrom(r, buf, unix.MSG_DONTWAIT)
  61. assert.ErrorIs(t, err, unix.EAGAIN, "extra datagrams delivered")
  62. }
  63. // forEachTunWritePath runs fn once over sendmsg_x and once over the per-packet writev fallback.
  64. func forEachTunWritePath(t *testing.T, fn func(t *testing.T, fallback bool)) {
  65. for _, fallback := range []bool{false, true} {
  66. name := "sendmsg_x"
  67. if fallback {
  68. name = "writev"
  69. }
  70. t.Run(name, func(t *testing.T) {
  71. noTunSendX.Store(fallback)
  72. t.Cleanup(func() { noTunSendX.Store(false) })
  73. fn(t, fallback)
  74. })
  75. }
  76. }
  77. // TestTunWriteBatch pins that WriteBatch delivers every packet whole, in order and behind the right
  78. // utun AF prefix, across several sendmsg_x calls, and returns nil.
  79. func TestTunWriteBatch(t *testing.T) {
  80. forEachTunWritePath(t, func(t *testing.T, fallback bool) {
  81. tn, r := newSocketpairTun(t, 1<<20)
  82. var pkts, want [][]byte
  83. for i := range 3*msgx.Batch + 5 {
  84. p, w := testTunPkt(i, 60+i)
  85. pkts = append(pkts, p)
  86. want = append(want, w)
  87. }
  88. require.NoError(t, tn.WriteBatch(pkts))
  89. assert.Equal(t, fallback, noTunSendX.Load())
  90. readTunBack(t, r, want)
  91. })
  92. }
  93. // TestTunWriteBatchSkipsInvalid pins that an empty packet or one with no IP version is skipped
  94. // without shifting the packets after it, and that the first such error is the one reported.
  95. func TestTunWriteBatchSkipsInvalid(t *testing.T) {
  96. forEachTunWritePath(t, func(t *testing.T, fallback bool) {
  97. tn, r := newSocketpairTun(t, 1<<20)
  98. var pkts, want [][]byte
  99. for i := range 20 {
  100. p, w := testTunPkt(i, 60)
  101. pkts = append(pkts, p)
  102. want = append(want, w)
  103. }
  104. pkts = append(pkts[:10], append([][]byte{{0x10, 1, 2}, {}}, pkts[10:]...)...)
  105. err := tn.WriteBatch(pkts)
  106. assert.ErrorContains(t, err, "IP version")
  107. readTunBack(t, r, want)
  108. })
  109. }
  110. // TestTunWriteBatchOversize pins that a packet larger than the socket's send buffer drops only
  111. // itself and is reported, whether sendmsg_x reaches it mid-call (a short count) or first (EMSGSIZE
  112. // for the whole call). The send buffer is 1024 bytes and the oversize packet 2000.
  113. func TestTunWriteBatchOversize(t *testing.T) {
  114. for _, tc := range []struct {
  115. name string
  116. at int
  117. }{
  118. {"mid-call", msgx.Batch + 6},
  119. {"first-in-call", msgx.Batch},
  120. } {
  121. t.Run(tc.name, func(t *testing.T) {
  122. forEachTunWritePath(t, func(t *testing.T, fallback bool) {
  123. tn, r := newSocketpairTun(t, 1024)
  124. var pkts, want [][]byte
  125. for i := range 2*msgx.Batch + 10 {
  126. size := 100
  127. if i == tc.at {
  128. size = 2000
  129. }
  130. p, w := testTunPkt(i, size)
  131. pkts = append(pkts, p)
  132. if i != tc.at {
  133. want = append(want, w)
  134. }
  135. }
  136. err := tn.WriteBatch(pkts)
  137. assert.ErrorIs(t, err, unix.EMSGSIZE)
  138. assert.Equal(t, fallback, noTunSendX.Load())
  139. readTunBack(t, r, want)
  140. })
  141. })
  142. }
  143. }
  144. // forEachTunReadPath runs fn once over recvmsg_x and once over the per-packet readv drain.
  145. func forEachTunReadPath(t *testing.T, fn func(t *testing.T, newQ func(*tun) *tunQueue)) {
  146. for _, recvX := range []bool{true, false} {
  147. name := "recvmsg_x"
  148. if !recvX {
  149. name = "readv"
  150. }
  151. t.Run(name, func(t *testing.T) {
  152. fn(t, func(tn *tun) *tunQueue {
  153. q := newTunQueue(tn)
  154. if !recvX {
  155. q.x = nil
  156. }
  157. return q
  158. })
  159. })
  160. }
  161. }
  162. // writeTunPkts writes n test packets of size bytes to the peer end w and returns the packets the tun should hand back
  163. // for them.
  164. func writeTunPkts(t *testing.T, w, n, size int) [][]byte {
  165. t.Helper()
  166. var want [][]byte
  167. for i := range n {
  168. pkt, wire := testTunPkt(i, size+i)
  169. _, err := unix.Write(w, wire)
  170. require.NoError(t, err)
  171. want = append(want, pkt)
  172. }
  173. return want
  174. }
  175. // newReadTun is newSocketpairTun with the tun's receive buffer big enough for every test burst, since an AF_UNIX
  176. // datagram send is bounded by the receiver's buffer, and the writer's send buffer big enough for the largest IP packet.
  177. func newReadTun(t *testing.T) (*tun, int) {
  178. t.Helper()
  179. tn, w := newSocketpairTun(t, 1<<20)
  180. require.NoError(t, unix.SetsockoptInt(w, unix.SOL_SOCKET, unix.SO_SNDBUF, 1<<18))
  181. rc, err := tn.f.SyscallConn()
  182. require.NoError(t, err)
  183. require.NoError(t, rc.Control(func(fd uintptr) {
  184. require.NoError(t, unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_RCVBUF, 1<<20))
  185. }))
  186. return tn, w
  187. }
  188. // readTunPkts calls q.Read once per entry of sizes, requires each to return that many packets, and returns copies of
  189. // them all.
  190. func readTunPkts(t *testing.T, q *tunQueue, sizes ...int) [][]byte {
  191. t.Helper()
  192. var got [][]byte
  193. for _, size := range sizes {
  194. pkts, err := q.Read()
  195. require.NoError(t, err)
  196. require.Len(t, pkts, size)
  197. for _, p := range pkts {
  198. got = append(got, p.Clone().Bytes)
  199. }
  200. }
  201. return got
  202. }
  203. // stubTunRecvmsgX swaps tunRecvmsgX for fn until the test ends and counts its calls. fn gets the real msgx.Recv to
  204. // delegate to.
  205. func stubTunRecvmsgX(t *testing.T, fn func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno)) *atomic.Int64 {
  206. t.Helper()
  207. sys := tunRecvmsgX
  208. var calls atomic.Int64
  209. tunRecvmsgX = func(fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) {
  210. calls.Add(1)
  211. return fn(sys, fd, hdrs)
  212. }
  213. t.Cleanup(func() { tunRecvmsgX = sys })
  214. return &calls
  215. }
  216. // TestTunQueueReadDrains pins that Read hands back every queued packet, AF prefix stripped and in order, capped at
  217. // tunReadBatch per call, and doesn't wait for more once the queue is empty.
  218. func TestTunQueueReadDrains(t *testing.T) {
  219. forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) {
  220. tn, w := newReadTun(t)
  221. q := newQ(tn)
  222. want := writeTunPkts(t, w, tunReadBatch+3, 60)
  223. assert.Equal(t, want, readTunPkts(t, q, tunReadBatch, 3))
  224. })
  225. }
  226. // TestTunQueueReadUsesRecvX pins that a burst is read with one recvmsg_x, not one call per packet.
  227. func TestTunQueueReadUsesRecvX(t *testing.T) {
  228. calls := stubTunRecvmsgX(t, func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) {
  229. return sys(fd, hdrs)
  230. })
  231. tn, w := newReadTun(t)
  232. q := newTunQueue(tn)
  233. want := writeTunPkts(t, w, 10, 60)
  234. assert.Equal(t, want, readTunPkts(t, q, 10))
  235. assert.EqualValues(t, 1, calls.Load())
  236. assert.NotNil(t, q.x)
  237. }
  238. // TestTunQueueReadLargePackets pins that packets up to the largest IP packet come back whole, which on the recvmsg_x
  239. // path means each lands in its own slot and none spills into the next.
  240. func TestTunQueueReadLargePackets(t *testing.T) {
  241. forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) {
  242. tn, w := newReadTun(t)
  243. q := newQ(tn)
  244. var want [][]byte
  245. for i, size := range []int{65535, 1, 9001, 65535} {
  246. pkt, wire := testTunPkt(i, max(size, 3))
  247. pkt, wire = pkt[:size], wire[:4+size]
  248. _, err := unix.Write(w, wire)
  249. require.NoError(t, err)
  250. want = append(want, pkt)
  251. }
  252. assert.Equal(t, want, readTunPkts(t, q, 4))
  253. })
  254. }
  255. // TestTunQueueReadClipsPackets pins that each returned packet's capacity ends at its own bytes, so appending to one
  256. // can't overwrite the next.
  257. func TestTunQueueReadClipsPackets(t *testing.T) {
  258. forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) {
  259. tn, w := newSocketpairTun(t, 1<<20)
  260. q := newQ(tn)
  261. writeTunPkts(t, w, 2, 100)
  262. pkts, err := q.Read()
  263. require.NoError(t, err)
  264. require.Len(t, pkts, 2)
  265. assert.Equal(t, len(pkts[0].Bytes), cap(pkts[0].Bytes))
  266. assert.Equal(t, len(pkts[1].Bytes), cap(pkts[1].Bytes))
  267. })
  268. }
  269. // TestTunQueueReadDoesNotAllocate pins both read paths to zero allocations per Read once the queue exists.
  270. func TestTunQueueReadDoesNotAllocate(t *testing.T) {
  271. forEachTunReadPath(t, func(t *testing.T, newQ func(*tun) *tunQueue) {
  272. tn, w := newReadTun(t)
  273. q := newQ(tn)
  274. _, wire := testTunPkt(0, 1300)
  275. allocs := testing.AllocsPerRun(1000, func() {
  276. if _, err := unix.Write(w, wire); err != nil {
  277. t.Fatal(err)
  278. }
  279. if pkts, err := q.Read(); err != nil || len(pkts) != 1 {
  280. t.Fatalf("read %d packets: %v", len(pkts), err)
  281. }
  282. })
  283. assert.Zero(t, allocs)
  284. })
  285. }
  286. // TestTunQueueReadXRefused pins that recvmsg_x being refused turns it off for good, and that the same Read still
  287. // hands back the queued packets through readv.
  288. func TestTunQueueReadXRefused(t *testing.T) {
  289. for _, errno := range []syscall.Errno{unix.ENOSYS, unix.EPERM, unix.EOPNOTSUPP, unix.EINVAL} {
  290. t.Run(errno.Error(), func(t *testing.T) {
  291. calls := stubTunRecvmsgX(t, func(func(uintptr, []msgx.Hdr) (int, syscall.Errno), uintptr, []msgx.Hdr) (int, syscall.Errno) {
  292. return 0, errno
  293. })
  294. tn, w := newReadTun(t)
  295. q := newTunQueue(tn)
  296. want := writeTunPkts(t, w, 5, 60)
  297. assert.Equal(t, want, readTunPkts(t, q, 5))
  298. assert.Nil(t, q.x)
  299. assert.EqualValues(t, 1, calls.Load())
  300. want = writeTunPkts(t, w, 2, 60)
  301. assert.Equal(t, want, readTunPkts(t, q, 2))
  302. assert.EqualValues(t, 1, calls.Load(), "recvmsg_x was tried again after it was refused")
  303. })
  304. }
  305. }
  306. // TestTunQueueReadXTransientError pins that an error from recvmsg_x that doesn't mean it is refused leaves this Read
  307. // to readv but keeps recvmsg_x on for the next.
  308. func TestTunQueueReadXTransientError(t *testing.T) {
  309. var failed atomic.Bool
  310. calls := stubTunRecvmsgX(t, func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) {
  311. if !failed.Swap(true) {
  312. return 0, unix.ENOBUFS
  313. }
  314. return sys(fd, hdrs)
  315. })
  316. tn, w := newReadTun(t)
  317. q := newTunQueue(tn)
  318. want := writeTunPkts(t, w, 3, 60)
  319. assert.Equal(t, want, readTunPkts(t, q, 3))
  320. assert.NotNil(t, q.x)
  321. assert.EqualValues(t, 1, calls.Load())
  322. want = writeTunPkts(t, w, 2, 60)
  323. assert.Equal(t, want, readTunPkts(t, q, 2))
  324. assert.EqualValues(t, 2, calls.Load(), "recvmsg_x wasn't tried again after a transient error")
  325. }
  326. // TestTunQueueReadXParks pins that EAGAIN from recvmsg_x parks the Read until the utun is readable, rather than
  327. // falling back or returning an empty batch.
  328. func TestTunQueueReadXParks(t *testing.T) {
  329. var failed atomic.Bool
  330. calls := stubTunRecvmsgX(t, func(sys func(uintptr, []msgx.Hdr) (int, syscall.Errno), fd uintptr, hdrs []msgx.Hdr) (int, syscall.Errno) {
  331. if !failed.Swap(true) {
  332. return 0, unix.EAGAIN
  333. }
  334. return sys(fd, hdrs)
  335. })
  336. tn, w := newReadTun(t)
  337. q := newTunQueue(tn)
  338. // A real EAGAIN means nothing is queued, so the packet arrives only once Read has parked. One packet, so the
  339. // wakeup can't split it across Reads.
  340. done := make(chan struct{})
  341. go func() {
  342. defer close(done)
  343. time.Sleep(50 * time.Millisecond)
  344. _, wire := testTunPkt(0, 60)
  345. _, _ = unix.Write(w, wire)
  346. }()
  347. t.Cleanup(func() { <-done })
  348. pkt, _ := testTunPkt(0, 60)
  349. assert.Equal(t, [][]byte{pkt}, readTunPkts(t, q, 1))
  350. assert.NotNil(t, q.x)
  351. // A spurious wakeup costs another EAGAIN, so only a floor.
  352. assert.GreaterOrEqual(t, calls.Load(), int64(2))
  353. }
  354. // TestTunQueueReadXBadBatch pins that a recvmsg_x batch no intact header could produce turns recvmsg_x off for good
  355. // and hands the Read to readv. The stub reports a bad batch without taking any packets, so readv then reads them all.
  356. func TestTunQueueReadXBadBatch(t *testing.T) {
  357. v4 := [4]byte{3: syscall.AF_INET}
  358. for _, tc := range []struct {
  359. name string
  360. mess func(x *tunRecvX) int
  361. }{
  362. {"too many", func(x *tunRecvX) int {
  363. for i := range x.hdrs {
  364. x.hdrs[i].Datalen, x.heads[i] = 64, v4
  365. }
  366. return len(x.hdrs) + 1
  367. }},
  368. {"past the buffer", func(x *tunRecvX) int {
  369. x.hdrs[0].Datalen, x.heads[0] = 4+tunRecvXSlot+1, v4
  370. return 1
  371. }},
  372. {"bad AF prefix", func(x *tunRecvX) int {
  373. x.hdrs[0].Datalen, x.heads[0] = 64, [4]byte{syscall.AF_INET}
  374. return 1
  375. }},
  376. } {
  377. t.Run(tc.name, func(t *testing.T) {
  378. var q *tunQueue
  379. stubTunRecvmsgX(t, func(func(uintptr, []msgx.Hdr) (int, syscall.Errno), uintptr, []msgx.Hdr) (int, syscall.Errno) {
  380. return tc.mess(q.x), 0
  381. })
  382. tn, w := newReadTun(t)
  383. q = newTunQueue(tn)
  384. want := writeTunPkts(t, w, 4, 60)
  385. assert.Equal(t, want, readTunPkts(t, q, 4))
  386. assert.Nil(t, q.x)
  387. })
  388. }
  389. }
  390. // TestTunRecvXCheck pins which recvmsg_x batches tunRecvXCheck accepts.
  391. func TestTunRecvXCheck(t *testing.T) {
  392. const slot = 1000
  393. v4, v6 := [4]byte{3: syscall.AF_INET}, [4]byte{3: syscall.AF_INET6}
  394. good := func() ([]msgx.Hdr, [][4]byte) {
  395. hdrs := make([]msgx.Hdr, 4)
  396. heads := make([][4]byte, 4)
  397. for i := range hdrs {
  398. hdrs[i].Datalen = uint64(4 + 20 + i)
  399. heads[i] = v4
  400. if i%2 == 1 {
  401. heads[i] = v6
  402. }
  403. }
  404. return hdrs, heads
  405. }
  406. for _, tc := range []struct {
  407. name string
  408. n int
  409. mess func(hdrs []msgx.Hdr, heads [][4]byte)
  410. ok bool
  411. }{
  412. {"all", 4, nil, true},
  413. {"none", 0, nil, true},
  414. {"prefix", 2, nil, true},
  415. {"garbage past n is ignored", 2, func(h []msgx.Hdr, hd [][4]byte) { h[3].Datalen, hd[3] = 1<<40, [4]byte{9} }, true},
  416. {"just the AF prefix", 1, func(h []msgx.Hdr, _ [][4]byte) { h[0].Datalen = 4 }, true},
  417. {"fills the slot", 1, func(h []msgx.Hdr, _ [][4]byte) { h[0].Datalen = 4 + slot }, true},
  418. {"more than headers", 5, nil, false},
  419. {"negative", -1, nil, false},
  420. {"past the slot", 2, func(h []msgx.Hdr, _ [][4]byte) { h[1].Datalen = 4 + slot + 1 }, false},
  421. {"shorter than the AF prefix", 1, func(h []msgx.Hdr, _ [][4]byte) { h[0].Datalen = 3 }, false},
  422. {"truncated", 3, func(h []msgx.Hdr, _ [][4]byte) { h[2].Flags = unix.MSG_TRUNC }, false},
  423. {"zero AF prefix", 1, func(_ []msgx.Hdr, hd [][4]byte) { hd[0] = [4]byte{} }, false},
  424. {"host order AF prefix", 1, func(_ []msgx.Hdr, hd [][4]byte) { hd[0] = [4]byte{syscall.AF_INET} }, false},
  425. {"other AF", 2, func(_ []msgx.Hdr, hd [][4]byte) { hd[1] = [4]byte{3: syscall.AF_UNIX} }, false},
  426. } {
  427. t.Run(tc.name, func(t *testing.T) {
  428. hdrs, heads := good()
  429. if tc.mess != nil {
  430. tc.mess(hdrs, heads)
  431. }
  432. err := tunRecvXCheck(tc.n, hdrs, heads, slot)
  433. if tc.ok {
  434. assert.NoError(t, err)
  435. } else {
  436. assert.Error(t, err)
  437. }
  438. })
  439. }
  440. }
  441. // TestTunRecvXRefused pins which recvmsg_x errnos turn it off for good.
  442. func TestTunRecvXRefused(t *testing.T) {
  443. for _, errno := range []syscall.Errno{unix.ENOSYS, unix.EPERM, unix.EOPNOTSUPP, unix.EINVAL, unix.EMSGSIZE} {
  444. assert.True(t, tunRecvXRefused(errno), "%v", errno)
  445. }
  446. for _, errno := range []syscall.Errno{0, unix.EAGAIN, unix.EINTR, unix.ENOBUFS, unix.ENOMEM, unix.EBADF, unix.ENOTCONN} {
  447. assert.False(t, tunRecvXRefused(errno), "%v", errno)
  448. }
  449. }
  450. // TestTunRaisePendingPackets pins that the pending limit is set at SYSPROTO_CONTROL level after the receive buffer is
  451. // grown to hold that many packets of the tun's MTU, and capped to what the buffer holds when it can't grow.
  452. func TestTunRaisePendingPackets(t *testing.T) {
  453. type call struct{ level, opt, value int }
  454. stub := func(t *testing.T, rcvbuf int, failLevel, failOpt int) *[]call {
  455. set, get := tunSetsockoptInt, tunGetsockoptInt
  456. var calls []call
  457. tunSetsockoptInt = func(fd, level, opt, value int) error {
  458. calls = append(calls, call{level, opt, value})
  459. if level == failLevel && opt == failOpt {
  460. return unix.ENOBUFS
  461. }
  462. return nil
  463. }
  464. tunGetsockoptInt = func(fd, level, opt int) (int, error) { return rcvbuf, nil }
  465. t.Cleanup(func() { tunSetsockoptInt, tunGetsockoptInt = set, get })
  466. return &calls
  467. }
  468. for _, tc := range []struct {
  469. name string
  470. rcvbuf int
  471. mtu int
  472. failLevel int
  473. failOpt int
  474. want []call
  475. }{
  476. {"grows rcvbuf first", 1 << 16, 9000, -1, -1, []call{
  477. {unix.SOL_SOCKET, unix.SO_RCVBUF, 64 * 9004},
  478. {2, 16, 64},
  479. }},
  480. {"rcvbuf already big enough", 1 << 20, 1300, -1, -1, []call{{2, 16, 64}}},
  481. {"rcvbuf refused caps pending", 10 * 1304, 1300, unix.SOL_SOCKET, unix.SO_RCVBUF, []call{
  482. {unix.SOL_SOCKET, unix.SO_RCVBUF, 64 * 1304},
  483. {2, 16, 10},
  484. }},
  485. } {
  486. t.Run(tc.name, func(t *testing.T) {
  487. calls := stub(t, tc.rcvbuf, tc.failLevel, tc.failOpt)
  488. tn, _ := newSocketpairTun(t, 1<<20)
  489. tn.DefaultMTU = tc.mtu
  490. _, err := tn.Queues(1)
  491. require.NoError(t, err)
  492. assert.Equal(t, tc.want, *calls)
  493. })
  494. }
  495. }
  496. func TestTun_WriteFraming(t *testing.T) {
  497. r, w, err := os.Pipe()
  498. require.NoError(t, err)
  499. defer r.Close()
  500. defer w.Close()
  501. tn := &tun{f: w}
  502. for _, tc := range []struct {
  503. name string
  504. pkt []byte
  505. af byte
  506. }{
  507. // Literal families rather than syscall.AF_*, so a wrong constant in Write can't also fix the expectation.
  508. {"ipv4", []byte{0x45, 1, 2, 3, 4, 5}, 2},
  509. {"ipv6", []byte{0x60, 1, 2, 3, 4, 5}, 30},
  510. } {
  511. t.Run(tc.name, func(t *testing.T) {
  512. n, err := tn.Write(tc.pkt)
  513. require.NoError(t, err)
  514. require.Equal(t, len(tc.pkt), n)
  515. got := make([]byte, 64)
  516. n, err = r.Read(got)
  517. require.NoError(t, err)
  518. require.Equal(t, append([]byte{0, 0, 0, tc.af}, tc.pkt...), got[:n])
  519. })
  520. }
  521. _, err = tn.Write([]byte{0x10, 1, 2, 3})
  522. require.Error(t, err)
  523. _, err = tn.Write([]byte{})
  524. require.ErrorIs(t, err, syscall.EIO)
  525. // The rejected writes must not have put anything on the fd, so the next read sees only this packet.
  526. pkt := []byte{0x45, 9, 9}
  527. _, err = tn.Write(pkt)
  528. require.NoError(t, err)
  529. got := make([]byte, 64)
  530. n, err := r.Read(got)
  531. require.NoError(t, err)
  532. require.Equal(t, append([]byte{0, 0, 0, 2}, pkt...), got[:n])
  533. }
  534. func TestTun_ReadStripsHeader(t *testing.T) {
  535. r, w, err := os.Pipe()
  536. require.NoError(t, err)
  537. defer r.Close()
  538. defer w.Close()
  539. tn := &tun{f: r}
  540. pkt := []byte{0x45, 1, 2, 3, 4, 5}
  541. _, err = w.Write(append([]byte{0, 0, 0, syscall.AF_INET}, pkt...))
  542. require.NoError(t, err)
  543. got := make([]byte, 64)
  544. n, err := tn.Read(got)
  545. require.NoError(t, err)
  546. require.Equal(t, pkt, got[:n])
  547. }
  548. func TestTun_ReadShort(t *testing.T) {
  549. r, w, err := os.Pipe()
  550. require.NoError(t, err)
  551. defer r.Close()
  552. defer w.Close()
  553. tn := &tun{f: r}
  554. // Fewer bytes than the protocol header is not a packet.
  555. _, err = w.Write([]byte{0, 0})
  556. require.NoError(t, err)
  557. n, err := tn.Read(make([]byte, 64))
  558. require.NoError(t, err)
  559. require.Zero(t, n)
  560. // The next packet reads cleanly only if the short read consumed exactly what was written.
  561. pkt := []byte{0x45, 1, 2, 3}
  562. _, err = w.Write(append([]byte{0, 0, 0, syscall.AF_INET}, pkt...))
  563. require.NoError(t, err)
  564. got := make([]byte, 64)
  565. n, err = tn.Read(got)
  566. require.NoError(t, err)
  567. require.Equal(t, pkt, got[:n])
  568. }
  569. // A heap allocation per packet rarely shows in a short throughput run, only across GC pauses, so
  570. // this pins Read and Write to zero allocations directly.
  571. func TestTun_ReadWriteDoNotAllocate(t *testing.T) {
  572. r, w, err := os.Pipe()
  573. require.NoError(t, err)
  574. defer r.Close()
  575. defer w.Close()
  576. tw := &tun{f: w}
  577. tr := &tun{f: r}
  578. pkts := [][]byte{bytes.Repeat([]byte{0x45}, 1300), bytes.Repeat([]byte{0x60}, 1300)}
  579. buf := make([]byte, 1500)
  580. i := 0
  581. allocs := testing.AllocsPerRun(1000, func() {
  582. i++
  583. if _, err := tw.Write(pkts[i%len(pkts)]); err != nil {
  584. t.Fatal(err)
  585. }
  586. if _, err := tr.Read(buf); err != nil {
  587. t.Fatal(err)
  588. }
  589. })
  590. require.Zero(t, allocs)
  591. }