diff --git a/fileplayback/player.go b/fileplayback/player.go index 90e492d..a3ad9ec 100644 --- a/fileplayback/player.go +++ b/fileplayback/player.go @@ -1,6 +1,7 @@ package fileplayback import ( + "context" "encoding/binary" "errors" "io" @@ -18,6 +19,8 @@ type Player struct { filename string audioChan chan gumble.AudioBuffer stopChan chan struct{} + ctx context.Context + cancel context.CancelFunc mutex sync.Mutex playing bool errorFunc func(error) @@ -73,6 +76,7 @@ func (p *Player) PlayFile(filename string) error { // Start the file reading goroutine p.playing = true p.stopChan = make(chan struct{}) + p.ctx, p.cancel = context.WithCancel(context.Background()) go p.readFileAudio() return nil @@ -88,17 +92,23 @@ func (p *Player) Stop() error { } close(p.stopChan) + if p.cancel != nil { + p.cancel() + p.cancel = nil + } p.playing = false + localPlayback := p.localPlayback + p.mutex.Unlock() - if p.localPlayback != nil { - p.localPlayback(nil) + if localPlayback != nil { + localPlayback(nil) } - // Drain the audio channel + // Drain the audio channel. for len(p.audioChan) > 0 { <-p.audioChan } - + p.mutex.Lock() return nil } @@ -138,7 +148,10 @@ func (p *Player) readFileAudio() { args := []string{"-loglevel", "error", "-i", p.filename} args = append(args, "-ac", "2", "-ar", strconv.Itoa(gumble.AudioSampleRate), "-f", "s16le", "-") - cmd := exec.Command("ffmpeg", args...) + p.mutex.Lock() + ctx := p.ctx + p.mutex.Unlock() + cmd := exec.CommandContext(ctx, "ffmpeg", args...) pipe, err := cmd.StdoutPipe() if err != nil { p.mutex.Lock() @@ -171,15 +184,21 @@ func (p *Player) readFileAudio() { case <-ticker.C: n, err := io.ReadFull(pipe, byteBuffer) if err != nil || n != len(byteBuffer) { - // File finished playing + select { + case <-p.stopChan: + cmd.Wait() + return + default: + } + // File finished playing. p.mutex.Lock() p.playing = false - if p.localPlayback != nil { - p.localPlayback(nil) - } + localPlayback := p.localPlayback p.mutex.Unlock() + if localPlayback != nil { + localPlayback(nil) + } cmd.Wait() - // Notify that file finished p.reportError(errors.New("file playback finished")) return } diff --git a/gumble/gumble/audio.go b/gumble/gumble/audio.go index 8dcea1f..fbf74a1 100644 --- a/gumble/gumble/audio.go +++ b/gumble/gumble/audio.go @@ -90,6 +90,7 @@ type AudioPacket struct { AudioBuffer - HasPosition bool - X, Y, Z float32 + HasPosition bool + X, Y, Z float32 + VolumeAdjustment float32 } diff --git a/gumble/gumble/audiolisteners.go b/gumble/gumble/audiolisteners.go index 7bf0e80..cc034d8 100644 --- a/gumble/gumble/audiolisteners.go +++ b/gumble/gumble/audiolisteners.go @@ -1,5 +1,7 @@ package gumble +import "sync" + type audioEventItem struct { parent *AudioListeners prev, next *audioEventItem @@ -8,6 +10,8 @@ type audioEventItem struct { } func (e *audioEventItem) Detach() { + e.parent.mu.Lock() + defer e.parent.mu.Unlock() if e.prev == nil { e.parent.head = e.next } else { @@ -23,11 +27,14 @@ func (e *audioEventItem) Detach() { // AudioListeners is a list of audio listeners. Each attached listener is // called in sequence when a new user audio stream begins. type AudioListeners struct { + mu sync.Mutex head, tail *audioEventItem } // Attach adds a new audio listener to the end of the current list of listeners. func (e *AudioListeners) Attach(listener AudioListener) Detacher { + e.mu.Lock() + defer e.mu.Unlock() item := &audioEventItem{ parent: e, prev: e.tail, diff --git a/gumble/gumble/client.go b/gumble/gumble/client.go index 06a3d54..fb44557 100644 --- a/gumble/gumble/client.go +++ b/gumble/gumble/client.go @@ -6,6 +6,7 @@ import ( "math" "net" "runtime" + "sync" "sync/atomic" "time" @@ -70,10 +71,17 @@ type Client struct { VoiceTarget *VoiceTarget // UDP transport for audio (lower latency than TCP-tunneled audio). - udpConn *net.UDPConn - udpActive bool - cryptOut cryptState // client→server encryption - cryptIn cryptState // server→client encryption + udpMu sync.RWMutex + udpWriteMu sync.Mutex + udpConn *net.UDPConn + udpActive bool + udpCryptoOut *cryptState15 + udpCryptoIn *cryptState15 + udpFrameNumber uint64 + udpFallbackLogged atomic.Bool + udpFirstRecv atomic.Bool + cryptOut cryptState // client→server encryption + cryptIn cryptState // server→client encryption state uint32 @@ -258,7 +266,10 @@ func (c *Client) readRoutine() { } // When UDP audio is active, ignore TCP-tunneled audio // (packet type 1) to avoid double-processing packets. - if pType == 1 && c.udpActive { + c.udpMu.RLock() + udpActive := c.udpActive + c.udpMu.RUnlock() + if pType == 1 && udpActive { continue } if int(pType) < len(handlers) { @@ -270,11 +281,15 @@ func (c *Client) readRoutine() { atomic.StoreUint32(&c.state, uint32(StateDisconnected)) close(c.end) - // Clean up UDP connection - if c.udpConn != nil { + // Clean up UDP connection. + c.udpMu.Lock() + udpConn := c.udpConn + c.udpConn = nil + c.udpActive = false + c.udpMu.Unlock() + if udpConn != nil { log.Debug("closing UDP connection") - c.udpConn.Close() - c.udpConn = nil + udpConn.Close() } if wasSynced { @@ -331,27 +346,29 @@ func (c *Client) EnableStereoEncoder() { // WriteAudio writes an audio packet, preferring UDP when encryption is // set up. Falls back to TCP-tunneled audio when UDP is unavailable. -var udpFallbackLogged bool func (c *Client) WriteAudio(format, target byte, sequence int64, final bool, data []byte, X, Y, Z *float32) error { // Try Mumble 1.5 native UDP first (unless disabled) if !c.Config.DisableUDP { - if sent, err := c.WriteAudioUDP15(data, final); sent { + if sent, err := c.WriteAudioUDP15(uint32(target), data, final); sent { if err != nil { log.Error("UDP15 send error: %v", err) } return err } } - // Fall back to TCP tunnel — log once per process - if !udpFallbackLogged { - udpFallbackLogged = true + // Fall back to the TCP tunnel. + c.udpMu.RLock() + udpConn := c.udpConn + udpCryptoOut := c.udpCryptoOut + c.udpMu.RUnlock() + if !c.udpFallbackLogged.Swap(true) { if c.Config.DisableUDP { log.Info("UDP disabled, audio using TCP tunnel") - } else if c.udpConn == nil { + } else if udpConn == nil { log.Info("no UDP socket, audio using TCP tunnel") - } else if udp15Client == nil || !udp15Client.initialized { - log.Info("1.5 crypto not ready, audio using TCP tunnel") + } else if udpCryptoOut == nil { + log.Info("UDP crypto not ready, audio using TCP tunnel") } } return c.Conn.WriteAudio(format, target, sequence, final, data, X, Y, Z) diff --git a/gumble/gumble/crypt.go b/gumble/gumble/crypt.go index f62c8fb..9d74ea1 100644 --- a/gumble/gumble/crypt.go +++ b/gumble/gumble/crypt.go @@ -253,11 +253,13 @@ func (cs *cryptState) setup(key, iv []byte) error { cs.initialized = true - log.Info("cryptState setup: key=%s iv=%s encIV=%s nonce_prefix=%s", - hex.EncodeToString(cs.key[:]), - hex.EncodeToString(iv), - hex.EncodeToString(encIV[:]), - hex.EncodeToString(cs.nonce[:])) + if log.Enabled(log.LevelDebug) { + log.Debug("cryptState setup: key=%s iv=%s encIV=%s nonce_prefix=%s", + hex.EncodeToString(cs.key[:]), + hex.EncodeToString(iv), + hex.EncodeToString(encIV[:]), + hex.EncodeToString(cs.nonce[:])) + } return nil } @@ -302,8 +304,10 @@ func (c *Client) handleCryptSetup(buffer []byte) error { c.cryptOut.setup(packet.Key, packet.ClientNonce) c.cryptIn.setup(packet.Key, packet.ServerNonce) - // Also set up Mumble 1.5 native UDP crypto. - SetUDP15Crypto(packet.Key, packet.ClientNonce, packet.ServerNonce) + // Also set up per-client Mumble 1.5 native UDP crypto. + if err := c.setUDP15Crypto(packet.Key, packet.ClientNonce, packet.ServerNonce); err != nil { + return err + } if wasInit { log.Debug("CryptSetup updated (key rotation)") @@ -316,17 +320,19 @@ func (c *Client) handleCryptSetup(buffer []byte) error { log.Debug("received CryptSetup with incomplete fields, waiting for full key exchange") } - if c.cryptOut.initialized && c.udpConn != nil && !c.udpActive { - // Only start UDP reader if 1.5 crypto is also ready. - if udp15Client != nil && udp15Client.initialized { - c.udpActive = true - log.Info("UDP crypto ready (1.5 native), starting UDP reader and pinger") - go c.udpReadRoutine() - go c.udpPingRoutine() - } else { - log.Info("legacy crypto ready, waiting for 1.5 crypto before starting UDP") - } - } else if c.cryptOut.initialized && c.udpConn == nil { + c.udpMu.Lock() + udpReady := c.udpCryptoOut != nil + startUDP := c.cryptOut.initialized && c.udpConn != nil && !c.udpActive && udpReady + if startUDP { + c.udpActive = true + } + noUDPConn := c.udpConn == nil + c.udpMu.Unlock() + if startUDP { + log.Info("UDP crypto ready (1.5 native), starting UDP reader and pinger") + go c.udpReadRoutine() + go c.udpPingRoutine() + } else if c.cryptOut.initialized && noUDPConn { log.Warn("crypto ready but no UDP socket — audio will use TCP tunnel") } diff --git a/gumble/gumble/handlers.go b/gumble/gumble/handlers.go index 3e87d99..5ac2c89 100644 --- a/gumble/gumble/handlers.go +++ b/gumble/gumble/handlers.go @@ -10,8 +10,8 @@ import ( "time" "git.stormux.org/storm/barnard/gumble/gumble/MumbleProto" - "git.stormux.org/storm/barnard/log" "git.stormux.org/storm/barnard/gumble/gumble/varint" + "git.stormux.org/storm/barnard/log" "google.golang.org/protobuf/proto" ) @@ -231,25 +231,41 @@ func (c *Client) dispatchPLC(user *User, audioTarget byte, decoder AudioDecoder) // dispatchAudio sends an audio packet to all registered audio listeners. func (c *Client) dispatchAudio(user *User, packet *AudioPacket) { - c.volatile.Lock() - for item := c.Config.AudioListeners.head; item != nil; item = item.next { - c.volatile.Unlock() + listeners := &c.Config.AudioListeners + listeners.mu.Lock() + type delivery struct { + listener AudioListener + ch chan *AudioPacket + new bool + } + var deliveries []delivery + for item := listeners.head; item != nil; item = item.next { ch := item.streams[user] - if ch == nil { - ch = make(chan *AudioPacket) + newStream := ch == nil + if newStream { + bufferSize := c.Config.Buffers + if bufferSize < 1 { + bufferSize = 1 + } + ch = make(chan *AudioPacket, bufferSize) item.streams[user] = ch + } + deliveries = append(deliveries, delivery{item.listener, ch, newStream}) + } + listeners.mu.Unlock() + + for _, delivery := range deliveries { + if delivery.new { log.Debug("new audio stream from %s (session=%d)", user.Name, user.Session) - streamEvent := AudioStreamEvent{ - Client: c, - User: user, - C: ch, - } - item.listener.OnAudioStream(&streamEvent) + delivery.listener.OnAudioStream(&AudioStreamEvent{Client: c, User: user, C: delivery.ch}) + } + select { + case delivery.ch <- packet: + default: + // Never allow a slow listener to block protocol processing. + log.Debug("dropping buffered audio for slow listener (session=%d)", user.Session) } - ch <- packet - c.volatile.Lock() } - c.volatile.Unlock() } func (c *Client) handleAuthenticate(buffer []byte) error { diff --git a/gumble/gumble/udp.go b/gumble/gumble/udp.go index 22fb675..8c9a767 100644 --- a/gumble/gumble/udp.go +++ b/gumble/gumble/udp.go @@ -32,7 +32,9 @@ func (c *Client) startUDP() error { return nil } + c.udpMu.Lock() c.udpConn = conn + c.udpMu.Unlock() log.Info("UDP socket connected to %s", conn.RemoteAddr()) // The UDP reader and pinger will be started once CryptSetup is received. return nil @@ -44,15 +46,23 @@ func (c *Client) udpReadRoutine() { log.Info("UDP reader started (1.5 native format)") buf := make([]byte, maxUDPPacketSize) var packetCount uint64 + c.udpMu.RLock() + udpConn := c.udpConn + c.udpMu.RUnlock() + if udpConn == nil { + return + } for { - n, addr, err := c.udpConn.ReadFromUDP(buf) + n, addr, err := udpConn.ReadFromUDP(buf) if err != nil { log.Warn("UDP read error (stopping reader): %v", err) return } packetCount++ - log.Info("UDP recv #%d: %d bytes from %s hex=%s", - packetCount, n, addr, hex.EncodeToString(buf[:n])) + if log.Enabled(log.LevelDebug) { + log.Debug("UDP recv #%d: %d bytes from %s hex=%s", + packetCount, n, addr, hex.EncodeToString(buf[:n])) + } packet := make([]byte, n) copy(packet, buf[:n]) c.HandleUDPPacket15(packet, packetCount) @@ -79,8 +89,12 @@ func (c *Client) udpPingRoutine() { // sendUDPPing sends a Mumble 1.5 native UDP ping. // Uses standard protobuf varint encoding (not Mumble's custom varint). func (c *Client) sendUDPPing() { - cs := udp15Client - if cs == nil || !cs.initialized || c.udpConn == nil { + c.udpWriteMu.Lock() + defer c.udpWriteMu.Unlock() + c.udpMu.RLock() + cs, udpConn := c.udpCryptoOut, c.udpConn + c.udpMu.RUnlock() + if cs == nil || udpConn == nil { return } // Type byte 0x01 = UDPPing, field 1 = timestamp (protobuf varint, milliseconds). @@ -98,6 +112,5 @@ func (c *Client) sendUDPPing() { if err != nil { return } - c.udpConn.Write(encrypted) + udpConn.Write(encrypted) } - diff --git a/gumble/gumble/udp15.go b/gumble/gumble/udp15.go index e4e8f98..2a0d24c 100644 --- a/gumble/gumble/udp15.go +++ b/gumble/gumble/udp15.go @@ -3,8 +3,11 @@ package gumble import ( "bytes" "crypto/aes" + "encoding/binary" "encoding/hex" "errors" + "math" + "sync" "git.stormux.org/storm/barnard/gumble/gumble/varint" "git.stormux.org/storm/barnard/log" @@ -35,13 +38,14 @@ func pbEncodeVarint(buf []byte, v uint64) int { // and the number of bytes consumed (0 on error). func pbDecodeVarint(buf []byte) (uint64, int) { var v uint64 - var s uint for i, b := range buf { - v |= uint64(b&0x7F) << s + if i == 10 || (i == 9 && b > 1) { + return 0, 0 // overflow + } + v |= uint64(b&0x7F) << (7 * i) if b < 0x80 { return v, i + 1 } - s += 7 } return 0, 0 // truncated } @@ -59,23 +63,16 @@ func pbDecodeVarint(buf []byte) (uint64, int) { // encodeUDPAudio builds a MumbleUDP.Audio protobuf message. // If session == 0, sender_session is omitted (used for outbound). // Uses standard protobuf varint encoding, not Mumble's custom varint. -func encodeUDPAudio(session uint32, frameNumber uint32, opusData []byte, terminator bool) []byte { +func encodeUDPAudio(target uint32, frameNumber uint64, opusData []byte, terminator bool) []byte { var buf bytes.Buffer var tmp [10]byte // max protobuf varint size - // Field 1: target = 0 (normal speech). Always encoded so the server - // sees the oneof Header choice, matching the reference client. + // Field 1 selects the target header oneof. n := pbEncodeVarint(tmp[:], uint64((1<<3)|0)) buf.Write(tmp[:n]) - n = pbEncodeVarint(tmp[:], 0) + n = pbEncodeVarint(tmp[:], uint64(target)) buf.Write(tmp[:n]) - if session != 0 { - n := pbEncodeVarint(tmp[:], uint64((3<<3)|0)) - buf.Write(tmp[:n]) - n = pbEncodeVarint(tmp[:], uint64(session)) - buf.Write(tmp[:n]) - } n = pbEncodeVarint(tmp[:], uint64((4<<3)|0)) buf.Write(tmp[:n]) n = pbEncodeVarint(tmp[:], uint64(frameNumber)) @@ -99,7 +96,9 @@ func encodeUDPAudio(session uint32, frameNumber uint32, opusData []byte, termina // decodeUDPAudio parses a MumbleUDP.Audio protobuf message. // Uses standard protobuf varint decoding, not Mumble's custom varint. -func decodeUDPAudio(data []byte) (session uint32, frameNumber uint32, opusData []byte, terminator bool) { +func decodeUDPAudio(data []byte) (session uint32, frameNumber uint64, opusData []byte, terminator bool, context uint32, position *[3]float32, volumeAdjustment float32) { + var positionValues [3]float32 + positionCount := 0 pos := 0 for pos < len(data) { key, n := pbDecodeVarint(data[pos:]) @@ -118,10 +117,12 @@ func decodeUDPAudio(data []byte) (session uint32, frameNumber uint32, opusData [ } pos += n switch fieldNum { + case 2: + context = uint32(val) case 3: session = uint32(val) case 4: - frameNumber = uint32(val) + frameNumber = val case 16: terminator = val != 0 } @@ -131,18 +132,43 @@ func decodeUDPAudio(data []byte) (session uint32, frameNumber uint32, opusData [ return } pos += n - if int(length) > len(data)-pos { + if length > uint64(len(data)-pos) { return } + end := pos + int(length) if fieldNum == 5 { - opusData = make([]byte, length) - copy(opusData, data[pos:pos+int(length)]) + opusData = append(opusData[:0], data[pos:end]...) + } else if fieldNum == 6 && length == 12 { + for i := range positionValues { + positionValues[i] = math.Float32frombits(binary.LittleEndian.Uint32(data[pos+i*4:])) + } + positionCount = 3 } - pos += int(length) + pos = end + case 1: + if len(data)-pos < 8 { + return + } + pos += 8 + case 5: + if len(data)-pos < 4 { + return + } + value := math.Float32frombits(binary.LittleEndian.Uint32(data[pos:])) + if fieldNum == 6 && positionCount < len(positionValues) { + positionValues[positionCount] = value + positionCount++ + } else if fieldNum == 7 { + volumeAdjustment = value + } + pos += 4 default: - return // unknown wire type, bail + return // invalid wire type } } + if positionCount == len(positionValues) { + position = &positionValues + } return } @@ -160,6 +186,7 @@ const ( // cryptState15 implements Mumble 1.5 native UDP encryption. type cryptState15 struct { + mu sync.Mutex key [16]byte encryptIV [16]byte decryptIV [16]byte @@ -170,6 +197,8 @@ type cryptState15 struct { // setup15 initializes 1.5-style crypto from CryptSetup. // clientNonce → encryptIV, serverNonce → decryptIV. func (cs *cryptState15) setup15(key, clientNonce, serverNonce []byte) error { + cs.mu.Lock() + defer cs.mu.Unlock() if len(key) != 16 || len(clientNonce) != 16 || len(serverNonce) != 16 { return errors.New("gumble: invalid crypto key/nonce") } @@ -178,10 +207,12 @@ func (cs *cryptState15) setup15(key, clientNonce, serverNonce []byte) error { copy(cs.decryptIV[:], serverNonce) cs.initialized = true - log.Info("cryptState15 setup: key=%s encryptIV=%s decryptIV=%s", - hex.EncodeToString(cs.key[:]), - hex.EncodeToString(cs.encryptIV[:]), - hex.EncodeToString(cs.decryptIV[:])) + if log.Enabled(log.LevelDebug) { + log.Debug("cryptState15 setup: key=%s encryptIV=%s decryptIV=%s", + hex.EncodeToString(cs.key[:]), + hex.EncodeToString(cs.encryptIV[:]), + hex.EncodeToString(cs.decryptIV[:])) + } return nil } @@ -189,6 +220,8 @@ func (cs *cryptState15) setup15(key, clientNonce, serverNonce []byte) error { // encrypt15 encrypts plaintext for Mumble 1.5 native UDP. // Returns [iv_byte(1)][tag(3)][ciphertext]. func (cs *cryptState15) encrypt15(plaintext []byte) ([]byte, error) { + cs.mu.Lock() + defer cs.mu.Unlock() if !cs.initialized { return nil, errors.New("gumble: crypto not initialized") } @@ -210,6 +243,8 @@ func (cs *cryptState15) encrypt15(plaintext []byte) ([]byte, error) { // decrypt15 decrypts a Mumble 1.5 native UDP packet. // Matches wumble's decrypt: advances IV when decrypt_iv[0]+1 == iv_byte. func (cs *cryptState15) decrypt15(packet []byte) ([]byte, error) { + cs.mu.Lock() + defer cs.mu.Unlock() if !cs.initialized { return nil, errors.New("gumble: crypto not initialized") } @@ -309,8 +344,6 @@ func backupIV(iv []byte) { } } - - // --------------------------------------------------------------------------- // OCB variant for Mumble 1.5 native UDP. // Matches the implementation in Wumble's crypt_state.cr. @@ -480,70 +513,46 @@ func xorBytes(dst, a, b []byte) { } } -// --------------------------------------------------------------------------- -// Global 1.5 crypto state. -// --------------------------------------------------------------------------- - -var udp15Client *cryptState15 -var udp15Server *cryptState15 - -// SetUDP15Crypto installs the 1.5 native UDP crypto state from CryptSetup. -// udp15Client: encrypts with client_nonce (outbound). -// udp15Server: decrypts with server_nonce (inbound). -func SetUDP15Crypto(key, clientNonce, serverNonce []byte) { - csClient := &cryptState15{} - if err := csClient.setup15(key, clientNonce, serverNonce); err != nil { - log.Error("SetUDP15Crypto client: %v", err) - return +// setUDP15Crypto installs per-client 1.5 UDP crypto state from CryptSetup. +func (c *Client) setUDP15Crypto(key, clientNonce, serverNonce []byte) error { + outbound := &cryptState15{} + if err := outbound.setup15(key, clientNonce, serverNonce); err != nil { + return err } - udp15Client = csClient - - // Server state: we only use decryptIV (to decrypt server→client packets). - // The server encrypts with server_nonce, so our decryptIV must be server_nonce. - csServer := &cryptState15{} - copy(csServer.key[:], key) - copy(csServer.decryptIV[:], serverNonce) - csServer.initialized = true - udp15Server = csServer - + inbound := &cryptState15{} + // Server-to-client packets use serverNonce as decryptIV. + if err := inbound.setup15(key, clientNonce, serverNonce); err != nil { + return err + } + c.udpWriteMu.Lock() + c.udpMu.Lock() + c.udpCryptoOut = outbound + c.udpCryptoIn = inbound + c.udpFrameNumber = 0 + c.udpMu.Unlock() + c.udpWriteMu.Unlock() log.Info("Mumble 1.5 native UDP crypto initialized") -} - -// --------------------------------------------------------------------------- -// Frame number tracking. -// --------------------------------------------------------------------------- - -var udp15FrameNumber uint32 - -// nextFrameNumber returns the next frame number for outbound 1.5 UDP audio. -func nextFrameNumber() uint32 { - n := udp15FrameNumber - udp15FrameNumber++ - return n + return nil } // --------------------------------------------------------------------------- // WriteAudioUDP15 writes an encrypted audio packet using Mumble 1.5 native UDP. // Returns true if sent, false if TCP should be used. -func (c *Client) WriteAudioUDP15(data []byte, final bool) (bool, error) { - cs := udp15Client - if cs == nil || !cs.initialized { +func (c *Client) WriteAudioUDP15(target uint32, data []byte, final bool) (bool, error) { + // Encryption and socket writes must remain ordered: otherwise a later + // packet can reach the server before the packet with the preceding IV. + c.udpWriteMu.Lock() + defer c.udpWriteMu.Unlock() + c.udpMu.Lock() + cs, udpConn := c.udpCryptoOut, c.udpConn + frameNum := c.udpFrameNumber + c.udpFrameNumber++ + c.udpMu.Unlock() + if cs == nil || udpConn == nil { return false, nil } - // Snapshot udpConn to avoid race with disconnect cleanup nil'ing it. - udpConn := c.udpConn - if udpConn == nil { - return false, nil - } - - frameNum := nextFrameNumber() - - // Build MumbleUDP.Audio: type byte 0x00 + protobuf. - // Matches reference client: 0x00 + target=0 + frame_number + opus_data [+ is_terminator] - payload := append([]byte{0x00}, encodeUDPAudio(0, frameNum, data, final)...) - - // Encrypt with 1.5 format. + payload := append([]byte{0x00}, encodeUDPAudio(target, frameNum, data, final)...) encrypted, err := cs.encrypt15(payload) if err != nil { log.Error("UDP15 encrypt failed: %v", err) @@ -562,21 +571,20 @@ func (c *Client) WriteAudioUDP15(data []byte, final bool) (bool, error) { } // HandleUDPPacket15 processes an incoming Mumble 1.5 native UDP packet. -var firstUDP15Recv bool - func (c *Client) HandleUDPPacket15(packet []byte, pktNum uint64) { if len(packet) < udp15HeaderSize { log.Warn("UDP15 #%d: packet too short (%d bytes)", pktNum, len(packet)) return } - if !firstUDP15Recv { - firstUDP15Recv = true - log.Info("UDP15 #%d: first packet received! hex=%s", pktNum, hex.EncodeToString(packet)) + if !c.udpFirstRecv.Swap(true) && log.Enabled(log.LevelDebug) { + log.Debug("UDP15 #%d: first packet received! hex=%s", pktNum, hex.EncodeToString(packet)) } - cs := udp15Server - if cs == nil || !cs.initialized { + c.udpMu.RLock() + cs := c.udpCryptoIn + c.udpMu.RUnlock() + if cs == nil { log.Warn("UDP15 #%d: crypto not initialized", pktNum) return } @@ -587,7 +595,7 @@ func (c *Client) HandleUDPPacket15(packet []byte, pktNum uint64) { return } - log.Info("UDP15 #%d: decrypt OK, plaintext_len=%d", pktNum, len(plaintext)) + log.Debug("UDP15 #%d: decrypt OK, plaintext_len=%d", pktNum, len(plaintext)) // Check type byte (0x00 = Audio, 0x01 = Ping) if len(plaintext) < 1 { @@ -604,8 +612,8 @@ func (c *Client) HandleUDPPacket15(packet []byte, pktNum uint64) { // MumbleUDP.Audio protobuf format (1.5 native). if msgType == 0x00 { - session, frameNum, opusData, terminator := decodeUDPAudio(plaintext) - c.dispatchOpus15(pktNum, session, int64(frameNum), opusData, terminator) + session, frameNum, opusData, terminator, context, position, volumeAdjustment := decodeUDPAudio(plaintext) + c.dispatchOpus15(pktNum, session, int64(frameNum), opusData, terminator, context, position, volumeAdjustment) return } @@ -620,7 +628,7 @@ func (c *Client) HandleUDPPacket15(packet []byte, pktNum uint64) { // dispatchOpus15 processes a decoded MumbleUDP.Audio frame and dispatches // the decoded PCM to audio listeners. -func (c *Client) dispatchOpus15(pktNum uint64, session uint32, frameNum int64, opusData []byte, terminator bool) { +func (c *Client) dispatchOpus15(pktNum uint64, session uint32, frameNum int64, opusData []byte, terminator bool, context uint32, position *[3]float32, volumeAdjustment float32) { if len(opusData) == 0 && !terminator { log.Info("UDP15 #%d: no opus data (session=%d frame=%d), skipping", pktNum, session, frameNum) return @@ -654,7 +662,7 @@ func (c *Client) dispatchOpus15(pktNum uint64, session uint32, frameNum int64, o return } - c.decodeAndDispatch(pktNum, user, decoder, frameNum, opusData) + c.decodeAndDispatch(pktNum, user, decoder, frameNum, opusData, context, position, volumeAdjustment) } // handleLegacyUDPVoice parses the legacy UDPVoice format (type byte 0x80) @@ -699,11 +707,11 @@ func (c *Client) handleLegacyUDPVoice(pktNum uint64, data []byte) { log.Info("UDP15 #%d: legacy voice session=%d seq=%d opus_len=%d term=%v", pktNum, session, seq, len(opusData), terminator) - c.dispatchOpus15(pktNum, uint32(session), seq, opusData, terminator) + c.dispatchOpus15(pktNum, uint32(session), seq, opusData, terminator, 0, nil, 0) } // decodeAndDispatch decodes an Opus frame and dispatches PCM to audio listeners. -func (c *Client) decodeAndDispatch(pktNum uint64, user *User, decoder AudioDecoder, frameNum int64, opusData []byte) { +func (c *Client) decodeAndDispatch(pktNum uint64, user *User, decoder AudioDecoder, frameNum int64, opusData []byte, context uint32, position *[3]float32, volumeAdjustment float32) { // Detect sequence gaps. if user.audioSequenceValid { gap := frameNum - user.audioSequence @@ -732,14 +740,19 @@ func (c *Client) decodeAndDispatch(pktNum uint64, user *User, decoder AudioDecod return } - log.Info("UDP15 #%d: Opus OK for %s, pcm_samples=%d", pktNum, user.Name, len(pcm)) + log.Debug("UDP15 #%d: Opus OK for %s, pcm_samples=%d", pktNum, user.Name, len(pcm)) event := AudioPacket{ - Client: c, - Sender: user, - Target: &VoiceTarget{ID: 0}, - Sequence: frameNum, - AudioBuffer: AudioBuffer(pcm), + Client: c, + Sender: user, + Target: &VoiceTarget{ID: context}, + Sequence: frameNum, + AudioBuffer: AudioBuffer(pcm), + VolumeAdjustment: volumeAdjustment, + } + if position != nil { + event.HasPosition = true + event.X, event.Y, event.Z = position[0], position[1], position[2] } c.dispatchAudio(user, &event) } diff --git a/gumble/gumble/udp15_test.go b/gumble/gumble/udp15_test.go index f23867c..e3ea5ce 100644 --- a/gumble/gumble/udp15_test.go +++ b/gumble/gumble/udp15_test.go @@ -2,7 +2,9 @@ package gumble import ( "bytes" + "encoding/binary" "encoding/hex" + "math" "testing" ) @@ -40,9 +42,9 @@ func TestOCB15RoundTrip(t *testing.T) { }{ {"empty", []byte{}}, {"1 byte", []byte{0x41}}, - {"15 bytes", []byte("hello world 1234")}, // 15 - {"16 bytes", []byte("hello world 12345")}, // exactly 1 block - {"17 bytes", []byte("hello world 123456")}, // 1 full + 1 partial + {"15 bytes", []byte("hello world 1234")}, // 15 + {"16 bytes", []byte("hello world 12345")}, // exactly 1 block + {"17 bytes", []byte("hello world 123456")}, // 1 full + 1 partial {"32 bytes", []byte("hello world 12345678901234567")}, // exactly 2 blocks {"33 bytes", []byte("hello world 123456789012345678")}, {"100 bytes", bytes.Repeat([]byte{0x41}, 100)}, @@ -140,13 +142,13 @@ func TestUDPAudioProtobuf(t *testing.T) { } for _, tt := range tests { - encoded := encodeUDPAudio(tt.session, tt.frameNumber, tt.opusData, tt.terminator) - session, frameNum, opusData, terminator := decodeUDPAudio(encoded) + encoded := encodeUDPAudio(uint32(tt.session), uint64(tt.frameNumber), tt.opusData, tt.terminator) + session, frameNum, opusData, terminator, _, _, _ := decodeUDPAudio(encoded) - if session != tt.session { - t.Errorf("session: got %d, want %d", session, tt.session) + if session != 0 { + t.Errorf("outbound packet unexpectedly contains session %d", session) } - if frameNum != tt.frameNumber { + if frameNum != uint64(tt.frameNumber) { t.Errorf("frameNumber: got %d, want %d", frameNum, tt.frameNumber) } if !bytes.Equal(opusData, tt.opusData) { @@ -158,6 +160,39 @@ func TestUDPAudioProtobuf(t *testing.T) { } } +func TestUDPAudioProtobufIncomingFields(t *testing.T) { + var packet bytes.Buffer + writeVarint := func(v uint64) { + var buf [10]byte + n := pbEncodeVarint(buf[:], v) + packet.Write(buf[:n]) + } + writeVarint(2<<3 | 0) // context + writeVarint(3) + writeVarint(3<<3 | 0) // sender_session + writeVarint(123) + writeVarint(4<<3 | 0) // frame_number + writeVarint(1 << 32) + writeVarint(5<<3 | 2) // opus_data + writeVarint(2) + packet.Write([]byte{0xaa, 0xbb}) + writeVarint(6<<3 | 2) // packed positional_data + writeVarint(12) + for _, f := range []uint32{math.Float32bits(1), math.Float32bits(2), math.Float32bits(3)} { + var buf [4]byte + binary.LittleEndian.PutUint32(buf[:], f) + packet.Write(buf[:]) + } + + session, frame, opusData, terminator, context, position, _ := decodeUDPAudio(packet.Bytes()) + if session != 123 || frame != 1<<32 || !bytes.Equal(opusData, []byte{0xaa, 0xbb}) || terminator || context != 3 { + t.Fatalf("decoded unexpected audio: session=%d frame=%d opus=%x terminator=%v context=%d", session, frame, opusData, terminator, context) + } + if position == nil || *position != [3]float32{1, 2, 3} { + t.Fatalf("position = %v, want [1 2 3]", position) + } +} + func mustDecodeHex(s string) []byte { b, err := hex.DecodeString(s) if err != nil { diff --git a/gumble/gumbleopenal/stream.go b/gumble/gumbleopenal/stream.go index 88eb976..b7ebdb0 100644 --- a/gumble/gumbleopenal/stream.go +++ b/gumble/gumbleopenal/stream.go @@ -536,7 +536,7 @@ func (s *Stream) processAudioPacket(packet *gumble.AudioPacket, user *gumble.Use rawPtr := 0 if isStereo { // Process stereo samples as pairs - for i := 0; i < samples*2; i += 2 { + for i := 0; i < samples; i += 2 { // Process left channel with saturation protection sample := packet.AudioBuffer[i] if boost > 1 { diff --git a/log/log.go b/log/log.go index c73aea6..4003911 100644 --- a/log/log.go +++ b/log/log.go @@ -7,6 +7,7 @@ import ( "io" "os" "sync" + "sync/atomic" "time" ) @@ -40,10 +41,16 @@ type Logger interface { Log(level Level, format string, args ...interface{}) } -var ( - mu sync.Mutex - logger Logger = &nopLogger{} -) +type loggerState struct { + logger Logger + level Level +} + +var logger atomic.Pointer[loggerState] + +func init() { + logger.Store(&loggerState{logger: &nopLogger{}, level: LevelError + 1}) +} type nopLogger struct{} @@ -51,21 +58,28 @@ func (n *nopLogger) Log(level Level, format string, args ...interface{}) {} // SetLogger sets the destination for log messages. Pass nil to disable. func SetLogger(l Logger) { - mu.Lock() - defer mu.Unlock() + state := &loggerState{logger: l, level: LevelDebug} if l == nil { - logger = &nopLogger{} - } else { - logger = l + state.logger = &nopLogger{} + state.level = LevelError + 1 + } else if writer, ok := l.(*WriterLogger); ok { + state.level = writer.level } + logger.Store(state) +} + +// Enabled reports whether messages at level will be emitted. Callers should +// use it to avoid computing expensive log arguments when logging is disabled. +func Enabled(level Level) bool { + return level >= logger.Load().level } // WriterLogger is a simple Logger that writes to an io.Writer. type WriterLogger struct { - mu sync.Mutex - w io.Writer - level Level - buf []byte + mu sync.Mutex + w io.Writer + level Level + buf []byte } // NewWriterLogger creates a logger that writes to w, filtering below level. @@ -88,29 +102,29 @@ func (wl *WriterLogger) Log(level Level, format string, args ...interface{}) { } func Debug(format string, args ...interface{}) { - mu.Lock() - l := logger - mu.Unlock() - l.Log(LevelDebug, format, args...) + state := logger.Load() + if LevelDebug >= state.level { + state.logger.Log(LevelDebug, format, args...) + } } func Info(format string, args ...interface{}) { - mu.Lock() - l := logger - mu.Unlock() - l.Log(LevelInfo, format, args...) + state := logger.Load() + if LevelInfo >= state.level { + state.logger.Log(LevelInfo, format, args...) + } } func Warn(format string, args ...interface{}) { - mu.Lock() - l := logger - mu.Unlock() - l.Log(LevelWarn, format, args...) + state := logger.Load() + if LevelWarn >= state.level { + state.logger.Log(LevelWarn, format, args...) + } } func Error(format string, args ...interface{}) { - mu.Lock() - l := logger - mu.Unlock() - l.Log(LevelError, format, args...) + state := logger.Load() + if LevelError >= state.level { + state.logger.Log(LevelError, format, args...) + } }