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