feat(bindings): read the session grant and present tokens in band

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<int>::of(&QComboBox::currentIndexChanged), this, &MoQDock::OnEncodingChanged);
 	connect(latencyMode, QOverload<int>::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<MoQOutput *>(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())

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;

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<std::mutex> lock(mutex);
+	return token_refusal;
+}
+
+bool MoQOutput::PresentToken(const std::string &token)
+{
+	std::lock_guard<std::recursive_mutex> signal_lock(signal_mutex);
+	if (!attempt || !attempt->session)
+		return false;
+	{
+		std::lock_guard<std::mutex> 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<moq::expected<std::shared_ptr<moq::AuthToken>>>(
+					   attempt, [this](const std::shared_ptr<Attempt> &current, 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<std::mutex> 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;

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<moq::Session> &current, 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<moq::AuthToken> token;
+		std::optional<moq::Continuation> presenting;
 	};
 
 	void OnConnect(const std::shared_ptr<Attempt> &attempt, moq::expected<std::shared_ptr<moq::Session>> 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<int> connect_time_ms{0};

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.

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;
 

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<Grant?> 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.

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.

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<UnsupportedMoqException>()),
+    );
+    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(