search

skycoin / dmsg

dmsg
7 stars 17 forks source link

issue with audio streaming over dmsghttp (dmsgweb) #255

Open 0pcom opened 8 months ago

0pcom commented 8 months ago

It should be possible to do anything via dmsghttp that is possible with regular http

tested with dmsgweb

Using VlC, a network audio stream was started on http://127.0.0.1:8080/music

An example html document is used to link to both the http and dmsghttp endpoints, as well as to access the audio stream directly for testing. Substituite with the public key of your dmsg server in the following html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Example HTML Document</title>
</head>
<body>
        <h1>Example HTML Document</h1>
        <a href='http://035c095f095c9ec2b35545a1a6d519aa7f4103cd4d52dea888c6a7e18c91721c90.dmsg'>http://035c095f095c9ec2b35545a1a6d519aa7f4103cd4d52dea888c6a7e18c91721c90.dmsg</a><br>
        <a href='http://127.0.0.1:8081/'>http://127.0.0.1:8081/</a><br>
        <audio controls><source src='http://127.0.0.1:8080/music' type='audio/mpeg'><source src='http://127.0.0.1:8080/music' type='audio/ogg'><source src='http://127.0.0.1:8080/music' type='audio/wav'>Your browser does not support the audio element.</audio><br>
        <audio controls><source src='http://127.0.0.1:8081/music' type='audio/mpeg'><source src='http://127.0.0.1:8081/music' type='audio/ogg'><source src='http://127.0.0.1:8081/music' type='audio/wav'>Your browser does not support the audio element.</audio><br>
</body>
</html>

example dmsghttp + http server which serves http on port 8081

dmsgweb keys-gen > dmusic.key

package main

import (
    "fmt"
    "context"
    "log"
  "net/http"
    "net/http/httputil"
    "net/url"
    "strings"
    "time"
    "sync"
    "net"

    "github.com/gin-gonic/gin"
    "github.com/spf13/cobra"
    cc "github.com/ivanpirog/coloredcobra"
    "github.com/skycoin/skywire-utilities/pkg/logging"
    "github.com/skycoin/skywire-utilities/pkg/cipher"
    "github.com/skycoin/skywire-utilities/pkg/cmdutil"
    "github.com/skycoin/dmsg/pkg/disc"
    dmsg "github.com/skycoin/dmsg/pkg/dmsg"
    "github.com/skycoin/skywire-utilities/pkg/skyenv"

)

func main() {
    Execute()
}

var (
    startTime = time.Now()
    runTime   time.Duration
    sk       cipher.SecKey
    pk          cipher.PubKey
    dmsgDisc string
    dmsgPort uint
    wl       string
    wlkeys   []cipher.PubKey
    webPort uint
)

func init() {
    rootCmd.Flags().UintVarP(&webPort, "port", "p", 8081, "port to serve")
    rootCmd.Flags().UintVarP(&dmsgPort, "dport", "d", 80, "dmsg port to serve")
    rootCmd.Flags().StringVarP(&wl, "wl", "w", "", "whitelisted keys for dmsg authenticated routes")
    rootCmd.Flags().StringVarP(&dmsgDisc, "dmsg-disc", "D", skyenv.DmsgDiscAddr, "dmsg discovery url")
    pk, _ = sk.PubKey()
    rootCmd.Flags().VarP(&sk, "sk", "s", "a random key is generated if unspecified\n\r")
    rootCmd.CompletionOptions.DisableDefaultCmd = true
    var helpflag bool
    rootCmd.SetUsageTemplate(help)
    rootCmd.PersistentFlags().BoolVarP(&helpflag, "help", "h", false, "help for "+rootCmd.Use)
    rootCmd.SetHelpCommand(&cobra.Command{Hidden: true})
    rootCmd.PersistentFlags().MarkHidden("help") //nolint
}

var rootCmd = &cobra.Command{
    Use:   "example",
    Short: "example http & dmsghttp server",
    Long: "example http & dmsghttp server",
    Run: func(_ *cobra.Command, _ []string) {
        Server()
    },
}

func Execute() {
    cc.Init(&cc.Config{
        RootCmd:       rootCmd,
        Headings:      cc.HiBlue + cc.Bold,
        Commands:      cc.HiBlue + cc.Bold,
        CmdShortDescr: cc.HiBlue,
        Example:       cc.HiBlue + cc.Italic,
        ExecName:      cc.HiBlue + cc.Bold,
        Flags:         cc.HiBlue + cc.Bold,
        FlagsDescr:      cc.HiBlue,
        NoExtraNewlines: true,
        NoBottomNewline: true,
    })
    if err := rootCmd.Execute(); err != nil {
        log.Fatal("Failed to execute command: ", err)
    }
}

