Live video on the AT Protocol
stream.place
seiso.moe
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 version := fs.Bool("version", false, "print version and exit")
171
172 if runtime.GOOS == "linux" {
173 fs.BoolVar(&cli.NoMist, "no-mist", true, "Disable MistServer")
174 fs.IntVar(&cli.MistAdminPort, "mist-admin-port", 14242, "MistServer admin port (internal use only)")
175 fs.IntVar(&cli.MistRTMPPort, "mist-rtmp-port", 11935, "MistServer RTMP port (internal use only)")
176 fs.IntVar(&cli.MistHTTPPort, "mist-http-port", 18080, "MistServer HTTP port (internal use only)")
177 }
178
179 err = cli.Parse(
180 fs, os.Args[1:],
181 )
182 if err != nil {
183 return err
184 }
185 err = flag.CommandLine.Parse(nil)
186 if err != nil {
187 return err
188 }
189 _ = vFlag.Value.Set(*verbosity)
190 log.SetColorLogger(cli.Color)
191 ctx := context.Background()
192 ctx = log.WithDebugValue(ctx, cli.Debug)
193
194 log.Log(ctx,
195 "streamplace",
196 "version", build.Version,
197 "buildTime", build.BuildTimeStr(),
198 "uuid", build.UUID,
199 "runtime.GOOS", runtime.GOOS,
200 "runtime.GOARCH", runtime.GOARCH,
201 "runtime.Version", runtime.Version())
202 if *version {
203 return nil
204 }
205 spmetrics.Version.WithLabelValues(build.Version).Inc()
206
207 aqhttp.UserAgent = fmt.Sprintf("streamplace/%s", build.Version)
208
209 err = os.MkdirAll(cli.DataDir, os.ModePerm)
210 if err != nil {
211 return fmt.Errorf("error creating streamplace dir at %s:%w", cli.DataDir, err)
212 }
213 schema, err := v0.MakeV0Schema()
214 if err != nil {
215 return err
216 }
217 eip712signer, err := eip712.MakeEIP712Signer(ctx, &eip712.EIP712SignerOptions{
218 Schema: schema,
219 EthKeystorePath: cli.EthKeystorePath,
220 EthAccountAddr: cli.EthAccountAddr,
221 EthKeystorePassword: cli.EthPassword,
222 })
223 if err != nil {
224 return err
225 }
226 var signer crypto.Signer = eip712signer
227 if cli.PKCS11ModulePath != "" {
228 conf := &crypto11.Config{
229 Path: cli.PKCS11ModulePath,
230 }
231 count := 0
232 for _, val := range []string{cli.PKCS11TokenSlot, cli.PKCS11TokenLabel, cli.PKCS11TokenSerial} {
233 if val != "" {
234 count += 1
235 }
236 }
237 if count != 1 {
238 return fmt.Errorf("need exactly one of pkcs11-token-slot, pkcs11-token-label, or pkcs11-token-serial (got %d)", count)
239 }
240 if cli.PKCS11TokenSlot != "" {
241 num, err := strconv.ParseInt(cli.PKCS11TokenSlot, 10, 16)
242 if err != nil {
243 return fmt.Errorf("error parsing pkcs11-slot: %w", err)
244 }
245 numint := int(num)
246 // why does crypto11 want this as a reference? odd.
247 conf.SlotNumber = &numint
248 }
249 if cli.PKCS11TokenLabel != "" {
250 conf.TokenLabel = cli.PKCS11TokenLabel
251 }
252 if cli.PKCS11TokenSerial != "" {
253 conf.TokenSerial = cli.PKCS11TokenSerial
254 }
255 pin := cli.PKCS11Pin
256 if pin == "" {
257 fmt.Printf("Please enter PKCS11 PIN: ")
258 password, err := term.ReadPassword(int(os.Stdin.Fd()))
259 fmt.Println("")
260 if err != nil {
261 return fmt.Errorf("error reading PKCS11 password: %w", err)
262 }
263 pin = string(password)
264 }
265 conf.Pin = pin
266
267 sc, err := crypto11.Configure(conf)
268 if err != nil {
269 return fmt.Errorf("error initalizing PKCS11 HSM: %w", err)
270 }
271 var id []byte = nil
272 var label []byte = nil
273 if cli.PKCS11KeypairID != "" {
274 num, err := strconv.ParseInt(cli.PKCS11KeypairID, 10, 8)
275 if err != nil {
276 return fmt.Errorf("error parsing pkcs11-keypair-id: %w", err)
277 }
278 id = []byte{byte(num)}
279 }
280 if cli.PKCS11KeypairLabel != "" {
281 label = []byte(cli.PKCS11KeypairLabel)
282 }
283 hwsigner, err := sc.FindKeyPair(id, label)
284 if err != nil {
285 return fmt.Errorf("error finding keypair on PKCS11 token: %w", err)
286 }
287 if hwsigner == nil {
288 return fmt.Errorf("keypair on token not found (tried id='%s' label='%s')", cli.PKCS11KeypairID, cli.PKCS11KeypairLabel)
289 }
290 addr, err := signers.HexAddrFromSigner(hwsigner)
291 if err != nil {
292 return fmt.Errorf("error getting ethereum address for hardware keypair: %w", err)
293 }
294 log.Log(ctx, "successfully initialized hardware signer", "address", addr)
295 signer = hwsigner
296 }
297 var rep replication.Replicator = &boring.BoringReplicator{Peers: cli.Peers}
298 mod, err := model.MakeDB(cli.DBPath)
299 if err != nil {
300 return err
301 }
302 var noter notifications.FirebaseNotifier
303 if cli.FirebaseServiceAccount != "" {
304 noter, err = notifications.MakeFirebaseNotifier(ctx, cli.FirebaseServiceAccount)
305 if err != nil {
306 return err
307 }
308 }
309
310 jwkPath := cli.DataFilePath([]string{"jwk.json"})
311 jwk, err := atproto.EnsureJWK(ctx, jwkPath)
312 if err != nil {
313 return err
314 }
315 cli.JWK = jwk
316
317 accessJWKPath := cli.DataFilePath([]string{"access-jwk.json"})
318 accessJWK, err := atproto.EnsureJWK(ctx, accessJWKPath)
319 if err != nil {
320 return err
321 }
322 cli.AccessJWK = accessJWK
323
324 b := bus.NewBus()
325 atsync := &atproto.ATProtoSynchronizer{
326 CLI: &cli,
327 Model: mod,
328 Noter: noter,
329 Bus: b,
330 }
331 mm, err := media.MakeMediaManager(ctx, &cli, signer, rep, mod, b, atsync)
332 if err != nil {
333 return err
334 }
335
336 ms, err := media.MakeMediaSigner(ctx, &cli, cli.StreamerName, signer)
337 if err != nil {
338 return err
339 }
340
341 clientMetadata := &oatproxy.OAuthClientMetadata{
342 Scope: "atproto transition:generic",
343 ClientName: "Streamplace",
344 RedirectURIs: []string{
345 fmt.Sprintf("https://%s/login", cli.PublicHost),
346 fmt.Sprintf("https://%s/api/app-return", cli.PublicHost),
347 },
348 }
349
350 op := oatproxy.New(&oatproxy.Config{
351 Host: cli.PublicHost,
352 CreateOAuthSession: mod.CreateOAuthSession,
353 UpdateOAuthSession: mod.UpdateOAuthSession,
354 GetOAuthSession: mod.LoadOAuthSession,
355 Scope: "atproto transition:generic",
356 UpstreamJWK: cli.JWK,
357 DownstreamJWK: cli.AccessJWK,
358 ClientMetadata: clientMetadata,
359 })
360 d := director.NewDirector(mm, mod, &cli, b, op)
361 a, err := api.MakeStreamplaceAPI(&cli, mod, eip712signer, noter, mm, ms, b, atsync, d, op)
362 if err != nil {
363 return err
364 }
365
366 group, ctx := TimeoutGroupWithContext(ctx)
367 ctx = log.WithLogValues(ctx, "version", build.Version)
368
369 group.Go(func() error {
370 return handleSignals(ctx)
371 })
372
373 if cli.TracingEndpoint != "" {
374 group.Go(func() error {
375 return startTelemetry(ctx, cli.TracingEndpoint)
376 })
377 }
378
379 if cli.Secure {
380 group.Go(func() error {
381 return a.ServeHTTPS(ctx)
382 })
383 group.Go(func() error {
384 return a.ServeHTTPRedirect(ctx)
385 })
386 if cli.RTMPServerAddon != "" {
387 group.Go(func() error {
388 return rtmps.ServeRTMPS(ctx, &cli)
389 })
390 }
391 } else {
392 group.Go(func() error {
393 return a.ServeHTTP(ctx)
394 })
395 }
396
397 group.Go(func() error {
398 return a.ServeInternalHTTP(ctx)
399 })
400
401 if !cli.NoFirehose {
402 group.Go(func() error {
403 return atsync.StartFirehose(ctx)
404 })
405 }
406
407 group.Go(func() error {
408 return spmetrics.ExpireSessions(ctx)
409 })
410
411 group.Go(func() error {
412 return mod.StartSegmentCleaner(ctx)
413 })
414
415 group.Go(func() error {
416 return d.Start(ctx)
417 })
418
419 if cli.TestStream {
420 testSigner, err := eip712.MakeEIP712Signer(ctx, &eip712.EIP712SignerOptions{
421 Schema: schema,
422 EthKeystorePath: filepath.Join(cli.DataDir, "test-signer"),
423 })
424 if err != nil {
425 return err
426 }
427 atkey, err := atproto.ParsePubKey(signer.Public())
428 if err != nil {
429 return err
430 }
431 did := atkey.DIDKey()
432 testMediaSigner, err := media.MakeMediaSigner(ctx, &cli, did, testSigner)
433 if err != nil {
434 return err
435 }
436 err = mod.UpdateIdentity(&model.Identity{
437 ID: testMediaSigner.Pub().String(),
438 Handle: "stream-self-tester",
439 DID: "",
440 })
441 if err != nil {
442 return err
443 }
444 cli.AllowedStreams = append(cli.AllowedStreams, did)
445 a.Aliases["self-test"] = did
446 group.Go(func() error {
447 return mm.TestSource(ctx, testMediaSigner)
448 })
449 }
450
451 for _, job := range platformJobs {
452 group.Go(func() error {
453 return job(ctx, &cli)
454 })
455 }
456
457 if cli.WHIPTest != "" {
458 group.Go(func() error {
459 err := WHIP(strings.Split(cli.WHIPTest, " "))
460 log.Warn(ctx, "WHIP test complete, sleeping for 3 seconds and shutting down gstreamer")
461 time.Sleep(time.Second * 3)
462 // gst.Deinit()
463 log.Warn(ctx, "gst deinit complete, exiting")
464 return err
465 })
466 }
467
468 return group.Wait()
469}
470
471var ErrCaughtSignal = errors.New("caught signal")
472
473func handleSignals(ctx context.Context) error {
474 c := make(chan os.Signal, 1)
475 signal.Notify(c, syscall.SIGQUIT, syscall.SIGTERM, syscall.SIGINT, syscall.SIGABRT)
476 for {
477 select {
478 case s := <-c:
479 if s == syscall.SIGABRT {
480 if err := pprof.Lookup("goroutine").WriteTo(os.Stderr, 2); err != nil {
481 log.Error(ctx, "failed to create pprof", "error", err)
482 }
483 }
484 log.Log(ctx, "caught signal, attempting clean shutdown", "signal", s)
485 return fmt.Errorf("%w signal=%v", ErrCaughtSignal, s)
486 case <-ctx.Done():
487 return nil
488 }
489 }
490}