fix: validate urls for external accounts (#1031)

* [Bug 193694420] adding validation of token url and service account impersonation url (#1)

[b-193694420] adding url validation for token url and service account impersonation url
  * adding coverage of empty hostname to invalid token url
  * Add more tests to enlarge the coverage
Co-authored-by: Jin Qin <qinjin@google.com>

* adding coverage for service account impersonate url validation

* using urllib3 instead of urllib since it's the project test requirements included

* fix format

* addressing comments

* remove redundant

* fix tests since python 3.6 cannot have same string constants interpret as 3.7+

* fix lint and fix the flaky test in an alternate way

* revert the debugging output

* fix lint

Co-authored-by: Leo <39062083+lsirac@users.noreply.github.com>
Co-authored-by: Jin Qin <qinjin@google.com>
Co-authored-by: arithmetic1728 <58957152+arithmetic1728@users.noreply.github.com>

google/auth/external_account.py

@@ -34,6 +34,7 @@
 import re
 
 import six
+from urllib3.util import parse_url
 
 from google.auth import _helpers
 from google.auth import credentials
@@ -114,6 +115,12 @@ def __init__(
         self._default_scopes = default_scopes
         self._workforce_pool_user_project = workforce_pool_user_project
 
+        Credentials.validate_token_url(token_url)
+        if service_account_impersonation_url:
+            Credentials.validate_service_account_impersonation_url(
+                service_account_impersonation_url
+            )
+
         if self._client_id:
             self._client_auth = utils.ClientAuthentication(
                 utils.ClientAuthType.basic, self._client_id, self._client_secret
@@ -413,3 +420,51 @@ def _initialize_impersonated_credentials(self):
             quota_project_id=self._quota_project_id,
             iam_endpoint_override=self._service_account_impersonation_url,
         )
+
+    @staticmethod
+    def validate_token_url(token_url):
+        _TOKEN_URL_PATTERNS = [
+            "^[^\\.\\s\\/\\\\]+\\.sts\\.googleapis\\.com$",
+            "^sts\\.googleapis\\.com$",
+            "^sts\\.[^\\.\\s\\/\\\\]+\\.googleapis\\.com$",
+            "^[^\\.\\s\\/\\\\]+\\-sts\\.googleapis\\.com$",
+        ]
+
+        if not Credentials.is_valid_url(_TOKEN_URL_PATTERNS, token_url):
+            raise ValueError("The provided token URL is invalid.")
+
+    @staticmethod
+    def validate_service_account_impersonation_url(url):
+        _SERVICE_ACCOUNT_IMPERSONATION_URL_PATTERNS = [
+            "^[^\\.\\s\\/\\\\]+\\.iamcredentials\\.googleapis\\.com$",
+            "^iamcredentials\\.googleapis\\.com$",
+            "^iamcredentials\\.[^\\.\\s\\/\\\\]+\\.googleapis\\.com$",
+            "^[^\\.\\s\\/\\\\]+\\-iamcredentials\\.googleapis\\.com$",
+        ]
+
+        if not Credentials.is_valid_url(
+            _SERVICE_ACCOUNT_IMPERSONATION_URL_PATTERNS, url
+        ):
+            raise ValueError(
+                "The provided service account impersonation URL is invalid."
+            )
+
+    @staticmethod
+    def is_valid_url(patterns, url):
+        """
+        Returns True if the provided URL's scheme is HTTPS and the host comforms to at least one of the provided patterns.
+        """
+        # Check specifically for whitespcaces:
+        #   Some python3.6 will parse the space character into %20 and pass the regex check which shouldn't be passed
+        if not url or len(str(url).split()) > 1:
+            return False
+
+        try:
+            uri = parse_url(url)
+        except Exception:
+            return False
+
+        if not uri.scheme or uri.scheme != "https" or not uri.hostname:
+            return False
+
+        return any(re.compile(p).match(uri.hostname.lower()) for p in patterns)

tests/test_external_account.py

@@ -275,9 +275,110 @@ def assert_resource_manager_request_kwargs(
         assert request_kwargs["headers"] == headers
         assert "body" not in request_kwargs
 
+    def test_valid_token_url_shall_pass_validation(self):
+        valid_urls = [
+            "https://sts.googleapis.com",
+            "https://us-east-1.sts.googleapis.com",
+            "https://US-EAST-1.sts.googleapis.com",
+            "https://sts.us-east-1.googleapis.com",
+            "https://sts.US-WEST-1.googleapis.com",
+            "https://us-east-1-sts.googleapis.com",
+            "https://US-WEST-1-sts.googleapis.com",
+            "https://us-west-1-sts.googleapis.com/path?query",
+        ]
+
+        for url in valid_urls:
+            # A valid url shouldn't throw exception and a None value should be returned
+            external_account.Credentials.validate_token_url(url)
+
+    def test_invalid_token_url_shall_throw_exceptions(self):
+        invalid_urls = [
+            "https://iamcredentials.googleapis.com",
+            "sts.googleapis.com",
+            "https://",
+            "http://sts.googleapis.com",
+            "https://st.s.googleapis.com",
+            "https://us-eas\t-1.sts.googleapis.com",
+            "https:/us-east-1.sts.googleapis.com",
+            "https://US-WE/ST-1-sts.googleapis.com",
+            "https://sts-us-east-1.googleapis.com",
+            "https://sts-US-WEST-1.googleapis.com",
+            "testhttps://us-east-1.sts.googleapis.com",
+            "https://us-east-1.sts.googleapis.comevil.com",
+            "https://us-east-1.us-east-1.sts.googleapis.com",
+            "https://us-ea.s.t.sts.googleapis.com",
+            "https://sts.googleapis.comevil.com",
+            "hhttps://us-east-1.sts.googleapis.com",
+            "https://us- -1.sts.googleapis.com",
+            "https://-sts.googleapis.com",
+            "https://us-east-1.sts.googleapis.com.evil.com",
+        ]
+
+        for url in invalid_urls:
+            # An invalid url should throw a ValueError exception
+            with pytest.raises(ValueError) as excinfo:
+                external_account.Credentials.validate_token_url(url)
+
+            assert excinfo.match("The provided token URL is invalid.")
+
+    def test_valid_service_account_impersonation_url_shall_pass_validation(self):
+        valid_urls = [
+            "https://iamcredentials.googleapis.com",
+            "https://us-east-1.iamcredentials.googleapis.com",
+            "https://US-EAST-1.iamcredentials.googleapis.com",
+            "https://iamcredentials.us-east-1.googleapis.com",
+            "https://iamcredentials.US-WEST-1.googleapis.com",
+            "https://us-east-1-iamcredentials.googleapis.com",
+            "https://US-WEST-1-iamcredentials.googleapis.com",
+            "https://us-west-1-iamcredentials.googleapis.com/path?query",
+        ]
+
+        for url in valid_urls:
+            # A valid url shouldn't throw exception and a None value should be returned
+            external_account.Credentials.validate_service_account_impersonation_url(url)
+
+    def test_invalid_service_account_impersonate_url_shall_throw_exceptions(self):
+        invalid_urls = [
+            "https://sts.googleapis.com",
+            "iamcredentials.googleapis.com",
+            "https://",
+            "http://iamcredentials.googleapis.com",
+            "https://iamcre.dentials.googleapis.com",
+            "https://us-eas\t-1.iamcredentials.googleapis.com",
+            "https:/us-east-1.iamcredentials.googleapis.com",
+            "https://US-WE/ST-1-iamcredentials.googleapis.com",
+            "https://iamcredentials-us-east-1.googleapis.com",
+            "https://iamcredentials-US-WEST-1.googleapis.com",
+            "testhttps://us-east-1.iamcredentials.googleapis.com",
+            "https://us-east-1.iamcredentials.googleapis.comevil.com",
+            "https://us-east-1.us-east-1.iamcredentials.googleapis.com",
+            "https://us-ea.s.t.iamcredentials.googleapis.com",
+            "https://iamcredentials.googleapis.comevil.com",
+            "hhttps://us-east-1.iamcredentials.googleapis.com",
+            "https://us- -1.iamcredentials.googleapis.com",
+            "https://-iamcredentials.googleapis.com",
+            "https://us-east-1.iamcredentials.googleapis.com.evil.com",
+        ]
+
+        for url in invalid_urls:
+            # An invalid url should throw a ValueError exception
+            with pytest.raises(ValueError) as excinfo:
+                external_account.Credentials.validate_service_account_impersonation_url(
+                    url
+                )
+
+            assert excinfo.match(
+                "The provided service account impersonation URL is invalid."
+            )
+
     def test_default_state(self):
-        credentials = self.make_credentials()
+        credentials = self.make_credentials(
+            service_account_impersonation_url=self.SERVICE_ACCOUNT_IMPERSONATION_URL
+        )
 
+        # Token url and service account impersonation url should be set
+        assert credentials._token_url
+        assert credentials._service_account_impersonation_url
         # Not token acquired yet
         assert not credentials.token
         assert not credentials.valid
@@ -289,6 +390,31 @@ def test_default_state(self):
         assert credentials.requires_scopes
         assert not credentials.quota_project_id
 
+    def test_invalid_token_url(self):
+        with pytest.raises(ValueError) as excinfo:
+            CredentialsImpl(
+                audience=self.AUDIENCE,
+                subject_token_type=self.SUBJECT_TOKEN_TYPE,
+                token_url="https:///v1/token",
+                credential_source=self.CREDENTIAL_SOURCE,
+            )
+
+        assert excinfo.match("The provided token URL is invalid.")
+
+    def test_invalid_service_account_impersonate_url(self):
+        with pytest.raises(ValueError) as excinfo:
+            CredentialsImpl(
+                audience=self.AUDIENCE,
+                subject_token_type=self.SUBJECT_TOKEN_TYPE,
+                token_url=self.TOKEN_URL,
+                credential_source=self.CREDENTIAL_SOURCE,
+                service_account_impersonation_url=12345,  # create an exception by sending to parse url
+            )
+
+        assert excinfo.match(
+            "The provided service account impersonation URL is invalid."
+        )
+
     def test_nonworkforce_with_workforce_pool_user_project(self):
         with pytest.raises(ValueError) as excinfo:
             CredentialsImpl(