diff --git a/internal/proxy/udprelay/loop.go b/internal/proxy/udprelay/loop.go index 2dda333..488259b 100644 --- a/internal/proxy/udprelay/loop.go +++ b/internal/proxy/udprelay/loop.go @@ -269,18 +269,28 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, return } + const maxPayload = 1600 + wg.Go(func() { defer turncancel() - buf := make([]byte, 1600) - var wireBuf []byte + // При obf читаем payload сразу в buf[HeaderLen:], чтобы WrapInPlace + // дописал заголовок+tag без копии payload. + var buf, readSlot []byte if obfConn != nil { - wireBuf = make([]byte, rtpopus.MaxWire(len(buf))) + buf = make([]byte, rtpopus.MaxWire(maxPayload)) + readSlot = buf[rtpopus.HeaderLen : rtpopus.HeaderLen+maxPayload] + } else { + buf = make([]byte, maxPayload) + readSlot = buf } + // Адрес внутреннего пайпа константен; фиксируем один раз вместо + // atomic-записи на каждый пакет. + addrStored := false for { if turnctx.Err() != nil { return } - n, addr1, err1 := conn2.ReadFrom(buf) + n, addr1, err1 := conn2.ReadFrom(readSlot) if err1 != nil { return } @@ -288,16 +298,19 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, return } - internalPipeAddr.Store(addr1) + if !addrStored { + internalPipeAddr.Store(addr1) + addrStored = true + } - out := buf[:n] + out := readSlot[:n] if obfConn != nil { - written, wErr := obfConn.WrapInto(wireBuf, out) + written, wErr := obfConn.WrapInPlace(buf, n) if wErr != nil { deps.log().Errorf("[STREAM %d] OBF wrap failed: %v", streamID, wErr) return } - out = wireBuf[:written] + out = buf[:written] } written, err1 := relayConn.WriteTo(out, peer) @@ -310,12 +323,11 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, wg.Go(func() { defer turncancel() - readBufLen := 1600 + readBufLen := maxPayload if obfConn != nil { - readBufLen = rtpopus.MaxWire(1600) + readBufLen = rtpopus.MaxWire(maxPayload) } buf := make([]byte, readBufLen) - plain := make([]byte, 1600) for { n, _, err1 := relayConn.ReadFrom(buf) if err1 != nil { @@ -329,12 +341,12 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr, if addr, ok := addr1.(net.Addr); ok { payload := buf[:n] if obfConn != nil { - m, uErr := obfConn.Unwrap(payload, plain) + p, uErr := obfConn.UnwrapInPlace(buf[:n]) if uErr != nil { deps.log().Errorf("[STREAM %d] OBF unwrap failed: %v (n=%d)", streamID, uErr, n) continue } - payload = plain[:m] + payload = p } st.AddRx(len(payload)) if _, err := conn2.WriteTo(payload, addr); err != nil { diff --git a/internal/wire/rtpopus/rtpopus.go b/internal/wire/rtpopus/rtpopus.go index c1c77e0..3b50c9a 100644 --- a/internal/wire/rtpopus/rtpopus.go +++ b/internal/wire/rtpopus/rtpopus.go @@ -41,15 +41,19 @@ import ( ) const ( - KeyLen = 32 - rtpHdrLen = 12 - nonceLen = 12 - tagLen = 16 - headerLen = rtpHdrLen + nonceLen // 24 - Overhead = headerLen + tagLen // 40 - rtpVersion = 0x80 // V=2, P=0, X=0, CC=0 - rtpPT = 0x6F // M=0, PT=111 (opus) - tsStep = 960 // 20ms @ 48kHz + KeyLen = 32 + rtpHdrLen = 12 + nonceLen = 12 + tagLen = 16 + headerLen = rtpHdrLen + nonceLen // 24 + // HeaderLen — offset, с которого начинается plaintext в wire-буфере. + // Экспонирован для in-place API (WrapInPlace/UnwrapInPlace): вызывающий + // читает payload сразу в buf[HeaderLen:], избегая копии. + HeaderLen = headerLen + Overhead = headerLen + tagLen // 40 + rtpVersion = 0x80 // V=2, P=0, X=0, CC=0 + rtpPT = 0x6F // M=0, PT=111 (opus) + tsStep = 960 // 20ms @ 48kHz ) func MaxWire(payloadLen int) int { return Overhead + payloadLen } @@ -125,41 +129,65 @@ func NewConnFromState(state *State, isServer bool) (*Conn, error) { // WrapInto кодирует payload в dst (минимум MaxWire(len(payload)) байт) // и возвращает число записанных wire-байт. func (c *Conn) WrapInto(dst, payload []byte) (int, error) { - wireLen := Overhead + len(payload) - if len(dst) < wireLen { + if len(dst) < Overhead+len(payload) { + return 0, errors.New("rtpopus:dst buffer too small") + } + copy(dst[headerLen:], payload) + return c.WrapInPlace(dst, len(payload)) +} + +// WrapInPlace кодирует plaintext, который вызывающий уже разместил в +// buf[HeaderLen:HeaderLen+plainLen], дописывая RTP-заголовок+nonce перед ним +// и AEAD-tag после — без копии payload. buf должен вмещать MaxWire(plainLen). +// Возвращает число записанных wire-байт. +func (c *Conn) WrapInPlace(buf []byte, plainLen int) (int, error) { + wireLen := Overhead + plainLen + if len(buf) < wireLen { return 0, errors.New("rtpopus:dst buffer too small") } // RTP-заголовок. - dst[0] = rtpVersion - dst[1] = rtpPT + buf[0] = rtpVersion + buf[1] = rtpPT seq := uint16(c.seq.Add(1) - 1) //nolint:gosec // RTP sequence number is intentionally mod 2^16 - binary.BigEndian.PutUint16(dst[2:4], seq) + binary.BigEndian.PutUint16(buf[2:4], seq) ts := c.timestamp.Add(tsStep) - tsStep - binary.BigEndian.PutUint32(dst[4:8], ts) - copy(dst[8:12], c.ssrc[:]) + binary.BigEndian.PutUint32(buf[4:8], ts) + copy(buf[8:12], c.ssrc[:]) // Явный nonce. noncePos := rtpHdrLen - copy(dst[noncePos:noncePos+4], c.sessionID[:]) + copy(buf[noncePos:noncePos+4], c.sessionID[:]) ctr := c.counter.Add(1) - 1 - binary.BigEndian.PutUint64(dst[noncePos+4:noncePos+nonceLen], ctr) + binary.BigEndian.PutUint64(buf[noncePos+4:noncePos+nonceLen], ctr) - nonce := dst[noncePos : noncePos+nonceLen] - aad := dst[:headerLen] + nonce := buf[noncePos : noncePos+nonceLen] + aad := buf[:headerLen] ctPos := headerLen - copy(dst[ctPos:], payload) - c.state.aead.Seal(dst[ctPos:ctPos], nonce, dst[ctPos:ctPos+len(payload)], aad) + c.state.aead.Seal(buf[ctPos:ctPos], nonce, buf[ctPos:ctPos+plainLen], aad) return wireLen, nil } // Unwrap декодирует wire-пакет в dst и возвращает длину plaintext. -// AEAD открывает in-place внутри wire — вызывающий должен считать wire -// потреблённым после вызова (содержимое нельзя переиспользовать). func (c *Conn) Unwrap(wire, dst []byte) (int, error) { + plain, err := c.UnwrapInPlace(wire) + if err != nil { + return 0, err + } + if len(plain) > len(dst) { + return 0, errors.New("rtpopus:dst buffer too small") + } + copy(dst[:len(plain)], plain) + return len(plain), nil +} + +// UnwrapInPlace декодирует wire-пакет на месте и возвращает subslice plaintext +// внутри wire (без копии в отдельный буфер). AEAD открывает in-place — wire +// после вызова считается потреблённым, результат валиден до следующей записи в wire. +func (c *Conn) UnwrapInPlace(wire []byte) ([]byte, error) { if len(wire) < Overhead { - return 0, errors.New("rtpopus:packet too short") + return nil, errors.New("rtpopus:packet too short") } nonce := wire[rtpHdrLen : rtpHdrLen+nonceLen] aad := wire[:headerLen] @@ -167,13 +195,9 @@ func (c *Conn) Unwrap(wire, dst []byte) (int, error) { plain, err := c.state.aead.Open(ct[:0], nonce, ct, aad) if err != nil { - return 0, fmt.Errorf("rtpopus:AEAD open: %w", err) + return nil, fmt.Errorf("rtpopus:AEAD open: %w", err) } - if len(plain) > len(dst) { - return 0, errors.New("rtpopus:dst buffer too small") - } - copy(dst[:len(plain)], plain) - return len(plain), nil + return plain, nil } func GenKeyHex() (string, error) { diff --git a/internal/wire/rtpopus/rtpopus_test.go b/internal/wire/rtpopus/rtpopus_test.go index e7697a9..c6389df 100644 --- a/internal/wire/rtpopus/rtpopus_test.go +++ b/internal/wire/rtpopus/rtpopus_test.go @@ -63,6 +63,58 @@ func TestConnRoundTrip(t *testing.T) { } } +func TestInPlaceRoundTrip(t *testing.T) { + key := bytes.Repeat([]byte{0x42}, KeyLen) + payload := []byte("dtls record bytes in-place") + + client, err := NewConn(key, false) + if err != nil { + t.Fatalf("NewConn(client): %v", err) + } + server, err := NewConn(key, true) + if err != nil { + t.Fatalf("NewConn(server): %v", err) + } + + // Клиент: payload уже в buf[HeaderLen:], WrapInPlace без копии. + buf := make([]byte, MaxWire(len(payload))) + copy(buf[HeaderLen:], payload) + n, err := client.WrapInPlace(buf, len(payload)) + if err != nil { + t.Fatalf("WrapInPlace: %v", err) + } + if n != MaxWire(len(payload)) { + t.Fatalf("WrapInPlace len = %d, want %d", n, MaxWire(len(payload))) + } + if bytes.Contains(buf[:n], payload) { + t.Fatalf("wrapped packet contains plaintext payload") + } + + // WrapInPlace и WrapInto должны давать совместимый wire (decode общим Unwrap). + dst := make([]byte, 1600) + m, err := server.Unwrap(buf[:n], dst) + if err != nil { + t.Fatalf("Unwrap: %v", err) + } + if !bytes.Equal(dst[:m], payload) { + t.Fatalf("WrapInPlace→Unwrap mismatch: got %q want %q", dst[:m], payload) + } + + // Сервер: WrapInto → клиент UnwrapInPlace (subslice внутрь wire, без копии). + wire := make([]byte, MaxWire(len(payload))) + wn, err := server.WrapInto(wire, payload) + if err != nil { + t.Fatalf("server wrap: %v", err) + } + plain, err := client.UnwrapInPlace(wire[:wn]) + if err != nil { + t.Fatalf("UnwrapInPlace: %v", err) + } + if !bytes.Equal(plain, payload) { + t.Fatalf("UnwrapInPlace mismatch: got %q want %q", plain, payload) + } +} + func TestRTPHeaderProgression(t *testing.T) { key := bytes.Repeat([]byte{0x42}, KeyLen) c, err := NewConn(key, false)