Live video on the AT Protocol
stream.place
1package cmd
2
3import (
4 "context"
5 "crypto"
6 "errors"
7 "flag"
8 "fmt"
9 "os"
10 "os/signal"
11 "path/filepath"
12 "runtime"
13 "runtime/pprof"
14 "strconv"
15 "strings"
16 "syscall"
17 "time"
18
19 "golang.org/x/term"
20 "stream.place/streamplace/pkg/aqhttp"
21 "stream.place/streamplace/pkg/atproto"
22 "stream.place/streamplace/pkg/bus"
23 "stream.place/streamplace/pkg/crypto/signers"
24 "stream.place/streamplace/pkg/crypto/signers/eip712"
25 "stream.place/streamplace/pkg/director"
26 "stream.place/streamplace/pkg/log"
27 "stream.place/streamplace/pkg/media"
28 "stream.place/streamplace/pkg/notifications"
29 "stream.place/streamplace/pkg/replication"
30 "stream.place/streamplace/pkg/replication/boring"
31 "stream.place/streamplace/pkg/rtmps"
32 v0 "stream.place/streamplace/pkg/schema/v0"
33 "stream.place/streamplace/pkg/spmetrics"
34
35 "github.com/ThalesGroup/crypto11"
36 _ "github.com/go-gst/go-glib/glib"
37 _ "github.com/go-gst/go-gst/gst"
38 "stream.place/streamplace/pkg/api"
39 "stream.place/streamplace/pkg/config"
40 "stream.place/streamplace/pkg/model"
41)
42
43// Additional jobs that can be injected by platforms
44type jobFunc func(ctx context.Context, cli *config.CLI) error
45
46// parse the CLI and fire up an streamplace node!
47func start(build *config.BuildFlags, platformJobs []jobFunc) error {
48 selfTest := len(os.Args) > 1 && os.Args[1] == "self-test"
49 err := media.RunSelfTest(context.Background())
50 if err != nil {
51 if selfTest {
52 fmt.Println(err.Error())
53 os.Exit(1)
54 } else {
55 retryCount, _ := strconv.Atoi(os.Getenv("STREAMPLACE_SELFTEST_RETRY"))
56 if retryCount >= 3 {
57 log.Error(context.Background(), "gstreamer self-test failed 3 times, giving up", "error", err)
58 return err
59 }
60 log.Log(context.Background(), "error in gstreamer self-test, attempting recovery", "error", err, "retry", retryCount+1)
61 os.Setenv("STREAMPLACE_SELFTEST_RETRY", strconv.Itoa(retryCount+1))
62 err := syscall.Exec(os.Args[0], os.Args[1:], os.Environ())
63 if err != nil {
64 log.Error(context.Background(), "error in gstreamer self-test, could not restart", "error", err)
65 return err
66 }
67 panic("invalid code path: exec succeeded but we're still here???")
68 }
69 }
70 if selfTest {
71 runtime.GC()
72 pprof.Lookup("goroutine").WriteTo(os.Stderr, 2)
73 fmt.Println("self-test successful!")
74 os.Exit(0)
75 }
76
77 if len(os.Args) > 1 && os.Args[1] == "stream" {
78 if len(os.Args) != 3 {
79 fmt.Println("usage: streamplace stream [user]")
80 os.Exit(1)
81 }
82 return Stream(os.Args[2])
83 }
84
85 if len(os.Args) > 1 && os.Args[1] == "live" {
86 if len(os.Args) != 3 {
87 fmt.Println("usage: streamplace live [stream-key]")
88 os.Exit(1)
89 }
90 return Live(os.Args[2])
91 }
92
93 if len(os.Args) > 1 && os.Args[1] == "sign" {
94 return Sign(context.Background())
95 }
96
97 if len(os.Args) > 1 && os.Args[1] == "whep" {
98 return WHEP(os.Args[2:])
99 }
100 if len(os.Args) > 1 && os.Args[1] == "whip" {
101 return WHIP(os.Args[2:])
102 }
103
104 if len(os.Args) > 1 && os.Args[1] == "self-test" {
105 err := media.RunSelfTest(context.Background())
106 if err != nil {
107 fmt.Println(err.Error())
108 os.Exit(1)
109 }
110 fmt.Println("self-test successful!")
111 os.Exit(0)
112 }
113 flag.Set("logtostderr", "true")
114 vFlag := flag.Lookup("v")
115 fs := flag.NewFlagSet("streamplace", flag.ExitOnError)
116 cli := config.CLI{Build: build}
117 fs.StringVar(&cli.DataDir, "data-dir", config.DefaultDataDir(), "directory for keeping all streamplace data")
118 fs.StringVar(&cli.HttpAddr, "http-addr", ":38080", "Public HTTP address")
119 fs.StringVar(&cli.HttpInternalAddr, "http-internal-addr", "127.0.0.1:39090", "Private, admin-only HTTP address")
120 fs.StringVar(&cli.HttpsAddr, "https-addr", ":38443", "Public HTTPS address")
121 fs.BoolVar(&cli.Secure, "secure", false, "Run with HTTPS. Required for WebRTC output")
122 cli.DataDirFlag(fs, &cli.TLSCertPath, "tls-cert", filepath.Join("tls", "tls.crt"), "Path to TLS certificate")
123 cli.DataDirFlag(fs, &cli.TLSKeyPath, "tls-key", filepath.Join("tls", "tls.key"), "Path to TLS key")
124 fs.StringVar(&cli.SigningKeyPath, "signing-key", "", "Path to signing key for pushing OTA updates to the app")
125 cli.DataDirFlag(fs, &cli.DBPath, "db-path", "db.sqlite", "path to sqlite database file")
126 fs.StringVar(&cli.AdminAccount, "admin-account", "", "ethereum account that administrates this streamplace node")
127 fs.StringVar(&cli.FirebaseServiceAccount, "firebase-service-account", "", "JSON string of a firebase service account key")
128 fs.StringVar(&cli.GitLabURL, "gitlab-url", "https://git.stream.place/api/v4/projects/1", "gitlab url for generating download links")
129 cli.DataDirFlag(fs, &cli.EthKeystorePath, "eth-keystore-path", "keystore", "path to ethereum keystore")
130 fs.StringVar(&cli.EthAccountAddr, "eth-account-addr", "", "ethereum account address to use (if keystore contains more than one)")
131 fs.StringVar(&cli.EthPassword, "eth-password", "", "password for encrypting keystore")
132 fs.StringVar(&cli.TAURL, "ta-url", "http://timestamp.digicert.com", "timestamp authority server for signing")
133 fs.StringVar(&cli.PKCS11ModulePath, "pkcs11-module-path", "", "path to a PKCS11 module for HSM signing, for example /usr/lib/x86_64-linux-gnu/opensc-pkcs11.so")
134 fs.StringVar(&cli.PKCS11Pin, "pkcs11-pin", "", "PIN for logging into PKCS11 token. if not provided, will be prompted interactively")
135 fs.StringVar(&cli.PKCS11TokenSlot, "pkcs11-token-slot", "", "slot number of PKCS11 token (only use one of slot, label, or serial)")
136 fs.StringVar(&cli.PKCS11TokenLabel, "pkcs11-token-label", "", "label of PKCS11 token (only use one of slot, label, or serial)")
137 fs.StringVar(&cli.PKCS11TokenSerial, "pkcs11-token-serial", "", "serial number of PKCS11 token (only use one of slot, label, or serial)")
138 fs.StringVar(&cli.PKCS11KeypairLabel, "pkcs11-keypair-label", "", "label of signing keypair on PKCS11 token")
139 fs.StringVar(&cli.PKCS11KeypairID, "pkcs11-keypair-id", "", "id of signing keypair on PKCS11 token")
140 fs.StringVar(&cli.AppBundleID, "app-bundle-id", "", "bundle id of an app that we facilitate oauth login for")
141 fs.StringVar(&cli.StreamerName, "streamer-name", "", "name of the person streaming from this streamplace node")
142 fs.StringVar(&cli.FrontendProxy, "dev-frontend-proxy", "", "(FOR DEVELOPMENT ONLY) proxy frontend requests to this address instead of using the bundled frontend")
143 fs.StringVar(&cli.LivepeerGatewayURL, "livepeer-gateway-url", "", "URL of the Livepeer Gateway to use for transcoding")
144 fs.BoolVar(&cli.WideOpen, "wide-open", false, "allow ALL streams to be uploaded to this node (not recommended for production)")
145 cli.StringSliceFlag(fs, &cli.AllowedStreams, "allowed-streams", "", "if set, only allow these addresses or atproto DIDs to upload to this node")
146 cli.StringSliceFlag(fs, &cli.Peers, "peers", "", "other streamplace nodes to replicate to")
147 cli.StringSliceFlag(fs, &cli.Redirects, "redirects", "", "http 302s /path/one:/path/two,/path/three:/path/four")
148 cli.DebugFlag(fs, &cli.Debug, "debug", "", "modified log verbosity for specific functions or files in form func=ToHLS:3,file=gstreamer.go:4")
149 fs.BoolVar(&cli.TestStream, "test-stream", false, "run a built-in test stream on boot")
150 fs.BoolVar(&cli.NoFirehose, "no-firehose", false, "disable the bluesky firehose")
151 fs.BoolVar(&cli.PrintChat, "print-chat", false, "print chat messages to stdout")
152 fs.StringVar(&cli.WHIPTest, "whip-test", "", "run a WHIP self-test with the given parameters")
153 verbosity := fs.String("v", "3", "log verbosity level")
154 fs.StringVar(&cli.RelayHost, "relay-host", "wss://bsky.network", "websocket url for relay firehose")
155 fs.Bool("insecure", false, "DEPRECATED, does nothing.")
156 fs.StringVar(&cli.Color, "color", "", "'true' to enable colorized logging, 'false' to disable")
157 fs.StringVar(&cli.PublicHost, "public-host", "", "public host for this streamplace node (excluding https:// e.g. stream.place)")
158 fs.BoolVar(&cli.Thumbnail, "thumbnail", true, "enable thumbnail generation")
159 fs.BoolVar(&cli.SmearAudio, "smear-audio", false, "enable audio smearing to create 'perfect' segment timestamps")
160 fs.BoolVar(&cli.ExternalSigning, "external-signing", false, "enable external signing via exec (prevents potential memory leak)")
161 fs.StringVar(&cli.TracingEndpoint, "tracing-endpoint", "", "gRPC endpoint to send traces to")
162 fs.IntVar(&cli.RateLimitPerSecond, "rate-limit-per-second", 10, "rate limit for requests per second per ip")
163 fs.IntVar(&cli.RateLimitBurst, "rate-limit-burst", 10, "rate limit burst for requests per ip")
164 fs.IntVar(&cli.RateLimitWebsocket, "rate-limit-websocket", 10, "number of concurrent websocket connections allowed per ip")
165 fs.StringVar(&cli.RTMPServerAddon, "rtmp-server-addon", "", "address of external RTMP server to forward streams to")
166 fs.StringVar(&cli.RtmpsAddr, "rtmps-addr", ":1935", "address to listen for RTMPS connections")
167 version := fs.Bool("version", false, "print version and exit")
168
169 if runtime.GOOS == "linux" {
170 fs.BoolVar(&cli.NoMist, "no-mist", true, "Disable MistServer")
171 fs.IntVar(&cli.MistAdminPort, "mist-admin-port", 14242, "MistServer admin port (internal use only)")
172 fs.IntVar(&cli.MistRTMPPort, "mist-rtmp-port", 11935, "MistServer RTMP port (internal use only)")
173 fs.IntVar(&cli.MistHTTPPort, "mist-http-port", 18080, "MistServer HTTP port (internal use only)")
174 }
175
176 err = cli.Parse(
177 fs, os.Args[1:],
178 )
179 if err != nil {
180 return err
181 }
182 err = flag.CommandLine.Parse(nil)
183 if err != nil {
184 return err
185 }
186 vFlag.Value.Set(*verbosity)
187 log.SetColorLogger(cli.Color)
188 ctx := context.Background()
189 ctx = log.WithDebugValue(ctx, cli.Debug)
190
191 log.Log(ctx,
192 "streamplace",
193 "version", build.Version,
194 "buildTime", build.BuildTimeStr(),
195 "uuid", build.UUID,
196 "runtime.GOOS", runtime.GOOS,
197 "runtime.GOARCH", runtime.GOARCH,
198 "runtime.Version", runtime.Version())
199 if *version {
200 return nil
201 }
202 spmetrics.Version.WithLabelValues(build.Version).Inc()
203
204 aqhttp.UserAgent = fmt.Sprintf("streamplace/%s", build.Version)
205
206 err = os.MkdirAll(cli.DataDir, os.ModePerm)
207 if err != nil {
208 return fmt.Errorf("error creating streamplace dir at %s:%w", cli.DataDir, err)
209 }
210 schema, err := v0.MakeV0Schema()
211 if err != nil {
212 return err
213 }
214 eip712signer, err := eip712.MakeEIP712Signer(ctx, &eip712.EIP712SignerOptions{
215 Schema: schema,
216 EthKeystorePath: cli.EthKeystorePath,
217 EthAccountAddr: cli.EthAccountAddr,
218 EthKeystorePassword: cli.EthPassword,
219 })
220 if err != nil {
221 return err
222 }
223 var signer crypto.Signer = eip712signer
224 if cli.PKCS11ModulePath != "" {
225 conf := &crypto11.Config{
226 Path: cli.PKCS11ModulePath,
227 }
228 count := 0
229 for _, val := range []string{cli.PKCS11TokenSlot, cli.PKCS11TokenLabel, cli.PKCS11TokenSerial} {
230 if val != "" {
231 count += 1
232 }
233 }
234 if count != 1 {
235 return fmt.Errorf("need exactly one of pkcs11-token-slot, pkcs11-token-label, or pkcs11-token-serial (got %d)", count)
236 }
237 if cli.PKCS11TokenSlot != "" {
238 num, err := strconv.ParseInt(cli.PKCS11TokenSlot, 10, 16)
239 if err != nil {
240 return fmt.Errorf("error parsing pkcs11-slot: %w", err)
241 }
242 numint := int(num)
243 // why does crypto11 want this as a reference? odd.
244 conf.SlotNumber = &numint
245 }
246 if cli.PKCS11TokenLabel != "" {
247 conf.TokenLabel = cli.PKCS11TokenLabel
248 }
249 if cli.PKCS11TokenSerial != "" {
250 conf.TokenSerial = cli.PKCS11TokenSerial
251 }
252 pin := cli.PKCS11Pin
253 if pin == "" {
254 fmt.Printf("Please enter PKCS11 PIN: ")
255 password, err := term.ReadPassword(int(os.Stdin.Fd()))
256 fmt.Println("")
257 if err != nil {
258 return fmt.Errorf("error reading PKCS11 password: %w", err)
259 }
260 pin = string(password)
261 }
262 conf.Pin = pin
263
264 sc, err := crypto11.Configure(conf)
265 if err != nil {
266 return fmt.Errorf("error initalizing PKCS11 HSM: %w", err)
267 }
268 var id []byte = nil
269 var label []byte = nil
270 if cli.PKCS11KeypairID != "" {
271 num, err := strconv.ParseInt(cli.PKCS11KeypairID, 10, 8)
272 if err != nil {
273 return fmt.Errorf("error parsing pkcs11-keypair-id: %w", err)
274 }
275 id = []byte{byte(num)}
276 }
277 if cli.PKCS11KeypairLabel != "" {
278 label = []byte(cli.PKCS11KeypairLabel)
279 }
280 hwsigner, err := sc.FindKeyPair(id, label)
281 if err != nil {
282 return fmt.Errorf("error finding keypair on PKCS11 token: %w", err)
283 }
284 if hwsigner == nil {
285 return fmt.Errorf("keypair on token not found (tried id='%s' label='%s')", cli.PKCS11KeypairID, cli.PKCS11KeypairLabel)
286 }
287 addr, err := signers.HexAddrFromSigner(hwsigner)
288 if err != nil {
289 return fmt.Errorf("error getting ethereum address for hardware keypair: %w", err)
290 }
291 log.Log(ctx, "successfully initialized hardware signer", "address", addr)
292 signer = hwsigner
293 }
294 var rep replication.Replicator = &boring.BoringReplicator{Peers: cli.Peers}
295 mod, err := model.MakeDB(cli.DBPath)
296 if err != nil {
297 return err
298 }
299 var noter notifications.FirebaseNotifier
300 if cli.FirebaseServiceAccount != "" {
301 noter, err = notifications.MakeFirebaseNotifier(ctx, cli.FirebaseServiceAccount)
302 if err != nil {
303 return err
304 }
305 }
306
307 jwkPath := cli.DataFilePath([]string{"jwk.json"})
308 jwk, err := atproto.EnsureJWK(ctx, jwkPath)
309 if err != nil {
310 return err
311 }
312 cli.JWK = jwk
313
314 accessJWKPath := cli.DataFilePath([]string{"access-jwk.json"})
315 accessJWK, err := atproto.EnsureJWK(ctx, accessJWKPath)
316 if err != nil {
317 return err
318 }
319 cli.AccessJWK = accessJWK
320
321 b := bus.NewBus()
322 atsync := &atproto.ATProtoSynchronizer{
323 CLI: &cli,
324 Model: mod,
325 Noter: noter,
326 Bus: b,
327 }
328 mm, err := media.MakeMediaManager(ctx, &cli, signer, rep, mod, b, atsync)
329 if err != nil {
330 return err
331 }
332
333 ms, err := media.MakeMediaSigner(ctx, &cli, cli.StreamerName, signer)
334 if err != nil {
335 return err
336 }
337
338 d := director.NewDirector(mm, mod, &cli, b)
339
340 a, err := api.MakeStreamplaceAPI(&cli, mod, eip712signer, noter, mm, ms, b, atsync, d)
341 if err != nil {
342 return err
343 }
344
345 group, ctx := TimeoutGroupWithContext(ctx)
346 ctx = log.WithLogValues(ctx, "version", build.Version)
347
348 group.Go(func() error {
349 return handleSignals(ctx)
350 })
351
352 if cli.TracingEndpoint != "" {
353 group.Go(func() error {
354 return startTelemetry(ctx, cli.TracingEndpoint)
355 })
356 }
357
358 if cli.Secure {
359 group.Go(func() error {
360 return a.ServeHTTPS(ctx)
361 })
362 group.Go(func() error {
363 return a.ServeHTTPRedirect(ctx)
364 })
365 if cli.RTMPServerAddon != "" {
366 group.Go(func() error {
367 return rtmps.ServeRTMPS(ctx, &cli)
368 })
369 }
370 } else {
371 group.Go(func() error {
372 return a.ServeHTTP(ctx)
373 })
374 }
375
376 group.Go(func() error {
377 return a.ServeInternalHTTP(ctx)
378 })
379
380 if !cli.NoFirehose {
381 group.Go(func() error {
382 return atsync.StartFirehose(ctx)
383 })
384 }
385
386 group.Go(func() error {
387 return spmetrics.ExpireSessions(ctx)
388 })
389
390 group.Go(func() error {
391 return mod.StartSegmentCleaner(ctx)
392 })
393
394 group.Go(func() error {
395 return d.Start(ctx)
396 })
397
398 if cli.TestStream {
399 testSigner, err := eip712.MakeEIP712Signer(ctx, &eip712.EIP712SignerOptions{
400 Schema: schema,
401 EthKeystorePath: filepath.Join(cli.DataDir, "test-signer"),
402 })
403 if err != nil {
404 return err
405 }
406 atkey, err := atproto.ParsePubKey(signer.Public())
407 if err != nil {
408 return err
409 }
410 did := atkey.DIDKey()
411 testMediaSigner, err := media.MakeMediaSigner(ctx, &cli, did, testSigner)
412 if err != nil {
413 return err
414 }
415 err = mod.UpdateIdentity(&model.Identity{
416 ID: testMediaSigner.Pub().String(),
417 Handle: "stream-self-tester",
418 DID: "",
419 })
420 if err != nil {
421 return err
422 }
423 cli.AllowedStreams = append(cli.AllowedStreams, did)
424 a.Aliases["self-test"] = did
425 group.Go(func() error {
426 return mm.TestSource(ctx, testMediaSigner)
427 })
428 }
429
430 for _, job := range platformJobs {
431 group.Go(func() error {
432 return job(ctx, &cli)
433 })
434 }
435
436 if cli.WHIPTest != "" {
437 group.Go(func() error {
438 err := WHIP(strings.Split(cli.WHIPTest, " "))
439 log.Warn(ctx, "WHIP test complete, sleeping for 3 seconds and shutting down gstreamer")
440 time.Sleep(time.Second * 3)
441 // gst.Deinit()
442 log.Warn(ctx, "gst deinit complete, exiting")
443 return err
444 })
445 }
446
447 return group.Wait()
448}
449
450var ErrCaughtSignal = errors.New("caught signal")
451
452func handleSignals(ctx context.Context) error {
453 c := make(chan os.Signal, 1)
454 signal.Notify(c, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT, syscall.SIGABRT)
455 for {
456 select {
457 case s := <-c:
458 if s == syscall.SIGABRT {
459 pprof.Lookup("goroutine").WriteTo(os.Stderr, 2)
460 }
461 log.Log(ctx, "caught signal, attempting clean shutdown", "signal", s)
462 return fmt.Errorf("%w signal=%v", ErrCaughtSignal, s)
463 case <-ctx.Done():
464 return nil
465 }
466 }
467}