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