apps/go2rtc
1
0
Fork 0
mirror of https://github.com/AlexxIT/go2rtc.git synced 2026-04-22 15:47:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Code refactoring for #1744

Browse Source
This commit is contained in:
Alex X
2025-10-07 13:25:42 +03:00
parent 670370056c
commit fe2cc4b525
20 changed files with 269 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/api.go
+3 -15
View File
@@ -1,7 +1,6 @@
package api
import (
"bytes"
"crypto/tls"
"encoding/json"
"fmt"
@@ -15,7 +14,6 @@ import (
"time"
"github.com/AlexxIT/go2rtc/internal/app"
"github.com/AlexxIT/go2rtc/pkg/shell"
"github.com/rs/zerolog"
)
@@ -167,19 +165,9 @@ func ResponseJSON(w http.ResponseWriter, v any) {
func ResponsePrettyJSON(w http.ResponseWriter, v any) {
w.Header().Set("Content-Type", MimeJSON)
var buf bytes.Buffer
enc := json.NewEncoder(&buf)
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
err := enc.Encode(v)
if err != nil {
http.Error(w, "Internal Server Error", http.StatusInternalServerError)
return
}
redactedJSON := shell.Redact(buf.String())
w.Write([]byte(redactedJSON))
_ = enc.Encode(v)
}
func Response(w http.ResponseWriter, body any, contentType string) {
@@ -202,7 +190,7 @@ var log zerolog.Logger
func middlewareLog(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
log.Trace().Msgf("[api] %s %s %s", r.Method, shell.Redact(r.URL.String()), r.RemoteAddr)
log.Trace().Msgf("[api] %s %s %s", r.Method, r.URL, r.RemoteAddr)
next.ServeHTTP(w, r)
})
}
internal/app/app.go
-1
View File
@@ -67,7 +67,6 @@ func Init() {
Info["revision"] = revision
initConfig(config)
initSecrets()
initLogger()
platform := fmt.Sprintf("%s/%s", runtime.GOOS, runtime.GOARCH)
internal/app/config.go
+4 -2
View File
@@ -7,7 +7,7 @@ import (
"strings"
"sync"
"github.com/AlexxIT/go2rtc/pkg/shell"
"github.com/AlexxIT/go2rtc/pkg/creds"
"github.com/AlexxIT/go2rtc/pkg/yaml"
)
@@ -71,13 +71,15 @@ func initConfig(confs flagConfig) {
// config as file
if ConfigPath == "" {
ConfigPath = conf
initStorage()
}
if data, _ = os.ReadFile(conf); data == nil {
continue
}
data = []byte(shell.ReplaceEnvVars(string(data)))
loadEnv(data)
data = creds.ReplaceVars(data)
configs = append(configs, data)
}
}
internal/app/log.go
+3
View File
@@ -6,6 +6,7 @@ import (
"strings"
"sync"
"github.com/AlexxIT/go2rtc/pkg/creds"
"github.com/mattn/go-isatty"
"github.com/rs/zerolog"
)
@@ -88,6 +89,8 @@ func initLogger() {
writer = MemoryLog
}
writer = creds.SecretWriter(writer)
lvl, _ := zerolog.ParseLevel(modules["level"])
Logger = zerolog.New(writer).Level(lvl)
internal/app/secrets.go
-136
View File
@@ -1,136 +0,0 @@
package app
import (
"sync"
"github.com/AlexxIT/go2rtc/pkg/secrets"
"github.com/AlexxIT/go2rtc/pkg/yaml"
)
var (
secretsMap = make(map[string]*Secret)
secretsMu sync.Mutex
)
// SecretsManager implements secrets.SecretsManager interface
type SecretsManager struct{}
func (m *SecretsManager) NewSecret(name string, values interface{}) (secrets.Secret, error) {
secretsMu.Lock()
defer secretsMu.Unlock()
if s, exists := secretsMap[name]; exists {
return s, nil
}
s := &Secret{Name: name, Values: make(map[string]string)}
data, err := yaml.Encode(values, 2)
if err != nil {
return nil, err
}
if err := yaml.Unmarshal(data, &s.Values); err != nil {
return nil, err
}
secretsMap[name] = s
return s, nil
}
func (m *SecretsManager) GetSecret(name string) secrets.Secret {
secretsMu.Lock()
defer secretsMu.Unlock()
return secretsMap[name]
}
// Secret implements secrets.Secret interface
type Secret struct {
Name string
Values map[string]string
}
func (s *Secret) Get(key string) any {
secretsMu.Lock()
defer secretsMu.Unlock()
if s.Values == nil {
return nil
}
return s.Values[key]
}
func (s *Secret) Set(key string, value string) {
secretsMu.Lock()
defer secretsMu.Unlock()
if s.Values == nil {
s.Values = make(map[string]string)
}
s.Values[key] = value
}
func (s *Secret) Marshal() (interface{}, error) {
secretsMu.Lock()
defer secretsMu.Unlock()
if s.Values == nil {
return make(map[string]any), nil
}
return s.Values, nil
}
func (s *Secret) Unmarshal(value any) error {
secretsMu.Lock()
defer secretsMu.Unlock()
if s.Values == nil {
s.Values = make(map[string]string)
}
data, err := yaml.Encode(value, 2)
if err != nil {
return err
}
if err := yaml.Unmarshal(data, value); err != nil {
return err
}
return nil
}
func (s *Secret) Save() error {
secretsMu.Lock()
defer secretsMu.Unlock()
return PatchConfig([]string{"secrets", s.Name}, s.Values)
}
func initSecrets() {
var cfg struct {
Secrets map[string]map[string]string `yaml:"secrets"`
}
LoadConfig(&cfg)
if cfg.Secrets == nil {
return
}
secretsMu.Lock()
defer secretsMu.Unlock()
for name, values := range cfg.Secrets {
secretsMap[name] = &Secret{
Name: name,
Values: values,
}
}
// Register
secrets.SetManager(&SecretsManager{})
}
internal/app/storage.go
+56
View File
@@ -0,0 +1,56 @@
package app
import (
"sync"
"github.com/AlexxIT/go2rtc/pkg/creds"
"github.com/AlexxIT/go2rtc/pkg/yaml"
)
func initStorage() {
storage = &envStorage{data: make(map[string]string)}
creds.SetStorage(storage)
}
func loadEnv(data []byte) {
var cfg struct {
Env map[string]string `yaml:"env"`
}
if err := yaml.Unmarshal(data, &cfg); err != nil {
return
}
storage.mu.Lock()
for name, value := range cfg.Env {
storage.data[name] = value
creds.AddSecret(value)
}
storage.mu.Unlock()
}
var storage *envStorage
type envStorage struct {
data map[string]string
mu sync.Mutex
}
func (s *envStorage) SetValue(name, value string) error {
if err := PatchConfig([]string{"env", name}, value); err != nil {
return err
}
s.mu.Lock()
s.data[name] = value
s.mu.Unlock()
return nil
}
func (s *envStorage) GetValue(name string) (value string, ok bool) {
s.mu.Lock()
value, ok = s.data[name]
s.mu.Unlock()
return
}
internal/echo/echo.go
+1 -1
View File
@@ -22,7 +22,7 @@ func Init() {
b = bytes.TrimSpace(b)
log.Debug().Str("url", shell.Redact(url)).Msgf("[echo] %s", b)
log.Debug().Str("url", url).Msgf("[echo] %s", b)
return string(b), nil
})
internal/expr/expr.go
+1 -2
View File
@@ -6,7 +6,6 @@ import (
"github.com/AlexxIT/go2rtc/internal/app"
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/expr"
"github.com/AlexxIT/go2rtc/pkg/shell"
)
func Init() {
@@ -18,7 +17,7 @@ func Init() {
return "", err
}
log.Debug().Msgf("[expr] url=%s", shell.Redact(url))
log.Debug().Msgf("[expr] url=%s", url)
if url = v.(string); url == "" {
return "", errors.New("expr: result is empty")
internal/hls/hls.go
+3 -4
View File
@@ -12,7 +12,6 @@ import (
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/mp4"
"github.com/AlexxIT/go2rtc/pkg/mpegts"
"github.com/AlexxIT/go2rtc/pkg/shell"
"github.com/rs/zerolog"
)
@@ -143,7 +142,7 @@ func handlerSegmentTS(w http.ResponseWriter, r *http.Request) {
data := session.Segment()
if data == nil {
log.Warn().Msgf("[hls] can't get segment %s", shell.Redact(r.URL.RawQuery))
log.Warn().Msgf("[hls] can't get segment %s", r.URL.RawQuery)
http.NotFound(w, r)
return
}
@@ -173,7 +172,7 @@ func handlerInit(w http.ResponseWriter, r *http.Request) {
data := session.Init()
if data == nil {
log.Warn().Msgf("[hls] can't get init %s", shell.Redact(r.URL.RawQuery))
log.Warn().Msgf("[hls] can't get init %s", r.URL.RawQuery)
http.NotFound(w, r)
return
}
@@ -207,7 +206,7 @@ func handlerSegmentMP4(w http.ResponseWriter, r *http.Request) {
data := session.Segment()
if data == nil {
log.Warn().Msgf("[hls] can't get segment %s", shell.Redact(r.URL.RawQuery))
log.Warn().Msgf("[hls] can't get segment %s", r.URL.RawQuery)
http.NotFound(w, r)
return
}
internal/onvif/onvif.go
+2 -3
View File
@@ -15,7 +15,6 @@ import (
"github.com/AlexxIT/go2rtc/internal/streams"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/onvif"
"github.com/AlexxIT/go2rtc/pkg/shell"
"github.com/rs/zerolog"
)
@@ -166,12 +165,12 @@ func apiOnvif(w http.ResponseWriter, r *http.Request) {
for _, rawURL := range urls {
u, err := url.Parse(rawURL)
if err != nil {
log.Warn().Str("url", shell.Redact(rawURL)).Msg("[onvif] broken")
log.Warn().Str("url", rawURL).Msg("[onvif] broken")
continue
}
if u.Scheme != "http" {
log.Warn().Str("url", shell.Redact(rawURL)).Msg("[onvif] unsupported")
log.Warn().Str("url", rawURL).Msg("[onvif] unsupported")
continue
}
internal/streams/api.go
+5
View File
@@ -5,10 +5,13 @@ import (
"github.com/AlexxIT/go2rtc/internal/api"
"github.com/AlexxIT/go2rtc/internal/app"
"github.com/AlexxIT/go2rtc/pkg/creds"
"github.com/AlexxIT/go2rtc/pkg/probe"
)
func apiStreams(w http.ResponseWriter, r *http.Request) {
w = creds.SecretResponse(w)
query := r.URL.Query()
src := query.Get("src")
@@ -120,5 +123,7 @@ func apiStreamsDOT(w http.ResponseWriter, r *http.Request) {
}
dot = append(dot, '}')
dot = []byte(creds.SecretString(string(dot)))
api.Response(w, dot, "text/vnd.graphviz")
}
internal/streams/dot.go
+1 -3
View File
@@ -4,8 +4,6 @@ import (
"encoding/json"
"fmt"
"strings"
"github.com/AlexxIT/go2rtc/pkg/shell"
)
func AppendDOT(dot []byte, stream *Stream) []byte {
@@ -168,7 +166,7 @@ func (c *conn) label() string {
sb.WriteString("\nsource=" + c.Source)
}
if c.URL != "" {
sb.WriteString("\nurl=" + shell.Redact(c.URL))
sb.WriteString("\nurl=" + c.URL)
}
if c.UserAgent != "" {
sb.WriteString("\nuser_agent=" + c.UserAgent)
internal/streams/producer.go
+5 -6
View File
@@ -8,7 +8,6 @@ import (
"time"
"github.com/AlexxIT/go2rtc/pkg/core"
"github.com/AlexxIT/go2rtc/pkg/shell"
)
type state byte
@@ -150,7 +149,7 @@ func (p *Producer) start() {
return
}
log.Debug().Msgf("[streams] start producer url=%s", shell.Redact(p.url))
log.Debug().Msgf("[streams] start producer url=%s", p.url)
p.state = stateStart
p.workerID++
@@ -168,7 +167,7 @@ func (p *Producer) worker(conn core.Producer, workerID int) {
return
}
log.Warn().Err(err).Str("url", shell.Redact(p.url)).Caller().Send()
log.Warn().Err(err).Str("url", p.url).Caller().Send()
}
p.reconnect(workerID, 0)
@@ -179,11 +178,11 @@ func (p *Producer) reconnect(workerID, retry int) {
defer p.mu.Unlock()
if p.workerID != workerID {
log.Trace().Msgf("[streams] stop reconnect url=%s", shell.Redact(p.url))
log.Trace().Msgf("[streams] stop reconnect url=%s", p.url)
return
}
log.Debug().Msgf("[streams] retry=%d to url=%s", retry, shell.Redact(p.url))
log.Debug().Msgf("[streams] retry=%d to url=%s", retry, p.url)
conn, err := GetProducer(p.url)
if err != nil {
@@ -258,7 +257,7 @@ func (p *Producer) stop() {
p.workerID++
}
log.Debug().Msgf("[streams] stop producer url=%s", shell.Redact(p.url))
log.Debug().Msgf("[streams] stop producer url=%s", p.url)
if p.conn != nil {
_ = p.conn.Stop()
internal/streams/streams.go
+1 -2
View File
@@ -9,7 +9,6 @@ import (
"github.com/AlexxIT/go2rtc/internal/api"
"github.com/AlexxIT/go2rtc/internal/app"
"github.com/AlexxIT/go2rtc/pkg/shell"
"github.com/rs/zerolog"
)
@@ -128,7 +127,7 @@ func GetOrPatch(query url.Values) *Stream {
// check if name param provided
if name := query.Get("name"); name != "" {
log.Info().Msgf("[streams] create new stream url=%s", shell.Redact(source))
log.Info().Msgf("[streams] create new stream url=%s", source)
return Patch(name, source)
}
pkg/creds/README.md
+7
View File
@@ -0,0 +1,7 @@
# Credentials
This module allows you to get variables:
- from custom storage (ex. config file)
- from [credential files](https://systemd.io/CREDENTIALS/)
- from environment variables
pkg/creds/creds.go
+79
View File
@@ -0,0 +1,79 @@
package creds
import (
"errors"
"os"
"path/filepath"
"regexp"
"strings"
)
type Storage interface {
SetValue(name, value string) error
GetValue(name string) (string, bool)
}
var storage Storage
func SetStorage(s Storage) {
storage = s
}
func SetValue(name, value string) error {
if storage == nil {
return errors.New("credentials: storage not initialized")
}
if err := storage.SetValue(name, value); err != nil {
return err
}
AddSecret(value)
return nil
}
func GetValue(name string) (value string, ok bool) {
value, ok = getValue(name)
AddSecret(value)
return
}
func getValue(name string) (string, bool) {
if storage != nil {
if value, ok := storage.GetValue(name); ok {
return value, true
}
}
if dir, ok := os.LookupEnv("CREDENTIALS_DIRECTORY"); ok {
if value, _ := os.ReadFile(filepath.Join(dir, name)); value != nil {
return strings.TrimSpace(string(value)), true
}
}
return os.LookupEnv(name)
}
// ReplaceVars - support format ${CAMERA_PASSWORD} and ${RTSP_USER:admin}
func ReplaceVars(data []byte) []byte {
re := regexp.MustCompile(`\${([^}{]+)}`)
return re.ReplaceAllFunc(data, func(match []byte) []byte {
key := string(match[2 : len(match)-1])
var def string
var defok bool
if i := strings.IndexByte(key, ':'); i > 0 {
key, def = key[:i], key[i+1:]
defok = true
}
if value, ok := GetValue(key); ok {
return []byte(value)
}
if defok {
return []byte(def)
}
return match
})
}
pkg/creds/secrets.go
+83
View File
@@ -0,0 +1,83 @@
package creds
import (
"io"
"net/http"
"slices"
"strings"
"sync"
)
func AddSecret(value string) {
if value == "" {
return
}
secretsMu.Lock()
defer secretsMu.Unlock()
if slices.Contains(secrets, value) {
return
}
secrets = append(secrets, value)
secretsReplacer = nil
}
var secrets []string
var secretsMu sync.Mutex
var secretsReplacer *strings.Replacer
func getReplacer() *strings.Replacer {
secretsMu.Lock()
defer secretsMu.Unlock()
if secretsReplacer == nil {
oldnew := make([]string, 0, 2*len(secrets))
for _, s := range secrets {
oldnew = append(oldnew, s, "***")
}
secretsReplacer = strings.NewReplacer(oldnew...)
}
return secretsReplacer
}
func SecretString(s string) string {
re := getReplacer()
return re.Replace(s)
}
func SecretWriter(w io.Writer) io.Writer {
return &secretWriter{w}
}
type secretWriter struct {
w io.Writer
}
func (s *secretWriter) Write(b []byte) (int, error) {
re := getReplacer()
return re.WriteString(s.w, string(b))
}
type secretResponse struct {
w http.ResponseWriter
}
func (s *secretResponse) Header() http.Header {
return s.w.Header()
}
func (s *secretResponse) Write(b []byte) (int, error) {
re := getReplacer()
return re.WriteString(s.w, string(b))
}
func (s *secretResponse) WriteHeader(statusCode int) {
s.w.WriteHeader(statusCode)
}
func SecretResponse(w http.ResponseWriter) http.ResponseWriter {
return &secretResponse{w}
}
pkg/creds/secrets_test.go
+15
View File
@@ -0,0 +1,15 @@
package creds
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestString(t *testing.T) {
AddSecret("admin")
AddSecret("pa$$word")
s := SecretString("rtsp://admin:pa$$word@192.168.1.123/stream1")
require.Equal(t, "rtsp://***:***@192.168.1.123/stream1", s)
}
pkg/secrets/secrets.go
-44
View File
@@ -1,44 +0,0 @@
package secrets
import (
"errors"
"sync"
)
type SecretsManager interface {
NewSecret(name string, defaultValues interface{}) (Secret, error)
GetSecret(name string) Secret
}
type Secret interface {
Get(key string) any
Set(key string, value string)
Marshal() (interface{}, error)
Unmarshal(value any) error
Save() error
}
var manager SecretsManager
var once sync.Once
func SetManager(m SecretsManager) {
once.Do(func() {
manager = m
})
}
// NewSecret creates or retrieves a secret
func NewSecret(name string, defaultValues interface{}) (Secret, error) {
if manager == nil {
return nil, errors.New("secrets manager not initialized")
}
return manager.NewSecret(name, defaultValues)
}
// GetSecret retrieves an existing secret
func GetSecret(name string) Secret {
if manager == nil {
return nil
}
return manager.GetSecret(name)
}
pkg/shell/shell.go
-143
View File
@@ -1,22 +1,10 @@
package shell
import (
"fmt"
"os"
"os/signal"
"path/filepath"
"regexp"
"strings"
"sync"
"syscall"
"github.com/AlexxIT/go2rtc/pkg/yaml"
)
var (
secretReplacer *strings.Replacer
secretValues map[string]bool
secretMutex sync.RWMutex
)
func QuoteSplit(s string) []string {
@@ -48,139 +36,8 @@ func QuoteSplit(s string) []string {
return a
}
// ReplaceEnvVars - support format ${CAMERA_PASSWORD} and ${RTSP_USER:admin}
func ReplaceEnvVars(text string) string {
var cfg struct {
Env map[string]string `yaml:"env"`
Secrets map[string]map[string]string `yaml:"secrets"`
}
yaml.Unmarshal([]byte(text), &cfg)
buildSecretReplacer(cfg)
re := regexp.MustCompile(`\${([^}{]+)}`)
return re.ReplaceAllStringFunc(text, func(match string) string {
key := match[2 : len(match)-1]
var def string
var dok bool
if i := strings.IndexByte(key, ':'); i > 0 {
key, def = key[:i], key[i+1:]
dok = true
}
if dir, vok := os.LookupEnv("CREDENTIALS_DIRECTORY"); vok {
value, err := os.ReadFile(filepath.Join(dir, key))
if err == nil {
return strings.TrimSpace(string(value))
}
}
if value, vok := os.LookupEnv(key); vok {
return value
}
if cfg.Env != nil {
if value, ok := cfg.Env[key]; ok {
return value
}
}
if cfg.Secrets != nil {
for secretName, secretValues := range cfg.Secrets {
for k, v := range secretValues {
name := fmt.Sprintf("%s_%s", secretName, k)
if key == name {
return v
}
}
}
}
if dok {
return def
}
return match
})
}
func RunUntilSignal() {
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
println("exit with signal:", (<-sigs).String())
}
func Redact(text string) string {
secretMutex.RLock()
defer secretMutex.RUnlock()
if secretReplacer == nil {
return text
}
return secretReplacer.Replace(text)
}
func buildSecretReplacer(cfg struct {
Env map[string]string `yaml:"env"`
Secrets map[string]map[string]string `yaml:"secrets"`
}) {
secretMutex.Lock()
defer secretMutex.Unlock()
if secretValues == nil {
secretValues = make(map[string]bool)
}
var newSecrets []string
if dir, ok := os.LookupEnv("CREDENTIALS_DIRECTORY"); ok {
entries, err := os.ReadDir(dir)
if err == nil {
for _, entry := range entries {
if !entry.IsDir() {
value, err := os.ReadFile(filepath.Join(dir, entry.Name()))
if err == nil {
cleanValue := strings.TrimSpace(string(value))
if len(cleanValue) > 0 && !secretValues[cleanValue] {
secretValues[cleanValue] = true
newSecrets = append(newSecrets, cleanValue)
}
}
}
}
}
}
if cfg.Secrets != nil {
for _, secretMap := range cfg.Secrets {
for _, value := range secretMap {
if len(value) > 0 && !secretValues[value] {
secretValues[value] = true
newSecrets = append(newSecrets, value)
}
}
}
}
if len(newSecrets) > 0 {
rebuildReplacer()
}
}
func rebuildReplacer() {
var replacements []string
for secret := range secretValues {
replacements = append(replacements, secret, "*****")
}
if len(replacements) > 0 {
secretReplacer = strings.NewReplacer(replacements...)
} else {
secretReplacer = nil
}
}