diff --git a/cpp/obs/src/moq-dock.cpp b/cpp/obs/src/moq-dock.cpp index c5b35528f71..b3b7e428565 100644 --- a/cpp/obs/src/moq-dock.cpp +++ b/cpp/obs/src/moq-dock.cpp @@ -414,7 +414,9 @@ MoQDock::MoQDock(QWidget *parent) : QWidget(parent) tokenEdit->setPlaceholderText("(optional) publish JWT"); tokenEdit->setEchoMode(QLineEdit::Password); const QString tokenHelp = "Optional relay publish token. Leave empty for public relays such as /anon. " - "Pasting a URL with ?jwt= into Relay URL fills this and strips it from the URL."; + "Pasting a URL with ?jwt= into Relay URL fills this and strips it from the URL. " + "Editing it while live presents the new token without reconnecting, on a session " + "that carries in-band authorization."; tokenEdit->setToolTip(tokenHelp); pathEdit = new QLineEdit(streamPage); @@ -609,7 +611,8 @@ MoQDock::MoQDock(QWidget *parent) : QWidget(parent) connect(pollTimer, &QTimer::timeout, this, &MoQDock::UpdateStatus); connect(urlEdit, &QLineEdit::editingFinished, this, &MoQDock::OnRelayUrlEdited); - connect(tokenEdit, &QLineEdit::editingFinished, this, &MoQDock::SaveSettings); + connect(tokenEdit, &QLineEdit::textEdited, this, [this] { tokenEdited = true; }); + connect(tokenEdit, &QLineEdit::editingFinished, this, &MoQDock::OnTokenEdited); connect(pathEdit, &QLineEdit::editingFinished, this, &MoQDock::SaveSettings); connect(encodingMode, QOverload::of(&QComboBox::currentIndexChanged), this, &MoQDock::OnEncodingChanged); connect(latencyMode, QOverload::of(&QComboBox::currentIndexChanged), this, &MoQDock::SaveSettings); @@ -675,6 +678,34 @@ void MoQDock::OnRelayUrlEdited() SaveSettings(); } +void MoQDock::OnTokenEdited() +{ + SaveSettings(); + + const QString token = tokenEdit->text().trimmed(); + if (!running || !output || token.isEmpty() || !tokenEdited) + return; + + // The same refusal as Go Live: the URL is fixed while running, but the token + // may be new. + const QUrl dial(urlEdit->text().trimmed()); + if (dial.isValid() && MoQDial::IsCleartext(dial.scheme().toStdString())) { + tokenNotice = "Publish token requires an encrypted relay URL (not http://, ws://, or tcp://)"; + status->setText(tokenNotice); + return; + } + + // Presented in band, so the stream keeps going while the relay checks it. + auto *moq = static_cast(obs_obj_get_data(output)); + if (!moq || !moq->PresentToken(token.toStdString())) { + tokenNotice = "Not connected yet; edit the token again once live"; + status->setText(tokenNotice); + return; + } + tokenNotice.clear(); + tokenEdited = false; +} + void MoQDock::PeelJwtFromRelayUrl() { QString text = urlEdit->text().trimmed(); @@ -1069,6 +1100,8 @@ void MoQDock::StartStream() const std::string url = ConnectUrl().toStdString(); const std::string path = pathEdit->text().toStdString(); + tokenEdited = false; + tokenNotice.clear(); SaveSettings(); @@ -1158,10 +1191,10 @@ void MoQDock::SetRunning(bool isRunning) .arg(isRunning ? "#e04434" : "#36a45e")); urlEdit->setEnabled(!isRunning); - tokenEdit->setEnabled(!isRunning); pathEdit->setEnabled(!isRunning); // The settings are read once at connect, so editing them mid-stream would look - // like it applied when it hadn't. + // like it applied when it hadn't. The token stays editable: a new one is + // presented on the live session. advancedButton->setEnabled(!isRunning); encodingMode->setEnabled(!isRunning); latencyMode->setEnabled(!isRunning); @@ -1260,14 +1293,24 @@ void MoQDock::UpdateStatus() return; } - status->setText(stats.reconnects > 0 ? QString("● Connected · reconnects %1").arg(stats.reconnects) - : QString("● Connected")); - status->setStyleSheet("color: #36a45e;"); + // Why the last token edit did not take, so the next refresh does not hide it. + QString tokenProblem = tokenNotice; + const QString tokenRefusal = QString::fromStdString(moq->TokenRefusal()); + if (tokenProblem.isEmpty() && !tokenRefusal.isEmpty()) + tokenProblem = QString("New publish token refused: %1").arg(tokenRefusal); + QString connectedText = stats.reconnects > 0 ? QString("● Connected · reconnects %1").arg(stats.reconnects) + : QString("● Connected"); + if (!tokenProblem.isEmpty()) + connectedText += QStringLiteral(" · new token not in use"); + status->setText(connectedText); + status->setStyleSheet(tokenProblem.isEmpty() ? "color: #36a45e;" : "color: #d08b1d;"); { QStringList lines; if (!publishSummary.isEmpty()) lines << publishSummary; + if (!tokenProblem.isEmpty()) + lines << tokenProblem; if (!stats.protocol.empty()) lines << QString("Protocol %1").arg(QString::fromStdString(stats.protocol)); if (!stats.dial.empty()) diff --git a/cpp/obs/src/moq-dock.h b/cpp/obs/src/moq-dock.h index 0d4e7d886f0..3beb13300fa 100644 --- a/cpp/obs/src/moq-dock.h +++ b/cpp/obs/src/moq-dock.h @@ -37,6 +37,7 @@ private slots: void OnEncodingChanged(int mode); void RefreshEncodingOptions(bool applyProfileDefaults = false); void OnRelayUrlEdited(); + void OnTokenEdited(); private: void StartStream(); @@ -96,6 +97,11 @@ private slots: OBSEncoderAutoRelease audioEncoder; bool running = false; + // Whether the token field was typed in since its token was last presented, so + // leaving it unchanged does not present it again while a retyped one does. + bool tokenEdited = false; + // Why the last token edit was not presented, kept until the next one is. + QString tokenNotice; // Filled when Go Live creates encoders; shown in Stats while publishing. QString publishSummary; QString latencySummary; diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index 6ea6deffe42..8a7763028e3 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -409,6 +409,7 @@ void MoQOutput::Reset() epoch.clear(); last_failure_code = MoQError::None; last_failure_reason.clear(); + token_refusal.clear(); } connect_time_ms = 0; @@ -597,6 +598,45 @@ void MoQOutput::CopyLastFailure(int *code, std::string *reason) *reason = last_failure_reason; } +std::string MoQOutput::TokenRefusal() +{ + std::lock_guard lock(mutex); + return token_refusal; +} + +bool MoQOutput::PresentToken(const std::string &token) +{ + std::lock_guard signal_lock(signal_mutex); + if (!attempt || !attempt->session) + return false; + { + std::lock_guard lock(mutex); + token_refusal.clear(); + } + + // Replacing `presenting` cancels a presentation still in flight, which the + // newer token supersedes. + attempt->presenting = attempt->session->auth()->add(token).then( + worker.Executor(), Current>>( + attempt, [this](const std::shared_ptr ¤t, auto result) { + if (!result) { + const std::string reason = result.error().to_string(); + LOG_WARNING("MoQ relay refused the new publish token: %s", + reason.c_str()); + std::lock_guard lock(mutex); + token_refusal = reason; + return; + } + // The old token's grant leaves the union only after the new one joined. + if (current->token) + current->token->cancel(); + current->token = *result; + LOG_INFO("MoQ relay accepted the new publish token: %s", + MoQRedactUrl(current->url).c_str()); + })); + return true; +} + bool MoQOutput::TryGetVideoRate(VideoRate *out) { VideoRate rate; diff --git a/cpp/obs/src/moq-output.h b/cpp/obs/src/moq-output.h index 13dd23d880f..230dcbb5d1e 100644 --- a/cpp/obs/src/moq-output.h +++ b/cpp/obs/src/moq-output.h @@ -76,6 +76,14 @@ class MoQOutput { }; bool TryGetVideoRate(VideoRate *out); + // Present a fresh publish token on the connected session without reconnecting, + // replacing the one presented here before once the relay accepts it. False when + // there is no session yet. The outcome is logged, and a refusal is kept for + // TokenRefusal(); a session without in-band AUTH refuses it. + bool PresentToken(const std::string &token); + // Why the relay refused the token PresentToken() last presented, or empty. + std::string TokenRefusal(); + private: friend struct MoQOutputStatsTest; bool CommitConnectionStats(const std::shared_ptr ¤t, ConnectionStats snapshot, @@ -99,6 +107,11 @@ class MoQOutput { bool connected = false; // False while the session is between reconnects. bool live = false; + // The latest token PresentToken() got accepted, presented on every reconnect + // until it is replaced or the attempt is dropped, and the call presenting the + // next one. + std::shared_ptr token; + std::optional presenting; }; void OnConnect(const std::shared_ptr &attempt, moq::expected> result); @@ -172,6 +185,7 @@ class MoQOutput { std::string epoch; int last_failure_code = 0; std::string last_failure_reason; + std::string token_refusal; // Written on the worker, read by GetConnectTime(). std::atomic connect_time_ms{0}; diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index 7ceae871ec8..f05f5848318 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -840,6 +840,27 @@ int main() } printf("OBS-driven restart: ok\n"); + // A token edited mid-stream goes to the live session only, and presenting a + // newer one or stopping cancels whatever is still in flight. + { + TestRelay relay; + reset(relay.Url()); + MoQOutput o(nullptr, OUTPUT); + CHECK(!o.PresentToken("early")); + CHECK(o.Start()); + CHECK(WaitFor([&] { return o.IsLiveSession(); })); + CHECK(o.PresentToken("fresh")); + CHECK(o.PresentToken("fresher")); + // The default protocol carries no in-band AUTH, so the relay's refusal is + // kept for the dock while the stream stays up. + CHECK(WaitFor([&] { return !o.TokenRefusal().empty(); })); + CHECK(o.IsLiveSession()); + CHECK(signalCount() == 0); + o.Stop(); + CHECK(!o.PresentToken("late")); + } + printf("present a token mid-stream: ok\n"); + // Restarts and teardown racing failures still in flight, repeatedly. Whatever // the order, each Start() reports at most once, and nothing reports once the // output is gone. diff --git a/dart/moq/lib/src/aliases.dart b/dart/moq/lib/src/aliases.dart index ee9d97750c1..c3747c84e2c 100644 --- a/dart/moq/lib/src/aliases.dart +++ b/dart/moq/lib/src/aliases.dart @@ -144,6 +144,15 @@ typedef Reservation = MoqReservation; /// A snapshot of transport connection statistics. typedef ConnectionStats = MoqConnectionStats; +/// A session's authorization: the grant its peer gave it, and more tokens to present. +typedef Auth = MoqAuth; + +/// One token presented with [Auth.add]; cancel it to withdraw the token. +typedef AuthToken = MoqAuthToken; + +/// What the peer lets this side do: publish and subscribe patterns, and an expiry. +typedef Grant = MoqGrant; + /// A connection lifecycle transition reported by [Session.status]. typedef ConnectionStatus = MoqConnectionStatus; diff --git a/dart/moq/lib/src/client.dart b/dart/moq/lib/src/client.dart index a95ecbf5b2e..c8d069dac90 100644 --- a/dart/moq/lib/src/client.dart +++ b/dart/moq/lib/src/client.dart @@ -254,6 +254,13 @@ final class Moq { /// the room back. Bandwidth bandwidth() => session.bandwidth(); + /// The session's authorization: the grant its peer gave it, and more tokens to present. + /// + /// Listen to [AuthGrants.grants] to follow the grant, and [Auth.add] a fresh + /// token before the current one expires; each added token is presented again + /// on every reconnect. + Auth auth() => session.auth(); + /// Drain within one second, then stop the client. /// /// Explicit `moq-lite-07-wip` waits for the peer to receive finished tails; older @@ -268,6 +275,27 @@ final class Moq { } } +/// A [Stream] view over a session's grant. +extension AuthGrants on Auth { + /// The current grant, then each change, null while there is none. + /// + /// Listens on one [Auth] share its place in the changes, so a second watcher + /// takes its own from [Moq.auth]. Ends when the session shuts down, and errors with its close reason otherwise. + Stream grants() async* { + yield grant(); + while (true) { + final Grant? next; + try { + next = await grantChanged(); + } on MoqException catch (e) { + if (e is CancelledMoqException || e is ClosedMoqException) return; + rethrow; + } + yield next; + } + } +} + /// A [Stream] view over an announcement cursor. extension AnnounceConsumerUpdates on AnnounceConsumer { /// Stream announce events until the cursor ends. diff --git a/dart/moq/lib/src/durations.dart b/dart/moq/lib/src/durations.dart index ca766268bbc..d5786188a44 100644 --- a/dart/moq/lib/src/durations.dart +++ b/dart/moq/lib/src/durations.dart @@ -11,6 +11,13 @@ extension ConnectionStatsDuration on MoqConnectionStats { Duration? get rtt => rttUs == null ? null : Duration(microseconds: rttUs!); } +/// Duration views over an authorization grant. +extension GrantDuration on MoqGrant { + /// How long until the grant lapses, measured when it was read, or null for never. + Duration? get expires => + expiresUs == null ? null : Duration(microseconds: expiresUs!); +} + /// Duration views over the reconnect pacing. extension BackoffDuration on MoqBackoff { /// Delay before the first reconnect attempt, or null for the default. diff --git a/dart/moq/test/moq_test.dart b/dart/moq/test/moq_test.dart index 3bcdcfa02fb..05b30c8b16c 100644 --- a/dart/moq/test/moq_test.dart +++ b/dart/moq/test/moq_test.dart @@ -139,6 +139,69 @@ void main() { }, ); + test('both sides read the grant their peer gave them', () async { + const version = ['moq-lite-07-wip']; + final server = MoqServer( + config: MoqServerConfig( + bind: '127.0.0.1:0', + versions: version, + tls: MoqServerTls(generate: ['localhost']), + quic: MoqQuicConfig(), + ), + ); + final address = await server.listen().timeout(timeout); + final accepted = () async { + final request = await server.accept().timeout(timeout); + if (request == null) throw StateError('server closed before accept'); + return request.accept().timeout(timeout); + }(); + + final client = await Moq.connect( + 'https://$address', + options: const ConnectOptions( + tlsVerify: false, + bind: '127.0.0.1:0', + versions: version, + reconnect: false, + ), + ).timeout(timeout); + final serverSession = await accepted; + + // Generated records compare by identity, so compare the fields. + void expectEverything(Grant? grant) { + expect(grant?.publish, ['**']); + expect(grant?.subscribe, ['**']); + expect(grant?.expires, isNull); + } + + expectEverything( + await client + .auth() + .grants() + .firstWhere((g) => g != null) + .timeout(timeout), + ); + expectEverything( + await serverSession + .auth() + .grants() + .firstWhere((g) => g != null) + .timeout(timeout), + ); + + // A peer that verifies no tokens of its own refuses another one. + await expectLater( + client.auth().add(token: 'extra').timeout(timeout), + throwsA(isA()), + ); + expectEverything(client.auth().grant()); + + // The client drains first: cancelling the server session would fail its shutdown. + await client.close(); + serverSession.cancel(code: 0); + server.cancel(); + }); + test('connects, announces, subscribes, and delivers a frame', () async { final relay = MoqOriginProducer(config: MoqOriginConfig()); final server = MoqServer( diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index f21651d2045..8e584b4f4e1 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -8,9 +8,7 @@ import "dart:ffi"; import "dart:io" show Platform, File, Directory; import "dart:isolate"; import "dart:typed_data"; - import "package:ffi/ffi.dart"; - import "uniffi_runtime.dart"; export "uniffi_runtime.dart"; @@ -497,6 +495,77 @@ class FfiConverterMoqAudioPacket { } } +class MoqGrant { + final List publish; + final List subscribe; + final int? expiresUs; + MoqGrant({required this.publish, required this.subscribe, this.expiresUs}); +} + +class FfiConverterMoqGrant { + static MoqGrant lift(RustBuffer buf) { + return FfiConverterMoqGrant.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final publish_lifted = FfiConverterSequenceString.read( + Uint8List.view(buf.buffer, new_offset), + ); + final publish = publish_lifted.value; + new_offset += publish_lifted.bytesRead; + final subscribe_lifted = FfiConverterSequenceString.read( + Uint8List.view(buf.buffer, new_offset), + ); + final subscribe = subscribe_lifted.value; + new_offset += subscribe_lifted.bytesRead; + final expiresUs_lifted = FfiConverterOptionalUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final expiresUs = expiresUs_lifted.value; + new_offset += expiresUs_lifted.bytesRead; + return LiftRetVal( + MoqGrant(publish: publish, subscribe: subscribe, expiresUs: expiresUs), + new_offset - buf.offsetInBytes, + ); + } + + static RustBuffer lower(MoqGrant value) { + final total_length = + FfiConverterSequenceString.allocationSize(value.publish) + + FfiConverterSequenceString.allocationSize(value.subscribe) + + FfiConverterOptionalUInt64.allocationSize(value.expiresUs) + + 0; + final buf = Uint8List(total_length); + write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqGrant value, Uint8List buf) { + int new_offset = buf.offsetInBytes; + new_offset += FfiConverterSequenceString.write( + value.publish, + Uint8List.view(buf.buffer, new_offset), + ); + new_offset += FfiConverterSequenceString.write( + value.subscribe, + Uint8List.view(buf.buffer, new_offset), + ); + new_offset += FfiConverterOptionalUInt64.write( + value.expiresUs, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset - buf.offsetInBytes; + } + + static int allocationSize(MoqGrant value) { + return FfiConverterSequenceString.allocationSize(value.publish) + + FfiConverterSequenceString.allocationSize(value.subscribe) + + FfiConverterOptionalUInt64.allocationSize(value.expiresUs) + + 0; + } +} + class MoqFetchGroupOptions { final int priority; final int? maxDelayUs; @@ -7802,6 +7871,198 @@ class FfiConverterMoqAudioProducer { } } +abstract class MoqAuthInterface { + Future add({required String token}); + MoqGrant? grant(); + Future grantChanged(); +} + +final _MoqAuthFinalizer = Finalizer>((ptr) { + rustCall((status) => uniffi_moq_ffi_fn_free_moqauth(ptr, status)); +}); + +class MoqAuth implements MoqAuthInterface { + late final Pointer _ptr; + bool _isDisposed = false; + MoqAuth._(this._ptr) { + _MoqAuthFinalizer.attach(this, _ptr, detach: this); + } + factory MoqAuth.lift(Pointer ptr) { + return MoqAuth._(ptr); + } + Pointer uniffiClonePointer() { + if (_isDisposed) { + throw StateError("Object has already been disposed"); + } + return rustCall((status) => uniffi_moq_ffi_fn_clone_moqauth(_ptr, status)); + } + + void dispose() { + if (_isDisposed) { + return; + } + _isDisposed = true; + _MoqAuthFinalizer.detach(this); + rustCall((status) => uniffi_moq_ffi_fn_free_moqauth(_ptr, status)); + } + + Future add({required String token}) { + return uniffiRustCallAsync( + () => uniffi_moq_ffi_fn_method_moqauth_add( + uniffiClonePointer(), + FfiConverterString.lower(token), + ), + ffi_moq_ffi_rust_future_poll_u64, + ffi_moq_ffi_rust_future_complete_u64, + ffi_moq_ffi_rust_future_free_u64, + (ptr) => FfiConverterMoqAuthToken.lift(Pointer.fromAddress(ptr)), + moqExceptionErrorHandler, + ); + } + + MoqGrant? grant() { + return rustCallWithLifter( + (status) => + uniffi_moq_ffi_fn_method_moqauth_grant(uniffiClonePointer(), status), + FfiConverterOptionalMoqGrant.lift, + null, + ); + } + + Future grantChanged() { + return uniffiRustCallAsync( + () => + uniffi_moq_ffi_fn_method_moqauth_grant_changed(uniffiClonePointer()), + ffi_moq_ffi_rust_future_poll_rust_buffer, + ffi_moq_ffi_rust_future_complete_rust_buffer, + ffi_moq_ffi_rust_future_free_rust_buffer, + FfiConverterOptionalMoqGrant.lift, + moqExceptionErrorHandler, + ); + } +} + +class FfiConverterMoqAuth { + static MoqAuth lift(Pointer ptr) { + return MoqAuth.lift(ptr); + } + + static Pointer lower(MoqAuth value) { + return value.uniffiClonePointer(); + } + + static int allocationSize(MoqAuth value) { + return 8; + } + + static LiftRetVal read(Uint8List buf) { + final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); + final pointer = Pointer.fromAddress(handle); + return LiftRetVal(MoqAuth.lift(pointer), 8); + } + + static int write(MoqAuth value, Uint8List buf) { + final handle = lower(value); + buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); + return 8; + } +} + +abstract class MoqAuthTokenInterface { + void cancel(); + Future closed(); + MoqGrant? grant(); +} + +final _MoqAuthTokenFinalizer = Finalizer>((ptr) { + rustCall((status) => uniffi_moq_ffi_fn_free_moqauthtoken(ptr, status)); +}); + +class MoqAuthToken implements MoqAuthTokenInterface { + late final Pointer _ptr; + bool _isDisposed = false; + MoqAuthToken._(this._ptr) { + _MoqAuthTokenFinalizer.attach(this, _ptr, detach: this); + } + factory MoqAuthToken.lift(Pointer ptr) { + return MoqAuthToken._(ptr); + } + Pointer uniffiClonePointer() { + if (_isDisposed) { + throw StateError("Object has already been disposed"); + } + return rustCall( + (status) => uniffi_moq_ffi_fn_clone_moqauthtoken(_ptr, status), + ); + } + + void dispose() { + if (_isDisposed) { + return; + } + _isDisposed = true; + _MoqAuthTokenFinalizer.detach(this); + rustCall((status) => uniffi_moq_ffi_fn_free_moqauthtoken(_ptr, status)); + } + + void cancel() { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqauthtoken_cancel( + uniffiClonePointer(), + status, + ); + }, null); + } + + Future closed() { + return uniffiRustCallAsync( + () => uniffi_moq_ffi_fn_method_moqauthtoken_closed(uniffiClonePointer()), + ffi_moq_ffi_rust_future_poll_void, + ffi_moq_ffi_rust_future_complete_void, + ffi_moq_ffi_rust_future_free_void, + (_) {}, + moqExceptionErrorHandler, + ); + } + + MoqGrant? grant() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqauthtoken_grant( + uniffiClonePointer(), + status, + ), + FfiConverterOptionalMoqGrant.lift, + null, + ); + } +} + +class FfiConverterMoqAuthToken { + static MoqAuthToken lift(Pointer ptr) { + return MoqAuthToken.lift(ptr); + } + + static Pointer lower(MoqAuthToken value) { + return value.uniffiClonePointer(); + } + + static int allocationSize(MoqAuthToken value) { + return 8; + } + + static LiftRetVal read(Uint8List buf) { + final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); + final pointer = Pointer.fromAddress(handle); + return LiftRetVal(MoqAuthToken.lift(pointer), 8); + } + + static int write(MoqAuthToken value, Uint8List buf) { + final handle = lower(value); + buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); + return 8; + } +} + abstract class MoqBandwidthInterface { MoqReservation reserve({required MoqTrackDemand demand, required int maxBps}); } @@ -12064,6 +12325,7 @@ class FfiConverterMoqClient { } abstract class MoqSessionInterface { + MoqAuth auth(); MoqBandwidth bandwidth(); void cancel({required int code}); Future closed(); @@ -12108,6 +12370,17 @@ class MoqSession implements MoqSessionInterface { rustCall((status) => uniffi_moq_ffi_fn_free_moqsession(_ptr, status)); } + MoqAuth auth() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqsession_auth( + uniffiClonePointer(), + status, + ), + FfiConverterMoqAuth.lift, + null, + ); + } + MoqBandwidth bandwidth() { return rustCallWithLifter( (status) => uniffi_moq_ffi_fn_method_moqsession_bandwidth( @@ -13695,6 +13968,51 @@ class FfiConverterOptionalMoqFrame { } } +class FfiConverterOptionalMoqGrant { + static MoqGrant? lift(RustBuffer buf) { + return FfiConverterOptionalMoqGrant.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { + return LiftRetVal(null, 1); + } + final result = FfiConverterMoqGrant.read( + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ); + return LiftRetVal(result.value, result.bytesRead + 1); + } + + static int allocationSize([MoqGrant? value]) { + if (value == null) { + return 1; + } + return FfiConverterMoqGrant.allocationSize(value) + 1; + } + + static RustBuffer lower(MoqGrant? value) { + if (value == null) { + return toRustBuffer(Uint8List.fromList([0])); + } + final buf = Uint8List(FfiConverterOptionalMoqGrant.allocationSize(value)); + FfiConverterOptionalMoqGrant.write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqGrant? value, Uint8List buf) { + if (value == null) { + buf[0] = 0; + return 1; + } + buf[0] = 1; + return FfiConverterMoqGrant.write( + value, + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ) + + 1; + } +} + class FfiConverterOptionalMoqGroupConsumer { static MoqGroupConsumer? lift(RustBuffer buf) { return FfiConverterOptionalMoqGroupConsumer.read(buf.asUint8List()).value; @@ -15284,6 +15602,80 @@ external void uniffi_moq_ffi_fn_method_moqaudioproducer_write( Pointer uniffiStatus, ); +@Native Function(Pointer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_clone_moqauth( + Pointer handle, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_free_moqauth( + Pointer handle, + Pointer uniffiStatus, +); + +@Native Function(Pointer, RustBuffer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_method_moqauth_add( + Pointer ptr, + RustBuffer token, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external RustBuffer uniffi_moq_ffi_fn_method_moqauth_grant( + Pointer ptr, + Pointer uniffiStatus, +); + +@Native Function(Pointer)>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_method_moqauth_grant_changed( + Pointer ptr, +); + +@Native Function(Pointer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_clone_moqauthtoken( + Pointer handle, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_free_moqauthtoken( + Pointer handle, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqauthtoken_cancel( + Pointer ptr, + Pointer uniffiStatus, +); + +@Native Function(Pointer)>(assetId: _uniffiAssetId) +external Pointer uniffi_moq_ffi_fn_method_moqauthtoken_closed( + Pointer ptr, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external RustBuffer uniffi_moq_ffi_fn_method_moqauthtoken_grant( + Pointer ptr, + Pointer uniffiStatus, +); + @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) @@ -17179,6 +17571,14 @@ external void uniffi_moq_ffi_fn_free_moqsession( Pointer uniffiStatus, ); +@Native Function(Pointer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_method_moqsession_auth( + Pointer ptr, + Pointer uniffiStatus, +); + @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) @@ -17930,6 +18330,24 @@ external int uniffi_moq_ffi_checksum_method_moqaudioproducer_reservation(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqaudioproducer_write(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqauth_add(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqauth_grant(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqauth_grant_changed(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqauthtoken_cancel(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqauthtoken_closed(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqauthtoken_grant(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqbandwidth_reserve(); @@ -18336,6 +18754,9 @@ external int uniffi_moq_ffi_checksum_method_moqclient_connect(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqclient_failure(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqsession_auth(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqsession_bandwidth(); @@ -18582,6 +19003,24 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqaudioproducer_write() != 22094) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqauth_add() != 1074) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqauth_grant() != 14588) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqauth_grant_changed() != 5195) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqauthtoken_cancel() != 59324) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqauthtoken_closed() != 31020) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqauthtoken_grant() != 31654) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqbandwidth_reserve() != 27807) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } @@ -19007,6 +19446,9 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqclient_failure() != 29341) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqsession_auth() != 1680) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqsession_bandwidth() != 8006) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } diff --git a/doc/bin/obs.md b/doc/bin/obs.md index 3f09b137291..5bce66618d5 100644 --- a/doc/bin/obs.md +++ b/doc/bin/obs.md @@ -23,7 +23,12 @@ The source plays Opus, AAC-LC (mono or stereo), and PCM audio. A channel count n Enter a relay URL explicitly. On the shared anonymous relay, use a unique path such as `https://cdn.moq.dev/anon/your-stream`. A URL with `?jwt=` fills the -token field. +token field. Editing the token while live presents the new one without +reconnecting, on a session that carries in-band authorization: a +`moq-transport` draft-17 or later version pinned under Advanced settings, with +a relay that negotiates the MoQ Auth extension. The default protocol does not +yet, so there the dock reports the new token as refused and it applies on the +next Go Live. By default the dock tunes the encoder for low latency on this stream only: no B-frames or lookahead on the encoders that support turning them off. This diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index a961d2b85ba..ab3b7c2e2f6 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -200,6 +200,10 @@ Hold the owned `BandwidthConsumer` returned by `session->recv_bandwidth()` to re Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. +## Authorization + +On `moq-lite-07-wip` (opt in with `ClientConfig::versions`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `session->auth()->grant()` is the `Grant` the relay sent back, empty until it answers and while a client session reconnects; `grant_changed()` waits for the next one. `add(token)` presents a fresh token without reconnecting and resolves to an `AuthToken` whose grant joins the union until `cancel()`; it is presented again on every reconnect. A refusal resolves to an `Error` carrying the relay's code. + ## Stats `session->stats()` reports the transport counters, each a `std::optional` that is empty when the transport does not report it, plus the negotiated protocol in `version`. While a client session reconnects, `session->failure()` returns why as a `moq::Error`, and while its first `connect` is still retrying, `client->failure()` does. diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 9b2f0831f3e..a7038d555d5 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -137,6 +137,7 @@ shows which module holds each. - **Decoded frames hold decoder buffers.** `dispose()` each frame from a `video.Consumer` promptly rather than waiting for the garbage collector, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `await moq.close()` or `session.shutdown()` drains for up to one second and reports any error. `session.cancel(code: 0)` closes at once. Finish or abort live tracks first. Cancelling a stream subscription releases its native cursor. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Stats.** `session.stats()` reports `rttUs`, `estimatedSendRateBps`, and the byte and packet counters (`bytesSent`, `bytesReceived`, `bytesLost`, `packetsSent`, `packetsReceived`, `packetsLost`). A field is `null` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol, and while a client session reconnects `session.failure()` says why. Before the first connect, `client.failure()` says why the dial is retrying. Hold the owned `BandwidthConsumer` returned by `session.recvBandwidth()` to receive bandwidth estimates. `peek()` reads the current estimate in bits per second and `changed()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. +- **Authorization.** On `moq-lite-07-wip` (opt in with `ConnectOptions(versions: ...)`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), listen to `moq.auth().grants()` for the `Grant` the relay sent back, `null` until it answers and while a client session reconnects. `await auth().add(token: ...)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal throws with the relay's code. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** Audio and video convenience decoders reject a catalog configuration with `enabled: false`. Select an enabled rendition from the latest catalog before opening a decoder. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index ab775825e9c..ae7cc6a3776 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -127,6 +127,7 @@ shows which module holds each. - **Decoded frames hold decoder buffers.** `Close` each `VideoDecodedFrame` promptly, or the decoder stalls. `Resize` is best effort, so read each frame's `Width()` and `Height()`. - **Closing.** `client.Close()` drains for up to one second and returns an error on failure. `session.Shutdown(ctx)` does the same, and `session.Cancel(code)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Stats.** `Session().Stats()` reports `RTT`, `EstimatedSendRateBps`, and the byte and packet counters (`BytesSent`, `BytesReceived`, `BytesLost`, `PacketsSent`, `PacketsReceived`, `PacketsLost`). A field is nil when the transport does not report it, which is not the same as zero. `Version` names the negotiated protocol, and while a client session reconnects `Session().Failure()` says why. Hold the owned `BandwidthConsumer` returned by `Session().RecvBandwidth()` to receive bandwidth estimates. `Peek()` reads the current estimate in bits per second and `Changed(ctx)` waits for a new one. Calling `Close()` on the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. +- **Authorization.** On `moq-lite-07-wip` (opt in with `moq.WithVersions`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `Session().Auth().Grant()` is the `Grant` the relay sent back, nil until it answers and while a client session reconnects; `GrantChanged(ctx)` waits for the next one. `Add(ctx, token)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `Close()`, and is presented again on every reconnect. A refusal returns the relay's code. - **Errors.** Read diagnostic payloads from `Reason` and structured protocol failures from `Details`. - **Disabled renditions.** Audio and video convenience decoders reject a catalog configuration with `enabled: false`. Select an enabled rendition from the latest catalog before opening a decoder. - **Track demand.** Obtain a dynamic track request’s `Demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. diff --git a/doc/lib/index.md b/doc/lib/index.md index 291995ff87d..fc8a5bc800b 100644 --- a/doc/lib/index.md +++ b/doc/lib/index.md @@ -45,8 +45,9 @@ how it looks in that language: - **Catalog extensions**: write your own section next to `video` and `audio`, and read others' back. - **Routes**: see which relays a broadcast came through, and advertise a cost as a standby publisher. - **Connection stats** and **errors** that tell an auth rejection (don't retry) from a shutdown or a transport failure. +- **Authorization in band**: read the grant the relay sent back (the patterns this side may publish and subscribe to, and when it lapses), and present a fresh token before the current one expires, without reconnecting. It needs `moq-lite-07-wip`, which is opt-in, or moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth). -C's audio decoder neither takes nor reports a playout delay. +C's audio decoder neither takes nor reports a playout delay, and C has no in-band authorization. ## Upgrading diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index 476759efae8..ea0004a54a9 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -99,6 +99,7 @@ shows which module holds each. - **Decoded frames hold decoder buffers.** `close()` each frame from a `dev.moq.video.Consumer` promptly (or `use { }` it), or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** Suspending `moq.shutdown()` drains for up to one second and throws on failure. `Moq.close()`, and so `use { }`, cancels at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Stats.** `session.stats()` reports `rttUs`, `estimatedSendRateBps`, and the byte and packet counters (`bytesSent`, `bytesReceived`, `bytesLost`, `packetsSent`, `packetsReceived`, `packetsLost`). A field is `null` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol, and while a client session reconnects `session.failure()` says why. Before the first connect, `client.failure()` says why the dial is retrying. Hold the owned `BandwidthConsumer` returned by `session.recvBandwidth()` to receive bandwidth estimates. `peek()` reads the current estimate in bits per second and `changed()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. +- **Authorization.** On `moq-lite-07-wip` (opt in with `ClientConfig(versions = ...)`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), collect `moq.auth().grants()` for the `Grant` the relay sent back, `null` until it answers and while a client session reconnects. Suspending `auth().add(token)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal throws with the relay's code. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** Audio and video convenience decoders reject a catalog configuration with `enabled: false`. Select an enabled rendition from the latest catalog before opening a decoder. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index 950468be57e..3b0974b412f 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -114,6 +114,7 @@ shows which module holds each. - **Decoded frames hold decoder buffers.** Drop each frame from a `moq.video.Consumer` promptly, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `await session.shutdown()` drains for up to one second and raises on failure. `cancel(code)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Stats.** `session.stats()` reports `rtt`, `estimated_send_rate_bps`, and the byte and packet counters (`bytes_sent`, `bytes_received`, `bytes_lost`, `packets_sent`, `packets_received`, `packets_lost`). A field is `None` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol, and while a client session reconnects `session.failure()` says why. Before the first connect, `client.failure()` says why the dial is retrying. Hold the owned `Bandwidth.Consumer` returned by `session.recv_bandwidth()` to receive bandwidth estimates. `peek()` reads the current estimate in bits per second and `changed()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. +- **Authorization.** On `moq-lite-07-wip` (opt in with `versions=`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `session.auth().grant()` is the `Grant` the relay sent back, `None` until it answers and while a client session reconnects; `await auth.grant_changed()` waits for the next one. `await auth.add(token)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal raises with the relay's code. - **Errors.** Read diagnostic payloads from `reason` and structured protocol failures from `details`. - **Disabled renditions.** Audio and video convenience decoders reject a catalog configuration with `enabled: false`. Select an enabled rendition from the latest catalog before opening a decoder. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index ac2b2286271..3fe646d2d46 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -98,6 +98,7 @@ shows which module holds each. - **Decoded frames hold decoder buffers.** Release each frame from a `Video.Consumer` promptly, or the decoder stalls. `resize` is best effort, so read each frame's `width()` and `height()`. - **Closing.** `try await session.shutdown()` drains for up to one second and throws on failure. `session.cancel(code:)` closes at once. Finish or abort live tracks first. Application delivery requires explicitly selecting `moq-lite-07-wip` on both peers; older Lite versions (including default lite-06) and IETF wait only for transport ACKs. - **Stats.** `session.stats()` reports `rttUs`, `estimatedSendRateBps`, and the byte and packet counters (`bytesSent`, `bytesReceived`, `bytesLost`, `packetsSent`, `packetsReceived`, `packetsLost`). A field is `nil` when the transport does not report it, which is not the same as zero. `version` names the negotiated protocol, and while a client session reconnects `session.failure()` says why. Before the first connect, `client.failure()` says why the dial is retrying. Hold the owned `Bandwidth.Consumer` returned by `session.recvBandwidth()` to receive bandwidth estimates. `peek()` reads the current estimate in bits per second and `changed()` waits for a new one. Release the consumer or call `cancel()` when finished; releasing the last consumer stops PROBE; the session and stats snapshots do not request it. Client consumers survive reconnects and report no estimate while disconnected. +- **Authorization.** On `moq-lite-07-wip` (opt in with `Client(versions:)`) and on moq-transport draft-17+ with the [MoQ Auth extension](/draft/moq-auth), `session.auth().grant()` is the `Grant` the relay sent back, `nil` until it answers and while a client session reconnects; `try await auth.grantChanged()` waits for the next one. `try await auth.add(token:)` presents a fresh token without reconnecting, returns an `AuthToken` whose grant joins the union until `cancel()`, and is presented again on every reconnect. A refusal throws with the relay's code. - **Errors.** Diagnostic cases use the `reason:` argument label; match their associated value to inspect it. Protocol errors carry `details`. - **Disabled renditions.** Audio and video convenience decoders reject a catalog configuration with `enabled: false`. Select an enabled rendition from the latest catalog before opening a decoder. - **Track demand.** Obtain a dynamic track request’s `demand()` before accepting it to watch subscribers through the handoff to its producer. The watch does not keep the track alive and ends when the track closes or the request is rejected. diff --git a/go/wrapper/auth_test.go b/go/wrapper/auth_test.go new file mode 100644 index 00000000000..66a762965aa --- /dev/null +++ b/go/wrapper/auth_test.go @@ -0,0 +1,80 @@ +package moq_test + +import ( + "context" + "errors" + "reflect" + "testing" + "time" + + "moq.dev/moq" +) + +// awaitGrant follows auth's changes until it reports want. +func awaitGrant(t *testing.T, ctx context.Context, auth *moq.Auth, want moq.Grant) { + t.Helper() + + for { + got := auth.Grant() + if got != nil && reflect.DeepEqual(*got, want) { + return + } + if _, err := auth.GrantChanged(ctx); err != nil { + t.Fatalf("grant = %+v, want %+v: %v", got, want, err) + } + } +} + +// Both sides read the grant their peer gave them, and a peer that verifies no +// tokens of its own refuses another one as unsupported. +func TestSessionAuthReadsTheGrant(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + const version = "moq-lite-07-wip" + server, err := moq.Listen(ctx, "127.0.0.1:0", moq.WithTLSGenerate("localhost"), moq.WithServerVersions(version)) + if err != nil { + t.Fatal(err) + } + defer server.Close() + + accepted := make(chan *moq.Session, 1) + go func() { + req, err := server.Accept(ctx) + if err != nil || req == nil { + return + } + session, err := req.Accept(ctx, nil, nil) + if err != nil { + return + } + accepted <- session + }() + + client, err := moq.Dial(ctx, "https://"+server.LocalAddr(), + moq.WithTLSVerify(false), moq.WithVersions(version), moq.WithReconnect(false)) + if err != nil { + t.Fatal(err) + } + defer client.Close() + + everything := moq.Grant{Publish: []string{"**"}, Subscribe: []string{"**"}} + auth := client.Session().Auth() + awaitGrant(t, ctx, auth, everything) + + var session *moq.Session + select { + case session = <-accepted: + case <-ctx.Done(): + t.Fatal("no session accepted") + } + defer session.Cancel(0) + awaitGrant(t, ctx, session.Auth(), everything) + + if _, err := auth.Add(ctx, "extra"); !errors.Is(err, moq.ErrUnsupported) { + t.Fatalf("add = %v, want ErrUnsupported", err) + } + if got := auth.Grant(); got == nil || !reflect.DeepEqual(*got, everything) { + t.Fatalf("grant after a refusal = %+v", got) + } +} diff --git a/go/wrapper/records.go b/go/wrapper/records.go index 175e446ea7c..455fd68729a 100644 --- a/go/wrapper/records.go +++ b/go/wrapper/records.go @@ -158,3 +158,21 @@ func connectionStatsFromFFI(s ffi.MoqConnectionStats) ConnectionStats { Version: s.Version, } } + +// Grant is what the peer lets this side do, in this side's own paths. An empty +// pattern list grants nothing in that direction; "**" grants everything. +type Grant struct { + // Publish lists the broadcast patterns this side may publish to the peer. + Publish []string + // Subscribe lists the broadcast patterns this side may subscribe to from the peer. + Subscribe []string + // Expires is how long until the grant lapses, measured when it was read, or nil for never. + Expires *time.Duration +} + +func grantFromFFI(g *ffi.MoqGrant) *Grant { + if g == nil { + return nil + } + return &Grant{Publish: g.Publish, Subscribe: g.Subscribe, Expires: optFromMicros(g.ExpiresUs)} +} diff --git a/go/wrapper/session.go b/go/wrapper/session.go index b653b491ebf..91eadb3e4ab 100644 --- a/go/wrapper/session.go +++ b/go/wrapper/session.go @@ -73,6 +73,70 @@ func (s *Session) Bandwidth() *Bandwidth { return &Bandwidth{inner: s.inner.Bandwidth()} } +// Auth is the session's authorization: the grant its peer gave it, and more +// tokens to present. Each call returns an independent watcher of the grant. A +// client session's handle survives reconnects, presenting each added token again +// on every new connection. +func (s *Session) Auth() *Auth { + return &Auth{inner: s.inner.Auth()} +} + +// Auth reads a session's grant and presents more tokens. +// +// AUTH is only on moq-lite-07-wip and on moq-transport sessions that negotiate +// the MoQ Auth extension; elsewhere the grant stays nil. +type Auth struct { + inner *ffi.MoqAuth +} + +// Grant returns the union of every token's grant, or nil until the peer first +// answers one. A client session reads nil while it reconnects. +func (a *Auth) Grant() *Grant { + return grantFromFFI(a.inner.Grant()) +} + +// GrantChanged blocks until the grant differs from the one this handle last +// reported, or held when it was made, and returns it. It errors once the session +// ends. +func (a *Auth) GrantChanged(ctx context.Context) (*Grant, error) { + grant, err := bridge.Call(ctx, nil, a.inner.GrantChanged) + return grantFromFFI(grant), err +} + +// Add presents another token, returning once the peer accepts it. Its grant +// joins the union until the token is closed. The peer's refusal is a +// ProtocolError carrying its code; a session that cannot carry another token +// returns an error for which errors.Is(err, ErrUnsupported) holds. Cancelling ctx +// withdraws the token. +func (a *Auth) Add(ctx context.Context, token string) (*AuthToken, error) { + inner, err := bridge.CallHandle(ctx, nil, func(ctx context.Context) (*ffi.MoqAuthToken, error) { + return a.inner.Add(ctx, token) + }) + if err != nil { + return nil, err + } + return &AuthToken{inner: inner}, nil +} + +// AuthToken is one token presented with Auth.Add. +type AuthToken struct { + inner *ffi.MoqAuthToken +} + +// Grant returns this token's own grant, or nil after it ended. +func (t *AuthToken) Grant() *Grant { + return grantFromFFI(t.inner.Grant()) +} + +// Closed blocks until the token ends, returning the peer's revocation or the +// session's close, or nil when the peer closed it without an error. +func (t *AuthToken) Closed(ctx context.Context) error { + return bridge.CallErr(ctx, nil, t.inner.Closed) +} + +// Close withdraws the token, removing its grant from the union. +func (t *AuthToken) Close() { t.inner.Cancel() } + // Bandwidth divides one connection's send estimate among the tracks sharing it. type Bandwidth struct { inner *ffi.MoqBandwidth diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index 6430cf04fc7..a47e258b1d8 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -153,6 +153,12 @@ typealias BandwidthConsumer = uniffi.moq.MoqBandwidthConsumer typealias Reservation = uniffi.moq.MoqReservation /** A snapshot of transport connection statistics. */ typealias ConnectionStats = uniffi.moq.MoqConnectionStats +/** A session's authorization: the grant its peer gave it, and more tokens to present. */ +typealias Auth = uniffi.moq.MoqAuth +/** One token presented with [Auth.add]; cancel it to withdraw the token. */ +typealias AuthToken = uniffi.moq.MoqAuthToken +/** What the peer lets this side do: publish and subscribe patterns, and an expiry. */ +typealias Grant = uniffi.moq.MoqGrant /** A connection lifecycle transition reported by [Session.status]. */ typealias ConnectionStatus = uniffi.moq.MoqConnectionStatus /** Retry pacing for the automatic reconnect: initial delay, multiplier, ceiling, and give-up window. */ diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt index 6dd59ce6e05..d771680811a 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Durations.kt @@ -13,6 +13,10 @@ import kotlin.time.Duration.Companion.microseconds val ConnectionStats.rtt: Duration? get() = rttUs?.toLong()?.microseconds +/** How long until the grant lapses, measured when it was read, or null for never. */ +val Grant.expires: Duration? + get() = expiresUs?.toLong()?.microseconds + /** Delay before the first reconnect attempt, or null for the default. */ val Backoff.initial: Duration? get() = initialUs?.toLong()?.microseconds diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt index 5dc5e412567..0b9000452ef 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt @@ -55,6 +55,26 @@ fun GroupConsumer.frames(): Flow = pull({ cancel() }) { readFrame() } */ fun AnnounceConsumer.updates(): Flow = pull { next() }.onCompletion { cancel() } +/** + * The session's grant: the current one, then each change, null while there is none. + * + * Collections of one [Auth] share its place in the changes, so a second watcher takes + * its own from [Moq.auth]. Completes when the session shuts down, and throws its close reason otherwise. + */ +fun Auth.grants(): Flow = flow { + emit(grant()) + while (true) { + currentCoroutineContext().ensureActive() + val grant = try { + grantChanged() + } catch (e: MoqException) { + if (e.isShutdown) break + throw e + } + emit(grant) + } +} + /** Emits [next] until it returns null, checking for cancellation before each pull. */ private fun pull(next: suspend () -> T?): Flow = flow { while (true) { diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt index 21d7628e621..22af44a7a77 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Moq.kt @@ -68,6 +68,14 @@ class Moq internal constructor( */ fun bandwidth(): Bandwidth = session.bandwidth() + /** + * The session's authorization: the grant its peer gave it, and more tokens to present. + * + * Collect [Auth.grants] to follow the grant, and [Auth.add] a fresh token before the + * current one expires; each added token is presented again on every reconnect. + */ + fun auth(): Auth = session.auth() + /** * Drain finished tracks within one second, throwing on failure. * Explicit `moq-lite-07-wip` waits for the peer to receive finished tails; older diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index 691a8799972..a94059c54e8 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -619,6 +619,36 @@ class SmokeTest { } } + /** + * Both sides read the grant their peer gave them, and a peer that verifies no tokens + * of its own refuses another one as unsupported. + */ + @Test + fun `session auth reads the grant`() = runTest { + val version = listOf("moq-lite-07-wip") + Server.listen( + ServerConfig(bind = "127.0.0.1:0", versions = version, tls = ServerTls(generate = listOf("localhost"))), + ).use { server -> + // Not requests().first(): ending that collection cancels the server. + val accepted = async { server.server.accept()!!.accept() } + val moq = Moq.connect( + "https://${server.localAddr}", + ClientConfig(bind = "127.0.0.1:0", versions = version, tls = ClientTls(insecure = true), once = true), + ) + moq.use { + val everything = Grant(publish = listOf("**"), subscribe = listOf("**"), expiresUs = null) + assertEquals(everything, moq.auth().grants().first { it != null }) + val session = accepted.await() + assertEquals(everything, session.auth().grants().first { it != null }) + assertNull(everything.expires) + + assertFailsWith { moq.auth().add("extra") } + assertEquals(everything, moq.auth().grant()) + session.cancel(0u) + } + } + } + /** * Cancelling the client aborts a connect parked on a server that never * accepts, and the connect reports it as [MoqException.Cancelled]. diff --git a/py/moq-rs/README.md b/py/moq-rs/README.md index 1b7e3caf71f..79acfdf15ca 100644 --- a/py/moq-rs/README.md +++ b/py/moq-rs/README.md @@ -158,6 +158,7 @@ client = moq.Client( - `.stats() → ConnectionStats`. Snapshot RTT, bandwidth estimates, and byte/packet counters. - `await .status() → ConnectionStatus`, `.connects()`, `.failure()`. Watch reconnects; `connects()` counts connections, 1 on the first, and `failure()` says why a reconnect is under way. Before the first connect, `Client.failure()` says why the dial is retrying. - `.bandwidth() → Bandwidth`. Divide the send estimate between encoders and app-owned tracks. + - `.auth() → Auth`. On `moq-lite-07-wip` and on moq-transport draft-17+ with the MoQ Auth extension, `.grant()` reads the peer's `Grant` and `await .add(token) → AuthToken` presents a fresh token without reconnecting. ### Publishing diff --git a/py/moq-rs/docs/index.md b/py/moq-rs/docs/index.md index 94050e1c24c..64e1347d2dd 100644 --- a/py/moq-rs/docs/index.md +++ b/py/moq-rs/docs/index.md @@ -40,6 +40,9 @@ asyncio.run(main()) Session Request Transport + Auth + AuthToken + Grant ``` ## Publishing diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index f564c37f84c..1fb96899186 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -41,7 +41,7 @@ TrackRequest, ) from .server import Request, Server, Transport -from .session import Bandwidth, Reservation, Session +from .session import Auth, AuthToken, Bandwidth, Reservation, Session from .subscribe import ( BroadcastConsumer, GroupConsumer, @@ -59,6 +59,7 @@ ErrorScope, FetchGroupOptions, Frame, + Grant, Order, ProtocolError, ProtocolKind, @@ -87,6 +88,8 @@ "AudioCodec", "AudioEncoderInput", "AudioSampleFormat", + "Auth", + "AuthToken", "Backoff", "Bandwidth", "BroadcastConsumer", @@ -104,6 +107,7 @@ "ProtocolError", "ProtocolKind", "FetchGroupOptions", + "Grant", "GroupConsumer", "GroupDemand", "GroupRequest", diff --git a/py/moq-rs/moq/_records.py b/py/moq-rs/moq/_records.py index 2b22b3a03f5..05e5342d789 100644 --- a/py/moq-rs/moq/_records.py +++ b/py/moq-rs/moq/_records.py @@ -16,6 +16,7 @@ MoqConnectionStats, MoqDatagram, MoqFrame, + MoqGrant, MoqOrder, MoqRange, MoqSubscription, @@ -211,3 +212,28 @@ def _from_ffi(stats: MoqConnectionStats) -> ConnectionStats: packets_lost=stats.packets_lost, version=stats.version, ) + + +@dataclass(frozen=True) +class Grant: + """What the peer lets this side do, in this side's own paths. + + An empty pattern list grants nothing in that direction; ``**`` grants everything. + """ + + publish: tuple[str, ...] + """The broadcast patterns this side may publish to the peer.""" + subscribe: tuple[str, ...] + """The broadcast patterns this side may subscribe to from the peer.""" + expires: timedelta | None = None + """How long until the grant lapses, measured when it was read, or ``None`` for never.""" + + @staticmethod + def _from_ffi(grant: MoqGrant | None) -> Grant | None: + if grant is None: + return None + return Grant( + publish=tuple(grant.publish), + subscribe=tuple(grant.subscribe), + expires=_opt_from_us(grant.expires_us), + ) diff --git a/py/moq-rs/moq/session.py b/py/moq-rs/moq/session.py index 465e759b047..c195b72af4b 100644 --- a/py/moq-rs/moq/session.py +++ b/py/moq-rs/moq/session.py @@ -2,11 +2,11 @@ from __future__ import annotations -from moq_ffi import MoqBandwidth, MoqBandwidthConsumer, MoqError, MoqReservation, MoqSession +from moq_ffi import MoqAuth, MoqAuthToken, MoqBandwidth, MoqBandwidthConsumer, MoqError, MoqReservation, MoqSession from .origin import OriginConsumer, OriginProducer from .publish import TrackDemand -from .types import ConnectionStats, ConnectionStatus +from .types import ConnectionStats, ConnectionStatus, Grant class Session: @@ -107,6 +107,14 @@ def stats(self) -> ConnectionStats: Individual fields are ``None`` when the transport backend doesn't report them.""" return ConnectionStats._from_ffi(self._inner.stats()) + def auth(self) -> Auth: + """The session's authorization: the grant its peer gave it, and more tokens to present. + + Each call returns an independent watcher of the grant. A client session's + handle survives reconnects, presenting each added token again on every new + connection.""" + return Auth(self._inner.auth()) + def recv_bandwidth(self) -> Bandwidth.Consumer | None: """Hold an owned receive-estimate consumer; releasing the last one stops PROBE.""" inner = self._inner.recv_bandwidth() @@ -123,6 +131,61 @@ def bandwidth(self) -> Bandwidth: return Bandwidth(self._inner.bandwidth()) +class Auth: + """A session's authorization, returned by :meth:`Session.auth`. + + AUTH is only on ``moq-lite-07-wip`` and on moq-transport sessions that negotiate + the MoQ Auth extension; elsewhere the grant stays ``None``:: + + auth = session.auth() + print(auth.grant()) + token = await auth.add(fresh_jwt) + """ + + def __init__(self, inner: MoqAuth) -> None: + self._inner = inner + + def grant(self) -> Grant | None: + """The union of every token's grant, or ``None`` until the peer first answers one. + + A client session reads ``None`` while it reconnects.""" + return Grant._from_ffi(self._inner.grant()) + + async def grant_changed(self) -> Grant | None: + """Wait for the grant to differ from the one last reported, or held when this + handle was made, returning it. + + Raises once nothing will change it again, because the session ended.""" + return Grant._from_ffi(await self._inner.grant_changed()) + + async def add(self, token: str) -> AuthToken: + """Present another token, returning once the peer accepts it. + + Its grant joins the union until :meth:`AuthToken.cancel`. The peer's refusal + raises ``moq.Error`` carrying its code; a session that cannot carry another + token raises ``moq.Error.Unsupported``.""" + return AuthToken(await self._inner.add(token)) + + +class AuthToken: + """One token presented with :meth:`Auth.add`, withdrawn by :meth:`cancel`.""" + + def __init__(self, inner: MoqAuthToken) -> None: + self._inner = inner + + def grant(self) -> Grant | None: + """This token's own grant, or ``None`` after it ended.""" + return Grant._from_ffi(self._inner.grant()) + + async def closed(self) -> None: + """Wait for the token to end, raising the peer's revocation or the session's close.""" + await self._inner.closed() + + def cancel(self) -> None: + """Withdraw the token, removing its grant from the union.""" + self._inner.cancel() + + class Bandwidth: """Divides one connection's send estimate among the tracks sharing it. diff --git a/py/moq-rs/moq/types.py b/py/moq-rs/moq/types.py index e9d3a3d237d..262b69d98ee 100644 --- a/py/moq-rs/moq/types.py +++ b/py/moq-rs/moq/types.py @@ -64,7 +64,7 @@ MoqVideoPixelFormat as VideoPixelFormat, ) -from ._records import Backoff, ConnectionStats, Datagram, Frame, Subscription, TrackInfo +from ._records import Backoff, ConnectionStats, Datagram, Frame, Grant, Subscription, TrackInfo __all__ = [ "Anchor", @@ -79,6 +79,7 @@ "ErrorScope", "Frame", "FetchGroupOptions", + "Grant", "Order", "ProtocolError", "ProtocolKind", diff --git a/py/moq-rs/tests/test_server.py b/py/moq-rs/tests/test_server.py index ee8aee91821..c769e4fd6ba 100644 --- a/py/moq-rs/tests/test_server.py +++ b/py/moq-rs/tests/test_server.py @@ -549,3 +549,48 @@ async def drain(reader: moq.TrackConsumer) -> None: await asyncio.gather(*reading, return_exceptions=True) if broadcast is not None: broadcast.close() + + +async def _await_grant(auth: moq.Auth, want: moq.Grant) -> None: + async def follow() -> None: + while auth.grant() != want: + await auth.grant_changed() + + await asyncio.wait_for(follow(), timeout=5.0) + + +async def test_session_auth_reads_the_grant(): + """Both sides read the grant their peer gave them; a peer that verifies no tokens refuses one.""" + version = ["moq-lite-07-wip"] + async with moq.Server("127.0.0.1:0", tls_generate=["localhost"], versions=version) as server: + sessions: list = [] + accepted = asyncio.Event() + + async def accept_loop() -> None: + async for request in server: + sessions.append(await request.accept()) + accepted.set() + + accept_task = asyncio.create_task(accept_loop()) + try: + async with moq.Client( + f"https://{server.local_addr}", tls_verify=False, bind="127.0.0.1:0", versions=version + ) as client: + session = client.session + assert session is not None + everything = moq.Grant(publish=("**",), subscribe=("**",)) + await _await_grant(session.auth(), everything) + await asyncio.wait_for(accepted.wait(), timeout=5.0) + await _await_grant(sessions[0].auth(), everything) + + with pytest.raises(moq.Error.Unsupported): + await asyncio.wait_for(session.auth().add("extra"), timeout=5.0) + assert session.auth().grant() == everything + finally: + accept_task.cancel() + try: + await accept_task + except asyncio.CancelledError: + pass + for session in sessions: + session.cancel(0) diff --git a/quest/m1/auth/README.md b/quest/m1/auth/README.md index 17f6a6024d9..43ad2a025ab 100644 --- a/quest/m1/auth/README.md +++ b/quest/m1/auth/README.md @@ -145,8 +145,6 @@ because `main` assigned 0x3A to NOT_FETCHABLE first. ## Required - [Auth outlives its session](/quest/m1/auth/issuer.md) - one `auth::Handle` spans a connection's sessions, so a route's auth survives a reconnect -- [Bindings](/quest/m1/auth/bindings.md) - grants and tokens reach every - binding through moq-ffi - [Token in band](/quest/m1/auth/token-in-band.md) - extra credentials ride AUTH streams: a session starts on the token its SETUP or URL carried and its AUTH streams add the rest diff --git a/quest/m1/auth/bindings.md b/quest/m1/auth/bindings.md deleted file mode 100644 index c19d3bb7d88..00000000000 --- a/quest/m1/auth/bindings.md +++ /dev/null @@ -1,45 +0,0 @@ -# [M] Bindings - -## Goal - -Every binding can read a session's grant and present more tokens: moq-ffi -exposes it, the generated C and C++ bindings pick it up from moq-ffi, and the -hand-written Python, Swift, Kotlin, Go, and Dart wrappers and their docs carry -it. An OBS user whose token is about to expire mid-stream can be handed a new -one without the plugin reconnecting. - -## Plan - -This quest exposes session AUTH only. Per-request setters and verifier APIs -remain deferred until a consumer needs them (confirmed 2026-10-12); do not -export the experimental Rust request hook through the bindings here. - -- `MoqSession` in `rs/moq-ffi/src/session.rs` gains `auth()` returning an - `Arc` object, the shape `publisher()` uses, with `grant() -> - Option` (a record: publish patterns, subscribe patterns, expiry - as a duration, matching `auth::Grant`) for the union, `async grant_changed() -> MoqGrant` so a caller - can await the next change without polling, and `async add(token: String) -> - Arc` whose object carries that token's own `grant()` and - `closed()` and whose release withdraws it, raising the structured error - moq-ffi already maps on refusal. `MoqClient` sessions reached through the - reconnecting connection use `moq_tokio::Connection::auth()`. -- No new libmoq API: the hand-written C ABI gets no more feature work. The - generated C and C++ bindings get this from moq-ffi, and OBS reaches it - through the generated C++; update `cpp/obs/src` - only if the plugin surfaces a token field. -- Interop: the Python, Go, and C++ interop clients print their grant as the - `auth granted publish=[...] subscribe=[...]` line and join `prints_grant` and - `enforces_grant` in `test/interop/interop.sh`, beside Rust and JS. -- Wrappers: `py/moq-rs/moq/session.py`, `swift/Sources/Moq`, - `kt/.../Flows.kt` (a `Flow` over `grant_changed`), `go/wrapper/session.go` - (context-cancellable like the rest), and `dart/moq/lib/moq.dart`. Kotlin - typealiases pick up the raw methods for free; the flow is the idiomatic add. -- Docs: `doc/lib/{py,swift,kt,go,dart}` each gain a short auth section, and - `doc/lib/rs` documents `Session::auth()`. -- Tests: each wrapper's existing session test reads a grant from a local - relay, adds a second token minted by `moq auth`, sees the union grow, and - releases it; the refusal path surfaces the structured error in each - language. - -Additive. Decided in the 2026-09-30 audit: libmoq is frozen, so the C -surface comes from moq-ffi rather than new `moq_session_auth_*` calls. diff --git a/quest/m1/auth/issuer.md b/quest/m1/auth/issuer.md index 1f4a1af0e06..6acc18cf6f3 100644 --- a/quest/m1/auth/issuer.md +++ b/quest/m1/auth/issuer.md @@ -47,4 +47,4 @@ none. ## Related - [#5695](https://github.com/moq-dev/moq/pull/5695) - adds `Route::auth`, bound to one session's handle -- [Bindings](/quest/m1/auth/bindings.md) - exposes `Connection::auth()` through moq-ffi, so it should follow this shape +- `rs/moq-ffi/src/auth.rs` - wraps both the session and the connection handle behind one `MoqAuth`; collapse it onto this shape diff --git a/quest/m1/auth/token-in-band.md b/quest/m1/auth/token-in-band.md index 3c189cd81d9..e5413ae7086 100644 --- a/quest/m1/auth/token-in-band.md +++ b/quest/m1/auth/token-in-band.md @@ -87,8 +87,3 @@ already carries the full pattern grant. cross-language harness runs with tokens configured. Additive. Wire: a lite-07-wip SETUP parameter. - -## Required - -- [Bindings](/quest/m1/auth/bindings.md) - supplies the client surface the new - token setters sit beside diff --git a/quest/m3/request-token-surfaces.md b/quest/m3/request-token-surfaces.md index a5cc15889ec..2d6f60dace7 100644 --- a/quest/m3/request-token-surfaces.md +++ b/quest/m3/request-token-surfaces.md @@ -18,7 +18,7 @@ or acceptor to mirror. So each surface here exposes that unified shape once Rust has it, and only for the surface a named consumer uses. Start by asking that consumer which surface it needs. Do not build all three. Through moq-ffi, follow the session -AUTH shape that [Bindings](/quest/m1/auth/bindings.md) settles. Update every +AUTH shape of `MoqAuth` in `rs/moq-ffi/src/auth.rs`. Update every wrapper, as the cross-package table requires. Public API: whatever the chosen surface needs to present and accept auths diff --git a/rs/moq-ffi/src/auth.rs b/rs/moq-ffi/src/auth.rs new file mode 100644 index 00000000000..f191e07e853 --- /dev/null +++ b/rs/moq-ffi/src/auth.rs @@ -0,0 +1,238 @@ +//! In-band authorization: what the peer lets a session do, and more tokens to present. + +use std::sync::Arc; + +use crate::error::MoqError; +use crate::ffi::Task; + +/// What the peer lets this side do, in this side's own paths. +/// +/// An empty pattern list grants nothing in that direction; `**` grants everything. +#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)] +pub struct MoqGrant { + /// The broadcast patterns this side may publish to the peer. + pub publish: Vec, + /// The broadcast patterns this side may subscribe to from the peer. + pub subscribe: Vec, + /// Microseconds until the grant lapses, measured when it was read, or `None` for never. + pub expires_us: Option, +} + +impl From for MoqGrant { + fn from(grant: moq_net::auth::Grant) -> Self { + let patterns = |patterns: &moq_net::Patterns| patterns.iter().map(ToString::to_string).collect(); + Self { + publish: patterns(&grant.publish), + subscribe: patterns(&grant.subscribe), + expires_us: grant.expires.map(|expires| { + let remaining = expires.saturating_duration_since(moq_net::time::now()); + u64::try_from(remaining.as_micros()).unwrap_or(u64::MAX) + }), + } + } +} + +/// Where a session's tokens live: on the session itself, or on a connection that presents +/// them again on every reconnect. +#[derive(Clone)] +pub(crate) enum Handle { + Session(moq_net::auth::Handle), + #[cfg(not(target_arch = "wasm32"))] + Connection(moq_tokio::connection::auth::Handle), +} + +impl Handle { + fn grant(&self) -> Watch { + match self { + Self::Session(handle) => Watch::Session(handle.grant()), + #[cfg(not(target_arch = "wasm32"))] + Self::Connection(handle) => Watch::Connection(handle.grant()), + } + } + + async fn add(self, token: String) -> Result { + match self { + Self::Session(handle) => { + let auth = handle.add(token).await?; + // A token the peer accepted carries a grant until it ends. Without AUTH, + // moq-transport accepts one at once for the requests that name it, but nothing + // here can name one, so it would never be sent. One already revoked is left for + // `closed` to report. moq-tokio refuses the same for a client's connection. + if auth.grant().peek().is_none() && poll_now(auth.closed()).is_pending() { + return Err(MoqError::Unsupported); + } + Ok(Token::Session(auth)) + } + #[cfg(not(target_arch = "wasm32"))] + Self::Connection(handle) => Ok(Token::Connection( + handle.add(token).await.map_err(crate::session::map_closed_error)?, + )), + } + } +} + +/// Poll `future` once without waiting. +fn poll_now(future: F) -> std::task::Poll { + let future = std::pin::pin!(future); + std::future::Future::poll(future, &mut std::task::Context::from_waker(std::task::Waker::noop())) +} + +enum Watch { + Session(moq_net::auth::Watch), + #[cfg(not(target_arch = "wasm32"))] + Connection(moq_tokio::connection::auth::Watch), +} + +impl Watch { + fn peek(&self) -> Option { + match self { + Self::Session(watch) => watch.peek(), + #[cfg(not(target_arch = "wasm32"))] + Self::Connection(watch) => watch.peek(), + } + .map(Into::into) + } + + async fn changed(&mut self) -> Result, MoqError> { + let grant = match self { + Self::Session(watch) => watch.changed().await?, + #[cfg(not(target_arch = "wasm32"))] + Self::Connection(watch) => watch.changed().await.map_err(crate::session::map_closed_error)?, + }; + Ok(grant.map(Into::into)) + } +} + +enum Token { + Session(moq_net::auth::Auth), + #[cfg(not(target_arch = "wasm32"))] + Connection(moq_tokio::connection::auth::Auth), +} + +impl Token { + fn grant(&self) -> Watch { + match self { + Self::Session(auth) => Watch::Session(auth.grant()), + #[cfg(not(target_arch = "wasm32"))] + Self::Connection(auth) => Watch::Connection(auth.grant()), + } + } + + async fn closed(&self) -> Result<(), MoqError> { + match self { + Self::Session(auth) => match auth.closed().await { + moq_net::Error::Cancel => Ok(()), + err => Err(err.into()), + }, + #[cfg(not(target_arch = "wasm32"))] + Self::Connection(auth) => auth.closed().await.map_err(crate::session::map_closed_error), + } + } +} + +/// A session's authorization: the grant its peer gave it, and more tokens to present. +/// +/// Returned by [`MoqSession::auth`](crate::session::MoqSession::auth). AUTH is only on +/// `moq-lite-07-wip` and on moq-transport sessions that negotiate the MoQ Auth extension; +/// elsewhere the grant stays `None`. +#[derive(uniffi::Object)] +pub struct MoqAuth { + handle: Handle, + /// The watch behind [`grant`](Self::grant), which reads without the lock. + peek: Watch, + /// The watch behind [`grant_changed`](Self::grant_changed), which tracks what this + /// handle last reported. Concurrent callers share it, so each watcher takes its own handle. + changed: Task, +} + +impl MoqAuth { + pub(crate) fn new(handle: Handle) -> Self { + // Start from the grant as it is now, so a watcher reads that with `grant` and waits + // in `grant_changed` only for what follows, never for the same grant twice. + let mut changed = handle.grant(); + let _ = poll_now(changed.changed()); + Self { + peek: handle.grant(), + changed: Task::new(changed), + handle, + } + } +} + +#[uniffi::export] +impl MoqAuth { + /// The union of every token's grant, or `None` until the peer first answers one. + /// + /// A client session reads `None` while it reconnects. + pub fn grant(&self) -> Option { + let _guard = crate::ffi::enter(); + self.peek.peek() + } + + /// Wait for the grant to differ from the one this handle last reported, or held when it was + /// made, returning it. + /// + /// Calls on one handle share what it last reported; an independent watcher takes its own + /// handle from [`MoqSession::auth`](crate::session::MoqSession::auth). + /// + /// Errors once nothing will change it again, because the session ended. + pub async fn grant_changed(&self) -> Result, MoqError> { + self.changed.run(|mut watch| async move { watch.changed().await }).await + } + + /// Present another token, resolving once the peer accepts it. + /// + /// Its grant joins the union until [`MoqAuthToken::cancel`]. A client session presents it + /// again on every reconnect. The peer's refusal is a [`MoqError::Protocol`] carrying its code. + /// + /// # Errors + /// + /// [`MoqError::Unsupported`] when the session cannot carry another token: a version + /// without AUTH, or moq-transport without the MoQ Auth extension. + pub async fn add(&self, token: String) -> Result, MoqError> { + let handle = self.handle.clone(); + let token = crate::ffi::detached(handle.add(token)).await?; + Ok(Arc::new(MoqAuthToken::new(token))) + } +} + +/// One token presented with [`MoqAuth::add`]. +/// +/// It stays presented until [`cancel`](Self::cancel), or until the handle is released. +#[derive(uniffi::Object)] +pub struct MoqAuthToken { + /// The watch behind [`grant`](Self::grant), which does not keep the token presented. + peek: Watch, + token: Task, +} + +impl MoqAuthToken { + fn new(token: Token) -> Self { + Self { + peek: token.grant(), + token: Task::new(token), + } + } +} + +#[uniffi::export] +impl MoqAuthToken { + /// This token's own grant, or `None` while the peer has not answered it or after it ended. + pub fn grant(&self) -> Option { + let _guard = crate::ffi::enter(); + self.peek.peek() + } + + /// Wait for the token to end: `Ok` when the peer closed it without an error, the peer's + /// revocation as [`MoqError::Protocol`], or the session's close. + pub async fn closed(&self) -> Result<(), MoqError> { + self.token.run(|token| async move { token.closed().await }).await + } + + /// Withdraw the token, removing its grant from the union. + /// + /// Terminal: the token is withdrawn here, not when the handle is. + pub fn cancel(&self) { + self.token.cancel(); + } +} diff --git a/rs/moq-ffi/src/lib.rs b/rs/moq-ffi/src/lib.rs index 5b741a8efce..fbe7a3906b7 100644 --- a/rs/moq-ffi/src/lib.rs +++ b/rs/moq-ffi/src/lib.rs @@ -18,6 +18,7 @@ mod android; // since they are the heaviest thing in the dependency graph. #[cfg(all(feature = "audio", not(target_arch = "wasm32")))] pub mod audio; +pub mod auth; pub mod bandwidth; pub mod consumer; pub mod demand; diff --git a/rs/moq-ffi/src/session.rs b/rs/moq-ffi/src/session.rs index 4e7175c1479..0f2c9048aee 100644 --- a/rs/moq-ffi/src/session.rs +++ b/rs/moq-ffi/src/session.rs @@ -2,6 +2,7 @@ use std::sync::Arc; use url::Url; +use crate::auth::MoqAuth; use crate::bandwidth::MoqBandwidth; use crate::error::MoqError; use crate::ffi::Task; @@ -1149,6 +1150,20 @@ impl MoqSession { self.bandwidth.clone() } + /// The session's authorization: the grant its peer gave it, and more tokens to present. + /// + /// Every call returns a new handle sharing the session's tokens, which tracks its own + /// [`grant_changed`](MoqAuth::grant_changed), so concurrent watchers each take one. A client + /// session's handle survives reconnects, presenting each added token again on every new + /// connection. + pub fn auth(&self) -> Arc { + Arc::new(MoqAuth::new(match &self.inner { + #[cfg(not(target_arch = "wasm32"))] + Inner::Connection(connection) => crate::auth::Handle::Connection(connection.auth()), + Inner::Session(session) => crate::auth::Handle::Session(session.auth()), + })) + } + /// Hold receive-estimate demand until the returned consumer is released. pub fn recv_bandwidth(&self) -> Option> { let consumer = match &self.inner { diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index f64b91adc8e..8e11daeb23b 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -6661,10 +6661,15 @@ async fn data_track_names_cannot_collide() { /// Real sockets exercise the FFI runtime and both accepted/client session handles. /// Only lite-07 acknowledges the received tail; default lite-06 closes after transport ACKs. async fn shutdown_pair() -> (Arc, Arc, Arc) { + session_pair("moq-lite-07-wip").await +} + +/// A server, a client session connected to it on `version`, and the session it accepted. +async fn session_pair(version: &str) -> (Arc, Arc, Arc) { let server = MoqServer::new(MoqServerConfig { bind: Some("127.0.0.1:0".into()), tls: localhost_tls(), - versions: vec!["moq-lite-07-wip".into()], + versions: vec![version.into()], ..Default::default() }) .unwrap(); @@ -6682,7 +6687,7 @@ async fn shutdown_pair() -> (Arc, Arc, Arc) { }); let client = MoqClient::new(MoqClientConfig { tls: insecure_tls(), - versions: vec!["moq-lite-07-wip".into()], + versions: vec![version.into()], bind: Some("127.0.0.1:0".into()), once: true, ..Default::default() @@ -6852,6 +6857,192 @@ async fn ascending_subscription_reads_a_cached_backlog_oldest_first() { assert_eq!(sequences, vec![0, 1, 2]); } +/// Wait until `auth` reports `want`, following its changes. +async fn await_grant(auth: &crate::auth::MoqAuth, want: crate::auth::MoqGrant) { + tokio::time::timeout(TIMEOUT, async { + while auth.grant().as_ref() != Some(&want) { + auth.grant_changed().await.expect("grant watch ended"); + } + }) + .await + .unwrap_or_else(|_| panic!("timed out waiting for {want:?}, last {:?}", auth.grant())); +} + +fn grant(publish: &[&str], subscribe: &[&str]) -> crate::auth::MoqGrant { + let patterns = |patterns: &[&str]| patterns.iter().map(|p| p.to_string()).collect(); + crate::auth::MoqGrant { + publish: patterns(publish), + subscribe: patterns(subscribe), + expires_us: None, + } +} + +/// Both sides of a session read the grant their peer gave them, on lite-07 and on moq-transport +/// with the Auth extension, and a peer that verifies no tokens of its own refuses another one as +/// unsupported. +#[tokio::test] +async fn auth_reads_the_peer_grant() { + for version in ["moq-lite-07-wip", "moq-transport-17"] { + auth_reads_the_peer_grant_on(version).await; + } +} + +async fn auth_reads_the_peer_grant_on(version: &str) { + let (server, client, accepted) = tokio::time::timeout(TIMEOUT, session_pair(version)) + .await + .expect("pair"); + + // Each side's origins span the root, so its peer grants it everything. + await_grant(&client.auth(), grant(&["**"], &["**"])).await; + await_grant(&accepted.auth(), grant(&["**"], &["**"])).await; + + let err = tokio::time::timeout(TIMEOUT, client.auth().add("extra".into())) + .await + .expect("add timed out") + .err() + .expect("the default acceptor refuses a token"); + assert!(matches!(err, MoqError::Unsupported), "refusal: {err:?}"); + assert_eq!( + client.auth().grant(), + Some(grant(&["**"], &["**"])), + "a refusal leaves the union" + ); + + client.cancel(0); + accepted.cancel(0); + server.cancel(); +} + +/// moq-transport without the Auth extension carries a token only on a request that names it, +/// which no binding can, so both sides refuse one rather than report a token nothing sends. +#[tokio::test] +async fn auth_refuses_a_token_nothing_carries() { + let (server, client, accepted) = tokio::time::timeout(TIMEOUT, session_pair("moq-transport-14")) + .await + .expect("pair"); + for session in [&client, &accepted] { + let err = tokio::time::timeout(TIMEOUT, session.auth().add("extra".into())) + .await + .expect("add timed out") + .err() + .expect("a token nothing carries is refused"); + assert!(matches!(err, MoqError::Unsupported), "refusal: {err:?}"); + assert_eq!(session.auth().grant(), None, "no AUTH, no grant"); + } + + client.cancel(0); + accepted.cancel(0); + server.cancel(); +} + +/// A token added through a client session joins the union with its own grant, leaves it on +/// cancel, and a refusal or revocation carries the peer's code. +#[tokio::test] +async fn auth_tokens_join_and_leave_the_union() { + let version: moq_net::Version = "moq-lite-07-wip".parse().unwrap(); + let mut config = moq_tokio::listen::Config::default(); + config.tcp.bind = Some("127.0.0.1:0".parse().unwrap()); + config.version = vec![version]; + let mut listener = config.init(Default::default()).unwrap().listen().await.unwrap(); + let addr = listener.tcp_local_addr().unwrap(); + + let publish = |pattern: &str| moq_net::auth::Grant { + publish: pattern.parse::().unwrap().into(), + ..Default::default() + }; + let serving = tokio::spawn(async move { + let request = listener.accept().await.expect("accept"); + let mut requests = request.auth().requests().unwrap(); + let (session, driver) = request.ok().await.unwrap(); + let mut issued = Vec::new(); + while let Some(request) = requests.next().await { + match request.token().as_ref() { + b"" => issued.push(request.accept(publish("setup/**"))), + b"extra" => issued.push(request.accept(publish("extra/**"))), + b"revoked" => request + .accept(publish("revoked/**")) + .revoke(moq_net::SessionError::Expired, "revoked"), + // Dropping the issued grant closes the token without an error. + b"closed" => drop(request.accept(publish("closed/**"))), + _ => request.reject(moq_net::SessionError::Unauthorized, "unknown token"), + } + } + drop((session, driver, issued)); + }); + + let client = MoqClient::new(MoqClientConfig { + versions: vec!["moq-lite-07-wip".into()], + bind: Some("127.0.0.1:0".into()), + once: true, + ..Default::default() + }) + .unwrap(); + let session = tokio::time::timeout(TIMEOUT, client.connect(format!("tcp://{addr}"))) + .await + .expect("connect timed out") + .unwrap(); + let auth = session.auth(); + await_grant(&auth, grant(&["setup/**"], &[])).await; + // A second handle watches on its own and starts from the grant it was made with. + let watcher = session.auth(); + + let extra = tokio::time::timeout(TIMEOUT, auth.add("extra".into())) + .await + .expect("add timed out") + .unwrap(); + assert_eq!(extra.grant(), Some(grant(&["extra/**"], &[]))); + // Patterns are kept sorted. + await_grant(&auth, grant(&["extra/**", "setup/**"], &[])).await; + let seen = tokio::time::timeout(TIMEOUT, watcher.grant_changed()) + .await + .expect("the second handle sees the change the first took") + .unwrap(); + assert_eq!(seen, Some(grant(&["extra/**", "setup/**"], &[]))); + + extra.cancel(); + await_grant(&auth, grant(&["setup/**"], &[])).await; + assert!(matches!(extra.closed().await, Err(MoqError::Cancelled))); + + let refused = tokio::time::timeout(TIMEOUT, auth.add("bogus".into())) + .await + .expect("add timed out") + .err() + .expect("refused"); + assert_protocol( + &refused, + crate::error::MoqErrorScope::Session, + MoqProtocolKind::Unauthorized, + ); + + let revoked = tokio::time::timeout(TIMEOUT, auth.add("revoked".into())) + .await + .expect("add timed out") + .unwrap(); + let closed = tokio::time::timeout(TIMEOUT, revoked.closed()) + .await + .expect("closed timed out") + .unwrap_err(); + assert_protocol(&closed, crate::error::MoqErrorScope::Session, MoqProtocolKind::Expired); + assert_eq!(revoked.grant(), None, "an ended token grants nothing"); + assert_eq!(auth.grant(), Some(grant(&["setup/**"], &[]))); + + let closed = tokio::time::timeout(TIMEOUT, auth.add("closed".into())) + .await + .expect("add timed out") + .unwrap(); + tokio::time::timeout(TIMEOUT, closed.closed()) + .await + .expect("closed timed out") + .expect("a clean close is not an error"); + assert_eq!(auth.grant(), Some(grant(&["setup/**"], &[]))); + + session.cancel(0); + tokio::time::timeout(TIMEOUT, serving) + .await + .expect("server ended") + .unwrap(); +} + /// A time anchor starts at the group containing it, and an untimed track refuses one. #[tokio::test] async fn time_anchors_start_at_the_containing_group() { diff --git a/rs/moq-tokio/src/connection/auth.rs b/rs/moq-tokio/src/connection/auth.rs index f6e5a194b24..ea686213d3a 100644 --- a/rs/moq-tokio/src/connection/auth.rs +++ b/rs/moq-tokio/src/connection/auth.rs @@ -29,6 +29,8 @@ impl Handle { } /// Present a token now and on later sessions until its handle is dropped or refused. + /// + /// A session that cannot send it, without AUTH, refuses it as unsupported. pub async fn add(&self, token: impl Into) -> crate::Result { let producer = kio::Producer::::default(); let state = producer.consume(); @@ -158,7 +160,8 @@ impl Auth { } } - /// Wait until the token is refused or its connection ends. + /// Wait until the token ends: `Ok` when the peer closed it without an error, otherwise its + /// refusal or revocation, or the connection's end. pub async fn closed(&self) -> crate::Result<()> { kio::wait(|waiter| self.state.poll_closed(waiter)).await; match &self.state.read().error { @@ -352,13 +355,20 @@ async fn present( Err(err) => return Err(err.into()), } }; - if let Some(answered) = answered.take() { - let _ = answered.send(Ok(())); - } let mut watch = token.grant(); + // Without AUTH, moq-transport accepts a token at once for the requests that name it, + // but a connection's token rides no request, so nothing would send it. One the peer + // already ended has a grant no longer and is left to the watch below. + if watch.peek().is_none() && token.closed().now_or_never().is_none() { + return Err(moq_net::Error::Unsupported.into()); + } if let Ok(mut state) = producer.write() { state.grant = watch.peek(); } + // After the grant, so `add` never returns a token that still reads none. + if let Some(answered) = answered.take() { + let _ = answered.send(Ok(())); + } loop { tokio::select! { biased; @@ -378,6 +388,8 @@ async fn present( snapshot = next(&connection, Some((connects, true))).await?; break; } + // The peer closed it without an error: the token is over, not refused. + Err(moq_net::Error::Cancel) => return Ok(()), Err(err) => return Err(err.into()), } } diff --git a/swift/Sources/Moq/Auth.swift b/swift/Sources/Moq/Auth.swift new file mode 100644 index 00000000000..6bca59e7aef --- /dev/null +++ b/swift/Sources/Moq/Auth.swift @@ -0,0 +1,52 @@ +import MoqFFI + +/// What the peer lets this side do, in this side's own paths: the broadcast +/// patterns it may publish and subscribe to, and `expiresUs` until it lapses. +/// An empty pattern list grants nothing in that direction; `**` grants everything. +public typealias Grant = MoqFFI.MoqGrant + +/// A session's authorization: the grant its peer gave it, and more tokens to present. +/// +/// Returned by `Session.auth()`. AUTH is only on `moq-lite-07-wip` and on +/// moq-transport sessions that negotiate the MoQ Auth extension; elsewhere the +/// grant stays `nil`. +public final class Auth: Sendable { + let ffi: MoqAuth + + init(_ ffi: MoqAuth) { self.ffi = ffi } + + /// The union of every token's grant, or `nil` until the peer first answers one. + /// A client session reads `nil` while it reconnects. + public func grant() -> Grant? { ffi.grant() } + + /// Suspend until the grant differs from the one last reported, or held when this + /// handle was made, returning it. + /// Throws once nothing will change it again, because the session ended. + public func grantChanged() async throws -> Grant? { try await ffi.grantChanged() } + + /// Present another token, returning once the peer accepts it. + /// + /// Its grant joins the union until `AuthToken.cancel()`; a client session presents + /// it again on every reconnect. The peer's refusal throws `MoqError.Protocol` + /// carrying its code, and a session that cannot carry another token throws + /// `MoqError.Unsupported`. + public func add(token: String) async throws -> AuthToken { + AuthToken(try await ffi.add(token: token)) + } +} + +/// One token presented with `Auth.add(token:)`, withdrawn by `cancel()`. +public final class AuthToken: Sendable { + let ffi: MoqAuthToken + + init(_ ffi: MoqAuthToken) { self.ffi = ffi } + + /// This token's own grant, or `nil` after it ended. + public func grant() -> Grant? { ffi.grant() } + + /// Suspend until the token ends, throwing the peer's revocation or the session's close. + public func closed() async throws { try await ffi.closed() } + + /// Withdraw the token, removing its grant from the union. + public func cancel() { ffi.cancel() } +} diff --git a/swift/Sources/Moq/Client.swift b/swift/Sources/Moq/Client.swift index 736e98b915a..4f0bb638118 100644 --- a/swift/Sources/Moq/Client.swift +++ b/swift/Sources/Moq/Client.swift @@ -166,6 +166,15 @@ public final class Session: Sendable { ffi.stats() } + /// The session's authorization: the grant its peer gave it, and more tokens to present. + /// + /// Each call returns an independent watcher of the grant. A client session's + /// handle survives reconnects, presenting each added token again on every new + /// connection. + public func auth() -> Auth { + Auth(ffi.auth()) + } + /// Hold receive-estimate demand until the returned consumer is released. public func recvBandwidth() -> Bandwidth.Consumer? { ffi.recvBandwidth().map(Bandwidth.Consumer.init) diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index d34f06f0f51..c2c06147bc7 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -101,6 +101,48 @@ final class SmokeTests: XCTestCase { } } + /// Both sides read the grant their peer gave them, and a peer that verifies no + /// tokens of its own refuses another one as unsupported. + func testSessionAuthReadsTheGrant() async throws { + let version = ["moq-lite-07-wip"] + let server = try Server(bind: "127.0.0.1:0", versions: version, tls: ServerTls(generate: ["localhost"])) + defer { server.cancel() } + let addr = try await server.listen() + async let accepted = server.accept() + + let client = try Client(bind: "127.0.0.1:0", versions: version, tls: ClientTls(insecure: true), once: true) + defer { client.cancel() } + let session = try await client.connect(to: "https://\(addr)") + guard let request = try await accepted else { return XCTFail("server closed before accept") } + let peer = try await request.accept() + + // Ending both sessions ends any wait below, so a grant that never arrives fails + // rather than hangs. + let watchdog = Task { + try await Task.sleep(nanoseconds: 10_000_000_000) + peer.cancel(code: 0) + session.cancel(code: 0) + } + defer { watchdog.cancel() } + + let everything = Grant(publish: ["**"], subscribe: ["**"], expiresUs: nil) + for auth in [session.auth(), peer.auth()] { + var grant = auth.grant() + while grant != everything { + grant = try await auth.grantChanged() + } + } + + do { + _ = try await session.auth().add(token: "extra") + XCTFail("the default acceptor refuses a token") + } catch MoqError.Unsupported {} + XCTAssertEqual(session.auth().grant(), everything) + + peer.cancel(code: 0) + session.cancel(code: 0) + } + /// `cancel()` releases the listening socket before it returns, so the same /// address binds again with no retry. func testServerCloseReleasesPort() async throws { diff --git a/test/interop/README.md b/test/interop/README.md index bd8c4aa7a8d..df8d6a65a4a 100644 --- a/test/interop/README.md +++ b/test/interop/README.md @@ -104,19 +104,20 @@ subtree and each subscriber with one to read it, so every cell also covers the Clients that print the grant the relay sent back over AUTH, as an `auth granted publish=[...] subscribe=[...]` line, must report exactly what -their token implies: the Rust CLI (a `moq_net::auth` debug log) and the native +their token implies: the Rust CLI (a `moq_net::auth` debug log), the Python, +Go, and C++ clients (read through moq-ffi before they start), and the native JS subscribers. A cell whose grant is missing or wrong fails even when media -flowed. The binding clients (Python, Go, C, GStreamer) have no grant to print -until moq-ffi exposes one, and the browser's shared connection keeps its session -private, so their cells check media alone. AUTH is only on the work-in-progress +flowed. moq-c's frozen ABI and GStreamer expose no grant, and the browser's +shared connection keeps its session private, so their cells check media alone. AUTH is only on the work-in-progress moq-lite-07, so the printing clients dial `moq-lite-07-wip` alone and a printing client that reports nothing never got its grant. The native JS subscribers dial WebTransport alone and offer it there. The rest keep their defaults, so the matrix also crosses versions through the relay, which accepts both. -After the matrix, each publisher whose refusal the harness can read (Rust) runs -once more with a token that excludes its broadcast. It must fail loud, logging +After the matrix, each publisher whose refusal the harness can read (Rust, +Python, Go, and C++, which log moq-net's error) runs once more with a token that +excludes its broadcast. It must fail loud, logging Unauthorized and naming the path, and every subscriber must time out. The browser publisher enforces its grant too, but its elements cannot offer `moq-lite-07-wip` yet. diff --git a/test/interop/clients/cpp/main.cpp b/test/interop/clients/cpp/main.cpp index 29083c62132..2dae8bf83b0 100644 --- a/test/interop/clients/cpp/main.cpp +++ b/test/interop/clients/cpp/main.cpp @@ -9,6 +9,9 @@ // // ffmpeg ... -f h264 - | cpp-interop publish --url http://localhost:4443 --broadcast b.hang // cpp-interop subscribe --url http://localhost:4443 --broadcast b.hang --timeout 20 +// +// --auth-version offers only that version, one that carries AUTH: the client waits for the +// grant the relay sends and prints it as an `auth granted` line before it starts. #include @@ -99,12 +102,38 @@ T until(moq::Future future, Clock::time_point deadline, const char *what) { return ok(future.get(), what); } -// The session owns its connection; the client that made it can go. -std::shared_ptr connect(const std::string &url, Clock::time_point deadline) { +// A JSON list of strings, the shape the Rust client logs its grant's patterns in. +std::string patterns(const std::vector &list) { + std::string out = "["; + for (size_t i = 0; i < list.size(); i++) { + out += (i == 0 ? "\"" : ", \"") + list[i] + "\""; + } + return out + "]"; +} + +// The session owns its connection; the client that made it can go. With an `auth_version`, +// waits for the relay's grant and prints it in the Rust client's `auth granted` shape. +std::shared_ptr connect( + const std::string &url, const std::string &auth_version, Clock::time_point deadline +) { moq::ClientConfig config; config.tls.insecure = true; + if (!auth_version.empty()) { + config.versions = {auth_version}; + } auto client = ok(moq::Client::init(config), "client init"); open_session = until(client->connect(url), deadline, "connect"); + if (!auth_version.empty()) { + auto auth = open_session->auth(); + auto grant = auth->grant(); + while (!grant) { + grant = until(auth->grant_changed(), deadline, "grant"); + } + std::fprintf( + stderr, "auth granted publish=%s subscribe=%s\n", patterns(grant->publish).c_str(), + patterns(grant->subscribe).c_str() + ); + } return open_session; } @@ -135,10 +164,10 @@ void publish_tone(const std::shared_ptr &audio, const std::a } } -int publish(const std::string &url, const std::string &path) { +int publish(const std::string &url, const std::string &path, const std::string &auth_version) { // No deadline to publish: the harness stops the publisher when the round ends. const auto forever = Clock::now() + std::chrono::hours(24); - auto session = connect(url, forever); + auto session = connect(url, auth_version, forever); // Hold the producer for the lifetime of the publish loop; close() unpublishes. auto broadcast = ok(session->publish()->create_broadcast(path), "create_broadcast"); @@ -189,10 +218,10 @@ int publish(const std::string &url, const std::string &path) { return 0; } -int subscribe(const std::string &url, const std::string &path, double timeout) { +int subscribe(const std::string &url, const std::string &path, const std::string &auth_version, double timeout) { const auto deadline = Clock::now() + std::chrono::duration_cast(std::chrono::duration(timeout)); - auto session = connect(url, deadline); + auto session = connect(url, auth_version, deadline); auto announced = ok(session->consume()->announced_broadcast(path), "announced_broadcast"); auto consumer = until(announced->available(), deadline, "available"); @@ -238,11 +267,12 @@ int subscribe(const std::string &url, const std::string &path, double timeout) { int main(int argc, char **argv) { if (argc < 2) { - fail("usage: cpp-interop publish|subscribe --url URL --broadcast PATH [--timeout SECONDS]"); + fail("usage: cpp-interop publish|subscribe --url URL --broadcast PATH [--timeout SECONDS] [--auth-version V]"); } const std::string role = argv[1]; std::string url; std::string broadcast; + std::string auth_version; double timeout = 20; for (int i = 2; i + 1 < argc; i += 2) { const std::string flag = argv[i]; @@ -252,6 +282,8 @@ int main(int argc, char **argv) { broadcast = argv[i + 1]; } else if (flag == "--timeout") { timeout = std::atof(argv[i + 1]); + } else if (flag == "--auth-version") { + auth_version = argv[i + 1]; } else { fail(("unknown flag " + flag).c_str()); } @@ -259,12 +291,14 @@ int main(int argc, char **argv) { if (url.empty() || broadcast.empty()) { fail("--url and --broadcast are required"); } + // Warnings and errors from the Rust layer, such as a publication outside the grant. + ok(moq::moq_log_level("warn"), "log level"); if (role == "publish") { - return publish(url, broadcast); + return publish(url, broadcast, auth_version); } if (role == "subscribe") { - return subscribe(url, broadcast, timeout); + return subscribe(url, broadcast, auth_version, timeout); } fail(("unknown role " + role).c_str()); } diff --git a/test/interop/clients/go/main.go b/test/interop/clients/go/main.go index 1763efd09df..291fb6e686a 100644 --- a/test/interop/clients/go/main.go +++ b/test/interop/clients/go/main.go @@ -10,11 +10,16 @@ // // ffmpeg ... -f h264 - | go-interop publish --url http://localhost:4443 --broadcast b.hang // go-interop subscribe --url http://localhost:4443 --broadcast b.hang --timeout 20 +// +// --auth-version offers only that version, one that carries AUTH: the client waits +// for the grant the relay sends and prints it as an `auth granted` line before it +// starts. package main import ( "context" "encoding/binary" + "encoding/json" "errors" "flag" "fmt" @@ -74,8 +79,43 @@ func publishTone(ctx context.Context, audio *moqaudio.Producer) { } } -func publish(ctx context.Context, url, broadcast string) error { - client, err := moq.Dial(ctx, url, moq.WithTLSVerify(false)) +// dial connects to url. With an authVersion, it offers only that version and waits +// for the relay's grant, printing it in the Rust client's `auth granted` shape. +func dial(ctx context.Context, url, authVersion string, timeout time.Duration) (*moq.Client, error) { + opts := []moq.ClientOption{moq.WithTLSVerify(false)} + if authVersion != "" { + opts = append(opts, moq.WithVersions(authVersion)) + } + client, err := moq.Dial(ctx, url, opts...) + if err != nil || authVersion == "" { + return client, err + } + + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + auth := client.Session().Auth() + grant := auth.Grant() + for grant == nil { + if grant, err = auth.GrantChanged(ctx); err != nil { + client.Close() + return nil, fmt.Errorf("grant: %w", err) + } + } + fmt.Fprintf(os.Stderr, "auth granted publish=%s subscribe=%s\n", patterns(grant.Publish), patterns(grant.Subscribe)) + return client, nil +} + +// patterns renders a grant's patterns as a JSON list, the shape the Rust client logs. +func patterns(list []string) string { + if list == nil { + list = []string{} + } + out, _ := json.Marshal(list) + return string(out) +} + +func publish(ctx context.Context, url, broadcast, authVersion string, timeout time.Duration) error { + client, err := dial(ctx, url, authVersion, timeout) if err != nil { return err } @@ -176,11 +216,11 @@ func catalogWithVideo(ctx context.Context, consumer *moq.BroadcastConsumer) (*mo } } -func subscribe(ctx context.Context, url, broadcast string, timeout time.Duration) error { +func subscribe(ctx context.Context, url, broadcast, authVersion string, timeout time.Duration) error { ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() - client, err := moq.Dial(ctx, url, moq.WithTLSVerify(false)) + client, err := dial(ctx, url, authVersion, timeout) if err != nil { return err } @@ -234,7 +274,7 @@ func subscribe(ctx context.Context, url, broadcast string, timeout time.Duration func run() error { if len(os.Args) < 2 { - return errors.New("usage: go-interop publish|subscribe --url URL --broadcast PATH [--timeout SECONDS]") + return errors.New("usage: go-interop publish|subscribe --url URL --broadcast PATH [--timeout SECONDS] [--auth-version V]") } role := os.Args[1] @@ -242,6 +282,7 @@ func run() error { url := flags.String("url", "", "relay URL") broadcast := flags.String("broadcast", "", "broadcast path") timeout := flags.Float64("timeout", 20, "seconds to wait for data") + authVersion := flags.String("auth-version", "", "offer only this version, which carries AUTH, and print the grant") if err := flags.Parse(os.Args[2:]); err != nil { return err } @@ -249,12 +290,17 @@ func run() error { return errors.New("--url and --broadcast are required") } + // Warnings and errors from the Rust layer, such as a publication outside the grant. + if err := moq.LogLevel("warn"); err != nil { + return err + } + wait := time.Duration(*timeout * float64(time.Second)) ctx := context.Background() switch role { case "publish": - return publish(ctx, *url, *broadcast) + return publish(ctx, *url, *broadcast, *authVersion, wait) case "subscribe": - return subscribe(ctx, *url, *broadcast, time.Duration(*timeout*float64(time.Second))) + return subscribe(ctx, *url, *broadcast, *authVersion, wait) default: return fmt.Errorf("unknown role %q", role) } diff --git a/test/interop/clients/python/interop.py b/test/interop/clients/python/interop.py index 2433090615d..b8e3ea078cf 100644 --- a/test/interop/clients/python/interop.py +++ b/test/interop/clients/python/interop.py @@ -7,6 +7,9 @@ subscribe: connect, find the video track in the catalog, and exit 0 as soon as any non-empty frame arrives (exit 1 on timeout / no data). +--auth-version offers only that version, one that carries AUTH: the client waits for +the grant the relay sends and prints it as an `auth granted` line before it starts. + ffmpeg ... -f h264 - | python interop.py publish --url http://localhost:4443 --broadcast b.hang python interop.py subscribe --url http://localhost:4443 --broadcast b.hang --timeout 20 """ @@ -14,6 +17,7 @@ import argparse import asyncio import contextlib +import json import math import struct import sys @@ -55,8 +59,33 @@ async def _publish_tone(audio: moq.audio.Producer) -> None: await asyncio.sleep(max(0.0, started + timestamp_us / 1_000_000 - loop.time())) -async def publish(url: str, broadcast: str) -> None: - async with moq.Client(url, tls_verify=False) as client: +def _client(url: str, auth_version: str | None) -> moq.Client: + return moq.Client(url, tls_verify=False, versions=[] if auth_version is None else [auth_version]) + + +async def _print_grant(client: moq.Client, auth_version: str | None, timeout: float) -> None: + """Wait for the relay's grant and print it, in the Rust client's `auth granted` shape.""" + if auth_version is None: + return + session = client.session + assert session is not None + auth = session.auth() + + async def first() -> moq.Grant: + while (grant := auth.grant()) is None: + await auth.grant_changed() + return grant + + grant = await asyncio.wait_for(first(), timeout) + print( + f"auth granted publish={json.dumps(list(grant.publish))} subscribe={json.dumps(list(grant.subscribe))}", + file=sys.stderr, + ) + + +async def publish(url: str, broadcast: str, auth_version: str | None, timeout: float) -> None: + async with _client(url, auth_version) as client: + await _print_grant(client, auth_version, timeout) # Hold the producer for the lifetime of the publish loop; finish() unpublishes. producer = client.create_broadcast(broadcast) catalog = moq.media.CatalogProducer(producer) @@ -108,8 +137,9 @@ async def _catalog_with_video(consumer: moq.BroadcastConsumer) -> moq.media.Cata raise RuntimeError("catalog stream ended without a video track") -async def subscribe(url: str, broadcast: str, timeout: float) -> None: - async with moq.Client(url, tls_verify=False) as client: +async def subscribe(url: str, broadcast: str, auth_version: str | None, timeout: float) -> None: + async with _client(url, auth_version) as client: + await _print_grant(client, auth_version, timeout) consumer = await asyncio.wait_for(client.announced_broadcast(broadcast), timeout) catalog = await asyncio.wait_for(_catalog_with_video(consumer), timeout) @@ -141,13 +171,16 @@ def main() -> None: parser.add_argument("--url", required=True) parser.add_argument("--broadcast", required=True) parser.add_argument("--timeout", type=float, default=20.0) + parser.add_argument("--auth-version") args = parser.parse_args() + # Warnings and errors from the Rust layer, such as a publication outside the grant. + moq.log_level("warn") try: if args.role == "publish": - asyncio.run(publish(args.url, args.broadcast)) + asyncio.run(publish(args.url, args.broadcast, args.auth_version, args.timeout)) else: - asyncio.run(subscribe(args.url, args.broadcast, args.timeout)) + asyncio.run(subscribe(args.url, args.broadcast, args.auth_version, args.timeout)) except KeyboardInterrupt: pass except (TimeoutError, asyncio.TimeoutError): diff --git a/test/interop/interop.sh b/test/interop/interop.sh index c691d545df0..59ecb277ef2 100755 --- a/test/interop/interop.sh +++ b/test/interop/interop.sh @@ -609,12 +609,12 @@ CLI_LOG="${RUST_LOG:-info},moq_net::auth=debug" AUTH_VERSION="${INTEROP_VERSION:-moq-lite-07-wip}" # The clients that print the grant they received as an `auth granted` line. The -# binding clients (python, go, c, gst) have no grant to print until moq-ffi -# exposes one, and the browser's shared connection keeps its session private. +# Python, Go, and C++ clients read it through moq-ffi; moq-c's frozen ABI and +# GStreamer expose none, and the browser's shared connection keeps its session private. prints_grant() { [[ "$COMPAT" -eq 0 ]] || return 1 case "$1" in - rust | js-native-node | js-native-bun) return 0 ;; + rust | python | go | cpp | js-native-node | js-native-bun) return 0 ;; *) return 1 ;; esac } @@ -665,6 +665,9 @@ ffmpeg_h264() { # shellcheck disable=SC2329 # invoked indirectly via 'harness_spawn' run_publisher() { local lang="$1" broadcast="$2" url="$3" + # The binding clients dial $AUTH_VERSION alone and print their grant only when it is checked. + local auth=() + if prints_grant "$lang"; then auth=(--auth-version "$AUTH_VERSION"); fi case "$lang" in rust) # A finished import stays up for its max-age; `stop_publisher` allows 10 s to close. @@ -673,13 +676,13 @@ run_publisher() { ;; python) ffmpeg_h264 | "$PY" "$CLIENTS/python/interop.py" \ - publish --url "$url" --broadcast "$broadcast" + publish --url "$url" --broadcast "$broadcast" ${auth[@]+"${auth[@]}"} ;; go) - ffmpeg_h264 | "$GO_INTEROP" publish --url "$url" --broadcast "$broadcast" + ffmpeg_h264 | "$GO_INTEROP" publish --url "$url" --broadcast "$broadcast" ${auth[@]+"${auth[@]}"} ;; cpp) - ffmpeg_h264 | "$CPP_INTEROP" publish --url "$url" --broadcast "$broadcast" + ffmpeg_h264 | "$CPP_INTEROP" publish --url "$url" --broadcast "$broadcast" ${auth[@]+"${auth[@]}"} ;; js-native) # exec, so the client leads the group and `stop_publisher` waits for its own close. @@ -760,6 +763,8 @@ observe_group() { # shellcheck disable=SC2329 # reached from a function 'harness_spawn' invokes run_subscriber() { local lang="$1" broadcast="$2" url="$3" publisher="${4:-}" + local auth=() + if prints_grant "$lang"; then auth=(--auth-version "$AUTH_VERSION"); fi case "$lang" in rust) # moq-cli only handles SIGINT, so -k forces SIGKILL if it ignores the @@ -820,13 +825,13 @@ PYCODE ;; python) "$PY" "$CLIENTS/python/interop.py" \ - subscribe --url "$url" --broadcast "$broadcast" --timeout "$TIMEOUT" + subscribe --url "$url" --broadcast "$broadcast" --timeout "$TIMEOUT" ${auth[@]+"${auth[@]}"} ;; go) - "$GO_INTEROP" subscribe --url "$url" --broadcast "$broadcast" --timeout "$TIMEOUT" + "$GO_INTEROP" subscribe --url "$url" --broadcast "$broadcast" --timeout "$TIMEOUT" ${auth[@]+"${auth[@]}"} ;; cpp) - "$CPP_INTEROP" subscribe --url "$url" --broadcast "$broadcast" --timeout "$TIMEOUT" + "$CPP_INTEROP" subscribe --url "$url" --broadcast "$broadcast" --timeout "$TIMEOUT" ${auth[@]+"${auth[@]}"} ;; c) "$C_INTEROP" subscribe --url "$url" --broadcast "$broadcast" --timeout "$TIMEOUT" @@ -1117,13 +1122,12 @@ run_media() { fi } -# The publishers whose refusal the harness can read: the Rust CLI's log. The -# binding publishers enforce the grant too, but surface it only through their -# bindings, and the browser elements have no way to offer $AUTH_VERSION. +# The publishers whose refusal the harness can read: each logs moq-net's error. +# The browser elements have no way to offer $AUTH_VERSION. enforces_grant() { [[ "$COMPAT" -eq 0 ]] || return 1 case "$1" in - rust) return 0 ;; + rust | python | go | cpp) return 0 ;; *) return 1 ;; esac }