Add StreamingRequestHandler.Audience for Streaming + Skills

libraries/Microsoft.Bot.Builder/Streaming/BotFrameworkHttpAdapterBase.cs

@@ -116,19 +116,30 @@ public async Task<InvokeResponse> ProcessStreamingActivityAsync(Activity activit
             BotAssert.ActivityNotNull(activity);
 
             Logger.LogInformation($"Received an incoming streaming activity. ActivityId: {activity.Id}");
+
+            // If a StreamingRequestHandler.Audience is a null value, then no callerId should have been generated
+            // and GetAudienceFromCallerId returns null.
+            // Thus we fallback to relying on the "original key", essentially $"{ServiceUrl}{Conversation.Id}",
+            // as opposed to $"{ServiceUrl}{Audience}{Conversation.Id}" and the StreamingRequestHandler implicitly does not support skills.
+            var audience = GetAudienceFromCallerId(activity);
 
             // If a conversation has moved from one connection to another for the same Channel or Skill and
             // hasn't been forgotten by the previous StreamingRequestHandler. The last requestHandler
             // the conversation has been associated with should always be the active connection.
-            var requestHandler = RequestHandlers.Where(x => x.ServiceUrl == activity.ServiceUrl).Where(y => y.HasConversation(activity.Conversation.Id)).LastOrDefault();
+            var requestHandler = RequestHandlers.Where(x => x.ServiceUrl == activity.ServiceUrl)
+                .Where(y => y.Audience == audience)
+                .Where(z => z.HasConversation(activity.Conversation.Id))
+                .LastOrDefault();
             using (var context = new TurnContext(this, activity))
             {
+                context.TurnState.Add<string>(OAuthScopeKey, audience);
+
                 // Pipes are unauthenticated. Pending to check that we are in pipes right now. Do not merge to master without that.
                 if (ClaimsIdentity != null)
                 {
                     context.TurnState.Add<IIdentity>(BotIdentityKey, ClaimsIdentity);
                 }
-
+                
                 var connectorClient = CreateStreamingConnectorClient(activity, requestHandler);
                 context.TurnState.Add(connectorClient);
 
@@ -217,9 +228,10 @@ public async Task<ResourceResponse> SendStreamingActivityAsync(Activity activity
         /// </summary>
         /// <param name="pipeName">The name of the Named Pipe to connect to.</param>
         /// <param name="bot">The bot to use when processing activities received over the Named Pipe.</param>
+        /// <param name="audience">The specified recipient of all outgoing activities.</param>
         /// <returns>A task that completes only once the StreamingRequestHandler has stopped listening
         /// for incoming requests on the Named Pipe.</returns>
-        public async Task ConnectNamedPipeAsync(string pipeName, IBot bot)
+        public async Task ConnectNamedPipeAsync(string pipeName, IBot bot, string audience = null)
         {
             if (string.IsNullOrEmpty(pipeName))
             {
@@ -234,7 +246,7 @@ public async Task ConnectNamedPipeAsync(string pipeName, IBot bot)
                 RequestHandlers = new List<StreamingRequestHandler>();
             }
 
-            var requestHandler = new StreamingRequestHandler(bot, this, pipeName, Logger);
+            var requestHandler = new StreamingRequestHandler(bot, this, pipeName, audience, Logger);
             RequestHandlers.Add(requestHandler);
 
             await requestHandler.ListenAsync().ConfigureAwait(false);
@@ -295,5 +307,27 @@ private IConnectorClient CreateStreamingConnectorClient(Activity activity, Strea
             var connectorClient = new ConnectorClient(new Uri(activity.ServiceUrl), emptyCredentials, customHttpClient: streamingClient);
             return connectorClient;
         }
+
+        /// <summary>
+        /// Attempts to get an audience from the <see cref="Activity.CallerId"/>.
+        /// </summary>
+        /// <param name="activity">The incoming activity to be processed by a <see cref="StreamingRequestHandler"/>.</param>
+        private string GetAudienceFromCallerId(Activity activity)
+        {
+            switch (activity.CallerId)
+            {
+                case CallerIdConstants.PublicAzureChannel:
+                    return AuthenticationConstants.ToChannelFromBotOAuthScope;
+                case CallerIdConstants.USGovChannel:
+                    return GovernmentAuthenticationConstants.ToChannelFromBotOAuthScope;
+                default:
+                    if (activity.CallerId.StartsWith(CallerIdConstants.BotToBotPrefix, StringComparison.InvariantCultureIgnoreCase))
+                    {
+                        return activity.CallerId.Substring(CallerIdConstants.BotToBotPrefix.Length);
+                    }
+
+                    return null;
+            }
+        }
     }
 }

libraries/Microsoft.Bot.Builder/Streaming/StreamingRequestHandler.cs

@@ -13,6 +13,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Bot.Connector;
+using Microsoft.Bot.Connector.Authentication;
 using Microsoft.Bot.Schema;
 using Microsoft.Bot.Streaming;
 using Microsoft.Bot.Streaming.Transport;
@@ -48,15 +49,38 @@ public class StreamingRequestHandler : RequestHandler
         /// <param name="socket">The base socket to use when connecting to the channel.</param>
         /// <param name="logger">Logger implementation for tracing and debugging information.</param>
         public StreamingRequestHandler(IBot bot, IStreamingActivityProcessor activityProcessor, WebSocket socket, ILogger logger = null)
+            : this(bot, activityProcessor, socket, null, logger)
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="StreamingRequestHandler"/> class and
+        /// establishes a connection over a WebSocket to a streaming channel.
+        /// </summary>
+        /// <remarks>
+        /// The audience represents the recipient at the other end of the StreamingRequestHandler's
+        /// streaming connection. Some acceptable audience values are as follows:
+        /// <list>
+        /// <item>- For Public Azure channels, use <see cref="Microsoft.Bot.Connector.Authentication.AuthenticationConstants.ToChannelFromBotOAuthScope"/>.</item>
+        /// <item>- For Azure Government channels, use <see cref="Microsoft.Bot.Connector.Authentication.GovernmentAuthenticationConstants.ToChannelFromBotOAuthScope"/>.</item>
+        /// </list>
+        /// </remarks>
+        /// <param name="bot">The bot for which we handle requests.</param>
+        /// <param name="activityProcessor">The processor for incoming requests.</param>
+        /// <param name="socket">The base socket to use when connecting to the channel.</param>
+        /// <param name="logger">Logger implementation for tracing and debugging information.</param>
+        /// <param name="audience">The specified recipient of all outgoing activities.</param>
+        public StreamingRequestHandler(IBot bot, IStreamingActivityProcessor activityProcessor, WebSocket socket, string audience = null, ILogger logger = null)
         {
             _bot = bot ?? throw new ArgumentNullException(nameof(bot));
             _activityProcessor = activityProcessor ?? throw new ArgumentNullException(nameof(activityProcessor));
-            
+
             if (socket == null)
             {
                 throw new ArgumentNullException(nameof(socket));
             }
 
+            Audience = audience;
             _logger = logger ?? NullLogger.Instance;
             _conversations = new ConcurrentDictionary<string, DateTime>();
             _userAgent = GetUserAgent();
@@ -74,6 +98,28 @@ public StreamingRequestHandler(IBot bot, IStreamingActivityProcessor activityPro
         /// <param name="pipeName">The name of the Named Pipe to use when connecting to the channel.</param>
         /// <param name="logger">Logger implementation for tracing and debugging information.</param>
         public StreamingRequestHandler(IBot bot, IStreamingActivityProcessor activityProcessor, string pipeName, ILogger logger = null)
+            : this(bot, activityProcessor, pipeName, null, logger)
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="StreamingRequestHandler"/> class and
+        /// establishes a connection over a Named Pipe to a streaming channel.
+        /// </summary>
+        /// <remarks>
+        /// The audience represents the recipient at the other end of the StreamingRequestHandler's
+        /// streaming connection. Some acceptable audience values are as follows:
+        /// <list>
+        /// <item>- For Public Azure channels, use <see cref="Microsoft.Bot.Connector.Authentication.AuthenticationConstants.ToChannelFromBotOAuthScope"/>.</item>
+        /// <item>- For Azure Government channels, use <see cref="Microsoft.Bot.Connector.Authentication.GovernmentAuthenticationConstants.ToChannelFromBotOAuthScope"/>.</item>
+        /// </list>
+        /// </remarks>
+        /// <param name="bot">The bot for which we handle requests.</param>
+        /// <param name="activityProcessor">The processor for incoming requests.</param>
+        /// <param name="pipeName">The name of the Named Pipe to use when connecting to the channel.</param>
+        /// <param name="logger">Logger implementation for tracing and debugging information.</param>
+        /// <param name="audience">The specified recipient of all outgoing activities.</param>
+        public StreamingRequestHandler(IBot bot, IStreamingActivityProcessor activityProcessor, string pipeName, string audience, ILogger logger = null)
         {
             _bot = bot ?? throw new ArgumentNullException(nameof(bot));
             _activityProcessor = activityProcessor ?? throw new ArgumentNullException(nameof(activityProcessor));
@@ -84,6 +130,7 @@ public StreamingRequestHandler(IBot bot, IStreamingActivityProcessor activityPro
                 throw new ArgumentNullException(nameof(pipeName));
             }
 
+            Audience = audience;
             _conversations = new ConcurrentDictionary<string, DateTime>();
             _userAgent = GetUserAgent();
             _server = new NamedPipeServer(pipeName, this);
@@ -101,6 +148,14 @@ public StreamingRequestHandler(IBot bot, IStreamingActivityProcessor activityPro
         public string ServiceUrl { get; private set; }
 #pragma warning restore CA1056 // Uri properties should not be strings
 
+        /// <summary>
+        /// Gets the intended recipient of <see cref="Activity">Activities</see> sent from this StreamingRequestHandler.
+        /// </summary>
+        /// <value>
+        /// The intended recipient of Activities sent from this StreamingRequestHandler.
+        /// </value>
+        public string Audience { get; private set; }
+
         /// <summary>
         /// Begins listening for incoming requests over this StreamingRequestHandler's server.
         /// </summary>
@@ -225,6 +280,43 @@ public override async Task<StreamingResponse> ProcessRequestAsync(ReceiveRequest
                     }
                 }
 
+                // Populate Activity.CallerId given the Audience value.
+                string callerId = null;
+                switch (Audience)
+                {
+                    case AuthenticationConstants.ToChannelFromBotOAuthScope:
+                        callerId = CallerIdConstants.PublicAzureChannel;
+                        break;
+                    case GovernmentAuthenticationConstants.ToChannelFromBotOAuthScope:
+                        callerId = CallerIdConstants.USGovChannel;
+                        break;
+                    default:
+                        if (!string.IsNullOrEmpty(Audience))
+                        {
+                            if (Guid.TryParse(Audience, out var result))
+                            {
+                                // Large assumption drawn here; any GUID is an AAD AppId. This is prohibitive towards bots not using the Bot Framework auth model
+                                // but still using GUIDs/UUIDs as identifiers.
+                                // It's also indicative of the tight coupling between the Bot Framework protocol, authentication and transport mechanism in the SDK.
+                                // In R12, this work will be re-implemented to better utilize the CallerId and Audience set on BotFrameworkAuthentication instances
+                                // and decouple the three concepts mentioned above.
+                                callerId = $"{CallerIdConstants.BotToBotPrefix}{Audience}";
+                            }
+                            else
+                            {
+                                // Fallback to using the raw Audience as the CallerId. The auth model being used by the Adapter using this StreamingRequestHandler
+                                // is not known to the SDK, therefore it is assumed the developer knows what they're doing. The SDK should not prevent
+                                // the developer from extending it to use custom auth models in Streaming contexts.
+                                callerId = Audience;
+                            }
+                        }
+
+                        // A null Audience is an implicit statement indicating the bot does not support skills.
+                        break;
+                }
+
+                activity.CallerId = callerId;
+
                 // Now that the request has been converted into an activity we can send it to the adapter.
                 var adapterResponse = await _activityProcessor.ProcessStreamingActivityAsync(activity, _bot.OnTurnAsync, cancellationToken).ConfigureAwait(false);
 

libraries/integration/Microsoft.Bot.Builder.Integration.AspNet.Core/ApplicationBuilderExtensions.cs

@@ -77,16 +77,17 @@ public static IApplicationBuilder UseBotFramework(this IApplicationBuilder appli
         /// </summary>
         /// <param name="applicationBuilder">The application builder that defines the bot's pipeline.<see cref="IApplicationBuilder"/>.</param>
         /// <param name="pipeName">The name of the named pipe to use when creating the server.</param>
+        /// <param name="audience">The specified recipient of all outgoing activities.</param>
         /// <returns>A reference to this instance after the operation has completed.</returns>
-        public static IApplicationBuilder UseNamedPipes(this IApplicationBuilder applicationBuilder, string pipeName = "bfv4.pipes")
+        public static IApplicationBuilder UseNamedPipes(this IApplicationBuilder applicationBuilder, string pipeName = "bfv4.pipes", string audience = null)
         {
             if (applicationBuilder == null)
             {
                 throw new ArgumentNullException(nameof(applicationBuilder));
             }
 
             var bot = applicationBuilder.ApplicationServices.GetService(typeof(IBot)) as IBot;
-            _ = (applicationBuilder.ApplicationServices.GetService(typeof(IBotFrameworkHttpAdapter)) as BotFrameworkHttpAdapter).ConnectNamedPipeAsync(pipeName, bot);
+            _ = (applicationBuilder.ApplicationServices.GetService(typeof(IBotFrameworkHttpAdapter)) as BotFrameworkHttpAdapter).ConnectNamedPipeAsync(pipeName, bot, audience);
 
             return applicationBuilder;
         }