From 2c588a4c72c2b2c3e50bf4346ecce6f1b8258bc4 Mon Sep 17 00:00:00 2001 From: Piotr Komborski <163394+combor@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:11:16 +0100 Subject: [PATCH 1/6] Reach each site through its own proxy Sites stream to their own country only, and one process-wide HTTPS_PROXY can exit in a single country. -tvp-proxy and -bbc-proxy (MAGNETOWID_TVP_PROXY, MAGNETOWID_BBC_PROXY) take an HTTP proxy's URL or "direct"; unset, a site follows the environment as before. Each site gets its own client, and its streams carry that client's transport, so playlists, segments, subtitles and probes go the same way as the site's API. ffmpeg, for the streams it fetches itself, is given the proxy that transport would use instead of inheriting http_proxy. --- cmd/magnetowid/main.go | 66 +++++++++++++++-- cmd/magnetowid/main_test.go | 67 ++++++++++++++++- internal/downloader/ffmpeg.go | 49 ++++++++++-- internal/downloader/ffmpeg_unix_test.go | 99 ++++++++++++++++++++++++- internal/downloader/queue.go | 2 +- internal/downloader/queue_test.go | 20 ++++- internal/downloader/segments.go | 3 +- internal/probe/probe.go | 2 +- internal/probe/probe_test.go | 23 +++++- internal/provider/bbc/bbc_test.go | 4 + internal/provider/bbc/stream.go | 3 +- internal/provider/provider.go | 13 ++++ internal/provider/tvp/tvp.go | 2 +- internal/provider/tvp/tvp_test.go | 4 + 14 files changed, 336 insertions(+), 21 deletions(-) diff --git a/cmd/magnetowid/main.go b/cmd/magnetowid/main.go index 6d80baa..f063fcb 100644 --- a/cmd/magnetowid/main.go +++ b/cmd/magnetowid/main.go @@ -50,6 +50,9 @@ func run(log *slog.Logger, level *slog.LevelVar) error { categories := flag.String("categories", envOr("MAGNETOWID_CATEGORIES", "tv,movies"), "comma-separated download categories") ffmpeg := flag.String("ffmpeg", envOr("MAGNETOWID_FFMPEG", "ffmpeg"), "ffmpeg binary") logLevel := flag.String("log-level", envOr("MAGNETOWID_LOG_LEVEL", "info"), "log level: debug, info, warn or error") + const proxyUsage = `'s HTTP proxy, e.g. http://127.0.0.1:8888, or "direct" for none (default: the environment's)` + tvpProxy := flag.String("tvp-proxy", os.Getenv("MAGNETOWID_TVP_PROXY"), "TVP VOD"+proxyUsage) + bbcProxy := flag.String("bbc-proxy", os.Getenv("MAGNETOWID_BBC_PROXY"), "BBC iPlayer"+proxyUsage) healthcheck := flag.Bool("healthcheck", false, "ask the magnetowid at the listen address whether it is healthy, and exit") flag.Parse() @@ -88,12 +91,20 @@ func run(log *slog.Logger, level *slog.LevelVar) error { } defer db.Close() - httpClient := &http.Client{Timeout: 30 * time.Second} - tvpProvider, err := tvp.New(httpClient, log, db) + // Each site streams to its own country, so each has its own connection. + tvpClient, err := siteClient(*tvpProxy) + if err != nil { + return fmt.Errorf("-tvp-proxy: %w", err) + } + bbcClient, err := siteClient(*bbcProxy) + if err != nil { + return fmt.Errorf("-bbc-proxy: %w", err) + } + tvpProvider, err := tvp.New(tvpClient, log, db) if err != nil { return err } - bbcProvider, err := bbc.New(httpClient, log, db) + bbcProvider, err := bbc.New(bbcClient, log, db) if err != nil { return err } @@ -111,7 +122,8 @@ func run(log *slog.Logger, level *slog.LevelVar) error { return err } mux := http.NewServeMux() - prober := &probe.Prober{Client: httpClient} + // Streams bring their site's connection. + prober := &probe.Prober{Client: &http.Client{Timeout: siteTimeout}} mux.Handle("/{provider}/api", &newznab.Handler{Providers: providers, APIKey: *apiKey, Probe: prober, Log: log}) mux.Handle("/api", &sabnzbd.Handler{Queue: queue, APIKey: *apiKey, Categories: cats, Log: log}) (&overrides.Handler{Store: overrideStore, APIKey: *apiKey, Log: log}).Register(mux) @@ -130,7 +142,8 @@ func run(log *slog.Logger, level *slog.LevelVar) error { srv := &http.Server{Addr: *listen, Handler: logRequests(log, mux), ReadHeaderTimeout: 10 * time.Second} errc := make(chan error, 1) go func() { errc <- srv.ListenAndServe() }() - log.Info("magnetowid listening", "version", version, "addr", *listen, "download_dir", dir, "providers", providers.Names()) + log.Info("magnetowid listening", "version", version, "addr", *listen, "download_dir", dir, "providers", providers.Names(), + "tvp_proxy", proxyName(*tvpProxy), "bbc_proxy", proxyName(*bbcProxy)) var serveErr error select { @@ -168,6 +181,49 @@ func logRequests(log *slog.Logger, next http.Handler) http.Handler { }) } +const ( + siteTimeout = 30 * time.Second + // Skips the environment's proxy. + directProxy = "direct" +) + +// siteClient returns the client for a site's API and streams. proxy is an HTTP +// proxy's URL, "direct", or empty for the environment's HTTPS_PROXY and NO_PROXY. +func siteClient(proxy string) (*http.Client, error) { + t := http.DefaultTransport.(*http.Transport).Clone() + // Keep a connection for each of a download's segment workers. + t.MaxIdleConnsPerHost = 4 + switch proxy { + case "": + case directProxy: + t.Proxy = nil + default: + // ffmpeg, which fetches some streams itself, tunnels through nothing else. + u, err := url.Parse(proxy) + if err != nil || u.Scheme != "http" || u.Host == "" { + // The URL may hold a password. + return nil, fmt.Errorf("want an http:// proxy URL or %q", directProxy) + } + t.Proxy = http.ProxyURL(u) + } + return &http.Client{Timeout: siteTimeout, Transport: t}, nil +} + +// proxyName describes a valid proxy setting for the log, without its password. +func proxyName(proxy string) string { + switch proxy { + case "": + return "environment" + case directProxy: + return proxy + } + u, err := url.Parse(proxy) + if err != nil { + return "invalid" + } + return u.Redacted() +} + func checkHealth(listen string) error { u, err := healthURL(listen) if err != nil { diff --git a/cmd/magnetowid/main_test.go b/cmd/magnetowid/main_test.go index 0874408..e9e74a4 100644 --- a/cmd/magnetowid/main_test.go +++ b/cmd/magnetowid/main_test.go @@ -1,6 +1,11 @@ package main -import "testing" +import ( + "net/http" + "reflect" + "strings" + "testing" +) func TestHealthURL(t *testing.T) { for listen, want := range map[string]string{ @@ -20,3 +25,63 @@ func TestHealthURL(t *testing.T) { t.Error("healthURL accepted a listen address without a port") } } + +func TestSiteClient(t *testing.T) { + req, err := http.NewRequest(http.MethodGet, "https://vod.example/playlist.m3u8", nil) + if err != nil { + t.Fatal(err) + } + transport := func(proxy string) *http.Transport { + t.Helper() + c, err := siteClient(proxy) + if err != nil { + t.Fatalf("siteClient(%q): %v", proxy, err) + } + return c.Transport.(*http.Transport) + } + + // The environment's proxy is read once per process, so compare functions. + if got := transport("").Proxy; reflect.ValueOf(got).Pointer() != reflect.ValueOf(http.ProxyFromEnvironment).Pointer() { + t.Error("an unset proxy doesn't follow the environment") + } + if transport("direct").Proxy != nil { + t.Error(`"direct" still uses a proxy`) + } + for proxy, name := range map[string]string{ + "http://proxy.example:3128": "http://proxy.example:3128", + "http://user:secret@proxy.example:3128": "http://user:xxxxx@proxy.example:3128", + } { + u, err := transport(proxy).Proxy(req) + if err != nil || u == nil || u.String() != proxy { + t.Errorf("siteClient(%q) proxies through %v, %v", proxy, u, err) + } + if got := proxyName(proxy); got != name { + t.Errorf("proxyName(%q) = %q, want %q", proxy, got, name) + } + } + if a, b := transport(""), transport(""); a == b { + t.Error("sites share a transport") + } + + // ffmpeg tunnels through HTTP proxies only. + for _, proxy := range []string{ + "socks5://user:secret@proxy.example:1080", + "https://proxy.example:3128", + "proxy.example:3128", + "http://", + "http://[proxy", + "DIRECT", + } { + _, err := siteClient(proxy) + if err == nil { + t.Errorf("siteClient(%q) accepted the proxy", proxy) + } else if strings.Contains(err.Error(), "secret") { + t.Errorf("siteClient(%q) reports the password: %v", proxy, err) + } + } + for proxy, want := range map[string]string{"": "environment", "direct": "direct"} { + if got := proxyName(proxy); got != want { + t.Errorf("proxyName(%q) = %q, want %q", proxy, got, want) + } + } +} diff --git a/internal/downloader/ffmpeg.go b/internal/downloader/ffmpeg.go index d375409..1eefc57 100644 --- a/internal/downloader/ffmpeg.go +++ b/internal/downloader/ffmpeg.go @@ -13,6 +13,7 @@ import ( "os" "os/exec" "path/filepath" + "slices" "strconv" "strings" "sync" @@ -33,8 +34,9 @@ const playlistTimeout = 30 * time.Second // Neither transcodes. type FFmpeg struct { Path string // binary; "ffmpeg" if empty - // Fetches playlists and segments; nil uses a default. Each download gets - // its own copy, with a cookie jar if the client has none. + // Fetches playlists and segments, on the stream's Transport if it has one; + // nil uses a default. Each download gets its own copy, with a cookie jar + // if the client has none. Client *http.Client Log *slog.Logger // nil discards // Zero defaults to 5 minutes, allowing for +faststart's silent final pass. @@ -81,6 +83,7 @@ func (f *FFmpeg) Download(ctx context.Context, s provider.Stream, out string, pr c := *f.Client client = &c } + client = s.Client(client) if client.Jar == nil { // ffmpeg sends a playlist's cookies with its segments too. client.Jar, _ = cookiejar.New(nil) @@ -108,13 +111,47 @@ func (f *FFmpeg) Download(ctx context.Context, s provider.Stream, out string, pr if err := os.RemoveAll(dir); err != nil { return err } + env, err := proxyEnv(s) + if err != nil { + return err + } inputs := directInputs(s, m, master) var args []string for _, in := range inputs { args = append(args, headerArgs(s.Header)...) args = append(args, "-i", in) } - return f.run(ctx, "", append(args, outputArgs(len(inputs), m.AudioLanguage, out)...), progress) + return f.run(ctx, "", env, append(args, outputArgs(len(inputs), m.AudioLanguage, out)...), progress) +} + +// proxyEnv returns the environment in which ffmpeg reaches the stream as its +// Transport does, or nil, magnetowid's own, for a stream without one. ffmpeg +// reads http_proxy for HTTP and HTTPS alike. +func proxyEnv(s provider.Stream) ([]string, error) { + t, ok := s.Transport.(*http.Transport) + if !ok { + return nil, nil + } + env := slices.DeleteFunc(os.Environ(), func(kv string) bool { + k, _, _ := strings.Cut(kv, "=") + return strings.EqualFold(k, "http_proxy") || strings.EqualFold(k, "no_proxy") + }) + if t.Proxy == nil { + return env, nil + } + req, err := http.NewRequest(http.MethodGet, s.URL, nil) + if err != nil { + return nil, err + } + // A proxy that can't be determined must not become a direct connection. + proxy, err := t.Proxy(req) + if err != nil { + return nil, fmt.Errorf("finding the stream's proxy: %w", err) + } + if proxy != nil { + env = append(env, "http_proxy="+proxy.String()) + } + return env, nil } func (f *FFmpeg) log() *slog.Logger { @@ -175,8 +212,9 @@ func (f *FFmpeg) binary() (string, error) { return bin, nil } -// run runs ffmpeg in dir, or the current folder if dir is empty. -func (f *FFmpeg) run(parent context.Context, dir string, args []string, progress func(time.Duration, int64)) error { +// run runs ffmpeg in dir, or the current folder if dir is empty, with env, or +// magnetowid's environment if env is nil. +func (f *FFmpeg) run(parent context.Context, dir string, env, args []string, progress func(time.Duration, int64)) error { bin, err := f.binary() if err != nil { return err @@ -191,6 +229,7 @@ func (f *FFmpeg) run(parent context.Context, dir string, args []string, progress args = append([]string{"-nostdin", "-hide_banner", "-loglevel", "warning", "-y"}, args...) cmd := exec.CommandContext(ctx, bin, args...) cmd.Dir = dir + cmd.Env = env cmd.WaitDelay = 5 * time.Second // Isolate ffmpeg from terminal signals so magnetowid can requeue interrupted jobs. cmd.SysProcAttr = ownProcessGroup() diff --git a/internal/downloader/ffmpeg_unix_test.go b/internal/downloader/ffmpeg_unix_test.go index 3ba6bf8..73e373b 100644 --- a/internal/downloader/ffmpeg_unix_test.go +++ b/internal/downloader/ffmpeg_unix_test.go @@ -8,11 +8,13 @@ import ( "io" "net/http" "net/http/httptest" + "net/url" "os" "path/filepath" "slices" "strconv" "strings" + "sync" "syscall" "testing" "time" @@ -58,11 +60,13 @@ func TestFFmpegRunsInOwnProcessGroup(t *testing.T) { } } -// fakeFFmpeg records its arguments and folder in log and creates its output. +// fakeFFmpeg records its folder, arguments and proxy settings in log and +// creates its output. func fakeFFmpeg(t *testing.T, dir string) (bin, log string) { t.Helper() bin, log = filepath.Join(dir, "ffmpeg"), filepath.Join(dir, "args") - script := "#!/bin/sh\npwd > '" + log + "'\nfor a; do echo \"$a\"; out=$a; done >> '" + log + "'\n: > \"$out\"\n" + script := "#!/bin/sh\npwd > '" + log + "'\nfor a; do echo \"$a\"; out=$a; done >> '" + log + "'\n" + + "echo \"http_proxy=$http_proxy no_proxy=$no_proxy\" >> '" + log + "'\n: > \"$out\"\n" if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { t.Fatal(err) } @@ -110,6 +114,97 @@ func TestFFmpegInputs(t *testing.T) { } } +// ffmpeg reads its proxy from the environment, which must match the stream's +// transport and not magnetowid's own settings. +func TestFFmpegFollowsStreamProxy(t *testing.T) { + dir := t.TempDir() + bin, log := fakeFFmpeg(t, dir) + t.Setenv("http_proxy", "http://environment.example:3128") + t.Setenv("no_proxy", "elsewhere.example") + proxy, err := url.Parse("http://proxy.example:3128") + if err != nil { + t.Fatal(err) + } + for _, tt := range []struct { + name string + transport http.RoundTripper + want string + }{ + {"no transport", nil, "http_proxy=http://environment.example:3128 no_proxy=elsewhere.example"}, + {"proxy", &http.Transport{Proxy: http.ProxyURL(proxy)}, "http_proxy=http://proxy.example:3128 no_proxy="}, + {"direct", &http.Transport{}, "http_proxy= no_proxy="}, + {"no proxy for the stream", &http.Transport{Proxy: func(*http.Request) (*url.URL, error) { return nil, nil }}, + "http_proxy= no_proxy="}, + } { + // Not a playlist, so ffmpeg fetches it. + s := provider.Stream{URL: "http://example.invalid/video.mp4", Transport: tt.transport} + if err := (&FFmpeg{Path: bin}).Download(context.Background(), s, filepath.Join(dir, "out.mp4"), + func(time.Duration, int64) {}); err != nil { + t.Fatalf("%s: %v", tt.name, err) + } + b, err := os.ReadFile(log) + if err != nil { + t.Fatal(err) + } + if lines := strings.Split(strings.TrimSpace(string(b)), "\n"); lines[len(lines)-1] != tt.want { + t.Errorf("%s: ffmpeg ran with %q, want %q", tt.name, lines[len(lines)-1], tt.want) + } + } + + // An unknown proxy must not become a direct connection. + broken := &http.Transport{Proxy: func(*http.Request) (*url.URL, error) { return nil, errors.New("bad proxy") }} + os.Remove(log) + err = (&FFmpeg{Path: bin}).Download(context.Background(), + provider.Stream{URL: "http://example.invalid/video.mp4", Transport: broken}, filepath.Join(dir, "out.mp4"), + func(time.Duration, int64) {}) + if err == nil || !strings.Contains(err.Error(), "bad proxy") { + t.Errorf("err = %v, want the proxy's", err) + } + if _, err := os.Stat(log); !os.IsNotExist(err) { + t.Errorf("ffmpeg ran without knowing its proxy: %v", err) + } +} + +// The stream's transport, e.g. its site's proxy, fetches playlists and segments. +func TestDownloadUsesStreamTransport(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/master.m3u8": + io.WriteString(w, "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\nvod.m3u8\n") + case "/vod.m3u8": + io.WriteString(w, "#EXTM3U\n#EXTINF:4,\n1.ts\n#EXT-X-ENDLIST\n") + default: + io.WriteString(w, "segment data") + } + })) + defer srv.Close() + dir := t.TempDir() + bin, _ := fakeFFmpeg(t, dir) + var mu sync.Mutex + var paths []string + site := roundTripFunc(func(req *http.Request) (*http.Response, error) { + mu.Lock() + paths = append(paths, req.URL.Path) + mu.Unlock() + return http.DefaultTransport.RoundTrip(req) + }) + // The engine's own client must stay unused. + other := &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + t.Errorf("%s fetched without the stream's transport", req.URL.Path) + return nil, errors.New("wrong transport") + })} + err := (&FFmpeg{Path: bin, Client: other}).Download(context.Background(), + provider.Stream{URL: srv.URL + "/master.m3u8", Transport: site}, filepath.Join(dir, "out.mp4"), + func(time.Duration, int64) {}) + if err != nil { + t.Fatal(err) + } + slices.Sort(paths) + if want := []string{"/1.ts", "/master.m3u8", "/vod.m3u8"}; !slices.Equal(paths, want) { + t.Errorf("the stream's transport fetched %q, want %q", paths, want) + } +} + // ffmpeg 5.x reports unreadable inputs without "Error opening input". func TestUnreadableSegmentsRemoved(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/internal/downloader/queue.go b/internal/downloader/queue.go index ea53ed3..709a4ac 100644 --- a/internal/downloader/queue.go +++ b/internal/downloader/queue.go @@ -650,7 +650,7 @@ func (q *Queue) saveSubtitles(ctx context.Context, j Job, s provider.Stream, wor if slices.Contains(paths, path) { continue // more in the same language } - srt, err := subtitles.Fetch(ctx, defaultClient, sub, s.Header) + srt, err := subtitles.Fetch(ctx, s.Client(defaultClient), sub, s.Header) if err == nil { err = os.WriteFile(path, srt, 0o666) } diff --git a/internal/downloader/queue_test.go b/internal/downloader/queue_test.go index ec27732..e9dc1af 100644 --- a/internal/downloader/queue_test.go +++ b/internal/downloader/queue_test.go @@ -15,6 +15,7 @@ import ( "slices" "strings" "sync" + "sync/atomic" "testing" "time" @@ -29,8 +30,13 @@ type fakeProvider struct { offline int err error subtitles []provider.Subtitle + transport http.RoundTripper } +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) } + func (f *fakeProvider) Name() string { return "fake" } func (f *fakeProvider) Search(context.Context, provider.Query) ([]provider.Item, error) { @@ -48,7 +54,8 @@ func (f *fakeProvider) Resolve(_ context.Context, id string) (provider.Stream, e if f.err != nil { return provider.Stream{}, f.err } - return provider.Stream{URL: fmt.Sprintf("http://example.invalid/%s/%d.m3u8", id, f.resolves), Subtitles: f.subtitles}, nil + return provider.Stream{URL: fmt.Sprintf("http://example.invalid/%s/%d.m3u8", id, f.resolves), Subtitles: f.subtitles, + Transport: f.transport}, nil } type fakeEngine struct { @@ -365,7 +372,13 @@ func TestSubtitlesSaved(t *testing.T) {
Dzień dobry.