func Server() {
    wg := new(sync.WaitGroup)
    wg.Add(1)

    log := logging.MustGetLogger("dmsghttp")

    ctx, cancel := cmdutil.SignalContext(context.Background(), log)
    defer cancel()
    pk, err := sk.PubKey()
    if err != nil {
        pk, sk = cipher.GenerateKeyPair()
    }
    if wl != "" {
        wlk := strings.Split(wl, ",")
        for _, key := range wlk {
            var pk1 cipher.PubKey
            err := pk1.Set(key)
            if err == nil {
                wlkeys = append(wlkeys, pk1)
            }
        }
    }
    if len(wlkeys) > 0 {
        if len(wlkeys) == 1 {
            log.Info(fmt.Sprintf("%d key whitelisted", len(wlkeys)))
        } else {
            log.Info(fmt.Sprintf("%d keys whitelisted", len(wlkeys)))
        }
    }

    dmsgclient := dmsg.NewClient(pk, sk, disc.NewHTTP(dmsgDisc, &http.Client{}, log), dmsg.DefaultConfig())
    defer func() {
        if err := dmsgclient.Close(); err != nil {
            log.WithError(err).Error()
        }
    }()

    go dmsgclient.Serve(context.Background())

    select {
    case <-ctx.Done():
        log.WithError(ctx.Err()).Warn()
        return

    case <-dmsgclient.Ready():
    }

    lis, err := dmsgclient.Listen(uint16(dmsgPort))
    if err != nil {
        log.WithError(err).Fatal()
    }
    go func() {
        <-ctx.Done()
        if err := lis.Close(); err != nil {
            log.WithError(err).Error()
        }
    }()

    r1 := gin.New()
    // Disable Gin's default logger middleware
    r1.Use(gin.Recovery())
    r1.Use(loggingMiddleware())
    r1.GET("/", func(c *gin.Context) {
  c.Writer.Header().Set("Server", "")
        c.Writer.WriteHeader(http.StatusOK)
//      l := "<!doctype html><html lang=en><head><title>Example Website</title></head><body style='background-color:black;color:white;'>\n<style type='text/css'>\npre {\n  font-family:Courier New;\n  font-size:10pt;\n}\n.af_line {\n  color: gray;\n  text-decoration: none;\n}\n.column {\n  float: left;\n  width: 30%;\n  padding: 10px;\n}\n.row:after {\n  content: '';\n  display: table;\n  clear: both;\n}\n</style>\n<pre>"
        l := "<!DOCTYPE html><html><head><meta name='viewport' content='initial-scale=1'></head><body style='background-color:black;color:white;'><audio controls><source src='/music' type='audio/mpeg'><source src='/music' type='audio/ogg'><source src='/music' type='audio/wav'>Your browser does not support the audio element.</audio></body></html>"
//      l += "</body></html>"
        c.Writer.Write([]byte(l))
        return
    })

  r1.GET("/music", func(c *gin.Context) {
        targetURL, _ := url.Parse("http://127.0.0.1:8080/")
        proxy := httputil.NewSingleHostReverseProxy(targetURL)
        proxy.ServeHTTP(c.Writer, c.Request)
    })
    // only whitelisted public keys can access authRoute(s)
    authRoute := r1.Group("/")
    if len(wlkeys) > 0 {
        authRoute.Use(whitelistAuth(wlkeys))
    }

    authRoute.GET("/auth", func(c *gin.Context) {
        //override the behavior of `public fallback` for this endpoint when no keys are whitelisted
        if len(wlkeys) == 0 {
            c.Writer.WriteHeader(http.StatusNotFound)
            return
        }
        c.Writer.WriteHeader(http.StatusOK)
        l := "<!doctype html><html lang=en><head><title>Example Website</title></head><body style='background-color:black;color:white;'>\n<style type='text/css'>\npre {\n  font-family:Courier New;\n  font-size:10pt;\n}\n.af_line {\n  color: gray;\n  text-decoration: none;\n}\n.column {\n  float: left;\n  width: 30%;\n  padding: 10px;\n}\n.row:after {\n  content: '';\n  display: table;\n  clear: both;\n}\n</style>\n<pre>"
        l += "<p>Hello World!</p>"
        l += "</body></html>"
        c.Writer.Write([]byte(l))
    })

r1.GET("/health", func(c *gin.Context) {
            runTime = time.Since(startTime)
    c.JSON(http.StatusOK, gin.H{
            "frontend_start_time": startTime,
            "frontend_run_time":   runTime.String(),
            "dmsg_address": fmt.Sprintf("%s:%d", pk.String(), dmsgPort),
        })
})

    // Start the server using the custom Gin handler
    serve := &http.Server{
        Handler:           &GinHandler{Router: r1},
        ReadHeaderTimeout: 5 * time.Second,
    ReadTimeout:  10 * time.Second,
    WriteTimeout: 10 * time.Second,
    }

    // Start serving
    go func() {
        log.WithField("dmsg_addr", lis.Addr().String()).Info("Serving...")
        if err := serve.Serve(lis); err != nil && err != http.ErrServerClosed {
            log.Fatalf("Serve: %v", err)
        }
        wg.Done()
    }()

    go func() {
        fmt.Printf("listening on http://127.0.0.1:%d using gin router\n", webPort)
        r1.Run(fmt.Sprintf(":%d", webPort))
        wg.Done()
    }()

    wg.Wait()
}

    func whitelistAuth(whitelistedPKs []cipher.PubKey) gin.HandlerFunc {
        return func(c *gin.Context) {
            // Get the remote PK.
            remotePK, _, err := net.SplitHostPort(c.Request.RemoteAddr)
            if err != nil {
                c.Writer.WriteHeader(http.StatusInternalServerError)
                c.Writer.Write([]byte("500 Internal Server Error"))
                c.AbortWithStatus(http.StatusInternalServerError)
                return
            }
            // Check if the remote PK is whitelisted.
            whitelisted := false
            if len(whitelistedPKs) == 0 {
                whitelisted = true
            } else {
                for _, whitelistedPK := range whitelistedPKs {
                    if remotePK == whitelistedPK.String() {
                        whitelisted = true
                        break
                    }
                }
            }
            if whitelisted {
                c.Next()
            } else {
                // Otherwise, return a 401 Unauthorized error.
                c.Writer.WriteHeader(http.StatusUnauthorized)
                c.Writer.Write([]byte("401 Unauthorized"))
                c.AbortWithStatus(http.StatusUnauthorized)
                return
            }
        }
    }

