udp_tester.go 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140
  1. //go:build e2e_testing
  2. // +build e2e_testing
  3. package udp
  4. import (
  5. "io"
  6. "net/netip"
  7. "sync/atomic"
  8. "github.com/sirupsen/logrus"
  9. "github.com/slackhq/nebula/config"
  10. "github.com/slackhq/nebula/header"
  11. )
  12. type Packet struct {
  13. To netip.AddrPort
  14. From netip.AddrPort
  15. Data []byte
  16. }
  17. func (u *Packet) Copy() *Packet {
  18. n := &Packet{
  19. To: u.To,
  20. From: u.From,
  21. Data: make([]byte, len(u.Data)),
  22. }
  23. copy(n.Data, u.Data)
  24. return n
  25. }
  26. type TesterConn struct {
  27. Addr netip.AddrPort
  28. RxPackets chan *Packet // Packets to receive into nebula
  29. TxPackets chan *Packet // Packets transmitted outside by nebula
  30. closed atomic.Bool
  31. l *logrus.Logger
  32. }
  33. func NewListener(l *logrus.Logger, ip netip.Addr, port int, _ bool, _ int) (Conn, error) {
  34. return &TesterConn{
  35. Addr: netip.AddrPortFrom(ip, uint16(port)),
  36. RxPackets: make(chan *Packet, 10),
  37. TxPackets: make(chan *Packet, 10),
  38. l: l,
  39. }, nil
  40. }
  41. // Send will place a UdpPacket onto the receive queue for nebula to consume
  42. // this is an encrypted packet or a handshake message in most cases
  43. // packets were transmitted from another nebula node, you can send them with Tun.Send
  44. func (u *TesterConn) Send(packet *Packet) {
  45. if u.closed.Load() {
  46. return
  47. }
  48. h := &header.H{}
  49. if err := h.Parse(packet.Data); err != nil {
  50. panic(err)
  51. }
  52. if u.l.Level >= logrus.DebugLevel {
  53. u.l.WithField("header", h).
  54. WithField("udpAddr", packet.From).
  55. WithField("dataLen", len(packet.Data)).
  56. Debug("UDP receiving injected packet")
  57. }
  58. u.RxPackets <- packet
  59. }
  60. // Get will pull a UdpPacket from the transmit queue
  61. // nebula meant to send this message on the network, it will be encrypted
  62. // packets were ingested from the tun side (in most cases), you can send them with Tun.Send
  63. func (u *TesterConn) Get(block bool) *Packet {
  64. if block {
  65. return <-u.TxPackets
  66. }
  67. select {
  68. case p := <-u.TxPackets:
  69. return p
  70. default:
  71. return nil
  72. }
  73. }
  74. //********************************************************************************************************************//
  75. // Below this is boilerplate implementation to make nebula actually work
  76. //********************************************************************************************************************//
  77. func (u *TesterConn) WriteTo(b []byte, addr netip.AddrPort) error {
  78. if u.closed.Load() {
  79. return io.ErrClosedPipe
  80. }
  81. p := &Packet{
  82. Data: make([]byte, len(b), len(b)),
  83. From: u.Addr,
  84. To: addr,
  85. }
  86. copy(p.Data, b)
  87. u.TxPackets <- p
  88. return nil
  89. }
  90. func (u *TesterConn) ListenOut(r EncReader) {
  91. for {
  92. p, ok := <-u.RxPackets
  93. if !ok {
  94. return
  95. }
  96. r(p.From, p.Data)
  97. }
  98. }
  99. func (u *TesterConn) ReloadConfig(*config.C) {}
  100. func NewUDPStatsEmitter(_ []Conn) func() {
  101. // No UDP stats for non-linux
  102. return func() {}
  103. }
  104. func (u *TesterConn) LocalAddr() (netip.AddrPort, error) {
  105. return u.Addr, nil
  106. }
  107. func (u *TesterConn) Rebind() error {
  108. return nil
  109. }
  110. func (u *TesterConn) Close() error {
  111. if u.closed.CompareAndSwap(false, true) {
  112. close(u.RxPackets)
  113. close(u.TxPackets)
  114. }
  115. return nil
  116. }