stream.place
Live video on the AT Protocol
1package media
2
3import (
4 "bytes"
5 "context"
6 "crypto"
7 "crypto/ecdsa"
8 "fmt"
9 "io"
10 "os"
11 "os/exec"
12 "time"
13
14 "github.com/decred/dcrd/dcrec/secp256k1"
15 "github.com/mr-tron/base58"
16 "go.opentelemetry.io/otel"
17 "stream.place/streamplace/pkg/config"
18 "stream.place/streamplace/pkg/crypto/aqpub"
19 "stream.place/streamplace/pkg/spmetrics"
20)
21
22type MediaSignerExt struct {
23 cli *config.CLI
24 signer crypto.Signer
25 pub aqpub.Pub
26 certPath string
27 streamer string
28 keyBs []byte
29 taURL string
30}
31
32func MakeMediaSignerExt(ctx context.Context, cli *config.CLI, streamer string, keyBs []byte) (MediaSigner, error) {
33 key, _ := secp256k1.PrivKeyFromBytes(keyBs)
34 if key == nil {
35 return nil, fmt.Errorf("invalid authorization key (not valid secp256k1)")
36 }
37 var signer crypto.Signer = key.ToECDSA()
38 _, certPath, err := prepareCert(ctx, cli, signer)
39 if err != nil {
40 return nil, err
41 }
42 pub, err := aqpub.FromPublicKey(signer.Public().(*ecdsa.PublicKey))
43 if err != nil {
44 return nil, err
45 }
46 return &MediaSignerExt{
47 // cli: cli,
48 signer: signer,
49 certPath: certPath,
50 streamer: streamer,
51 pub: pub,
52 keyBs: keyBs,
53 taURL: cli.TAURL,
54 }, nil
55}
56
57func (ms *MediaSignerExt) SignMP4(ctx context.Context, input io.ReadSeeker, start int64) ([]byte, error) {
58 startTime := time.Now()
59 ctx, span := otel.Tracer("signer").Start(ctx, "SignMP4_Ext")
60 defer span.End()
61 // Get the path to the current executable
62 execPath, err := os.Executable()
63 if err != nil {
64 return nil, fmt.Errorf("failed to get executable path: %w", err)
65 }
66
67 enc := base58.Encode(ms.keyBs)
68
69 // Prepare command
70 cmd := exec.Command(execPath, "sign",
71 "--key", enc,
72 "--cert", ms.certPath,
73 "--ta-url", ms.taURL,
74 "--streamer", ms.streamer,
75 "--start-time", fmt.Sprintf("%d", start))
76
77 // overwrite so that our subprocesses don't do their own leak checking
78 cmd.Env = append(os.Environ(), "LD_PRELOAD=")
79
80 // Set up pipes for stdin and stdout
81 stdin, err := cmd.StdinPipe()
82 if err != nil {
83 return nil, fmt.Errorf("failed to create stdin pipe: %w", err)
84 }
85
86 stdout := &bytes.Buffer{}
87 cmd.Stdout = stdout
88 stderr := &bytes.Buffer{}
89 cmd.Stderr = stderr
90
91 // Start the command
92 if err := cmd.Start(); err != nil {
93 return nil, fmt.Errorf("failed to start command: %w", err)
94 }
95
96 // Copy input to stdin
97 _, err = io.Copy(stdin, input)
98 if err != nil {
99 return nil, fmt.Errorf("failed to write to stdin: %w stderr=%s", err, stderr.String())
100 }
101 stdin.Close()
102
103 // Wait for the command to complete
104 if err := cmd.Wait(); err != nil {
105 return nil, fmt.Errorf("command failed: %w, stderr: %s", err, stderr.String())
106 }
107 spmetrics.SigningDuration.WithLabelValues(ms.streamer).Observe(float64(time.Since(startTime).Milliseconds()))
108 return stdout.Bytes(), nil
109}
110
111func (ms *MediaSignerExt) Pub() aqpub.Pub {
112 return ms.pub
113}
114
115func (ms *MediaSignerExt) Streamer() string {
116 return ms.streamer
117}