Auth configuration logging with redaction for sensitive values

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

@@ -33,5 +33,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-hosting-core/microsoft_agents/hosting/core/app/oauth/authorization.py

@@ -93,16 +93,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") or {}
+            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(

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_log_config.py

@@ -0,0 +1,172 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import json
+
+from logging import Logger
+from urllib.parse import urlparse
+
+from microsoft_agents.activity._model_utils import pick_model_dict, SkipNone
+
+from .agent_auth_configuration import AgentAuthConfiguration
+
+_REDACTION_PEEK_LENGTH = 3
+_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 Exception:
+        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 connection_name, config in config_map.items():
+        summary.append(
+            pick_model_dict(
+                CONNECTION=connection_name,
+                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)),
+                ANONYMOUS_ALLOWED=str(config.ANONYMOUS_ALLOWED),
+            )
+        )
+    return json.dumps(summary, indent=2)
+
+
+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)
+            obj["SERVICEURL"] = service_url
+
+        connections_map_output.append(obj)
+
+    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)
+
+    logger.info(
+        "\nConnections:\n%s\n\nConnections Map:\n%s",
+        connections_output,
+        connections_map_output,
+    )

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/connection_manager.py

@@ -2,13 +2,19 @@
 # Licensed under the MIT License.
 
 import re
+import logging
+
 from collections.abc import Callable
 
 from .agent_auth_configuration import AgentAuthConfiguration
 from .access_token_provider_base import AccessTokenProviderBase
 from .claims_identity import ClaimsIdentity
 from .connections import Connections
 
+from ._log_config import _log_config
+
+logger = logging.getLogger(__name__)
+
 
 class ConnectionManager(Connections):
     """
@@ -59,6 +65,7 @@ def __init__(
             if connections_map is not None
             else kwargs.get("CONNECTIONSMAP", [])
         )
+
         if isinstance(raw_connections_map, dict):
             raw_connections_map = list(raw_connections_map.values())
         self._connections_map = raw_connections_map or []
@@ -89,6 +96,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: str | None) -> AccessTokenProviderBase:
         """
         Get the OAuth connection for the agent.

tests/hosting_core/test_log_config.py

@@ -0,0 +1,146 @@
+import json
+from unittest.mock import Mock
+
+import pytest
+
+from microsoft_agents.hosting.core import AgentAuthConfiguration
+from microsoft_agents.hosting.core.authorization._log_config import (
+    _log_config,
+    _redact_scopes,
+    _redact_str,
+    _redact_str_or_none,
+    _redact_url,
+    _redact_url_or_none,
+    _summarize_auth_configs,
+    _summarize_connections_map,
+)
+
+
+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) == "cli..."
+
+    @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) == "ten..."
+
+    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/..."
+        )
+
+
+class TestLogConfig:
+    def test_summarize_auth_configs_formats_connections_as_json_array(self):
+        summary = _summarize_auth_configs(
+            {
+                "SERVICE_CONNECTION": AgentAuthConfiguration(
+                    client_id="client-id-secret",
+                    tenant_id="tenant-id-secret",
+                    client_secret="client-secret",
+                    connection_name="configured-name",
+                    authority="https://login.microsoftonline.com/tenant/oauth2/v2.0/token",
+                    scopes=["scope1", "scope2"],
+                )
+            }
+        )
+
+        parsed_summary = json.loads(summary)
+
+        assert parsed_summary == [
+            {
+                "CONNECTION": "SERVICE_CONNECTION",
+                "CONNECTION_NAME": "configured-name",
+                "CLIENTID": "cli...",
+                "TENANTID": "ten...",
+                "CLIENTSECRET": "...",
+                "AUTHORITY": "https://login.microsoftonline.com/...",
+                "SCOPES": "... [2 scope(s)]",
+                "ANONYMOUS_ALLOWED": "False",
+            }
+        ]
+        assert "client-secret" not in summary
+        assert "oauth2/v2.0/token" not in summary
+
+    def test_summarize_connections_map_redacts_service_urls(self):
+        summary = _summarize_connections_map(
+            [
+                {
+                    "CONNECTION": "SERVICE_CONNECTION",
+                    "AUDIENCE": "api://service",
+                    "SERVICEURL": "https://service.example.com/path?secret=value",
+                },
+                {"CONNECTION": "AGENTIC", "SERVICEURL": "*"},
+            ]
+        )
+
+        parsed_summary = json.loads(summary)
+
+        assert parsed_summary == [
+            {
+                "CONNECTION": "SERVICE_CONNECTION",
+                "AUDIENCE": "api://service",
+                "SERVICEURL": "https://service.example.com/...",
+            },
+            {"CONNECTION": "AGENTIC", "SERVICEURL": "*"},
+        ]
+        assert "path?secret=value" not in summary
+
+    def test_log_config_uses_clean_parameterized_format(self):
+        logger = Mock()
+        config_map = {
+            "SERVICE_CONNECTION": AgentAuthConfiguration(client_id="client-id-secret")
+        }
+        connections_map = [{"CONNECTION": "SERVICE_CONNECTION", "SERVICEURL": "*"}]
+
+        _log_config(logger, config_map, connections_map)
+
+        logger.info.assert_called_once()
+        message, connections_output, connections_map_output = logger.info.call_args.args
+        assert message == "\nConnections:\n%s\n\nConnections Map:\n%s"
+        assert json.loads(connections_output)[0]["CONNECTION"] == "SERVICE_CONNECTION"
+        assert json.loads(connections_map_output)[0]["CONNECTION"] == (
+            "SERVICE_CONNECTION"
+        )