type GinHandler struct {
Router *gin.Engine
}

func (h *GinHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
h.Router.ServeHTTP(w, r)
}

func loggingMiddleware() gin.HandlerFunc {
    return func(c *gin.Context) {
        start := time.Now()
        c.Next()
        latency := time.Since(start)
        if latency > time.Minute {
            latency = latency.Truncate(time.Second)
        }
        statusCode := c.Writer.Status()
        method := c.Request.Method
        path := c.Request.URL.Path

        // Get the background color based on the status code
        statusCodeBackgroundColor := getBackgroundColor(statusCode)

        // Get the method color
        methodColor := getMethodColor(method)

        fmt.Printf("[EXAMPLE] %s |%s %3d %s| %13v | %15s | %72s |%s %-7s %s %s\n",
            time.Now().Format("2006/01/02 - 15:04:05"),
            statusCodeBackgroundColor,
            statusCode,
            resetColor(),
            latency,
            c.ClientIP(),
            c.Request.RemoteAddr,
            methodColor,
            method,
            resetColor(),
            path,
        )
    }
}
func getBackgroundColor(statusCode int) string {
    switch {
    case statusCode >= http.StatusOK && statusCode < http.StatusMultipleChoices:
        return green
    case statusCode >= http.StatusMultipleChoices && statusCode < http.StatusBadRequest:
        return white
    case statusCode >= http.StatusBadRequest && statusCode < http.StatusInternalServerError:
        return yellow
    default:
        return red
    }
}

func getMethodColor(method string) string {
    switch method {
    case http.MethodGet:
        return blue
    case http.MethodPost:
        return cyan
    case http.MethodPut:
        return yellow
    case http.MethodDelete:
        return red
    case http.MethodPatch:
        return green
    case http.MethodHead:
        return magenta
    case http.MethodOptions:
        return white
    default:
        return reset
    }
}

func resetColor() string {
    return reset
}

type consoleColorModeValue int
var consoleColorMode = autoColor
const (
    autoColor consoleColorModeValue = iota
    disableColor
    forceColor
)

const (
    green   = "\033[97;42m"
    white   = "\033[90;47m"
    yellow  = "\033[90;43m"
    red     = "\033[97;41m"
    blue    = "\033[97;44m"
    magenta = "\033[97;45m"
    cyan    = "\033[97;46m"
    reset   = "\033[0m"
)

