From 1b99fcbe790e312781f2a66904c6042d2b3ea5f6 Mon Sep 17 00:00:00 2001 From: Winlin Date: Sat, 30 Dec 2023 08:55:01 +0800 Subject: [PATCH] A demo for SRT proxy. (#3869) See https://www.figma.com/file/kItb5HWOI4HimjDp62pas3/SRT-Proxy --- trunk/research/srt-proxy/.gitignore | 2 + trunk/research/srt-proxy/go.mod | 3 + trunk/research/srt-proxy/main.go | 400 ++++++++++++++++++++++++++++ 3 files changed, 405 insertions(+) create mode 100644 trunk/research/srt-proxy/.gitignore create mode 100644 trunk/research/srt-proxy/go.mod create mode 100644 trunk/research/srt-proxy/main.go diff --git a/trunk/research/srt-proxy/.gitignore b/trunk/research/srt-proxy/.gitignore new file mode 100644 index 000000000..5281c15ee --- /dev/null +++ b/trunk/research/srt-proxy/.gitignore @@ -0,0 +1,2 @@ +.idea +gosrt \ No newline at end of file diff --git a/trunk/research/srt-proxy/go.mod b/trunk/research/srt-proxy/go.mod new file mode 100644 index 000000000..fa94f6a67 --- /dev/null +++ b/trunk/research/srt-proxy/go.mod @@ -0,0 +1,3 @@ +module srt-proxy + +go 1.18 diff --git a/trunk/research/srt-proxy/main.go b/trunk/research/srt-proxy/main.go new file mode 100644 index 000000000..5aff5d0a8 --- /dev/null +++ b/trunk/research/srt-proxy/main.go @@ -0,0 +1,400 @@ +package main + +import ( + "context" + "encoding/binary" + "fmt" + "net" + "os" + "time" +) + +// Use FFmpeg to push stream to this proxy: +// ffmpeg -re -i ~/git/srs/trunk/doc/source.flv -c copy -pes_payload_size 0 -f mpegts 'srt://localhost:10081?streamid=#!::r=live/livestream?m=publish' +// Play by SRT from this proxy: +// ffplay 'srt://localhost:10081?streamid=#!::r=live/livestream,latency=20,m=request' +var listenAddress = "127.0.0.1:10081" + +// Proxy to backend SRS Server. +// Play by HTTP-FLV from SRS: +// ffplay http://localhost:8080/live/livestream.flv +// Play by SRT from SRS: +// ffplay 'srt://localhost:10080?streamid=#!::r=live/livestream,latency=20,m=request' +var backendAddress = "127.0.0.1:10080" + +func main() { + fmt.Println("Hello, SRT!") + if err := doMain(context.Background()); err != nil { + fmt.Println(fmt.Sprintf("err %+v", err)) + os.Exit(1) + } +} + +func doMain(ctx context.Context) error { + serverAddr, err := net.ResolveUDPAddr("udp", listenAddress) + if err != nil { + return err + } + + server, err := net.ListenUDP("udp", serverAddr) + if err != nil { + return err + } + defer server.Close() + fmt.Println("UDP server listening on", server.LocalAddr().String()) + + start := time.Now() + + buf := make([]byte, 4096) + connections := make(map[string]*SRTConnection) + for { + n, clientAddr, err := server.ReadFromUDP(buf) + if err != nil { + return err + } + + connection, ok := connections[clientAddr.String()] + if !ok { + connection = &SRTConnection{ + start: start, + server: server, + serverAddr: serverAddr, + clientAddr: clientAddr, + } + connections[clientAddr.String()] = connection + fmt.Println("New connection from", clientAddr.String()) + } + + if err := connection.Consume(buf[:n]); err != nil { + return err + } + + fmt.Println(fmt.Sprintf("Received %v bytes from %s", n, clientAddr.String())) + } + + return nil +} + +type SRTConnection struct { + // Listener start time. + start time.Time + // Local UDP server connection. + server *net.UDPConn + // Local UDP server listen address. + serverAddr *net.UDPAddr + // Client remote address. + clientAddr *net.UDPAddr + // Backend server connection. + backend *net.UDPConn + + // Handshake packets with client. + handshake0 *SRTHandshakePacket + handshake1 *SRTHandshakePacket + handshake2 *SRTHandshakePacket + handshake3 *SRTHandshakePacket +} + +func (v *SRTConnection) Close() error { + if v.backend != nil { + return v.backend.Close() + } + return nil +} + +func (v *SRTConnection) Consume(b []byte) error { + pkt := &SRTHandshakePacket{} + if err := pkt.UnmarshalBinary(b); err != nil { + return err + } + + // Handle handshake messages. + if pkt.IsHandshake() { + if pkt.SynCookie == 0 { + // Save handshake packet. + v.handshake0 = pkt + fmt.Println(fmt.Sprintf("Handshake 0: %v", v.handshake0.String())) + + // Response handshake 1. + v.handshake1 = &SRTHandshakePacket{ + ControlFlag: pkt.ControlFlag, + ControlType: 0, + SubType: 0, + AdditionalInfo: 0, + Timestamp: uint32(time.Since(v.start).Microseconds()), + SocketID: pkt.SRTSocketID, + Version: 5, + EncryptionField: 0, + ExtensionField: 0x4A17, + InitSequence: pkt.InitSequence, + MTU: pkt.MTU, + FlowWindow: pkt.FlowWindow, + HandshakeType: 1, + SRTSocketID: pkt.SRTSocketID, + SynCookie: 0x418d5e4e, + PeerIP: v.serverAddr.IP, + } + fmt.Println(fmt.Sprintf("Handshake 1: %v", v.handshake1.String())) + + if b, err := v.handshake1.MarshalBinary(); err != nil { + return err + } else if _, err = v.server.WriteToUDP(b, v.clientAddr); err != nil { + return err + } + + return nil + } else { + // Save handshake packet. + v.handshake2 = pkt + fmt.Println(fmt.Sprintf("Handshake 2: %v", v.handshake2.String())) + + // Ignore if already connected. + if v.backend == nil { + remoteAddress, err := net.ResolveUDPAddr("udp", backendAddress) + if err != nil { + return err + } + + if v.backend, err = net.DialUDP("udp", nil, remoteAddress); err != nil { + return err + } + } + + // Proxy handshake 0 to backend server. + if b, err := v.handshake0.MarshalBinary(); err != nil { + return err + } else if _, err = v.backend.Write(b); err != nil { + return err + } + fmt.Println(fmt.Sprintf("Proxy send handshake 0: %v", v.handshake0.String())) + + // Read handshake 1 from backend server. + b := make([]byte, 4096) + handshake1p := &SRTHandshakePacket{} + if nn, err := v.backend.Read(b); err != nil { + return err + } else if err := handshake1p.UnmarshalBinary(b[:nn]); err != nil { + return err + } + fmt.Println(fmt.Sprintf("Proxy got handshake 1: %v", handshake1p.String())) + + // Proxy handshake 2 to backend server. + handshake2p := *v.handshake2 + handshake2p.SynCookie = handshake1p.SynCookie + if b, err := handshake2p.MarshalBinary(); err != nil { + return err + } else if _, err = v.backend.Write(b); err != nil { + return err + } + fmt.Println(fmt.Sprintf("Proxy send handshake 2: %v", handshake2p.String())) + + // Read handshake 3 from backend server. + handshake3p := &SRTHandshakePacket{} + if nn, err := v.backend.Read(b); err != nil { + return err + } else if err := handshake3p.UnmarshalBinary(b[:nn]); err != nil { + return err + } + fmt.Println(fmt.Sprintf("Proxy got handshake 3: %v", handshake3p.String())) + + // Response handshake 3 to client. + v.handshake3 = &*handshake3p + v.handshake3.SynCookie = v.handshake1.SynCookie + fmt.Println(fmt.Sprintf("Handshake 3: %v", v.handshake3.String())) + + if b, err := v.handshake3.MarshalBinary(); err != nil { + return err + } else if _, err = v.server.WriteToUDP(b, v.clientAddr); err != nil { + return err + } + + // Start a goroutine to proxy message from backend to client. + go func() { + for { + nn, err := v.backend.Read(b) + if err != nil { + return + } else if _, err = v.server.WriteToUDP(b[:nn], v.clientAddr); err != nil { + return + } + fmt.Println(fmt.Sprintf("Proxy got %d bytes from backend server.", nn)) + } + }() + return nil + } + } + + // Proxy all other messages to backend server. + if _, err := v.backend.Write(b); err != nil { + return err + } + + fmt.Println(fmt.Sprintf("Packet: %v", pkt)) + return nil +} + +// See https://datatracker.ietf.org/doc/html/draft-sharabayko-srt-01#section-3.2 +// See https://datatracker.ietf.org/doc/html/draft-sharabayko-srt-01#section-3.2.1 +type SRTHandshakePacket struct { + // F: 1 bit. Packet Type Flag. The control packet has this flag set to + // "1". The data packet has this flag set to "0". + ControlFlag uint8 + // Control Type: 15 bits. Control Packet Type. The use of these bits + // is determined by the control packet type definition. + // Handshake control packets (Control Type = 0x0000) are used to + // exchange peer configurations, to agree on connection parameters, and + // to establish a connection. + ControlType uint16 + // Subtype: 16 bits. This field specifies an additional subtype for + // specific packets. + SubType uint16 + // Type-specific Information: 32 bits. The use of this field depends on + // the particular control packet type. Handshake packets do not use + // this field. + AdditionalInfo uint32 + // Timestamp: 32 bits. + Timestamp uint32 + // Destination Socket ID: 32 bits. + SocketID uint32 + + // Version: 32 bits. A base protocol version number. Currently used + // values are 4 and 5. Values greater than 5 are reserved for future + // use. + Version uint32 + // Encryption Field: 16 bits. Block cipher family and key size. The + // values of this field are described in Table 2. The default value + // is AES-128. + // 0 | No Encryption Advertised + // 2 | AES-128 + // 3 | AES-192 + // 4 | AES-256 + EncryptionField uint16 + // Extension Field: 16 bits. This field is message specific extension + // related to Handshake Type field. The value MUST be set to 0 + // except for the following cases. (1) If the handshake control + // packet is the INDUCTION message, this field is sent back by the + // Listener. (2) In the case of a CONCLUSION message, this field + // value should contain a combination of Extension Type values. + // 0x00000001 | HSREQ + // 0x00000002 | KMREQ + // 0x00000004 | CONFIG + // 0x4A17 if HandshakeType is INDUCTION, see https://datatracker.ietf.org/doc/html/draft-sharabayko-srt-01#section-4.3.1.1 + ExtensionField uint16 + // Initial Packet Sequence Number: 32 bits. The sequence number of the + // very first data packet to be sent. + InitSequence uint32 + // Maximum Transmission Unit Size: 32 bits. This value is typically set + // to 1500, which is the default Maximum Transmission Unit (MTU) size + // for Ethernet, but can be less. + MTU uint32 + // Maximum Flow Window Size: 32 bits. The value of this field is the + // maximum number of data packets allowed to be "in flight" (i.e. the + // number of sent packets for which an ACK control packet has not yet + // been received). + FlowWindow uint32 + // Handshake Type: 32 bits. This field indicates the handshake packet + // type. + // 0xFFFFFFFD | DONE + // 0xFFFFFFFE | AGREEMENT + // 0xFFFFFFFF | CONCLUSION + // 0x00000000 | WAVEHAND + // 0x00000001 | INDUCTION + HandshakeType uint32 + // SRT Socket ID: 32 bits. This field holds the ID of the source SRT + // socket from which a handshake packet is issued. + SRTSocketID uint32 + // SYN Cookie: 32 bits. Randomized value for processing a handshake. + // The value of this field is specified by the handshake message + // type. + SynCookie uint32 + // Peer IP Address: 128 bits. IPv4 or IPv6 address of the packet's + // sender. The value consists of four 32-bit fields. + PeerIP net.IP + // Extensions. + // Extension Type: 16 bits. The value of this field is used to process + // an integrated handshake. Each extension can have a pair of + // request and response types. + // Extension Length: 16 bits. The length of the Extension Contents + // field in four-byte blocks. + // Extension Contents: variable length. The payload of the extension. + ExtraData []byte +} + +func (v *SRTHandshakePacket) IsControl() bool { + return v.ControlFlag == 0x80 +} + +func (v *SRTHandshakePacket) IsHandshake() bool { + return v.IsControl() && v.ControlType == 0x00 && v.SubType == 0x00 +} + +func (v *SRTHandshakePacket) String() string { + return fmt.Sprintf("Control=%v, CType=%v, SType=%v, Timestamp=%v, SocketID=%v, Version=%v, Encrypt=%v, Extension=%v, InitSequence=%v, MTU=%v, FlowWnd=%v, HSType=%v, SRTSocketID=%v, Cookie=%v, Peer=%vB, Extra=%vB", + v.IsControl(), v.ControlType, v.SubType, v.Timestamp, v.SocketID, v.Version, v.EncryptionField, v.ExtensionField, v.InitSequence, v.MTU, v.FlowWindow, v.HandshakeType, v.SRTSocketID, v.SynCookie, len(v.PeerIP), len(v.ExtraData)) +} + +func (v *SRTHandshakePacket) UnmarshalBinary(b []byte) error { + if len(b) < 4 { + return fmt.Errorf("Invalid packet length %v", len(b)) + } + v.ControlFlag = b[0] & 0x80 + v.ControlType = binary.BigEndian.Uint16(b[0:2]) & 0x7fff + v.SubType = binary.BigEndian.Uint16(b[2:4]) + + if !v.IsHandshake() { + return nil + } + + if len(b) < 64 { + return fmt.Errorf("Invalid packet length %v", len(b)) + } + v.AdditionalInfo = binary.BigEndian.Uint32(b[4:]) + v.Timestamp = binary.BigEndian.Uint32(b[8:]) + v.SocketID = binary.BigEndian.Uint32(b[12:]) + v.Version = binary.BigEndian.Uint32(b[16:]) + v.EncryptionField = binary.BigEndian.Uint16(b[20:]) + v.ExtensionField = binary.BigEndian.Uint16(b[22:]) + v.InitSequence = binary.BigEndian.Uint32(b[24:]) + v.MTU = binary.BigEndian.Uint32(b[28:]) + v.FlowWindow = binary.BigEndian.Uint32(b[32:]) + v.HandshakeType = binary.BigEndian.Uint32(b[36:]) + v.SRTSocketID = binary.BigEndian.Uint32(b[40:]) + v.SynCookie = binary.BigEndian.Uint32(b[44:]) + + // Only support IPv4. + v.PeerIP = net.IPv4(b[51], b[50], b[49], b[48]) + + v.ExtraData = b[64:] + + return nil +} + +func (v *SRTHandshakePacket) MarshalBinary() ([]byte, error) { + b := make([]byte, 64+len(v.ExtraData)) + binary.BigEndian.PutUint16(b, uint16(v.ControlFlag)<<8|v.ControlType) + binary.BigEndian.PutUint16(b[2:], v.SubType) + binary.BigEndian.PutUint32(b[4:], v.AdditionalInfo) + binary.BigEndian.PutUint32(b[8:], v.Timestamp) + binary.BigEndian.PutUint32(b[12:], v.SocketID) + binary.BigEndian.PutUint32(b[16:], v.Version) + binary.BigEndian.PutUint16(b[20:], v.EncryptionField) + binary.BigEndian.PutUint16(b[22:], v.ExtensionField) + binary.BigEndian.PutUint32(b[24:], v.InitSequence) + binary.BigEndian.PutUint32(b[28:], v.MTU) + binary.BigEndian.PutUint32(b[32:], v.FlowWindow) + binary.BigEndian.PutUint32(b[36:], v.HandshakeType) + binary.BigEndian.PutUint32(b[40:], v.SRTSocketID) + binary.BigEndian.PutUint32(b[44:], v.SynCookie) + + // Only support IPv4. + ip := v.PeerIP.To4() + b[48] = ip[3] + b[49] = ip[2] + b[50] = ip[1] + b[51] = ip[0] + + if len(v.ExtraData) > 0 { + copy(b[64:], v.ExtraData) + } + + return b, nil +}