Auth configuration logging with redaction for sensitive values

libraries/microsoft-agents-activity/microsoft_agents/activity/config/_load_configuration.py

@@ -1,10 +1,14 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
+import logging
+
 from typing import Any
 
 from ._configure_logging import _configure_logging
 
+logger = logging.getLogger(__name__)
+
 
 def load_configuration_from_env(env_vars: dict[str, Any]) -> dict:
     """
@@ -33,5 +37,5 @@ def load_configuration_from_env(env_vars: dict[str, Any]) -> dict:
     return {
         "AGENTAPPLICATION": result.get("AGENTAPPLICATION", {}),
         "CONNECTIONS": result.get("CONNECTIONS", {}),
-        "CONNECTIONSMAP": result.get("CONNECTIONSMAP", {}),
+        "CONNECTIONSMAP": result.get("CONNECTIONSMAP", []),
     }

libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/_utils.py

@@ -0,0 +1,178 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+# this file will be moved to hosting_core once the sidecar changes are in.
+
+import json
+
+from logging import Logger
+from urllib.parse import urlparse
+
+from microsoft_agents.activity._model_utils import pick_model_dict, SkipNone
+from microsoft_agents.hosting.core import (
+    AgentAuthConfiguration,
+    AccessTokenProviderBase,
+    ClaimsIdentity,
+    Connections,
+)
+
+_REDACTION_PEEK_LENGTH = 2
+_REDACTION_THRESH = _REDACTION_PEEK_LENGTH + 6
+
+
+def _redact_str(s: str, peek: bool = False) -> str:
+    """Redact a string for logging purposes.
+
+    :arg s: The string to redact.
+    :type s: str
+    :arg peek: Whether to show a peek of the string. Defaults to False.
+    :type peek: bool
+    :return: The redacted string.
+    """
+    if peek and len(s) > _REDACTION_THRESH:
+        return f"{s[:_REDACTION_PEEK_LENGTH]}..."
+    else:
+        return "..."
+
+
+def _redact_str_or_none(s: str | None, peek: bool = False) -> str | None:
+    """Redact a string or None for logging purposes.
+
+    :arg s: The string to redact or None.
+    :type s: str | None
+    :arg peek: Whether to show a peek of the string. Defaults to False.
+    :type peek: bool
+    :return: The redacted string or None.
+    :rtype: str | None
+    """
+    if s is None:
+        return None
+    return _redact_str(s, peek=peek)
+
+
+def _redact_scopes(scopes: list[str] | None) -> str | None:
+    """Redact a list of scopes for logging purposes.
+
+    :arg scopes: The list of scopes to redact.
+    :type scopes: list[str] | None
+    :return: A string summarizing the scopes.
+    :rtype: str | None
+    """
+    if scopes is None:
+        return None
+    return f"... [{len(scopes)} scope(s)]"
+
+
+def _redact_url(url: str) -> str:
+    """
+    Redact a URL for logging purposes.
+
+    :arg url: The URL to redact.
+    :type url: str
+    :return: The redacted URL.
+    :rtype: str
+    """
+    url = url.strip()
+    if not url:
+        return ""
+
+    try:
+        url_parsed = urlparse(url)
+        return f"{url_parsed.scheme}://{url_parsed.netloc}/..."
+    except:
+        return "..."
+
+
+def _redact_url_or_none(url: str | None) -> str | None:
+    """Redact a URL or None for logging purposes.
+
+    :arg url: The URL to redact or None.
+    :type url: str | None
+    :return: The redacted URL or None.
+    :rtype: str | None
+    """
+    if url is None:
+        return None
+    return _redact_url(url)
+
+
+def _summarize_auth_configs(config_map: dict[str, AgentAuthConfiguration]) -> str:
+    """
+    Summarize the authentication configuration for logging.
+
+    :arg config_map: A dictionary of connection configurations.
+    :type config_map: dict[str, :class:`microsoft_agents.hosting.core.AgentAuthConfiguration`]
+    :return: A string summarizing the authentication configuration.
+    :rtype: str
+    """
+    summary = []
+    for config in config_map.values():
+        summary.append(
+            json.dumps(
+                pick_model_dict(
+                    CONNECTION_NAME=SkipNone(config.CONNECTION_NAME),
+                    CLIENTID=SkipNone(_redact_str_or_none(config.CLIENT_ID, peek=True)),
+                    TENANTID=SkipNone(_redact_str_or_none(config.TENANT_ID, peek=True)),
+                    CLIENTSECRET=SkipNone(_redact_str_or_none(config.CLIENT_SECRET)),
+                    AUTHORITY=SkipNone(_redact_url_or_none(config.AUTHORITY)),
+                    SCOPES=SkipNone(_redact_scopes(config.SCOPES)),
+                    FEDERATED_CLIENT_ID=SkipNone(
+                        _redact_str_or_none(config.FEDERATED_CLIENT_ID, peek=True)
+                    ),
+                    CERT_PFX_FILE=SkipNone(_redact_str_or_none(config.CERT_PFX_FILE)),
+                    ALT_BLUEPRINT_ID=SkipNone(
+                        _redact_str_or_none(config.ALT_BLUEPRINT_ID, peek=True)
+                    ),
+                    IDPM_RESOURCE=SkipNone(_redact_url_or_none(config.IDPM_RESOURCE)),
+                    AZURE_REGION=SkipNone(_redact_url_or_none(config.AZURE_REGION)),
+                    ANNONYMOUS_ALLOWED=str(config.ANONYMOUS_ALLOWED),
+                )
+            )
+        )
+    return "\n".join(summary)
+
+
+def _summarize_connections_map(connections_map: list[dict[str, str]]) -> str:
+
+    connections_map_output = []
+    for mapping in connections_map:
+
+        obj = {
+            "CONNECTION": mapping.get("CONNECTION", ""),
+        }
+
+        if "AUDIENCE" in mapping:
+            obj["AUDIENCE"] = mapping["AUDIENCE"]
+
+        if "SERVICEURL" in mapping:
+
+            service_url = mapping.get("SERVICEURL", "").strip()
+            if service_url != "*":
+                service_url = _redact_url(service_url)
+
+        connections_map_output.append(mapping)
+
+    return json.dumps(connections_map_output, indent=2)
+
+
+def _log_config(
+    logger: Logger,
+    config_map: dict[str, AgentAuthConfiguration],
+    connections_map: list[dict[str, str]],
+) -> None:
+    """
+    Log the configuration of the MSAL connection manager.
+
+    :arg logger: The logger to use for logging.
+    :type logger: :class:`logging.Logger`
+    :arg connections_map: A list of connection mappings.
+    :type connections_map: list[dict[str, str]]
+    :arg config_map: A dictionary of connection configurations.
+    :type config_map: dict[str, :class:`microsoft_agents.hosting.core.AgentAuthConfiguration`]
+    """
+
+    connections_output = _summarize_auth_configs(config_map)
+    connections_map_output = _summarize_connections_map(connections_map)
+
+    output = f"Connections: \n{connections_output}\nConnections Map: {connections_map_output}"
+    logger.info(output)

libraries/microsoft-agents-authentication-msal/microsoft_agents/authentication/msal/msal_connection_manager.py

@@ -1,8 +1,10 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
+import logging
 import re
-from typing import Dict, Optional
+
+from typing import Optional
 from microsoft_agents.hosting.core import (
     AgentAuthConfiguration,
     AccessTokenProviderBase,
@@ -11,6 +13,9 @@
 )
 
 from .msal_auth import MsalAuth
+from ._utils import _log_config
+
+logger = logging.getLogger(__name__)
 
 
 class MsalConnectionManager(Connections):
@@ -34,8 +39,14 @@ def __init__(
         :raises ValueError: If no service connection configuration is provided.
         """
 