var (
    err error
)
const help = "Usage:\r\n" +
    "  {{.UseLine}}{{if .HasAvailableSubCommands}}{{end}} {{if gt (len .Aliases) 0}}\r\n\r\n" +
    "{{.NameAndAliases}}{{end}}{{if .HasAvailableSubCommands}}\r\n\r\n" +
    "Available Commands:{{range .Commands}}{{if (or .IsAvailableCommand)}}\r\n  " +
    "{{rpad .Name .NamePadding }} {{.Short}}{{end}}{{end}}{{end}}{{if .HasAvailableLocalFlags}}\r\n\r\n" +
    "Flags:\r\n" +
    "{{.LocalFlags.FlagUsages | trimTrailingWhitespaces}}{{end}}{{if .HasAvailableInheritedFlags}}\r\n\r\n" +
    "Global Flags:\r\n" +
    "{{.InheritedFlags.FlagUsages | trimTrailingWhitespaces}}{{end}}\r\n\r\n"

start the server with

go run dmusic.go -s $(tail -n1 dmusic.key)

The result - accessing the http endpoint, the music streaming works well.

Both audio elements in the html document at the start of this ticket work.

When accessing the dmsg url in a browser with dmsgweb, the music plays for one or two seconds, then stops.

When the dmsghttp web server application is stopped, the music resumes for a few more seconds, oddly.

the following panic is observed some time after the music initially stops:


2024/03/08 13:51:53 [Recovery] 2024/03/08 - 13:51:53 panic recovered:
GET /music HTTP/1.1
Host: 127.0.0.1:8081
Accept: */*
Accept-Encoding: identity;q=1, *;q=0
Accept-Language: en-US,en;q=0.9
Connection: keep-alive
Range: bytes=0-
Sec-Ch-Ua: "Not A(Brand";v="99", "Brave";v="121", "Chromium";v="121"
Sec-Ch-Ua-Mobile: ?0
Sec-Ch-Ua-Platform: "Linux"
Sec-Fetch-Dest: audio
Sec-Fetch-Mode: no-cors
Sec-Fetch-Site: cross-site
Sec-Gpc: 1
User-Agent: Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36

net/http: abort Handler
/usr/lib/go/src/net/http/httputil/reverseproxy.go:516 (0x71da84)
    (*ReverseProxy).ServeHTTP: panic(http.ErrAbortHandler)
/home/user/go/src/github.com/user/dmusic/d.go:160 (0xa6e372)
    Server.func4: proxy.ServeHTTP(c.Writer, c.Request)
/home/user/go/src/github.com/user/dmusic/vendor/github.com/gin-gonic/gin/context.go:174 (0x92fa6a)
    (*Context).Next: c.handlers[c.index](c)
/home/user/go/src/github.com/user/dmusic/d.go:265 (0xa6e54b)
    Server.loggingMiddleware.func9: c.Next()
/home/user/go/src/github.com/user/dmusic/vendor/github.com/gin-gonic/gin/context.go:174 (0x93bf19)
    (*Context).Next: c.handlers[c.index](c)
/home/user/go/src/github.com/user/dmusic/vendor/github.com/gin-gonic/gin/recovery.go:102 (0x93bf07)
    CustomRecoveryWithWriter.func1: c.Next()
/home/user/go/src/github.com/user/dmusic/vendor/github.com/gin-gonic/gin/context.go:174 (0x93ad2d)
    (*Context).Next: c.handlers[c.index](c)
/home/user/go/src/github.com/user/dmusic/vendor/github.com/gin-gonic/gin/gin.go:620 (0x93a9bc)
    (*Engine).handleHTTPRequest: c.Next()
/home/user/go/src/github.com/user/dmusic/vendor/github.com/gin-gonic/gin/gin.go:576 (0x93a4f1)
    (*Engine).ServeHTTP: engine.handleHTTPRequest(c)
/usr/lib/go/src/net/http/server.go:3137 (0x6f6b6d)
    serverHandler.ServeHTTP: handler.ServeHTTP(rw, req)
/usr/lib/go/src/net/http/server.go:2039 (0x6f1e47)
    (*conn).serve: serverHandler{c.server}.ServeHTTP(w, w.req)
/usr/lib/go/src/runtime/asm_amd64.s:1695 (0x479b40)
    goexit: BYTE    $0x90   // NOP
0pcom commented 8 months ago

this actually appears to be an issue with dmsgweb ; when I resolve the dmsg address with the -t flag for dmsgweb to the local web interface, the streaming works as it should. So it's the resolving proxy implementation which appears at issue