fix(abstractions): normalize allowed hosts, reject userinfo, and validate URL structure

packages/abstractions/kiota_abstractions/authentication/allowed_hosts_validator.py

@@ -13,14 +13,7 @@ def __init__(self, allowed_hosts: list[str]) -> None:
             allowed_hosts (list[str]): A list of valid hosts.  If the list is empty, all hosts
             are valid.
         """
-        if not isinstance(allowed_hosts, list):
-            raise TypeError("Allowed hosts must be a list of strings")
-
-        for host in allowed_hosts:
-            if host.startswith("https://") or host.startswith("http://"):
-                raise ValueError("Allowed host value cannot contain 'https://' or 'http://' prefix")
-
-        self.allowed_hosts: set[str] = {x.lower() for x in allowed_hosts}
+        self.set_allowed_hosts(allowed_hosts)
 
     def get_allowed_hosts(self) -> list[str]:
         """Gets the list of valid hosts.  If the list is empty, all hosts are valid.
@@ -40,11 +33,19 @@ def set_allowed_hosts(self, allowed_hosts: list[str]) -> None:
         if not isinstance(allowed_hosts, list):
             raise TypeError("Allowed hosts must be a list of strings")
 
+        normalized_hosts: set[str] = set()
         for host in allowed_hosts:
-            if host.startswith("https://") or host.startswith("http://"):
+            if not isinstance(host, str):
+                raise TypeError("Allowed host value must be a string")
+            cleaned_host = host.strip()
+            if not cleaned_host:
+                raise ValueError("Allowed host value cannot be empty or whitespace only")
+            lower_host = cleaned_host.lower()
+            if lower_host.startswith("https://") or lower_host.startswith("http://"):
                 raise ValueError("Allowed host value cannot contain 'https://' or 'http://' prefix")
+            normalized_hosts.add(lower_host)
 
-        self.allowed_hosts = {x.lower() for x in allowed_hosts}
+        self.allowed_hosts: set[str] = normalized_hosts
 
     def is_url_host_valid(self, url: str) -> bool:
         """Checks whether the provided host is valid.
@@ -53,25 +54,28 @@ def is_url_host_valid(self, url: str) -> bool:
             url (str): The url to check.
 
         Returns:
-            bool: [description]
+            bool: True if the host is valid, False otherwise.
         """
-        if not url:
+        if not url or not isinstance(url, str) or not url.strip():
             return False
-        if not self.get_allowed_hosts():
-            return True
+        cleaned_url = url.strip()
+
         # Format: urlparse("scheme://netloc/path;parameters?query#fragment")
         # Returns: ParseResult(scheme='scheme', netloc='netloc', path='/path;parameters', params='',
         #    query='query', fragment='fragment')
-        o = urlparse(url)
-        if not all([o.scheme, o.netloc]):
+        try:
+            o = urlparse(cleaned_url)
+        except ValueError:
             return False
-        if not o.hostname:
+
+        if not (o.scheme and o.netloc
+                and o.hostname) or (o.username is not None or o.password is not None):
             return False
-        hostname = o.hostname.lower()
-        # Existing exact match
-        if hostname in self.allowed_hosts:
+
+        if not self.allowed_hosts:
             return True
