gilles/go2rtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'AlexxIT:master' into wyze

Browse Source
This commit is contained in:
seydx
2026-01-01 05:25:58 +01:00
committed by GitHub
parent a42ab88dbd 4dae65a535
commit 7d4b3fe65b
19 changed files with 2826 additions and 511 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/webrtc/api.go
+123 -38
View File
@@ -1,7 +1,9 @@
package webrtc
import (
"fmt"
"net"
"slices"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/xnet"
@@ -27,6 +29,69 @@ type Filters struct {
UDPPorts []uint16 `yaml:"udp_ports"`
}
func (f *Filters) Network(protocol string) string {
if f == nil || f.Networks == nil {
return protocol
}
v4 := slices.Contains(f.Networks, protocol+"4")
v6 := slices.Contains(f.Networks, protocol+"6")
if v4 && v6 {
return protocol
} else if v4 {
return protocol + "4"
} else if v6 {
return protocol + "6"
}
return ""
}
func (f *Filters) NetIPs() (ips []net.IP) {
itfs, _ := net.Interfaces()
for _, itf := range itfs {
if itf.Flags&net.FlagUp == 0 {
continue
}
if !f.IncludeLoopback() && itf.Flags&net.FlagLoopback != 0 {
continue
}
if !f.InterfaceFilter(itf.Name) {
continue
}
addrs, _ := itf.Addrs()
for _, addr := range addrs {
ip := parseNetAddr(addr)
if ip == nil || !f.IPFilter(ip) {
continue
}
ips = append(ips, ip)
}
}
return
}
func parseNetAddr(addr net.Addr) net.IP {
switch addr := addr.(type) {
case *net.IPNet:
return addr.IP
case *net.IPAddr:
return addr.IP
}
return nil
}
func (f *Filters) IncludeLoopback() bool {
return f != nil && f.Loopback
}
func (f *Filters) InterfaceFilter(name string) bool {
return f == nil || f.Interfaces == nil || slices.Contains(f.Interfaces, name)
}
func (f *Filters) IPFilter(ip net.IP) bool {
return f == nil || f.IPs == nil || core.Contains(f.IPs, ip.String())
}
func NewServerAPI(network, address string, filters *Filters) (*webrtc.API, error) {
// for debug logs add to env: `PION_LOG_DEBUG=all`
m := &webrtc.MediaEngine{}
@@ -99,48 +164,17 @@ func NewServerAPI(network, address string, filters *Filters) (*webrtc.API, error
_ = s.SetEphemeralUDPPortRange(filters.UDPPorts[0], filters.UDPPorts[1])
}
//if len(hosts) != 0 {
// // support only: host, srflx
// if candidateType, err := webrtc.NewICECandidateType(hosts[0]); err == nil {
// s.SetNAT1To1IPs(hosts[1:], candidateType)
// } else {
// s.SetNAT1To1IPs(hosts, 0) // 0 = host
// }
//}
// If you don't specify an address, this won't cause an error.
// Connections can still be established using random UDP addresses.
if address != "" {
// Both newMux functions respect filters and do not raise an error
// if the port cannot be listened on.
if network == "" || network == "tcp" {
if ln, err := net.Listen("tcp", address); err == nil {
tcpMux := webrtc.NewICETCPMux(nil, ln, 8)
s.SetICETCPMux(tcpMux)
}
tcpMux := newTCPMux(address, filters)
s.SetICETCPMux(tcpMux)
}
if network == "" || network == "udp" {
// UDPMuxDefault should not listening on unspecified address, use NewMultiUDPMuxFromPort instead
var udpMux ice.UDPMux
if port := xnet.ParseUnspecifiedPort(address); port != 0 {
var networks []ice.NetworkType
for _, ntype := range networkTypes {
networks = append(networks, ice.NetworkType(ntype))
}
var err error
if udpMux, err = ice.NewMultiUDPMuxFromPort(
port,
ice.UDPMuxFromPortWithInterfaceFilter(interfaceFilter),
ice.UDPMuxFromPortWithIPFilter(ipFilter),
ice.UDPMuxFromPortWithNetworks(networks...),
); err != nil {
return nil, err
}
} else {
ln, err := net.ListenPacket("udp", address)
if err != nil {
return nil, err
}
udpMux = ice.NewUDPMuxDefault(ice.UDPMuxParams{UDPConn: ln})
}
udpMux := newUDPMux(address, filters)
s.SetICEUDPMux(udpMux)
}
}
@@ -152,6 +186,57 @@ func NewServerAPI(network, address string, filters *Filters) (*webrtc.API, error
), nil
}
// OnNewListener temporary ugly solution for log
var OnNewListener = func(ln any) {}
func newTCPMux(address string, filters *Filters) ice.TCPMux {
networkTCP := filters.Network("tcp") // tcp or tcp4 or tcp6
if ln, _ := net.Listen(networkTCP, address); ln != nil {
OnNewListener(ln)
return webrtc.NewICETCPMux(nil, ln, 8)
}
return nil
}
func newUDPMux(address string, filters *Filters) ice.UDPMux {
host, port, err := net.SplitHostPort(address)
if err != nil {
return nil
}
// UDPMux should not listening on unspecified address.
// So we will create a listener on all available interfaces.
// We can't use ice.NewMultiUDPMuxFromPort, because it sometimes crashes with an error:
// listen udp [***]:8555: bind: cannot assign requested address
var addrs []string
if host == "" {
for _, ip := range filters.NetIPs() {
addrs = append(addrs, fmt.Sprintf("%s:%s", ip, port))
}
} else {
addrs = []string{address}
}
networkUDP := filters.Network("udp") // udp or udp4 or udp6
var muxes []ice.UDPMux
for _, addr := range addrs {
if ln, _ := net.ListenPacket(networkUDP, addr); ln != nil {
OnNewListener(ln)
mux := ice.NewUDPMuxDefault(ice.UDPMuxParams{UDPConn: ln})
muxes = append(muxes, mux)
}
}
switch len(muxes) {
case 0:
return nil
case 1:
return muxes[0]
}
return ice.NewMultiUDPMuxDefault(muxes...)
}
func RegisterDefaultCodecs(m *webrtc.MediaEngine) error {
for _, codec := range []webrtc.RTPCodecParameters{
{
pkg/xiaomi/cs2/conn.go
+201 -84
View File
@@ -1,6 +1,7 @@
package cs2
import (
"bufio"
"encoding/binary"
"fmt"
"io"
@@ -10,33 +11,27 @@ import (
"time"
)
func Dial(host string) (*Conn, error) {
conn, err := net.ListenUDP("udp", nil)
func Dial(host, transport string) (*Conn, error) {
conn, err := handshake(host, transport)
if err != nil {
return nil, err
}
_, isTCP := conn.(*tcpConn)
c := &Conn{
conn: conn,
addr: &net.UDPAddr{IP: net.ParseIP(host), Port: 32108},
conn: conn,
isTCP: isTCP,
rawCh0: make(chan []byte, 10),
rawCh2: make(chan []byte, 100),
}
if err = c.handshake(); err != nil {
_ = conn.Close()
return nil, err
}
c.rawCh0 = make(chan []byte, 10)
c.rawCh2 = make(chan []byte, 100)
go c.worker()
return c, nil
}
type Conn struct {
conn *net.UDPConn
addr *net.UDPAddr
conn net.Conn
isTCP bool
err error
seqCh0 uint16
@@ -53,30 +48,58 @@ const (
magicDrw = 0xD1
msgLanSearch = 0x30
msgPunchPkt = 0x41
msgP2PRdy = 0x42
msgP2PRdyUDP = 0x42
msgP2PRdyTCP = 0x43
msgDrw = 0xD0
msgDrwAck = 0xD1
msgAlive = 0xE0
msgPing = 0xE0
msgPong = 0xE1
msgClose = 0xF1
)
func (c *Conn) handshake() error {
_ = c.SetDeadline(time.Now().Add(5 * time.Second))
buf, err := c.WriteAndWait([]byte{magic, msgLanSearch, 0, 0}, msgPunchPkt)
func handshake(host, transport string) (net.Conn, error) {
conn, err := newUDPConn(host, 32108)
if err != nil {
return fmt.Errorf("%s: read punch: %w", "cs2", err)
return nil, err
}
_, err = c.WriteAndWait(buf, msgP2PRdy)
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
req := []byte{magic, msgLanSearch, 0, 0}
res, err := conn.(*udpConn).WriteUntil(req, func(res []byte) bool {
return res[1] == msgPunchPkt
})
if err != nil {
return fmt.Errorf("%s: read ready: %w", "cs2", err)
_ = conn.Close()
return nil, err
}
_ = c.Write([]byte{magic, msgAlive, 0, 0})
var msgUDP, msgTCP byte
_ = c.SetDeadline(time.Time{})
if transport == "" || transport == "udp" {
msgUDP = msgP2PRdyUDP
}
if transport == "" || transport == "tcp" {
msgTCP = msgP2PRdyTCP
}
return nil
res, err = conn.(*udpConn).WriteUntil(res, func(res []byte) bool {
return res[1] == msgUDP || res[1] == msgTCP
})
if err != nil {
_ = conn.Close()
return nil, err
}
_ = conn.SetDeadline(time.Time{})
if res[1] == msgTCP {
_ = conn.Close()
//host := fmt.Sprintf("%d.%d.%d.%d:%d", b[31], b[30], b[29], b[28], uint16(b[27])<<8|uint16(b[26]))
return newTCPConn(conn.RemoteAddr().String())
}
return conn, nil
}
func (c *Conn) worker() {
@@ -85,38 +108,41 @@ func (c *Conn) worker() {
close(c.rawCh2)
}()
chAck := make([]uint16, 4)
chAck := make([]uint16, 4) // only for UDP
buf := make([]byte, 1200)
var ch2WaitSize int
var ch2WaitData []byte
var keepaliveTS time.Time
for {
n, addr, err := c.conn.ReadFromUDP(buf)
n, err := c.conn.Read(buf)
if err != nil {
c.err = fmt.Errorf("%s: %w", "cs2", err)
return
}
if string(addr.IP) != string(c.addr.IP) || n < 8 || buf[0] != magic {
continue // skip messages from another IP
}
//log.Printf("<- %x", buf[:n])
switch buf[1] {
case msgDrw:
ch := buf[5]
seqHI := buf[6]
seqLO := buf[7]
if chAck[ch] != uint16(seqHI)<<8|uint16(seqLO) {
continue
}
chAck[ch]++
if c.isTCP {
// For TCP we should using ping/pong.
if now := time.Now(); now.After(keepaliveTS) {
_, _ = c.conn.Write([]byte{magic, msgPing, 0, 0})
keepaliveTS = now.Add(5 * time.Second)
}
} else {
// For UDP we should using ack.
seqHI := buf[6]
seqLO := buf[7]
ack := []byte{magic, msgDrwAck, 0, 6, magicDrw, ch, 0, 1, seqHI, seqLO}
if _, err = c.conn.WriteToUDP(ack, c.addr); err != nil {
return
if chAck[ch] != uint16(seqHI)<<8|uint16(seqLO) {
continue
}
chAck[ch]++
ack := []byte{magic, msgDrwAck, 0, 6, magicDrw, ch, 0, 1, seqHI, seqLO}
_, _ = c.conn.Write(ack)
}
switch ch {
@@ -152,9 +178,12 @@ func (c *Conn) worker() {
continue
}
case msgP2PRdy: // skip it
case msgPing:
_, _ = c.conn.Write([]byte{magic, msgPong, 0, 0})
continue
case msgDrwAck:
case msgPong, msgP2PRdyUDP, msgP2PRdyTCP, msgClose:
continue // skip it
case msgDrwAck: // only for UDP
if c.cmdAck != nil {
c.cmdAck()
}
@@ -165,42 +194,15 @@ func (c *Conn) worker() {
}
}
func (c *Conn) Write(req []byte) error {
//log.Printf("-> %x", req)
_, err := c.conn.WriteToUDP(req, c.addr)
return err
}
func (c *Conn) WriteAndWait(req []byte, waitMsg uint8) ([]byte, error) {
var t *time.Timer
t = time.AfterFunc(1, func() {
if err := c.Write(req); err == nil && t != nil {
t.Reset(time.Second)
}
})
defer t.Stop()
buf := make([]byte, 1200)
for {
n, addr, err := c.conn.ReadFromUDP(buf)
if err != nil {
return nil, err
}
if string(addr.IP) != string(c.addr.IP) || n < 16 {
continue // skip messages from another IP
}
if buf[0] == magic && buf[1] == waitMsg {
c.addr.Port = addr.Port
return buf[:n], nil
}
func (c *Conn) Protocol() string {
if c.isTCP {
return "cs2+tcp"
}
return "cs2+udp"
}
func (c *Conn) RemoteAddr() net.Addr {
return c.addr
return c.conn.RemoteAddr()
}
func (c *Conn) SetDeadline(t time.Time) error {
@@ -232,6 +234,14 @@ func (c *Conn) WriteCommand(cmd uint16, data []byte) error {
c.cmdMu.Lock()
defer c.cmdMu.Unlock()
req := marshalCmd(0, c.seqCh0, uint32(cmd), data)
c.seqCh0++
if c.isTCP {
_, err := c.conn.Write(req)
return err
}
var repeat atomic.Int32
repeat.Store(5)
@@ -243,11 +253,8 @@ func (c *Conn) WriteCommand(cmd uint16, data []byte) error {
timeout.Reset(1)
}
req := marshalCmd(0, c.seqCh0, uint32(cmd), data)
c.seqCh0++
for {
if err := c.Write(req); err != nil {
if _, err := c.conn.Write(req); err != nil {
return err
}
<-timeout.C
@@ -285,7 +292,8 @@ func (c *Conn) WritePacket(data []byte) error {
binary.BigEndian.PutUint32(req[8:], n)
copy(req[offset:], data)
return c.Write(req)
_, err := c.conn.Write(req)
return err
}
func marshalCmd(channel byte, seq uint16, cmd uint32, payload []byte) []byte {
@@ -313,3 +321,112 @@ func marshalCmd(channel byte, seq uint16, cmd uint32, payload []byte) []byte {
return req
}
func newUDPConn(host string, port int) (net.Conn, error) {
// We using raw net.UDPConn, because RemoteAddr should be changed during handshake.
conn, err := net.ListenUDP("udp", nil)
if err != nil {
return nil, err
}
addr, err := net.ResolveUDPAddr("udp", host)
if err != nil {
addr = &net.UDPAddr{IP: net.ParseIP(host), Port: port}
}
return &udpConn{UDPConn: conn, addr: addr}, nil
}
type udpConn struct {
*net.UDPConn
addr *net.UDPAddr
}
func (c *udpConn) Read(p []byte) (n int, err error) {
var addr *net.UDPAddr
for {
n, addr, err = c.UDPConn.ReadFromUDP(p)
if err != nil {
return 0, err
}
if string(addr.IP) == string(c.addr.IP) || n >= 8 {
return
}
}
}
func (c *udpConn) Write(req []byte) (n int, err error) {
//log.Printf("-> %x", req)
return c.UDPConn.WriteToUDP(req, c.addr)
}
func (c *udpConn) RemoteAddr() net.Addr {
return c.addr
}
func (c *udpConn) WriteUntil(req []byte, ok func(res []byte) bool) ([]byte, error) {
var t *time.Timer
t = time.AfterFunc(1, func() {
if _, err := c.Write(req); err == nil && t != nil {
t.Reset(time.Second)
}
})
defer t.Stop()
buf := make([]byte, 1200)
for {
n, addr, err := c.UDPConn.ReadFromUDP(buf)
if err != nil {
return nil, err
}
if string(addr.IP) != string(c.addr.IP) || n < 16 {
continue // skip messages from another IP
}
if ok(buf[:n]) {
c.addr.Port = addr.Port
return buf[:n], nil
}
}
}
func newTCPConn(addr string) (net.Conn, error) {
conn, err := net.DialTimeout("tcp", addr, 3*time.Second)
if err != nil {
return nil, err
}
return &tcpConn{conn.(*net.TCPConn), bufio.NewReader(conn)}, nil
}
type tcpConn struct {
*net.TCPConn
rd *bufio.Reader
}
func (c *tcpConn) Read(p []byte) (n int, err error) {
tmp := make([]byte, 8)
if _, err = io.ReadFull(c.rd, tmp); err != nil {
return
}
n = int(binary.BigEndian.Uint16(tmp))
if len(p) < n {
return 0, fmt.Errorf("tcp: buffer too small")
}
_, err = io.ReadFull(c.rd, p[:n])
//log.Printf("<- %x%x", tmp, p[:n])
return
}
func (c *tcpConn) Write(req []byte) (n int, err error) {
n = len(req)
buf := make([]byte, 8+n)
binary.BigEndian.PutUint16(buf, uint16(n))
buf[2] = 0x68
copy(buf[8:], req)
//log.Printf("-> %x", buf)
_, err = c.TCPConn.Write(buf)
return
}
pkg/xiaomi/miss/client.go
+24 -11
View File
@@ -33,7 +33,7 @@ func Dial(rawURL string) (*Client, error) {
switch s := query.Get("vendor"); s {
case "cs2":
c.conn, err = cs2.Dial(u.Host)
c.conn, err = cs2.Dial(u.Host, query.Get("transport"))
case "tutk":
c.conn, err = tutk.Dial(u.Host, query.Get("uid"))
default:
@@ -63,6 +63,7 @@ const (
)
type Conn interface {
Protocol() string
ReadCommand() (cmd uint16, data []byte, err error)
WriteCommand(cmd uint16, data []byte) error
ReadPacket() ([]byte, error)
@@ -77,6 +78,10 @@ type Client struct {
key []byte
}
func (c *Client) Protocol() string {
return c.conn.Protocol()
}
func (c *Client) RemoteAddr() net.Addr {
return c.conn.RemoteAddr()
}
@@ -89,16 +94,6 @@ func (c *Client) Close() error {
return c.conn.Close()
}
func (c *Client) Protocol() string {
switch c.conn.(type) {
case *cs2.Conn:
return "cs2+udp"
case *tutk.Conn:
return "tutk+udp"
}
return ""
}
const (
cmdAuthReq = 0x100
cmdAuthRes = 0x101
@@ -276,6 +271,24 @@ type Packet struct {
Payload []byte
}
func (p *Packet) SampleRate() uint32 {
// flag: 1 0011 000 - sample rate 16000
// flag: 100 00 01 0000 000 - sample rate 8000
v := (p.Flags >> 3) & 0b1111
if v != 0 {
return 16000
}
return 8000
}
//func (p *Packet) AudioUnknown1() byte {
// return byte((p.Flags >> 7) & 0b11)
//}
//
//func (p *Packet) AudioUnknown2() byte {
// return byte((p.Flags >> 9) & 0b11)
//}
func GenerateKey() ([]byte, []byte, error) {
public, private, err := box.GenerateKey(rand.Reader)
if err != nil {
pkg/xiaomi/producer.go
+1 -1
View File
@@ -110,7 +110,7 @@ func probe(client *miss.Client, channel, quality, audio uint8) ([]*core.Media, e
}
case miss.CodecPCMA:
if acodec == nil {
acodec = &core.Codec{Name: core.CodecPCMA, ClockRate: 8000}
acodec = &core.Codec{Name: core.CodecPCMA, ClockRate: pkt.SampleRate()}
}
case miss.CodecOPUS:
if acodec == nil {
pkg/xiaomi/tutk/conn.go
+4
View File
@@ -249,6 +249,10 @@ func (c *Conn) WriteAndWait(req []byte, ok func(req, res []byte) bool) ([]byte,
}
}
func (c *Conn) Protocol() string {
return "tutk+udp"
}
func (c *Conn) RemoteAddr() net.Addr {
return c.addr
}