From dc1fe0f19c5b6ebfbfe3bcab97161119245abc67 Mon Sep 17 00:00:00 2001 From: zarazaex69 Date: Mon, 11 May 2026 13:31:07 +0300 Subject: [PATCH] refactor: replace -carrier with -auth/-engine/-url/-token (stage E) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Break CLI backwards compatibility as planned for refactor/universal-carrier: - Drop -carrier flag; add -auth (auth provider name), -engine (engine name for -auth none), -url and -token (SFU endpoint + access token for direct/none auth mode). - session.Config.Carrier → Auth + Engine + URL + Token. - session.Gen() is now generic: auth.Get(cfg.Auth).(auth.RoomCreator) replaces the hard-coded switch on carrier names. - Register a "none" carrier in builtin (registerDirect) that bypasses auth and connects directly to any engine with caller-supplied URL+Token. - auth/telemost.Provider.Issue now accepts a raw room-ID hash in addition to a full https://telemost.yandex.ru/j/ URL. - Plumb Engine/URL/Token from session.Config through server.Run, client.Run/RunWithReady, bringUpLink, link.Config, transport.Config, and carrier.Config so the "none" carrier has access to them end-to-end. - Update all tests and mobile.go call sites. Co-Authored-By: Claude Opus 4.7 --- cmd/olcrtc/main.go | 23 ++- cmd/olcrtc/main_test.go | 20 +-- internal/app/session/session.go | 147 +++++++------------ internal/app/session/session_test.go | 48 ++---- internal/auth/telemost/telemost.go | 20 ++- internal/carrier/builtin/engine_adapter.go | 25 ++++ internal/carrier/builtin/register.go | 1 + internal/carrier/carrier.go | 4 + internal/client/client.go | 8 + internal/e2e/tunnel_test.go | 8 +- internal/link/direct/direct.go | 3 + internal/link/link.go | 4 + internal/server/server.go | 6 + internal/transport/datachannel/transport.go | 3 + internal/transport/seichannel/transport.go | 3 + internal/transport/transport.go | 5 + internal/transport/videochannel/transport.go | 3 + internal/transport/vp8channel/transport.go | 3 + mobile/mobile.go | 3 + mobile/mobile_test.go | 5 + 20 files changed, 191 insertions(+), 151 deletions(-) diff --git a/cmd/olcrtc/main.go b/cmd/olcrtc/main.go index 1e14779..d643dec 100644 --- a/cmd/olcrtc/main.go +++ b/cmd/olcrtc/main.go @@ -36,7 +36,10 @@ type config struct { mode string link string transport string - carrier string + auth string + engine string + url string + token string roomID string clientID string socksPort int @@ -175,7 +178,10 @@ func parseFlagsFrom(args []string, errorHandling flag.ErrorHandling) (config, er fs.StringVar(&cfg.mode, "mode", "", "Mode: srv or cnc") fs.StringVar(&cfg.link, "link", "", "Link: direct (p2p connection type)") fs.StringVar(&cfg.transport, "transport", "", "Transport: datachannel, videochannel, seichannel") - fs.StringVar(&cfg.carrier, "carrier", "", "Carrier: telemost, jazz, wbstream") + fs.StringVar(&cfg.auth, "auth", "", "Auth provider: telemost, jazz, wbstream, none") + fs.StringVar(&cfg.engine, "engine", "", "Engine (required when -auth none): livekit, goolom, salutejazz") + fs.StringVar(&cfg.url, "url", "", "SFU WebSocket URL (required when -auth none)") + fs.StringVar(&cfg.token, "token", "", "Access token (required when -auth none)") fs.StringVar(&cfg.roomID, "id", "", "Room ID") fs.StringVar(&cfg.clientID, "client-id", "", "Client ID: binds one srv to one cnc (required)") fs.IntVar(&cfg.socksPort, "socks-port", 0, "SOCKS5 port (client only)") @@ -252,11 +258,14 @@ func loadNames(dataDir string) error { func toSessionConfig(cfg config) session.Config { return session.Config{ - Mode: cfg.mode, - Link: cfg.link, - Transport: cfg.transport, - Carrier: cfg.carrier, - RoomID: cfg.roomID, + Mode: cfg.mode, + Link: cfg.link, + Transport: cfg.transport, + Auth: cfg.auth, + Engine: cfg.engine, + URL: cfg.url, + Token: cfg.token, + RoomID: cfg.roomID, ClientID: cfg.clientID, KeyHex: cfg.keyHex, SOCKSHost: cfg.socksHost, diff --git a/cmd/olcrtc/main_test.go b/cmd/olcrtc/main_test.go index 2199eaa..26526c9 100644 --- a/cmd/olcrtc/main_test.go +++ b/cmd/olcrtc/main_test.go @@ -20,7 +20,7 @@ func TestToSessionConfig(t *testing.T) { mode: "cnc", link: "direct", //nolint:goconst // test literal, repetition is intentional transport: "vp8channel", - carrier: "jazz", //nolint:goconst // test literal, repetition is intentional + auth: "jazz", //nolint:goconst // test literal, repetition is intentional roomID: "room", //nolint:goconst // test literal, repetition is intentional clientID: "client", //nolint:goconst // test literal, repetition is intentional keyHex: "key", //nolint:goconst // test literal, repetition is intentional @@ -49,7 +49,7 @@ func TestToSessionConfig(t *testing.T) { } got := toSessionConfig(cfg) - if got.Mode != cfg.mode || got.Carrier != "jazz" || got.SOCKSPort != cfg.socksPort || + if got.Mode != cfg.mode || got.Auth != "jazz" || got.SOCKSPort != cfg.socksPort || got.VideoTileRS != cfg.videoTileRS || got.VP8BatchSize != cfg.vp8BatchSize || got.SEIFPS != cfg.seiFPS || got.SEIBatchSize != cfg.seiBatchSize || got.SEIFragmentSize != cfg.seiFragmentSize || got.SEIAckTimeoutMS != cfg.seiAckTimeoutMS || @@ -64,7 +64,7 @@ func TestParseFlagsFrom(t *testing.T) { "-mode", "srv", //nolint:goconst // test literal, repetition is intentional "-link", "direct", "-transport", "vp8channel", - "-carrier", "telemost", + "-auth", "telemost", "-id", "room", "-client-id", "client", "-socks-port", "1080", @@ -96,7 +96,7 @@ func TestParseFlagsFrom(t *testing.T) { if err != nil { t.Fatalf("parseFlagsFrom() error = %v", err) } - if cfg.mode != "srv" || cfg.carrier != "telemost" || cfg.roomID != "room" || + if cfg.mode != "srv" || cfg.auth != "telemost" || cfg.roomID != "room" || cfg.debug != true || cfg.videoCodec != "tile" || cfg.videoTileRS != 40 || cfg.vp8FPS != 24 || cfg.vp8BatchSize != 3 || cfg.seiFPS != 40 || cfg.seiBatchSize != 4 || cfg.seiFragmentSize != 512 || cfg.seiAckTimeoutMS != 1500 || @@ -117,7 +117,7 @@ func TestRunGenModeValidationErrors(t *testing.T) { t.Fatal("runWithConfig(gen, no carrier) error = nil") } - if err := runWithConfig(config{mode: "gen", carrier: "wbstream", dnsServer: "1.1.1.1:53"}); err == nil { //nolint:goconst,lll // test literal, repetition is intentional + if err := runWithConfig(config{mode: "gen", auth: "wbstream", dnsServer: "1.1.1.1:53"}); err == nil { //nolint:goconst,lll // test literal, repetition is intentional t.Fatal("runWithConfig(gen, amount=0) error = nil") } } @@ -129,14 +129,14 @@ func TestRunGenModeCallsGen(t *testing.T) { oldRunGen := runGen t.Cleanup(func() { runGen = oldRunGen }) runGen = func(cfg config) error { - if cfg.carrier != "wbstream" || cfg.dnsServer != "1.1.1.1:53" || cfg.amount != 3 { + if cfg.auth != "wbstream" || cfg.dnsServer != "1.1.1.1:53" || cfg.amount != 3 { t.Fatalf("runGen cfg = %+v", cfg) } collected = append(collected, "ok") return nil } - err := runWithConfig(config{mode: "gen", carrier: "wbstream", dnsServer: "1.1.1.1:53", amount: 3}) + err := runWithConfig(config{mode: "gen", auth: "wbstream", dnsServer: "1.1.1.1:53", amount: 3}) if err != nil { t.Fatalf("runWithConfig(gen) error = %v", err) } @@ -151,7 +151,7 @@ func TestRunWithConfigValidationAndDataDirErrors(t *testing.T) { mode: "srv", link: "direct", transport: "datachannel", - carrier: "jazz", + auth: "jazz", clientID: "client", keyHex: "key", dnsServer: "1.1.1.1:53", @@ -183,7 +183,7 @@ func TestRunWithArgsSuccessfulSessionReturn(t *testing.T) { called := false runSession = func(ctx context.Context, cfg session.Config) error { called = true - if cfg.Mode != "srv" || cfg.Carrier != "jazz" || cfg.ClientID != "client" { + if cfg.Mode != "srv" || cfg.Auth != "jazz" || cfg.ClientID != "client" { t.Fatalf("session config = %+v", cfg) } select { @@ -198,7 +198,7 @@ func TestRunWithArgsSuccessfulSessionReturn(t *testing.T) { "-mode", "srv", "-link", "direct", "-transport", "datachannel", - "-carrier", "jazz", + "-auth", "jazz", "-client-id", "client", "-key", "key", "-dns", "1.1.1.1:53", diff --git a/internal/app/session/session.go b/internal/app/session/session.go index a72ae54..804e34b 100644 --- a/internal/app/session/session.go +++ b/internal/app/session/session.go @@ -9,8 +9,6 @@ import ( "time" "github.com/openlibrecommunity/olcrtc/internal/auth" - authSaluteJazz "github.com/openlibrecommunity/olcrtc/internal/auth/salutejazz" - authWBStream "github.com/openlibrecommunity/olcrtc/internal/auth/wbstream" "github.com/openlibrecommunity/olcrtc/internal/carrier" "github.com/openlibrecommunity/olcrtc/internal/carrier/builtin" "github.com/openlibrecommunity/olcrtc/internal/client" @@ -26,19 +24,16 @@ import ( ) const ( - modeSRV = "srv" - modeCNC = "cnc" - modeGen = "gen" - carrierJazz = "jazz" - carrierTelemost = "telemost" - carrierWBStream = "wbstream" - transportVideo = "videochannel" - transportVP8 = "vp8channel" - transportSEI = "seichannel" - videoCodecQRCode = "qrcode" - videoCodecTile = "tile" - roomURLAny = "any" - telemostRoomURLPrefix = "https://telemost.yandex.ru/j/" + modeSRV = "srv" + modeCNC = "cnc" + modeGen = "gen" + authJazz = "jazz" + authNone = "none" + transportVideo = "videochannel" + transportVP8 = "vp8channel" + transportSEI = "seichannel" + videoCodecQRCode = "qrcode" + videoCodecTile = "tile" ) var ( @@ -48,9 +43,9 @@ var ( ErrModeRequired = errors.New("mode required (use -mode srv, -mode cnc or -mode gen)") // ErrAmountRequired indicates that -amount is required for gen mode. ErrAmountRequired = errors.New("amount required for gen mode (use -amount )") - // ErrCarrierRequired indicates that no carrier was selected. - ErrCarrierRequired = errors.New( - "carrier required (use -carrier telemost, -carrier jazz or -carrier wbstream)") + // ErrAuthRequired indicates that no auth provider was selected. + ErrAuthRequired = errors.New( + "auth provider required (use -auth telemost, -auth jazz, -auth wbstream or -auth none)") // ErrUnsupportedCarrier indicates that carrier is not registered. ErrUnsupportedCarrier = errors.New("unsupported carrier") // ErrUnsupportedLink indicates that link is not registered. @@ -113,7 +108,10 @@ type Config struct { Mode string Link string Transport string - Carrier string + Auth string + Engine string + URL string + Token string RoomID string ClientID string KeyHex string @@ -158,7 +156,7 @@ func Validate(cfg Config) error { if err := validateMode(cfg); err != nil { return err } - if err := validateCarrier(cfg); err != nil { + if err := validateAuth(cfg); err != nil { return err } if err := validateLink(cfg); err != nil { @@ -185,12 +183,12 @@ func validateMode(cfg Config) error { } } -func validateCarrier(cfg Config) error { - if cfg.Carrier == "" { - return ErrCarrierRequired +func validateAuth(cfg Config) error { + if cfg.Auth == "" { + return ErrAuthRequired } - if !slices.Contains(carrier.Available(), cfg.Carrier) { - return fmt.Errorf("%w: %s (available: %v)", ErrUnsupportedCarrier, cfg.Carrier, carrier.Available()) + if !slices.Contains(carrier.Available(), cfg.Auth) { + return fmt.Errorf("%w: %s (available: %v)", ErrUnsupportedCarrier, cfg.Auth, carrier.Available()) } return nil } @@ -216,7 +214,7 @@ func validateTransportRegistration(cfg Config) error { } func validateCommon(cfg Config) error { - if cfg.RoomID == "" && cfg.Carrier != carrierJazz { + if cfg.RoomID == "" && cfg.Auth != authJazz && cfg.Auth != authNone { return ErrRoomIDRequired } if cfg.ClientID == "" { @@ -314,7 +312,7 @@ func validateModeConfig(cfg Config) error { // Run starts the configured mode. func Run(ctx context.Context, cfg Config) error { - roomURL := buildRoomURL(cfg.Carrier, cfg.RoomID) + roomURL := cfg.RoomID switch cfg.Mode { case modeSRV: @@ -322,7 +320,7 @@ func Run(ctx context.Context, cfg Config) error { ctx, cfg.Link, cfg.Transport, - cfg.Carrier, + cfg.Auth, roomURL, cfg.KeyHex, cfg.ClientID, @@ -345,6 +343,7 @@ func Run(ctx context.Context, cfg Config) error { cfg.SEIBatchSize, cfg.SEIFragmentSize, cfg.SEIAckTimeoutMS, + cfg.Engine, cfg.URL, cfg.Token, ); err != nil { return fmt.Errorf("server: %w", err) } @@ -354,7 +353,7 @@ func Run(ctx context.Context, cfg Config) error { ctx, cfg.Link, cfg.Transport, - cfg.Carrier, + cfg.Auth, roomURL, cfg.KeyHex, cfg.ClientID, @@ -378,6 +377,7 @@ func Run(ctx context.Context, cfg Config) error { cfg.SEIBatchSize, cfg.SEIFragmentSize, cfg.SEIAckTimeoutMS, + cfg.Engine, cfg.URL, cfg.Token, ); err != nil { return fmt.Errorf("client: %w", err) } @@ -387,29 +387,13 @@ func Run(ctx context.Context, cfg Config) error { } } -func buildRoomURL(carrierName, roomID string) string { - switch carrierName { - case carrierTelemost: - return telemostRoomURLPrefix + roomID - case carrierJazz: - if roomID == "" { - return roomURLAny - } - return roomID - case carrierWBStream: - return roomID - default: - return roomID - } -} - // ValidateGen validates that the config contains enough fields to run gen mode. func ValidateGen(cfg Config) error { - if cfg.Carrier == "" { - return ErrCarrierRequired + if cfg.Auth == "" { + return ErrAuthRequired } - if !slices.Contains(carrier.Available(), cfg.Carrier) { - return fmt.Errorf("%w: %s (available: %v)", ErrUnsupportedCarrier, cfg.Carrier, carrier.Available()) + if !slices.Contains(carrier.Available(), cfg.Auth) { + return fmt.Errorf("%w: %s (available: %v)", ErrUnsupportedCarrier, cfg.Auth, carrier.Available()) } if cfg.DNSServer == "" { return ErrDNSServerRequired @@ -443,53 +427,30 @@ func genRetry(ctx context.Context, fn func(context.Context) error) error { return lastErr } -// Gen creates cfg.Amount rooms for the configured carrier and writes each room ID to out. -// -//nolint:cyclop // transitional; refactor/universal-carrier replaces this with auth.RoomCreator dispatch +// Gen creates cfg.Amount rooms for the configured auth provider and writes each room ID to out. func Gen(ctx context.Context, cfg Config, out func(string)) error { - switch cfg.Carrier { - case carrierJazz: - creator, ok := any(authSaluteJazz.Provider{}).(auth.RoomCreator) - if !ok { - return fmt.Errorf("%w: jazz auth provider does not implement RoomCreator", ErrUnsupportedCarrier) - } - for i := range cfg.Amount { - var roomID string - err := genRetry(ctx, func(ctx context.Context) error { - var err error - roomID, err = creator.CreateRoom(ctx, auth.Config{Name: names.Generate()}) - if err != nil { - return fmt.Errorf("jazz CreateRoom: %w", err) - } - return nil - }) - if err != nil { - return fmt.Errorf("gen jazz room %d: %w", i+1, err) + p, err := auth.Get(cfg.Auth) + if err != nil { + return fmt.Errorf("%w: %s", ErrUnsupportedCarrier, cfg.Auth) + } + creator, ok := p.(auth.RoomCreator) + if !ok { + return fmt.Errorf("%w: %s does not support room generation", ErrUnsupportedCarrier, cfg.Auth) + } + for i := range cfg.Amount { + var roomID string + err := genRetry(ctx, func(ctx context.Context) error { + var genErr error + roomID, genErr = creator.CreateRoom(ctx, auth.Config{Name: names.Generate()}) + if genErr != nil { + return fmt.Errorf("CreateRoom: %w", genErr) } - out(roomID) + return nil + }) + if err != nil { + return fmt.Errorf("gen room %d: %w", i+1, err) } - case carrierWBStream: - creator, ok := any(authWBStream.Provider{}).(auth.RoomCreator) - if !ok { - return fmt.Errorf("%w: wbstream auth provider does not implement RoomCreator", ErrUnsupportedCarrier) - } - for i := range cfg.Amount { - var roomID string - err := genRetry(ctx, func(ctx context.Context) error { - var err error - roomID, err = creator.CreateRoom(ctx, auth.Config{Name: names.Generate()}) - if err != nil { - return fmt.Errorf("wbstream CreateRoom: %w", err) - } - return nil - }) - if err != nil { - return fmt.Errorf("gen wbstream room %d: %w", i+1, err) - } - out(roomID) - } - default: - return fmt.Errorf("%w: %s does not support room generation", ErrUnsupportedCarrier, cfg.Carrier) + out(roomID) } return nil } diff --git a/internal/app/session/session_test.go b/internal/app/session/session_test.go index c027e5a..ab705fe 100644 --- a/internal/app/session/session_test.go +++ b/internal/app/session/session_test.go @@ -14,7 +14,7 @@ func TestValidate(t *testing.T) { Mode: modeSRV, Link: "direct", Transport: "datachannel", - Carrier: "telemost", //nolint:goconst // test literal, repetition is intentional + Auth: "telemost", RoomID: "room-1", ClientID: "client-1", KeyHex: "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff", @@ -31,7 +31,7 @@ func TestValidate(t *testing.T) { name: "jazz allows empty room id", cfg: func() Config { cfg := base - cfg.Carrier = "jazz" //nolint:goconst // test literal, repetition is intentional + cfg.Auth = "jazz" cfg.RoomID = "" return cfg }(), @@ -59,7 +59,7 @@ func TestValidate(t *testing.T) { name: "unsupported carrier", cfg: func() Config { cfg := base - cfg.Carrier = "unknown" //nolint:goconst // test literal, repetition is intentional + cfg.Auth = "unknown" //nolint:goconst // test literal, repetition is intentional return cfg }(), want: ErrUnsupportedCarrier, @@ -338,25 +338,7 @@ func TestValidate(t *testing.T) { } } -func TestBuildRoomURL(t *testing.T) { - tests := []struct { - carrier string - roomID string - want string - }{ - {carrier: "telemost", roomID: "abc", want: "https://telemost.yandex.ru/j/abc"}, - {carrier: "jazz", roomID: "", want: "any"}, - {carrier: "jazz", roomID: "room", want: "room"}, - {carrier: "wbstream", roomID: "wb", want: "wb"}, //nolint:goconst // test literal, repetition is intentional - {carrier: "other", roomID: "raw", want: "raw"}, - } - - for _, tt := range tests { - if got := buildRoomURL(tt.carrier, tt.roomID); got != tt.want { - t.Fatalf("buildRoomURL(%q, %q) = %q, want %q", tt.carrier, tt.roomID, got, tt.want) - } - } -} +const testAuthWBStream = "wbstream" func TestValidateGen(t *testing.T) { RegisterDefaults() @@ -368,35 +350,35 @@ func TestValidateGen(t *testing.T) { }{ { name: "valid wbstream", - cfg: Config{Carrier: "wbstream", DNSServer: "1.1.1.1:53", Amount: 3}, + cfg: Config{Auth: testAuthWBStream, DNSServer: "1.1.1.1:53", Amount: 3}, }, { name: "valid jazz", - cfg: Config{Carrier: "jazz", DNSServer: "1.1.1.1:53", Amount: 1}, + cfg: Config{Auth: "jazz", DNSServer: "1.1.1.1:53", Amount: 1}, }, { - name: "missing carrier", + name: "missing auth", cfg: Config{DNSServer: "1.1.1.1:53", Amount: 1}, - want: ErrCarrierRequired, + want: ErrAuthRequired, }, { - name: "unsupported carrier", - cfg: Config{Carrier: "unknown", DNSServer: "1.1.1.1:53", Amount: 1}, + name: "unsupported auth", + cfg: Config{Auth: "unknown", DNSServer: "1.1.1.1:53", Amount: 1}, want: ErrUnsupportedCarrier, }, { name: "missing dns", - cfg: Config{Carrier: "wbstream", Amount: 1}, + cfg: Config{Auth: testAuthWBStream, Amount: 1}, want: ErrDNSServerRequired, }, { name: "amount zero", - cfg: Config{Carrier: "wbstream", DNSServer: "1.1.1.1:53", Amount: 0}, + cfg: Config{Auth: testAuthWBStream, DNSServer: "1.1.1.1:53", Amount: 0}, want: ErrAmountRequired, }, { name: "amount negative", - cfg: Config{Carrier: "wbstream", DNSServer: "1.1.1.1:53", Amount: -1}, + cfg: Config{Auth: testAuthWBStream, DNSServer: "1.1.1.1:53", Amount: -1}, want: ErrAmountRequired, }, } @@ -417,9 +399,9 @@ func TestValidateGen(t *testing.T) { } } -func TestGenUnsupportedCarrier(t *testing.T) { +func TestGenUnsupportedAuth(t *testing.T) { RegisterDefaults() - cfg := Config{Carrier: "telemost", DNSServer: "1.1.1.1:53", Amount: 1} + cfg := Config{Auth: "telemost", DNSServer: "1.1.1.1:53", Amount: 1} err := Gen(context.Background(), cfg, func(string) {}) if !errors.Is(err, ErrUnsupportedCarrier) { t.Fatalf("Gen(telemost) error = %v, want ErrUnsupportedCarrier", err) diff --git a/internal/auth/telemost/telemost.go b/internal/auth/telemost/telemost.go index 6774db1..048b348 100644 --- a/internal/auth/telemost/telemost.go +++ b/internal/auth/telemost/telemost.go @@ -3,10 +3,13 @@ package telemost import ( "context" "fmt" + "strings" "github.com/openlibrecommunity/olcrtc/internal/auth" ) +const roomURLPrefix = "https://telemost.yandex.ru/j/" + // Provider produces Goolom credentials for the Yandex Telemost service. type Provider struct{} @@ -15,14 +18,19 @@ func (Provider) Engine() string { return "goolom" } // Issue fetches connection info for a Telemost room and returns engine credentials. // -// cfg.RoomURL must be a Telemost conference URL (e.g. -// https://telemost.yandex.ru/j/). Room creation is not supported by the -// Telemost API; rooms originate in the Yandex UI. +// cfg.RoomURL accepts either a full Telemost conference URL +// (https://telemost.yandex.ru/j/) or just the room ID hash. Room +// creation is not supported by the Telemost API; rooms originate in the +// Yandex UI. func (Provider) Issue(ctx context.Context, cfg auth.Config) (auth.Credentials, error) { if cfg.RoomURL == "" { return auth.Credentials{}, auth.ErrRoomIDRequired } - info, err := GetConnectionInfo(ctx, cfg.RoomURL, cfg.Name) + roomURL := cfg.RoomURL + if !strings.HasPrefix(roomURL, "https://") { + roomURL = roomURLPrefix + roomURL + } + info, err := GetConnectionInfo(ctx, roomURL, cfg.Name) if err != nil { return auth.Credentials{}, fmt.Errorf("get connection info: %w", err) } @@ -32,8 +40,8 @@ func (Provider) Issue(ctx context.Context, cfg auth.Config) (auth.Credentials, e Extra: map[string]string{ "roomID": info.RoomID, "credentials": info.Credentials, - "roomURL": cfg.RoomURL, - "telemetryReferer": cfg.RoomURL, + "roomURL": roomURL, + "telemetryReferer": roomURL, }, }, nil } diff --git a/internal/carrier/builtin/engine_adapter.go b/internal/carrier/builtin/engine_adapter.go index 9827623..09bdb69 100644 --- a/internal/carrier/builtin/engine_adapter.go +++ b/internal/carrier/builtin/engine_adapter.go @@ -10,6 +10,31 @@ import ( "github.com/pion/webrtc/v4" ) +// registerDirect registers a carrier that skips auth entirely — the caller +// supplies the engine name, SFU URL, and access token directly via +// carrier.Config.Engine / carrier.Config.URL / carrier.Config.Token. +func registerDirect(carrierName string) { + carrier.Register(carrierName, func(ctx context.Context, cfg carrier.Config) (carrier.Session, error) { + engineName := cfg.Engine + if engineName == "" { + engineName = "livekit" + } + sess, err := engine.New(ctx, engineName, engine.Config{ + URL: cfg.URL, + Token: cfg.Token, + Name: cfg.Name, + OnData: cfg.OnData, + DNSServer: cfg.DNSServer, + ProxyAddr: cfg.ProxyAddr, + ProxyPort: cfg.ProxyPort, + }) + if err != nil { + return nil, fmt.Errorf("engine new: %w", err) + } + return &engineSession{session: sess}, nil + }) +} + // registerEngineAuth registers a carrier name that resolves credentials // through an auth provider and connects via the engine the auth provider // reports. diff --git a/internal/carrier/builtin/register.go b/internal/carrier/builtin/register.go index 73a815c..28c3885 100644 --- a/internal/carrier/builtin/register.go +++ b/internal/carrier/builtin/register.go @@ -15,4 +15,5 @@ func Register() { registerEngineAuth("wbstream", authWBStream.Provider{}) registerEngineAuth("jazz", authSaluteJazz.Provider{}) registerEngineAuth("telemost", authTelemost.Provider{}) + registerDirect("none") } diff --git a/internal/carrier/carrier.go b/internal/carrier/carrier.go index 3dde979..cf57987 100644 --- a/internal/carrier/carrier.go +++ b/internal/carrier/carrier.go @@ -44,6 +44,10 @@ type Config struct { DNSServer string ProxyAddr string ProxyPort int + // URL, Token, and Engine are used by the "none" auth carrier (direct engine access). + URL string + Token string + Engine string } // Factory creates a new carrier session. diff --git a/internal/client/client.go b/internal/client/client.go index 21cc7ec..fcbe01c 100644 --- a/internal/client/client.go +++ b/internal/client/client.go @@ -84,6 +84,7 @@ func Run( seiBatchSize int, seiFragmentSize int, seiAckTimeoutMS int, + engine, url, token string, ) error { return RunWithReady( ctx, linkName, transportName, carrierName, roomURL, keyHex, clientID, localAddr, @@ -92,6 +93,7 @@ func Run( videoQRSize, videoQRRecovery, videoCodec, videoTileModule, videoTileRS, vp8FPS, vp8BatchSize, seiFPS, seiBatchSize, seiFragmentSize, seiAckTimeoutMS, + engine, url, token, ) } @@ -125,6 +127,7 @@ func RunWithReady( seiBatchSize int, seiFragmentSize int, seiAckTimeoutMS int, + engine, url, token string, ) error { runCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -143,6 +146,7 @@ func RunWithReady( videoQRSize, videoQRRecovery, videoCodec, videoTileModule, videoTileRS, vp8FPS, vp8BatchSize, seiFPS, seiBatchSize, seiFragmentSize, seiAckTimeoutMS, + engine, url, token, ); err != nil { return err } @@ -181,11 +185,15 @@ func (c *Client) bringUpLink( videoTileModule, videoTileRS int, vp8FPS, vp8BatchSize int, seiFPS, seiBatchSize, seiFragmentSize, seiAckTimeoutMS int, + engine, url, token string, ) error { ln, err := link.New(ctx, linkName, link.Config{ Transport: transportName, Carrier: carrierName, RoomURL: roomURL, + Engine: engine, + URL: url, + Token: token, ClientID: c.clientID, Name: names.Generate(), OnData: c.onData, diff --git a/internal/e2e/tunnel_test.go b/internal/e2e/tunnel_test.go index 3144654..fa74c54 100644 --- a/internal/e2e/tunnel_test.go +++ b/internal/e2e/tunnel_test.go @@ -398,7 +398,7 @@ func validSessionConfig(mode, carrierName, transportName string) session.Config Mode: mode, Link: "direct", Transport: transportName, - Carrier: carrierName, + Auth: carrierName, RoomID: "room", ClientID: "client-1", KeyHex: testKeyHex, @@ -426,7 +426,7 @@ func validLinkConfig(carrierName, transportName string) link.Config { cfg := validSessionConfig("cnc", carrierName, transportName) return link.Config{ Transport: cfg.Transport, - Carrier: cfg.Carrier, + Carrier: cfg.Auth, RoomURL: "room", ClientID: cfg.ClientID, Name: "e2e-" + carrierName + "-" + transportName, @@ -542,6 +542,7 @@ func startTunnel(t *testing.T, serverClientID, clientClientID string) *tunnelRun 0, 0, 0, + "", "", "", ) }() room.waitConnected(t, 1) @@ -578,6 +579,7 @@ func startTunnel(t *testing.T, serverClientID, clientClientID string) *tunnelRun 0, 0, 0, + "", "", "", ) }() waitForReady(t, ready) @@ -633,6 +635,7 @@ func startRealTunnel( 4, 512, 1500, + "", "", "", ) }() @@ -678,6 +681,7 @@ func startRealTunnel( 4, 512, 1500, + "", "", "", ) }() diff --git a/internal/link/direct/direct.go b/internal/link/direct/direct.go index 0b40d5f..26b44fe 100644 --- a/internal/link/direct/direct.go +++ b/internal/link/direct/direct.go @@ -18,6 +18,9 @@ func New(ctx context.Context, cfg link.Config) (link.Link, error) { tr, err := transport.New(ctx, cfg.Transport, transport.Config{ Carrier: cfg.Carrier, RoomURL: cfg.RoomURL, + Engine: cfg.Engine, + URL: cfg.URL, + Token: cfg.Token, ClientID: cfg.ClientID, Name: cfg.Name, OnData: cfg.OnData, diff --git a/internal/link/link.go b/internal/link/link.go index 23606a2..9989e51 100644 --- a/internal/link/link.go +++ b/internal/link/link.go @@ -28,6 +28,10 @@ type Config struct { Transport string Carrier string RoomURL string + // Engine, URL, Token are forwarded for the "none" auth carrier. + Engine string + URL string + Token string ClientID string Name string OnData func([]byte) diff --git a/internal/server/server.go b/internal/server/server.go index 51e96f3..8f6327a 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -86,6 +86,7 @@ func Run( seiBatchSize int, seiFragmentSize int, seiAckTimeoutMS int, + engine, url, token string, ) error { runCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -110,6 +111,7 @@ func Run( videoQRSize, videoQRRecovery, videoCodec, videoTileModule, videoTileRS, vp8FPS, vp8BatchSize, seiFPS, seiBatchSize, seiFragmentSize, seiAckTimeoutMS, + engine, url, token, ); err != nil { return err } @@ -183,11 +185,15 @@ func (s *Server) bringUpLink( videoTileModule, videoTileRS int, vp8FPS, vp8BatchSize int, seiFPS, seiBatchSize, seiFragmentSize, seiAckTimeoutMS int, + engine, url, token string, ) error { ln, err := link.New(ctx, linkName, link.Config{ Transport: transportName, Carrier: carrierName, RoomURL: roomURL, + Engine: engine, + URL: url, + Token: token, ClientID: s.clientID, Name: names.Generate(), OnData: s.onData, diff --git a/internal/transport/datachannel/transport.go b/internal/transport/datachannel/transport.go index b361b3b..8a4f783 100644 --- a/internal/transport/datachannel/transport.go +++ b/internal/transport/datachannel/transport.go @@ -24,6 +24,9 @@ func New(ctx context.Context, cfg transport.Config) (transport.Transport, error) DNSServer: cfg.DNSServer, ProxyAddr: cfg.ProxyAddr, ProxyPort: cfg.ProxyPort, + Engine: cfg.Engine, + URL: cfg.URL, + Token: cfg.Token, }) if err != nil { return nil, fmt.Errorf("create carrier transport: %w", err) diff --git a/internal/transport/seichannel/transport.go b/internal/transport/seichannel/transport.go index c1b8ac7..78203f3 100644 --- a/internal/transport/seichannel/transport.go +++ b/internal/transport/seichannel/transport.go @@ -106,6 +106,9 @@ func New(ctx context.Context, cfg transport.Config) (transport.Transport, error) DNSServer: cfg.DNSServer, ProxyAddr: cfg.ProxyAddr, ProxyPort: cfg.ProxyPort, + Engine: cfg.Engine, + URL: cfg.URL, + Token: cfg.Token, }) if err != nil { return nil, fmt.Errorf("create carrier transport: %w", err) diff --git a/internal/transport/transport.go b/internal/transport/transport.go index 3061526..90153a2 100644 --- a/internal/transport/transport.go +++ b/internal/transport/transport.go @@ -36,6 +36,11 @@ type Transport interface { type Config struct { Carrier string RoomURL string + // Engine, URL, Token are forwarded to carrier.Config for the "none" auth + // carrier (direct engine access without a service-specific auth flow). + Engine string + URL string + Token string ClientID string Name string OnData func([]byte) diff --git a/internal/transport/videochannel/transport.go b/internal/transport/videochannel/transport.go index 6410d85..2c35b33 100644 --- a/internal/transport/videochannel/transport.go +++ b/internal/transport/videochannel/transport.go @@ -85,6 +85,9 @@ func New(ctx context.Context, cfg transport.Config) (transport.Transport, error) DNSServer: cfg.DNSServer, ProxyAddr: cfg.ProxyAddr, ProxyPort: cfg.ProxyPort, + Engine: cfg.Engine, + URL: cfg.URL, + Token: cfg.Token, }) if err != nil { return nil, fmt.Errorf("create carrier transport: %w", err) diff --git a/internal/transport/vp8channel/transport.go b/internal/transport/vp8channel/transport.go index c1504ff..13875b3 100644 --- a/internal/transport/vp8channel/transport.go +++ b/internal/transport/vp8channel/transport.go @@ -120,6 +120,9 @@ func New(ctx context.Context, cfg transport.Config) (transport.Transport, error) DNSServer: cfg.DNSServer, ProxyAddr: cfg.ProxyAddr, ProxyPort: cfg.ProxyPort, + Engine: cfg.Engine, + URL: cfg.URL, + Token: cfg.Token, }) if err != nil { return nil, fmt.Errorf("create carrier transport: %w", err) diff --git a/mobile/mobile.go b/mobile/mobile.go index ffb7a9c..a7b78ea 100644 --- a/mobile/mobile.go +++ b/mobile/mobile.go @@ -247,6 +247,7 @@ func Check( 0, 0, 0, + "", "", "", ) }() @@ -344,6 +345,7 @@ func Ping( 0, 0, 0, + "", "", "", ) }() @@ -603,6 +605,7 @@ func startWithConfig( 0, 0, 0, + "", "", "", ) mu.Lock() diff --git a/mobile/mobile_test.go b/mobile/mobile_test.go index 1db8990..45606d1 100644 --- a/mobile/mobile_test.go +++ b/mobile/mobile_test.go @@ -190,6 +190,7 @@ func TestStartWithInjectedRunnerLifecycle(t *testing.T) { _ int, _ int, _ int, + _, _, _ string, ) error { if linkName != defaultLink || transportName != dataTransport || carrierName != carrierJazz || roomURL != "any" || clientID != "client" || localAddr != "127.0.0.1:1080" || @@ -246,6 +247,7 @@ func TestStartUsesDefaultsAndCheckWithInjectedRunner(t *testing.T) { _ int, _ int, _ int, + _, _, _ string, ) error { if transportName != defaultTransport || roomURL != "https://telemost.yandex.ru/j/room" || localAddr != "127.0.0.1:1081" || socksUser != "u" || socksPass != "p" { @@ -287,6 +289,7 @@ func TestStartUsesDefaultsAndCheckWithInjectedRunner(t *testing.T) { _ int, _ int, _ int, + _, _, _ string, ) error { if transportName != dataTransport || vp8FPS != 1 || vp8BatchSize != 64 { t.Fatalf("Check args mismatch: transport=%q vp8=%d/%d", transportName, vp8FPS, vp8BatchSize) @@ -332,6 +335,7 @@ func TestCheckTimeoutAndRunError(t *testing.T) { _ int, _ int, _ int, + _, _, _ string, ) error { <-ctx.Done() return nil @@ -361,6 +365,7 @@ func TestCheckTimeoutAndRunError(t *testing.T) { int, int, int, + string, string, string, ) error { return want }