// SPDX-FileCopyrightText: 2026 The Pion community // SPDX-License-Identifier: MIT //go:build !js // sctp-interleaving demonstrates how to inspect plaintext SCTP packets between // DTLS and SCTP while still interoperating with a browser RTCPeerConnection. package main import ( "context" "crypto/sha256" "crypto/tls" "crypto/x509" "encoding/hex" "encoding/json" "errors" "fmt" "io" "net" "net/http" "runtime/debug" "slices" "strconv" "strings" "sync" "time" "github.com/pion/datachannel" "github.com/pion/dtls/v3" "github.com/pion/dtls/v3/pkg/crypto/selfsign" "github.com/pion/ice/v4" "github.com/pion/logging" "github.com/pion/sctp" "github.com/pion/sdp/v3" ) const ( listenAddr = ":8080" sctpPort = 5000 maxMessageSize = 64 * 1024 * 1024 outboundMTU = 1191 readBufferSize = 20 * 1024 * 1024 ) var ( errPeerProvidedNoDTLSCertificate = errors.New("peer provided no DTLS certificate") errNoApplicationMediaSection = errors.New("offer has no application media section") errMissingICECredentials = errors.New("offer missing ICE credentials") errNoICECandidates = errors.New("offer has no ICE candidates; wait for ICE gathering to complete") errMissingDTLSFingerprint = errors.New("offer missing DTLS fingerprint") errUnsupportedFingerprint = errors.New("unsupported fingerprint algorithm") errRemoteFingerprintMismatch = errors.New("remote certificate fingerprint mismatch") errDataChannelsNotReady = errors.New("control and bulk channels are not ready") errTrafficShaperNotReady = errors.New("traffic shaper is not ready") ) func main() { events := newEventHub() http.HandleFunc("/", func(responseWriter http.ResponseWriter, request *http.Request) { http.ServeFile(responseWriter, request, "./index.html") }) http.HandleFunc("/events", events.serveHTTP) http.HandleFunc("/status", serveStatus) http.HandleFunc("/server-pressure", func(responseWriter http.ResponseWriter, request *http.Request) { if request.Method != http.MethodPost { http.Error(responseWriter, "method not allowed", http.StatusMethodNotAllowed) return } sessionID := request.URL.Query().Get("session") runID := request.URL.Query().Get("run") if err := events.startServerPressure(sessionID, runID); err != nil { http.Error(responseWriter, err.Error(), http.StatusBadRequest) return } responseWriter.WriteHeader(http.StatusNoContent) }) http.HandleFunc("/offer", func(responseWriter http.ResponseWriter, request *http.Request) { if request.Method != http.MethodPost { http.Error(responseWriter, "method not allowed", http.StatusMethodNotAllowed) return } offer, err := io.ReadAll(request.Body) if err != nil { http.Error(responseWriter, err.Error(), http.StatusBadRequest) return } sessionID := request.URL.Query().Get("session") answer, err := acceptBrowserOffer(request.Context(), sessionID, string(offer), events) if err != nil { events.publish(sessionID, logEvent("error", err.Error())) http.Error(responseWriter, err.Error(), http.StatusInternalServerError) return } responseWriter.Header().Set("Content-Type", "application/sdp") // The SDP answer is generated by this server from parsed offer fields. //nolint:gosec if _, err = io.WriteString(responseWriter, answer); err != nil { events.publish(sessionID, logEvent("error", err.Error())) } }) fmt.Printf("Listening on %s\n", listenAddr) //nolint:gosec if err := http.ListenAndServe(listenAddr, nil); err != nil { panic(err) } } func acceptBrowserOffer(ctx context.Context, sessionID string, rawOffer string, events *eventHub) (string, error) { offer, err := parseOffer(rawOffer) if err != nil { return "", err } cert, err := selfsign.GenerateSelfSigned() if err != nil { return "", err } agent, localCandidates, err := gatherICECandidates(ctx, sessionID, events) if err != nil { return "", err } localUfrag, localPwd, err := agent.GetLocalUserCredentials() if err != nil { return "", err } for _, candidate := range offer.candidates { if addErr := agent.AddRemoteCandidate(candidate); addErr != nil { return "", addErr } } answer := buildAnswer(localUfrag, localPwd, localCandidates, cert, offer) go func() { stackCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute) defer cancel() if err := runStack(stackCtx, agent, offer, cert, sessionID, events); err != nil { events.publish(sessionID, logEvent("error", err.Error())) } }() return answer, nil } func gatherICECandidates(ctx context.Context, sessionID string, events *eventHub) (*ice.Agent, []ice.Candidate, error) { agent, err := ice.NewAgentWithOptions( ice.WithNetworkTypes([]ice.NetworkType{ice.NetworkTypeUDP4, ice.NetworkTypeUDP6}), ) if err != nil { return nil, nil, err } var ( mu sync.Mutex candidates []ice.Candidate done = make(chan struct{}) doneOnce sync.Once ) if err = agent.OnCandidate(func(c ice.Candidate) { if c == nil { doneOnce.Do(func() { close(done) }) return } mu.Lock() candidates = append(candidates, c) mu.Unlock() events.publish(sessionID, logEvent("ice-candidate", c.Marshal())) }); err != nil { return nil, nil, err } if err = agent.GatherCandidates(); err != nil { return nil, nil, err } select { case <-done: case <-ctx.Done(): return nil, nil, ctx.Err() case <-time.After(3 * time.Second): events.publish(sessionID, logEvent("warn", "ICE gathering timed out; continuing with gathered candidates")) } mu.Lock() defer mu.Unlock() return agent, append([]ice.Candidate(nil), candidates...), nil } func runStack( ctx context.Context, agent *ice.Agent, offer browserOffer, cert tls.Certificate, sessionID string, events *eventHub, ) error { events.publish(sessionID, logEvent("ice", "starting controlled ICE")) iceConn, err := agent.Accept(ctx, offer.iceUfrag, offer.icePwd) if err != nil { return err } events.publish(sessionID, logEvent("ice", "connected: "+iceConn.RemoteAddr().String())) packetConn := &connectedPacketConn{Conn: iceConn, remote: iceConn.RemoteAddr()} dtlsConn, err := dtls.ServerWithOptions( packetConn, packetConn.remote, dtls.WithCertificates(cert), dtls.WithSRTPProtectionProfiles( dtls.SRTP_AEAD_AES_256_GCM, dtls.SRTP_AEAD_AES_128_GCM, dtls.SRTP_AES128_CM_HMAC_SHA1_80, ), dtls.WithSupportedProtocols("webrtc"), dtls.WithClientAuth(dtls.RequireAnyClientCert), dtls.WithInsecureSkipVerify(true), dtls.WithVerifyPeerCertificate(func(rawCerts [][]byte, _ [][]*x509.Certificate) error { if len(rawCerts) == 0 { return errPeerProvidedNoDTLSCertificate } return verifyFingerprint(rawCerts[0], offer.fingerprint) }), ) if err != nil { return err } events.publish(sessionID, logEvent("dtls", "waiting for browser DTLS handshake")) if err = dtlsConn.Handshake(); err != nil { return err } if state, ok := dtlsConn.ConnectionState(); ok { events.publish(sessionID, logEvent("dtls", "connected; ALPN="+state.NegotiatedProtocol)) } else { events.publish(sessionID, logEvent("dtls", "connected")) } shaper := events.registerShaper(sessionID) defer events.unregisterShaper(sessionID, shaper) tap := &sctpTapConn{ Conn: dtlsConn, shaper: shaper, inspect: func(direction string, packet []byte) { events.publish(sessionID, packetEvent(direction, packet)) }, } assoc, err := sctp.ClientWithOptions( sctp.WithNetConn(tap), sctp.WithMTU(outboundMTU), sctp.WithMaxMessageSize(maxMessageSize), sctp.WithLoggerFactory(logging.NewDefaultLoggerFactory()), ) if err != nil { return err } events.publish(sessionID, logEvent("sctp", "association established")) for { dc, err := datachannel.Accept(assoc, &datachannel.Config{ LoggerFactory: logging.NewDefaultLoggerFactory(), }) if err != nil { return err } events.publish(sessionID, logEvent("datachannel", "accepted "+dc.Label)) events.registerDataChannel(sessionID, dc) go echoDataChannel(dc, sessionID, events) } } func echoDataChannel(dc *datachannel.DataChannel, sessionID string, events *eventHub) { defer events.unregisterDataChannel(sessionID, dc) buffer := make([]byte, readBufferSize) for { n, isString, err := dc.ReadDataChannel(buffer) if err != nil { events.publish(sessionID, logEvent("datachannel", dc.Label+" closed: "+err.Error())) return } kind := "binary" if isString { kind = "text" } events.publish(sessionID, logEvent("message", fmt.Sprintf("%s received %d byte %s message", dc.Label, n, kind))) response := fmt.Appendf(nil, "echo from Pion on %s: %d bytes", dc.Label, n) if isString { response = append([]byte(nil), buffer[:n]...) } if _, err = dc.WriteDataChannel(response, true); err != nil { events.publish(sessionID, logEvent("datachannel", dc.Label+" echo failed: "+err.Error())) return } } } type browserOffer struct { raw *sdp.SessionDescription media *sdp.MediaDescription iceUfrag string icePwd string fingerprint fingerprint candidates []ice.Candidate } type fingerprint struct { algorithm string value string } func parseOffer(rawOffer string) (browserOffer, error) { var desc sdp.SessionDescription if err := desc.UnmarshalString(rawOffer); err != nil { return browserOffer{}, err } var media *sdp.MediaDescription for _, md := range desc.MediaDescriptions { if md.MediaName.Media == "application" { media = md break } } if media == nil { return browserOffer{}, errNoApplicationMediaSection } iceUfrag := attr(&desc, media, "ice-ufrag") icePwd := attr(&desc, media, "ice-pwd") if iceUfrag == "" || icePwd == "" { return browserOffer{}, errMissingICECredentials } fp, err := parseFingerprint(attr(&desc, media, "fingerprint")) if err != nil { return browserOffer{}, err } candidates, err := parseCandidates(media) if err != nil { return browserOffer{}, err } if len(candidates) == 0 { return browserOffer{}, errNoICECandidates } return browserOffer{ raw: &desc, media: media, iceUfrag: iceUfrag, icePwd: icePwd, fingerprint: fp, candidates: candidates, }, nil } func attr(desc *sdp.SessionDescription, media *sdp.MediaDescription, key string) string { if value, ok := media.Attribute(key); ok { return value } value, _ := desc.Attribute(key) return value } func parseFingerprint(raw string) (fingerprint, error) { parts := strings.Fields(raw) if len(parts) != 2 { return fingerprint{}, errMissingDTLSFingerprint } if !strings.EqualFold(parts[0], "sha-256") { return fingerprint{}, fmt.Errorf("%w: %q", errUnsupportedFingerprint, parts[0]) } return fingerprint{algorithm: strings.ToLower(parts[0]), value: strings.ToUpper(parts[1])}, nil } func parseCandidates(media *sdp.MediaDescription) ([]ice.Candidate, error) { var candidates []ice.Candidate for _, attribute := range media.Attributes { if attribute.Key != "candidate" { continue } candidate, err := ice.UnmarshalCandidate(attribute.Value) if err != nil { return nil, err } candidates = append(candidates, candidate) } return candidates, nil } func buildAnswer( localUfrag string, localPwd string, candidates []ice.Candidate, cert tls.Certificate, offer browserOffer, ) string { fp := certificateFingerprint(cert) mid := attr(offer.raw, offer.media, "mid") if mid == "" { mid = "0" } lines := []string{ "v=0", "o=- " + strconv.FormatInt(time.Now().UnixNano(), 10) + " 0 IN IP4 127.0.0.1", "s=-", "t=0 0", "a=group:BUNDLE " + mid, "a=msid-semantic: WMS", "m=application 9 UDP/DTLS/SCTP webrtc-datachannel", "c=IN IP4 0.0.0.0", "a=mid:" + mid, "a=sendrecv", "a=ice-ufrag:" + localUfrag, "a=ice-pwd:" + localPwd, "a=ice-options:trickle", "a=fingerprint:sha-256 " + fp, "a=setup:passive", "a=sctp-port:" + strconv.Itoa(sctpPort), "a=max-message-size:" + strconv.Itoa(maxMessageSize), } for _, candidate := range candidates { lines = append(lines, "a="+candidate.Marshal()) } lines = append(lines, "a=end-of-candidates", "") return strings.Join(lines, "\r\n") } func certificateFingerprint(cert tls.Certificate) string { sum := sha256.Sum256(cert.Certificate[0]) encoded := strings.ToUpper(hex.EncodeToString(sum[:])) pairs := make([]string, 0, len(sum)) for offset := 0; offset < len(encoded); offset += 2 { pairs = append(pairs, encoded[offset:offset+2]) } return strings.Join(pairs, ":") } func verifyFingerprint(rawCert []byte, expected fingerprint) error { if !strings.EqualFold(expected.algorithm, "sha-256") { return fmt.Errorf("%w: %q", errUnsupportedFingerprint, expected.algorithm) } sum := sha256.Sum256(rawCert) actual := strings.ToUpper(hex.EncodeToString(sum[:])) expectedValue := strings.ToUpper(strings.ReplaceAll(expected.value, ":", "")) if actual != expectedValue { return fmt.Errorf("%w: got %s expected %s", errRemoteFingerprintMismatch, actual, expectedValue) } return nil } type connectedPacketConn struct { net.Conn remote net.Addr } func (c *connectedPacketConn) ReadFrom(p []byte) (int, net.Addr, error) { n, err := c.Conn.Read(p) return n, c.remote, err } func (c *connectedPacketConn) WriteTo(p []byte, _ net.Addr) (int, error) { return c.Conn.Write(p) } func (c *connectedPacketConn) LocalAddr() net.Addr { return c.Conn.LocalAddr() } type sctpTapConn struct { net.Conn inspect func(direction string, packet []byte) shaper *trafficShaper } func (c *sctpTapConn) Read(p []byte) (int, error) { n, err := c.Conn.Read(p) if n > 0 && c.inspect != nil { c.inspect("rx", append([]byte(nil), p[:n]...)) } if n > 0 && c.shaper != nil { c.shaper.throttle(n) } return n, err } func (c *sctpTapConn) Write(p []byte) (int, error) { if len(p) > 0 && c.inspect != nil { c.inspect("tx", append([]byte(nil), p...)) } n, err := c.Conn.Write(p) if n > 0 && c.shaper != nil { c.shaper.throttle(n) } return n, err } type eventHub struct { mu sync.Mutex clients map[string]map[chan []byte]struct{} shapers map[string]*trafficShaper connections map[string]map[string]*datachannel.DataChannel } func newEventHub() *eventHub { return &eventHub{ clients: map[string]map[chan []byte]struct{}{}, shapers: map[string]*trafficShaper{}, connections: map[string]map[string]*datachannel.DataChannel{}, } } func (h *eventHub) serveHTTP(responseWriter http.ResponseWriter, request *http.Request) { responseWriter.Header().Set("Content-Type", "text/event-stream") responseWriter.Header().Set("Cache-Control", "no-cache") responseWriter.Header().Set("Connection", "keep-alive") flusher, ok := responseWriter.(http.Flusher) if !ok { http.Error(responseWriter, "streaming unsupported", http.StatusInternalServerError) return } sessionID := request.URL.Query().Get("session") ch := make(chan []byte, 64) h.mu.Lock() if h.clients[sessionID] == nil { h.clients[sessionID] = map[chan []byte]struct{}{} } h.clients[sessionID][ch] = struct{}{} h.mu.Unlock() defer func() { h.mu.Lock() delete(h.clients[sessionID], ch) if len(h.clients[sessionID]) == 0 { delete(h.clients, sessionID) } h.mu.Unlock() close(ch) }() for { select { case <-request.Context().Done(): return case msg := <-ch: _, _ = fmt.Fprintf(responseWriter, "data: %s\n\n", msg) flusher.Flush() } } } func (h *eventHub) publish(sessionID string, event any) { payload, err := json.Marshal(event) if err != nil { return } h.mu.Lock() defer h.mu.Unlock() for ch := range h.clients[sessionID] { select { case ch <- payload: default: } } } func (h *eventHub) registerShaper(sessionID string) *trafficShaper { shaper := &trafficShaper{bytesPerSecond: 128 * 1024} h.mu.Lock() h.shapers[sessionID] = shaper h.mu.Unlock() return shaper } func (h *eventHub) unregisterShaper(sessionID string, shaper *trafficShaper) { h.mu.Lock() if h.shapers[sessionID] == shaper { delete(h.shapers, sessionID) } h.mu.Unlock() } func (h *eventHub) registerDataChannel(sessionID string, dc *datachannel.DataChannel) { h.mu.Lock() if h.connections[sessionID] == nil { h.connections[sessionID] = map[string]*datachannel.DataChannel{} } h.connections[sessionID][dc.Label] = dc h.mu.Unlock() } func (h *eventHub) unregisterDataChannel(sessionID string, dc *datachannel.DataChannel) { h.mu.Lock() if h.connections[sessionID][dc.Label] == dc { delete(h.connections[sessionID], dc.Label) } if len(h.connections[sessionID]) == 0 { delete(h.connections, sessionID) } h.mu.Unlock() } func (h *eventHub) startServerPressure(sessionID string, runID string) error { h.mu.Lock() channels := h.connections[sessionID] control := channels["control"] bulk := channels["bulk"] shaper := h.shapers[sessionID] h.mu.Unlock() if control == nil || bulk == nil { return errDataChannelsNotReady } if shaper == nil { return errTrafficShaperNotReady } if runID == "" { runID = strconv.FormatInt(time.Now().UnixNano(), 10) } go h.runServerPressure(sessionID, runID, shaper, control, bulk) return nil } func (h *eventHub) runServerPressure( sessionID string, runID string, shaper *trafficShaper, control *datachannel.DataChannel, bulk *datachannel.DataChannel, ) { shaper.setEnabled(true) defer shaper.setEnabled(false) const ( bulkSize = 2 * 1024 * 1024 probeDuration = 25 * time.Second ) h.publish(sessionID, logEvent("pressure", "server queueing one 2 MiB bulk message and control probes")) payload := make([]byte, bulkSize) for i := range payload { payload[i] = byte(i) } if _, err := bulk.WriteDataChannel(payload, false); err != nil { h.publish(sessionID, logEvent("pressure", "server bulk queue failed: "+err.Error())) return } h.publish(sessionID, logEvent("pressure", "server bulk queued; sending control probes while it drains")) ticker := time.NewTicker(250 * time.Millisecond) defer ticker.Stop() deadline := time.NewTimer(probeDuration) defer deadline.Stop() seq := 0 for { select { case <-deadline.C: h.publish(sessionID, logEvent("pressure", fmt.Sprintf("server pressure probes complete: %d sent", seq))) return case <-ticker.C: msg := fmt.Appendf(nil, "server-pressure:%s:%d", runID, seq) if _, err := control.WriteDataChannel(msg, true); err != nil { h.publish(sessionID, logEvent("pressure", "server control probe failed: "+err.Error())) return } seq++ } } } type trafficShaper struct { mu sync.Mutex enabled bool bytesPerSecond int } func (s *trafficShaper) setEnabled(enabled bool) { s.mu.Lock() s.enabled = enabled s.mu.Unlock() } func (s *trafficShaper) throttle(n int) { s.mu.Lock() enabled := s.enabled bytesPerSecond := s.bytesPerSecond s.mu.Unlock() if !enabled || bytesPerSecond <= 0 { return } time.Sleep(time.Duration(n) * time.Second / time.Duration(bytesPerSecond)) } func logEvent(kind, message string) map[string]any { return map[string]any{ "type": "log", "kind": kind, "message": message, } } func packetEvent(direction string, raw []byte) map[string]any { chunks := parseSCTPChunks(raw) event := map[string]any{ "type": "packet", "direction": direction, "size": len(raw), "chunks": chunks, } return event } type sctpChunkView struct { Type uint8 `json:"type"` Name string `json:"name"` Length uint16 `json:"length"` InterleavingAdvertised bool `json:"interleavingAdvertised,omitempty"` } func parseSCTPChunks(raw []byte) []sctpChunkView { if len(raw) < 12 { return nil } var chunks []sctpChunkView for offset := 12; offset+4 <= len(raw); { chunkType := raw[offset] chunkLength := int(raw[offset+2])<<8 | int(raw[offset+3]) if chunkLength < 4 || offset+chunkLength > len(raw) { break } chunk := sctpChunkView{ Type: chunkType, Name: chunkTypeName(chunkType), Length: uint16(chunkLength), //nolint:gosec // chunkLength is uint16 on the wire. } if chunkType == 1 || chunkType == 2 { chunk.InterleavingAdvertised = chunkAdvertisesInterleaving(raw[offset : offset+chunkLength]) } chunks = append(chunks, chunk) offset += (chunkLength + 3) &^ 3 } return chunks } func chunkAdvertisesInterleaving(chunk []byte) bool { const ( initValueLength = 16 paramHeaderLength = 4 paramTypeSupportedExtensions = 0x8008 chunkTypeInterleavedPayloadData = 64 ) paramsOffset := 4 + initValueLength for offset := paramsOffset; offset+paramHeaderLength <= len(chunk); { paramType := int(chunk[offset])<<8 | int(chunk[offset+1]) paramLength := int(chunk[offset+2])<<8 | int(chunk[offset+3]) if paramLength < paramHeaderLength || offset+paramLength > len(chunk) { return false } if paramType == paramTypeSupportedExtensions { if slices.Contains(chunk[offset+paramHeaderLength:offset+paramLength], chunkTypeInterleavedPayloadData) { return true } } offset += (paramLength + 3) &^ 3 } return false } func chunkTypeName(chunkType uint8) string { names := map[uint8]string{ 0: "DATA", 1: "INIT", 2: "INIT-ACK", 3: "SACK", 10: "COOKIE-ECHO", 11: "COOKIE-ACK", 64: "I-DATA", 130: "RECONFIG", 192: "FORWARD-TSN", } if name, ok := names[chunkType]; ok { return name } return "chunk-" + strconv.Itoa(int(chunkType)) } func serveStatus(responseWriter http.ResponseWriter, _ *http.Request) { status := map[string]string{"sctp": "(unknown)"} if info, ok := debug.ReadBuildInfo(); ok { for _, dep := range info.Deps { if dep.Path == "github.com/pion/sctp" { status["sctp"] = dep.Version if dep.Replace != nil { status["sctp"] = dep.Replace.Path + " " + dep.Replace.Version } break } } } responseWriter.Header().Set("Content-Type", "application/json") if err := json.NewEncoder(responseWriter).Encode(status); err != nil { http.Error(responseWriter, err.Error(), http.StatusInternalServerError) } }