-        self._connections: dict[str, MsalAuth] = {}
-        self._connections_map = connections_map or kwargs.get("CONNECTIONSMAP", {})
+        self._connections = {}
+        self._connections_map = connections_map or kwargs.get("CONNECTIONSMAP", [])
+        if not isinstance(self._connections_map, list) or (
+            len(self._connections_map) > 0
+            and not isinstance(self._connections_map[0], dict)
+        ):
+            raise ValueError("CONNECTIONSMAP must be a list of dictionaries.")
+
         self._config_map: dict[str, AgentAuthConfiguration] = {}
 
         if connections_configurations:
@@ -61,6 +72,8 @@ def __init__(
         if not self._connections.get("SERVICE_CONNECTION", None):
             raise ValueError("No service connection configuration provided.")
 
+        _log_config(logger, self._config_map, self._connections_map)
+
     def get_connection(self, connection_name: Optional[str]) -> AccessTokenProviderBase:
         """
         Get the OAuth connection for the agent.

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py

@@ -45,7 +45,7 @@
 
 from ._type_defs import RouteHandler, RouteSelector
 from ._routes import _RouteList, _Route, RouteRank, _agentic_selector
-from .proactive import Proactive, ProactiveOptions
+from .proactive import Proactive
 
 logger = logging.getLogger(__name__)
 
@@ -104,11 +104,6 @@ def __init__(
 
         configuration = kwargs
 
-        logger.debug(f"Initializing AgentApplication with options: {options}")
-        logger.debug(
-            f"Initializing AgentApplication with configuration: {configuration}"
-        )
-
         if not options:
             # TODO: consolidate configuration story
             # Take the options from the kwargs and create an ApplicationOptions instance

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/oauth/authorization.py

@@ -83,16 +83,15 @@ def __init__(
         )
         if not auth_handlers:
             # get from config
-            handlers_config: dict[str, dict] = auth_configuration.get("HANDLERS")
-            if not auth_handlers and handlers_config:
-                auth_handlers = {
-                    handler_name: AuthHandler(
-                        name=handler_name,
-                        auth_type=config.get("TYPE", None),
-                        **config.get("SETTINGS", {}),
-                    )
-                    for handler_name, config in handlers_config.items()
-                }
+            handlers_config: dict[str, dict] = auth_configuration.get("HANDLERS", {})
+            auth_handlers = {
+                handler_name: AuthHandler(
+                    name=handler_name,
+                    auth_type=config.get("TYPE", ""),
+                    **config.get("SETTINGS", {}),
+                )
+                for handler_name, config in handlers_config.items()
+            }
 
         self._handler_settings = auth_handlers
         self._auto_sign_in = auto_sign_in or bool(

test_samples/app_style/empty_agent.py

@@ -19,7 +19,6 @@
 
 from shared import start_server
 
-logging.basicConfig(level=logging.INFO)
 load_dotenv(path.join(path.dirname(__file__), ".env"))
 
 agents_sdk_config = load_configuration_from_env(environ)

tests/authentication_msal/test_utils.py

@@ -0,0 +1,65 @@
+import pytest
+
+from microsoft_agents.authentication.msal._utils import (
+    _redact_scopes,
+    _redact_str,
+    _redact_str_or_none,
+    _redact_url,
+    _redact_url_or_none,
+)
+
+
+class TestRedactionUtils:
+    @pytest.mark.parametrize("value", ["", "short", "12345678"])
+    def test_redact_str_without_peek(self, value):
+        assert _redact_str(value) == "..."
+
+    def test_redact_str_with_peek_for_long_value(self):
+        assert _redact_str("client-id-secret", peek=True) == "cl..."
+
+    @pytest.mark.parametrize("value", ["", "short", "12345678"])
+    def test_redact_str_with_peek_for_short_values(self, value):
+        assert _redact_str(value, peek=True) == "..."
+
+    def test_redact_str_or_none_returns_none_for_none(self):
+        assert _redact_str_or_none(None) is None
+
+    def test_redact_str_or_none_redacts_value(self):
+        assert _redact_str_or_none("tenant-id-secret", peek=True) == "te..."
+
+    def test_redact_scopes_returns_none_for_none(self):
+        assert _redact_scopes(None) is None
+
+    @pytest.mark.parametrize(
+        "scopes, expected",
+        [
+            ([], "... [0 scope(s)]"),
+            (["scope1"], "... [1 scope(s)]"),
+            (["scope1", "scope2"], "... [2 scope(s)]"),
+        ],
+    )
+    def test_redact_scopes_summarizes_count(self, scopes, expected):
+        assert _redact_scopes(scopes) == expected
+
+    @pytest.mark.parametrize(
+        "url, expected",
+        [
+            (
+                "https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+                "https://login.microsoftonline.com/...",
+            ),
+            ("  https://example.com/path?secret=value  ", "https://example.com/..."),
+            ("", ""),
+            ("   ", ""),
+        ],
+    )
+    def test_redact_url(self, url, expected):
+        assert _redact_url(url) == expected
+
+    def test_redact_url_or_none_returns_none_for_none(self):
+        assert _redact_url_or_none(None) is None
+
+    def test_redact_url_or_none_redacts_value(self):
+        assert (
+            _redact_url_or_none("https://example.com/path") == "https://example.com/..."
+        )