-        # Suffix match for entries starting with "."
-        return any(
+
+        hostname = o.hostname.lower()
+        return hostname in self.allowed_hosts or any(
             suffix.startswith(".") and hostname.endswith(suffix) for suffix in self.allowed_hosts
         )

packages/abstractions/tests/authentication/test_allowed_hosts_validator.py

@@ -12,9 +12,31 @@ def test_lowercases_hosts(self):
         validator = AllowedHostsValidator(["Example.COM", "API.Example.com"])
         assert set(validator.get_allowed_hosts()) == {"example.com", "api.example.com"}
 
+    def test_trims_whitespace_from_hosts(self):
+        validator = AllowedHostsValidator(["  example.com  ", "  api.example.com "])
+        assert set(validator.get_allowed_hosts()) == {"example.com", "api.example.com"}
+
     def test_raises_on_non_list(self):
         with pytest.raises(TypeError):
-            AllowedHostsValidator("example.com")
+            AllowedHostsValidator("example.com")  # type: ignore
+
+    def test_raises_on_non_string_element(self):
+        with pytest.raises(TypeError):
+            AllowedHostsValidator(["example.com", 123])  # type: ignore
+
+    def test_raises_on_empty_string_entry(self):
+        with pytest.raises(ValueError):
+            AllowedHostsValidator([""])
+
+    def test_raises_on_whitespace_only_entry(self):
+        with pytest.raises(ValueError):
+            AllowedHostsValidator(["   "])
+
+    def test_raises_on_mixed_valid_and_blank_entry(self):
+        with pytest.raises(ValueError):
+            AllowedHostsValidator(["example.com", ""])
+        with pytest.raises(ValueError):
+            AllowedHostsValidator(["example.com", "   "])
 
     def test_raises_on_http_prefix(self):
         with pytest.raises(ValueError):
@@ -24,6 +46,16 @@ def test_raises_on_https_prefix(self):
         with pytest.raises(ValueError):
             AllowedHostsValidator(["https://example.com"])
 
+    def test_raises_on_uppercase_prefix(self):
+        with pytest.raises(ValueError):
+            AllowedHostsValidator(["HTTP://example.com"])
+        with pytest.raises(ValueError):
+            AllowedHostsValidator(["HTTPS://example.com"])
+
+    def test_raises_on_prefix_with_whitespace(self):
+        with pytest.raises(ValueError):
+            AllowedHostsValidator(["  http://example.com  "])
+
 
 class TestIsUrlHostValid:
     def test_returns_false_for_empty_url(self):
@@ -32,12 +64,44 @@ def test_returns_false_for_empty_url(self):
 
     def test_returns_false_for_none_url(self):
         validator = AllowedHostsValidator(["example.com"])
-        assert validator.is_url_host_valid(None) is False
+        assert validator.is_url_host_valid(None) is False  # type: ignore
+
+    def test_returns_false_for_non_string_url(self):
+        validator = AllowedHostsValidator(["example.com"])
+        assert validator.is_url_host_valid(123) is False  # type: ignore
 
     def test_returns_true_when_allowed_hosts_empty(self):
         validator = AllowedHostsValidator([])
         assert validator.is_url_host_valid("https://anything.com/path") is True
 
+    def test_returns_false_for_malformed_url_even_when_allowed_hosts_empty(self):
+        validator = AllowedHostsValidator([])
+        assert validator.is_url_host_valid("not-a-valid-url") is False
+        assert validator.is_url_host_valid("https://") is False
+
+    def test_returns_false_for_unmatched_ipv6_bracket(self):
+        validator = AllowedHostsValidator(["example.com"])
+        assert validator.is_url_host_valid("https://[example.com/path") is False
+
+        empty_validator = AllowedHostsValidator([])
+        assert empty_validator.is_url_host_valid("https://[example.com/path") is False
+
+    def test_returns_false_for_userinfo_even_when_allowed_hosts_empty(self):
+        validator = AllowedHostsValidator([])
+        assert validator.is_url_host_valid("https://user:password@example.com/path") is False
+        assert validator.is_url_host_valid("https://@example.com/path") is False
+        assert validator.is_url_host_valid("https://:@example.com/path") is False
+
+    def test_returns_false_for_userinfo_url(self):
+        validator = AllowedHostsValidator(["example.com"])
+        assert validator.is_url_host_valid("https://user:password@example.com/path") is False
+        assert validator.is_url_host_valid("https://user@example.com/path") is False
+
+    def test_returns_false_for_explicitly_empty_userinfo(self):
+        validator = AllowedHostsValidator(["example.com"])
+        assert validator.is_url_host_valid("https://@example.com/path") is False
+        assert validator.is_url_host_valid("https://:@example.com/path") is False
+
     def test_returns_true_for_allowed_host(self):
         validator = AllowedHostsValidator(["example.com"])
         assert validator.is_url_host_valid("https://example.com/path") is True
@@ -93,6 +157,10 @@ def test_handles_url_with_query_and_fragment(self):
         validator = AllowedHostsValidator(["example.com"])
         assert validator.is_url_host_valid("https://example.com/path?q=1#frag") is True
 
+    def test_handles_url_with_whitespace(self):
+        validator = AllowedHostsValidator(["example.com"])
+        assert validator.is_url_host_valid("  https://example.com/path  ") is True
+
 
 class TestSetAllowedHosts:
     def test_updates_allowed_hosts(self):
@@ -104,7 +172,17 @@ def test_updates_allowed_hosts(self):
     def test_raises_on_non_list(self):
         validator = AllowedHostsValidator(["example.com"])
         with pytest.raises(TypeError):
-            validator.set_allowed_hosts("example.com")
+            validator.set_allowed_hosts("example.com")  # type: ignore
+
+    def test_raises_on_empty_string_entry(self):
+        validator = AllowedHostsValidator(["example.com"])
+        with pytest.raises(ValueError):
+            validator.set_allowed_hosts([""])
+
+    def test_raises_on_whitespace_only_entry(self):
+        validator = AllowedHostsValidator(["example.com"])
+        with pytest.raises(ValueError):
+            validator.set_allowed_hosts(["   "])
 
     def test_raises_on_http_prefix(self):
         validator = AllowedHostsValidator(["example.com"])

packages/authentication/azure/tests/test_azure_identity_access_token_provider.py

@@ -133,8 +133,8 @@ async def test_get_authorization_token_invalid_url():
     token_provider = AzureIdentityAccessTokenProvider(DummyAsyncAzureTokenCredential(), None)
     token = await token_provider.get_authorization_token('')
     assert token == ""
-    with pytest.raises(Exception):
-        token = await token_provider.get_authorization_token('https://')
+    token = await token_provider.get_authorization_token('https://')
+    assert token == ""
 
 
 @pytest.mark.asyncio
@@ -162,21 +162,38 @@ def get_token(self, *scopes, **kwargs):
 
 
 @pytest.mark.asyncio
-async def test_derived_scope_strips_userinfo_and_port():
+async def test_derived_scope_strips_port():
     """The default `.default` scope passed to `get_token` must be derived
-    from the hostname only — never include userinfo or
-    a `:port` (which Entra ID rejects for `.default` scopes).
+    from the hostname only — never include a `:port` (which Entra ID rejects
+    for `.default` scopes).
     """
     credential = RecordingSyncAzureTokenCredential()
     token_provider = AzureIdentityAccessTokenProvider(credential, None)
 
-    await token_provider.get_authorization_token(
-        'https://alice:secret@graph.microsoft.com:8443/v1.0/me'
+    token = await token_provider.get_authorization_token(
+        'https://graph.microsoft.com:8443/v1.0/me'
     )
 
+    assert token == "This is a dummy token"
     assert credential.received_scopes == [('https://graph.microsoft.com/.default',)]
 
 
+@pytest.mark.asyncio
+async def test_userinfo_url_is_rejected_without_requesting_token():
+    """URLs containing userinfo must be rejected by the validator before
+    credential acquisition and return an empty token without invoking get_token.
+    """
+    credential = RecordingSyncAzureTokenCredential()
+    token_provider = AzureIdentityAccessTokenProvider(credential, None)
+
+    token = await token_provider.get_authorization_token(
+        'https://alice:secret@graph.microsoft.com:8443/v1.0/me'
+    )
+
+    assert token == ""
+    assert credential.received_scopes == []
+
+
 @pytest.mark.asyncio
 async def test_derived_scope_is_not_cached_across_hosts():
     """The first URL's derived scope must not be reused for later URLs.
@@ -217,14 +234,14 @@ async def test_explicit_scopes_are_respected():
 
 @pytest.mark.asyncio
 async def test_derived_scope_rejects_url_without_hostname():
-    """A URI whose netloc has no hostname (e.g. `https://@/path`) must not
-    silently derive a scope like `https://None/.default`; it must raise.
+    """A URI whose netloc has no hostname (e.g. `https://@/path`) must be rejected
+    by the validator, returning an empty token without invoking get_token.
     """
     credential = RecordingSyncAzureTokenCredential()
     token_provider = AzureIdentityAccessTokenProvider(credential, None)
 
-    with pytest.raises(Exception):
-        await token_provider.get_authorization_token('https://@/path')
+    token = await token_provider.get_authorization_token('https://@/path')
+    assert token == ""
     assert credential.received_scopes == []