diff --git a/google/auth/_default.py b/google/auth/_default.py
index 0860c67fe..195388c9d 100644
--- a/google/auth/_default.py
+++ b/google/auth/_default.py
@@ -434,7 +434,7 @@ def _get_impersonated_service_account_credentials(filename, info, scopes):
                 filename, source_credentials_info
             )
         else:
-            raise ValueError(
+            raise exceptions.InvalidType(
                 "source credential of type {} is not supported.".format(
                     source_credentials_type
                 )
@@ -443,7 +443,7 @@ def _get_impersonated_service_account_credentials(filename, info, scopes):
         start_index = impersonation_url.rfind("/")
         end_index = impersonation_url.find(":generateAccessToken")
         if start_index == -1 or end_index == -1 or start_index > end_index:
-            raise ValueError(
+            raise exceptions.InvalidValue(
                 "Cannot extract target principal from {}".format(impersonation_url)
             )
         target_principal = impersonation_url[start_index + 1 : end_index]
diff --git a/google/auth/_helpers.py b/google/auth/_helpers.py
index 1b08ab87f..30fbafb64 100644
--- a/google/auth/_helpers.py
+++ b/google/auth/_helpers.py
@@ -22,6 +22,7 @@
 import six
 from six.moves import urllib
 
+from google.auth import exceptions
 
 # Token server doesn't provide a new a token when doing refresh unless the
 # token is expiring within 30 seconds, so refresh threshold should not be
@@ -51,10 +52,10 @@ def decorator(method):
             Callable: the same method passed in with an updated docstring.
 
         Raises:
-            ValueError: if the method already has a docstring.
+            google.auth.exceptions.InvalidOperation: if the method already has a docstring.
         """
         if method.__doc__:
-            raise ValueError("Method already has a docstring.")
+            raise exceptions.InvalidOperation("Method already has a docstring.")
 
         source_method = getattr(source_class, method.__name__)
         method.__doc__ = source_method.__doc__
@@ -101,13 +102,15 @@ def to_bytes(value, encoding="utf-8"):
             passed in if it started out as bytes.
 
     Raises:
-        ValueError: If the value could not be converted to bytes.
+        google.auth.exceptions.InvalidValue: If the value could not be converted to bytes.
     """
     result = value.encode(encoding) if isinstance(value, six.text_type) else value
     if isinstance(result, six.binary_type):
         return result
     else:
-        raise ValueError("{0!r} could not be converted to bytes".format(value))
+        raise exceptions.InvalidValue(
+            "{0!r} could not be converted to bytes".format(value)
+        )
 
 
 def from_bytes(value):
@@ -121,13 +124,15 @@ def from_bytes(value):
             if it started out as unicode.
 
     Raises:
-        ValueError: If the value could not be converted to unicode.
+        google.auth.exceptions.InvalidValue: If the value could not be converted to unicode.
     """
     result = value.decode("utf-8") if isinstance(value, six.binary_type) else value
     if isinstance(result, six.text_type):
         return result
     else:
-        raise ValueError("{0!r} could not be converted to unicode".format(value))
+        raise exceptions.InvalidValue(
+            "{0!r} could not be converted to unicode".format(value)
+        )
 
 
 def update_query(url, params, remove=None):
diff --git a/google/auth/_service_account_info.py b/google/auth/_service_account_info.py
index 157099273..b17f34f5c 100644
--- a/google/auth/_service_account_info.py
+++ b/google/auth/_service_account_info.py
@@ -20,6 +20,7 @@
 import six
 
 from google.auth import crypt
+from google.auth import exceptions
 
 
 def from_dict(data, require=None, use_rsa_signer=True):
@@ -40,7 +41,7 @@ def from_dict(data, require=None, use_rsa_signer=True):
             service account file.
 
     Raises:
-        ValueError: if the data was in the wrong format, or if one of the
+        MalformedError: if the data was in the wrong format, or if one of the
             required keys is missing.
     """
     keys_needed = set(require if require is not None else [])
@@ -48,7 +49,7 @@ def from_dict(data, require=None, use_rsa_signer=True):
     missing = keys_needed.difference(six.iterkeys(data))
 
     if missing:
-        raise ValueError(
+        raise exceptions.MalformedError(
             "Service account info was not in the expected format, missing "
             "fields {}.".format(", ".join(missing))
         )
diff --git a/google/auth/api_key.py b/google/auth/api_key.py
index 49c6ffd2d..4fdf7f276 100644
--- a/google/auth/api_key.py
+++ b/google/auth/api_key.py
@@ -20,6 +20,7 @@
 
 from google.auth import _helpers
 from google.auth import credentials
+from google.auth import exceptions
 
 
 class Credentials(credentials.Credentials):
@@ -36,7 +37,7 @@ def __init__(self, token):
         """
         super(Credentials, self).__init__()
         if not token:
-            raise ValueError("Token must be a non-empty API key string")
+            raise exceptions.InvalidValue("Token must be a non-empty API key string")
         self.token = token
 
     @property
diff --git a/google/auth/app_engine.py b/google/auth/app_engine.py
index 1460a7d1a..7083ee614 100644
--- a/google/auth/app_engine.py
+++ b/google/auth/app_engine.py
@@ -27,6 +27,7 @@
 from google.auth import _helpers
 from google.auth import credentials
 from google.auth import crypt
+from google.auth import exceptions
 
 # pytype: disable=import-error
 try:
@@ -67,13 +68,13 @@ def get_project_id():
         str: The project ID
 
     Raises:
-        EnvironmentError: If the App Engine APIs are unavailable.
+        google.auth.exceptions.OSError: If the App Engine APIs are unavailable.
     """
     # pylint: disable=missing-raises-doc
-    # Pylint rightfully thinks EnvironmentError is OSError, but doesn't
+    # Pylint rightfully thinks google.auth.exceptions.OSError is OSError, but doesn't
     # realize it's a valid alias.
     if app_identity is None:
-        raise EnvironmentError("The App Engine APIs are not available.")
+        raise exceptions.OSError("The App Engine APIs are not available.")
     return app_identity.get_application_id()
 
 
@@ -107,13 +108,13 @@ def __init__(
                 and billing.
 
         Raises:
-            EnvironmentError: If the App Engine APIs are unavailable.
+            google.auth.exceptions.OSError: If the App Engine APIs are unavailable.
         """
         # pylint: disable=missing-raises-doc
-        # Pylint rightfully thinks EnvironmentError is OSError, but doesn't
+        # Pylint rightfully thinks google.auth.exceptions.OSError is OSError, but doesn't
         # realize it's a valid alias.
         if app_identity is None:
-            raise EnvironmentError("The App Engine APIs are not available.")
+            raise exceptions.OSError("The App Engine APIs are not available.")
 
         super(Credentials, self).__init__()
         self._scopes = scopes
diff --git a/google/auth/aws.py b/google/auth/aws.py
index 04c5e7c5d..d3038b281 100644
--- a/google/auth/aws.py
+++ b/google/auth/aws.py
@@ -123,7 +123,7 @@ def get_request_options(
         )
         # Validate provided URL.
         if not uri.hostname or uri.scheme != "https":
-            raise ValueError("Invalid AWS service URL")
+            raise exceptions.InvalidResource("Invalid AWS service URL")
 
         header_map = _generate_authentication_header_map(
             host=uri.hostname,
@@ -408,9 +408,11 @@ def __init__(
             env_id, env_version = (None, None)
 
         if env_id != "aws" or self._cred_verification_url is None:
-            raise ValueError("No valid AWS 'credential_source' provided")
+            raise exceptions.InvalidResource(
+                "No valid AWS 'credential_source' provided"
+            )
         elif int(env_version or "") != 1:
-            raise ValueError(
+            raise exceptions.InvalidValue(
                 "aws version '{}' is not supported in the current build.".format(
                     env_version
                 )
@@ -428,7 +430,7 @@ def validate_metadata_server_url_if_any(url_string, name_of_data):
         if url_string:
             url = urlparse(url_string)
             if url.hostname != "169.254.169.254" and url.hostname != "fd00:ec2::254":
-                raise ValueError(
+                raise exceptions.InvalidResource(
                     "Invalid hostname '{}' for '{}'".format(url.hostname, name_of_data)
                 )
 
diff --git a/google/auth/compute_engine/credentials.py b/google/auth/compute_engine/credentials.py
index e97fabea9..618fa5a2d 100644
--- a/google/auth/compute_engine/credentials.py
+++ b/google/auth/compute_engine/credentials.py
@@ -221,7 +221,7 @@ def __init__(
 
         if use_metadata_identity_endpoint:
             if token_uri or additional_claims or service_account_email or signer:
-                raise ValueError(
+                raise exceptions.MalformedError(
                     "If use_metadata_identity_endpoint is set, token_uri, "
                     "additional_claims, service_account_email, signer arguments"
                     " must not be set"
@@ -312,7 +312,7 @@ def with_token_uri(self, token_uri):
         # since the signer is already instantiated,
         # the request is not needed
         if self._use_metadata_identity_endpoint:
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "If use_metadata_identity_endpoint is set, token_uri" " must not be set"
             )
         else:
@@ -423,7 +423,7 @@ def sign_bytes(self, message):
                 Signer is not available if metadata identity endpoint is used.
         """
         if self._use_metadata_identity_endpoint:
-            raise ValueError(
+            raise exceptions.InvalidOperation(
                 "Signer is not available if metadata identity endpoint is used"
             )
         return self._signer.sign(message)
diff --git a/google/auth/credentials.py b/google/auth/credentials.py
index ca1032a14..4c0af7a6b 100644
--- a/google/auth/credentials.py
+++ b/google/auth/credentials.py
@@ -21,6 +21,7 @@
 import six
 
 from google.auth import _helpers, environment_vars
+from google.auth import exceptions
 
 
 @six.add_metaclass(abc.ABCMeta)
@@ -190,9 +191,9 @@ def valid(self):
         return True
 
     def refresh(self, request):
-        """Raises :class:`ValueError``, anonymous credentials cannot be
+        """Raises :class:``InvalidOperation``, anonymous credentials cannot be
         refreshed."""
-        raise ValueError("Anonymous credentials cannot be refreshed.")
+        raise exceptions.InvalidOperation("Anonymous credentials cannot be refreshed.")
 
     def apply(self, headers, token=None):
         """Anonymous credentials do nothing to the request.
@@ -200,10 +201,10 @@ def apply(self, headers, token=None):
         The optional ``token`` argument is not supported.
 
         Raises:
-            ValueError: If a token was specified.
+            google.auth.exceptions.InvalidValue: If a token was specified.
         """
         if token is not None:
-            raise ValueError("Anonymous credentials don't support tokens.")
+            raise exceptions.InvalidValue("Anonymous credentials don't support tokens.")
 
     def before_request(self, request, method, url, headers):
         """Anonymous credentials do nothing to the request."""
diff --git a/google/auth/crypt/_python_rsa.py b/google/auth/crypt/_python_rsa.py
index 797a2592b..e8595440c 100644
--- a/google/auth/crypt/_python_rsa.py
+++ b/google/auth/crypt/_python_rsa.py
@@ -29,6 +29,7 @@
 import six
 
 from google.auth import _helpers
+from google.auth import exceptions
 from google.auth.crypt import base
 
 _POW2 = (128, 64, 32, 16, 8, 4, 2, 1)
@@ -101,7 +102,7 @@ def from_string(cls, public_key):
             der = rsa.pem.load_pem(public_key, "CERTIFICATE")
             asn1_cert, remaining = decoder.decode(der, asn1Spec=Certificate())
             if remaining != b"":
-                raise ValueError("Unused bytes", remaining)
+                raise exceptions.InvalidValue("Unused bytes", remaining)
 
             cert_info = asn1_cert["tbsCertificate"]["subjectPublicKeyInfo"]
             key_bytes = _bit_list_to_bytes(cert_info["subjectPublicKey"])
@@ -162,12 +163,12 @@ def from_string(cls, key, key_id=None):
         elif marker_id == 1:
             key_info, remaining = decoder.decode(key_bytes, asn1Spec=_PKCS8_SPEC)
             if remaining != b"":
-                raise ValueError("Unused bytes", remaining)
+                raise exceptions.InvalidValue("Unused bytes", remaining)
             private_key_info = key_info.getComponentByName("privateKey")
             private_key = rsa.key.PrivateKey.load_pkcs1(
                 private_key_info.asOctets(), format="DER"
             )
         else:
-            raise ValueError("No key could be detected.")
+            raise exceptions.MalformedError("No key could be detected.")
 
         return cls(private_key, key_id=key_id)
diff --git a/google/auth/crypt/base.py b/google/auth/crypt/base.py
index c98d5bf64..573211d7c 100644
--- a/google/auth/crypt/base.py
+++ b/google/auth/crypt/base.py
@@ -20,6 +20,7 @@
 
 import six
 
+from google.auth import exceptions
 
 _JSON_FILE_PRIVATE_KEY = "private_key"
 _JSON_FILE_PRIVATE_KEY_ID = "private_key_id"
@@ -106,7 +107,7 @@ def from_service_account_info(cls, info):
             ValueError: If the info is not in the expected format.
         """
         if _JSON_FILE_PRIVATE_KEY not in info:
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "The private_key field was not found in the service account " "info."
             )
 
diff --git a/google/auth/downscoped.py b/google/auth/downscoped.py
index a1d7b6e46..a84ac4af6 100644
--- a/google/auth/downscoped.py
+++ b/google/auth/downscoped.py
@@ -54,6 +54,7 @@
 
 from google.auth import _helpers
 from google.auth import credentials
+from google.auth import exceptions
 from google.oauth2 import sts
 
 # The maximum number of access boundary rules a Credential Access Boundary can
@@ -86,8 +87,8 @@ def __init__(self, rules=[]):
                 access boundary rules limiting the access that a downscoped credential
                 will have.
         Raises:
-            TypeError: If any of the rules are not a valid type.
-            ValueError: If the provided rules exceed the maximum allowed.
+            InvalidType: If any of the rules are not a valid type.
+            InvalidValue: If the provided rules exceed the maximum allowed.
         """
         self.rules = rules
 
@@ -113,18 +114,18 @@ def rules(self, value):
                 access boundary rules limiting the access that a downscoped credential
                 will have.
         Raises:
-            TypeError: If any of the rules are not a valid type.
-            ValueError: If the provided rules exceed the maximum allowed.
+            InvalidType: If any of the rules are not a valid type.
+            InvalidValue: If the provided rules exceed the maximum allowed.
         """
         if len(value) > _MAX_ACCESS_BOUNDARY_RULES_COUNT:
-            raise ValueError(
+            raise exceptions.InvalidValue(
                 "Credential access boundary rules can have a maximum of {} rules.".format(
                     _MAX_ACCESS_BOUNDARY_RULES_COUNT
                 )
             )
         for access_boundary_rule in value:
             if not isinstance(access_boundary_rule, AccessBoundaryRule):
-                raise TypeError(
+                raise exceptions.InvalidType(
                     "List of rules provided do not contain a valid 'google.auth.downscoped.AccessBoundaryRule'."
                 )
         # Make a copy of the original list.
@@ -138,17 +139,17 @@ def add_rule(self, rule):
                 limiting the access that a downscoped credential will have, to be added to
                 the existing rules.
         Raises:
-            TypeError: If any of the rules are not a valid type.
-            ValueError: If the provided rules exceed the maximum allowed.
+            InvalidType: If any of the rules are not a valid type.
+            InvalidValue: If the provided rules exceed the maximum allowed.
         """
         if len(self.rules) == _MAX_ACCESS_BOUNDARY_RULES_COUNT:
-            raise ValueError(
+            raise exceptions.InvalidValue(
                 "Credential access boundary rules can have a maximum of {} rules.".format(
                     _MAX_ACCESS_BOUNDARY_RULES_COUNT
                 )
             )
         if not isinstance(rule, AccessBoundaryRule):
-            raise TypeError(
+            raise exceptions.InvalidType(
                 "The provided rule does not contain a valid 'google.auth.downscoped.AccessBoundaryRule'."
             )
         self._rules.append(rule)
@@ -197,8 +198,8 @@ def __init__(
                 specific Cloud Storage objects.
 
         Raises:
-            TypeError: If any of the parameters are not of the expected types.
-            ValueError: If any of the parameters are not of the expected values.
+            InvalidType: If any of the parameters are not of the expected types.
+            InvalidValue: If any of the parameters are not of the expected values.
         """
         self.available_resource = available_resource
         self.available_permissions = available_permissions
@@ -221,10 +222,12 @@ def available_resource(self, value):
             value (str): The updated value of the available resource.
 
         Raises:
-            TypeError: If the value is not a string.
+            google.auth.exceptions.InvalidType: If the value is not a string.
         """
         if not isinstance(value, six.string_types):
-            raise TypeError("The provided available_resource is not a string.")
+            raise exceptions.InvalidType(
+                "The provided available_resource is not a string."
+            )
         self._available_resource = value
 
     @property
@@ -245,16 +248,16 @@ def available_permissions(self, value):
             value (Sequence[str]): The updated value of the available permissions.
 
         Raises:
-            TypeError: If the value is not a list of strings.
-            ValueError: If the value is not valid.
+            InvalidType: If the value is not a list of strings.
+            InvalidValue: If the value is not valid.
         """
         for available_permission in value:
             if not isinstance(available_permission, six.string_types):
-                raise TypeError(
+                raise exceptions.InvalidType(
                     "Provided available_permissions are not a list of strings."
                 )
             if available_permission.find("inRole:") != 0:
-                raise ValueError(
+                raise exceptions.InvalidValue(
                     "available_permissions must be prefixed with 'inRole:'."
                 )
         # Make a copy of the original list.
@@ -279,11 +282,11 @@ def availability_condition(self, value):
                 value of the availability condition.
 
         Raises:
-            TypeError: If the value is not of type google.auth.downscoped.AvailabilityCondition
+            google.auth.exceptions.InvalidType: If the value is not of type google.auth.downscoped.AvailabilityCondition
                 or None.
         """
         if not isinstance(value, AvailabilityCondition) and value is not None:
-            raise TypeError(
+            raise exceptions.InvalidType(
                 "The provided availability_condition is not a 'google.auth.downscoped.AvailabilityCondition' or None."
             )
         self._availability_condition = value
@@ -326,8 +329,8 @@ def __init__(self, expression, title=None, description=None):
             description (Optional[str]): Optional details about the purpose of the condition.
 
         Raises:
-            TypeError: If any of the parameters are not of the expected types.
-            ValueError: If any of the parameters are not of the expected values.
+            InvalidType: If any of the parameters are not of the expected types.
+            InvalidValue: If any of the parameters are not of the expected values.
         """
         self.expression = expression
         self.title = title
@@ -350,10 +353,10 @@ def expression(self, value):
             value (str): The updated value of the condition expression.
 
         Raises:
-            TypeError: If the value is not of type string.
+            google.auth.exceptions.InvalidType: If the value is not of type string.
         """
         if not isinstance(value, six.string_types):
-            raise TypeError("The provided expression is not a string.")
+            raise exceptions.InvalidType("The provided expression is not a string.")
         self._expression = value
 
     @property
@@ -373,10 +376,10 @@ def title(self, value):
             value (Optional[str]): The updated value of the title.
 
         Raises:
-            TypeError: If the value is not of type string or None.
+            google.auth.exceptions.InvalidType: If the value is not of type string or None.
         """
         if not isinstance(value, six.string_types) and value is not None:
-            raise TypeError("The provided title is not a string or None.")
+            raise exceptions.InvalidType("The provided title is not a string or None.")
         self._title = value
 
     @property
@@ -396,10 +399,12 @@ def description(self, value):
             value (Optional[str]): The updated value of the description.
 
         Raises:
-            TypeError: If the value is not of type string or None.
+            google.auth.exceptions.InvalidType: If the value is not of type string or None.
         """
         if not isinstance(value, six.string_types) and value is not None:
-            raise TypeError("The provided description is not a string or None.")
+            raise exceptions.InvalidType(
+                "The provided description is not a string or None."
+            )
         self._description = value
 
     def to_json(self):
diff --git a/google/auth/exceptions.py b/google/auth/exceptions.py
index 7760c87b8..fcbe61b74 100644
--- a/google/auth/exceptions.py
+++ b/google/auth/exceptions.py
@@ -74,3 +74,27 @@ def __init__(self, message=None, **kwargs):
 
 class ReauthSamlChallengeFailError(ReauthFailError):
     """An exception for SAML reauth challenge failures."""
+
+
+class MalformedError(DefaultCredentialsError, ValueError):
+    """An exception for malformed data."""
+
+
+class InvalidResource(DefaultCredentialsError, ValueError):
+    """An exception for URL error."""
+
+
+class InvalidOperation(DefaultCredentialsError, ValueError):
+    """An exception for invalid operation."""
+
+
+class InvalidValue(DefaultCredentialsError, ValueError):
+    """Used to wrap general ValueError of python."""
+
+
+class InvalidType(DefaultCredentialsError, TypeError):
+    """Used to wrap general TypeError of python."""
+
+
+class OSError(DefaultCredentialsError, EnvironmentError):
+    """Used to wrap EnvironmentError(OSError after python3.3)."""
diff --git a/google/auth/external_account.py b/google/auth/external_account.py
index 4249529e8..d24b22837 100644
--- a/google/auth/external_account.py
+++ b/google/auth/external_account.py
@@ -151,7 +151,7 @@ def __init__(
 
         if not self.is_workforce_pool and self._workforce_pool_user_project:
             # Workload identity pools do not support workforce pool user projects.
-            raise ValueError(
+            raise exceptions.InvalidValue(
                 "workforce_pool_user_project should not be set for non-workforce pool "
                 "credentials"
             )
@@ -445,7 +445,9 @@ def validate_token_url(token_url, url_type="token"):
         ]
 
         if not Credentials.is_valid_url(_TOKEN_URL_PATTERNS, token_url):
-            raise ValueError("The provided {} URL is invalid.".format(url_type))
+            raise exceptions.InvalidResource(
+                "The provided {} URL is invalid.".format(url_type)
+            )
 
     @staticmethod
     def validate_service_account_impersonation_url(url):
@@ -460,7 +462,7 @@ def validate_service_account_impersonation_url(url):
         if not Credentials.is_valid_url(
             _SERVICE_ACCOUNT_IMPERSONATION_URL_PATTERNS, url
         ):
-            raise ValueError(
+            raise exceptions.InvalidResource(
                 "The provided service account impersonation URL is invalid."
             )
 
@@ -498,7 +500,7 @@ def from_info(cls, info, **kwargs):
                 credentials.
 
         Raises:
-            ValueError: For invalid parameters.
+            InvalidValue: For invalid parameters.
         """
         return cls(
             audience=info.get("audience"),
diff --git a/google/auth/external_account_authorized_user.py b/google/auth/external_account_authorized_user.py
index 51e7f2058..04a7f5b91 100644
--- a/google/auth/external_account_authorized_user.py
+++ b/google/auth/external_account_authorized_user.py
@@ -118,7 +118,7 @@ def __init__(
         self._scopes = scopes
 
         if not self.valid and not self.can_refresh:
-            raise ValueError(
+            raise exceptions.InvalidOperation(
                 "Token should be created with fields to make it valid (`token` and "
                 "`expiry`), or fields to allow it to refresh (`refresh_token`, "
                 "`token_url`, `client_id`, `client_secret`)."
diff --git a/google/auth/identity_pool.py b/google/auth/identity_pool.py
index 5fa9faef9..ebe50883c 100644
--- a/google/auth/identity_pool.py
+++ b/google/auth/identity_pool.py
@@ -122,11 +122,11 @@ def __init__(
             # environment_id is only supported in AWS or dedicated future external
             # account credentials.
             if "environment_id" in credential_source:
-                raise ValueError(
+                raise exceptions.MalformedError(
                     "Invalid Identity Pool credential_source field 'environment_id'"
                 )
             if self._credential_source_format_type not in ["text", "json"]:
-                raise ValueError(
+                raise exceptions.MalformedError(
                     "Invalid credential_source format '{}'".format(
                         self._credential_source_format_type
                     )
@@ -137,18 +137,18 @@ def __init__(
                     "subject_token_field_name"
                 )
                 if self._credential_source_field_name is None:
-                    raise ValueError(
+                    raise exceptions.MalformedError(
                         "Missing subject_token_field_name for JSON credential_source format"
                     )
             else:
                 self._credential_source_field_name = None
 
         if self._credential_source_file and self._credential_source_url:
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "Ambiguous credential_source. 'file' is mutually exclusive with 'url'."
             )
         if not self._credential_source_file and not self._credential_source_url:
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "Missing credential_source. A 'file' or 'url' must be provided."
             )
 
diff --git a/google/auth/jwt.py b/google/auth/jwt.py
index 21de8fe95..390368958 100644
--- a/google/auth/jwt.py
+++ b/google/auth/jwt.py
@@ -122,7 +122,9 @@ def _decode_jwt_segment(encoded_section):
     try:
         return json.loads(section_bytes.decode("utf-8"))
     except ValueError as caught_exc:
-        new_exc = ValueError("Can't parse segment: {0}".format(section_bytes))
+        new_exc = exceptions.MalformedError(
+            "Can't parse segment: {0}".format(section_bytes)
+        )
         six.raise_from(new_exc, caught_exc)
 
 
@@ -137,13 +139,14 @@ def _unverified_decode(token):
             signature.
 
     Raises:
-        ValueError: if there are an incorrect amount of segments in the token or
-        segments of the wrong type.
+        google.auth.exceptions.MalformedError: if there are an incorrect amount of segments in the token or segments of the wrong type.
     """
     token = _helpers.to_bytes(token)
 
     if token.count(b".") != 2:
-        raise ValueError("Wrong number of segments in token: {0}".format(token))
+        raise exceptions.MalformedError(
+            "Wrong number of segments in token: {0}".format(token)
+        )
 
     encoded_header, encoded_payload, signature = token.split(b".")
     signed_section = encoded_header + b"." + encoded_payload
@@ -154,12 +157,12 @@ def _unverified_decode(token):
     payload = _decode_jwt_segment(encoded_payload)
 
     if not isinstance(header, Mapping):
-        raise ValueError(
+        raise exceptions.MalformedError(
             "Header segment should be a JSON object: {0}".format(encoded_header)
         )
 
     if not isinstance(payload, Mapping):
-        raise ValueError(
+        raise exceptions.MalformedError(
             "Payload segment should be a JSON object: {0}".format(encoded_payload)
         )
 
@@ -193,14 +196,17 @@ def _verify_iat_and_exp(payload, clock_skew_in_seconds=0):
             validation.
 
     Raises:
-        ValueError: if any checks failed.
+        google.auth.exceptions.InvalidValue: if value validation failed.
+        google.auth.exceptions.MalformedError: if schema validation failed.
     """
     now = _helpers.datetime_to_secs(_helpers.utcnow())
 
     # Make sure the iat and exp claims are present.
     for key in ("iat", "exp"):
         if key not in payload:
-            raise ValueError("Token does not contain required claim {}".format(key))
+            raise exceptions.MalformedError(
+                "Token does not contain required claim {}".format(key)
+            )
 
     # Make sure the token wasn't issued in the future.
     iat = payload["iat"]
@@ -208,7 +214,7 @@ def _verify_iat_and_exp(payload, clock_skew_in_seconds=0):
     # for clock skew.
     earliest = iat - clock_skew_in_seconds
     if now < earliest:
-        raise ValueError(
+        raise exceptions.InvalidValue(
             "Token used too early, {} < {}. Check that your computer's clock is set correctly.".format(
                 now, iat
             )
@@ -220,7 +226,7 @@ def _verify_iat_and_exp(payload, clock_skew_in_seconds=0):
     # to account for clow skew.
     latest = exp + clock_skew_in_seconds
     if latest < now:
-        raise ValueError("Token expired, {} < {}".format(latest, now))
+        raise exceptions.InvalidValue("Token expired, {} < {}".format(latest, now))
 
 
 def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=0):
@@ -246,7 +252,8 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=
         Mapping[str, str]: The deserialized JSON payload in the JWT.
 
     Raises:
-        ValueError: if any verification checks failed.
+        google.auth.exceptions.InvalidValue: if value validation failed.
+        google.auth.exceptions.MalformedError: if schema validation failed.
     """
     header, payload, signed_section, signature = _unverified_decode(token)
 
@@ -263,7 +270,7 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=
     except KeyError as exc:
         if key_alg in _CRYPTOGRAPHY_BASED_ALGORITHMS:
             six.raise_from(
-                ValueError(
+                exceptions.InvalidValue(
                     "The key algorithm {} requires the cryptography package "
                     "to be installed.".format(key_alg)
                 ),
@@ -271,7 +278,10 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=
             )
         else:
             six.raise_from(
-                ValueError("Unsupported signature algorithm {}".format(key_alg)), exc
+                exceptions.InvalidValue(
+                    "Unsupported signature algorithm {}".format(key_alg)
+                ),
+                exc,
             )
 
     # If certs is specified as a dictionary of key IDs to certificates, then
@@ -279,7 +289,9 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=
     if isinstance(certs, Mapping):
         if key_id:
             if key_id not in certs:
-                raise ValueError("Certificate for key id {} not found.".format(key_id))
+                raise exceptions.MalformedError(
+                    "Certificate for key id {} not found.".format(key_id)
+                )
             certs_to_check = [certs[key_id]]
         # If there's no key id in the header, check against all of the certs.
         else:
@@ -291,7 +303,7 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=
     if not crypt.verify_signature(
         signed_section, signature, certs_to_check, verifier_cls
     ):
-        raise ValueError("Could not verify token signature.")
+        raise exceptions.MalformedError("Could not verify token signature.")
 
     # Verify the issued at and created times in the payload.
     _verify_iat_and_exp(payload, clock_skew_in_seconds)
@@ -302,7 +314,7 @@ def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=
         if isinstance(audience, str):
             audience = [audience]
         if claim_audience not in audience:
-            raise ValueError(
+            raise exceptions.InvalidValue(
                 "Token has wrong audience {}, expected one of {}".format(
                     claim_audience, audience
                 )
@@ -414,7 +426,7 @@ def _from_signer_and_info(cls, signer, info, **kwargs):
             google.auth.jwt.Credentials: The constructed credentials.
 
         Raises:
-            ValueError: If the info is not in the expected format.
+            google.auth.exceptions.MalformedError: If the info is not in the expected format.
         """
         kwargs.setdefault("subject", info["client_email"])
         kwargs.setdefault("issuer", info["client_email"])
@@ -433,7 +445,7 @@ def from_service_account_info(cls, info, **kwargs):
             google.auth.jwt.Credentials: The constructed credentials.
 
         Raises:
-            ValueError: If the info is not in the expected format.
+            google.auth.exceptions.MalformedError: If the info is not in the expected format.
         """
         signer = _service_account_info.from_dict(info, require=["client_email"])
         return cls._from_signer_and_info(signer, info, **kwargs)
@@ -651,7 +663,7 @@ def _from_signer_and_info(cls, signer, info, **kwargs):
             google.auth.jwt.OnDemandCredentials: The constructed credentials.
 
         Raises:
-            ValueError: If the info is not in the expected format.
+            google.auth.exceptions.MalformedError: If the info is not in the expected format.
         """
         kwargs.setdefault("subject", info["client_email"])
         kwargs.setdefault("issuer", info["client_email"])
@@ -670,7 +682,7 @@ def from_service_account_info(cls, info, **kwargs):
             google.auth.jwt.OnDemandCredentials: The constructed credentials.
 
         Raises:
-            ValueError: If the info is not in the expected format.
+            google.auth.exceptions.MalformedError: If the info is not in the expected format.
         """
         signer = _service_account_info.from_dict(info, require=["client_email"])
         return cls._from_signer_and_info(signer, info, **kwargs)
diff --git a/google/auth/pluggable.py b/google/auth/pluggable.py
index b4fa448b8..7fef36112 100644
--- a/google/auth/pluggable.py
+++ b/google/auth/pluggable.py
@@ -93,7 +93,8 @@ def __init__(
         Raises:
             google.auth.exceptions.RefreshError: If an error is encountered during
                 access token retrieval logic.
-            ValueError: For invalid parameters.
+            google.auth.exceptions.InvalidValue: For invalid parameters.
+            google.auth.exceptions.MalformedError: For invalid parameters.
 
         .. note:: Typically one of the helper constructors
             :meth:`from_file` or
@@ -111,12 +112,12 @@ def __init__(
         )
         if not isinstance(credential_source, Mapping):
             self._credential_source_executable = None
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "Missing credential_source. The credential_source is not a dict."
             )
         self._credential_source_executable = credential_source.get("executable")
         if not self._credential_source_executable:
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "Missing credential_source. An 'executable' must be provided."
             )
         self._credential_source_executable_command = self._credential_source_executable.get(
@@ -136,7 +137,7 @@ def __init__(
         self._tokeninfo_username = ""
 
         if not self._credential_source_executable_command:
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "Missing command field. Executable command must be provided."
             )
         if not self._credential_source_executable_timeout_millis:
@@ -149,7 +150,7 @@ def __init__(
             or self._credential_source_executable_timeout_millis
             > EXECUTABLE_TIMEOUT_MILLIS_UPPER_BOUND
         ):
-            raise ValueError("Timeout must be between 5 and 120 seconds.")
+            raise exceptions.InvalidValue("Timeout must be between 5 and 120 seconds.")
 
         if self._credential_source_executable_interactive_timeout_millis:
             if (
@@ -158,7 +159,7 @@ def __init__(
                 or self._credential_source_executable_interactive_timeout_millis
                 > EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_UPPER_BOUND
             ):
-                raise ValueError(
+                raise exceptions.InvalidValue(
                     "Interactive timeout must be between 30 seconds and 30 minutes."
                 )
 
@@ -183,7 +184,7 @@ def retrieve_subject_token(self, request):
                         "expiration_time" not in response
                     ):  # Always treat missing expiration_time as expired and proceed to executable run.
                         raise exceptions.RefreshError
-                except ValueError:
+                except (exceptions.MalformedError, exceptions.InvalidValue):
                     raise
                 except exceptions.RefreshError:
                     pass
@@ -247,7 +248,9 @@ def revoke(self, request):
 
         """
         if not self.interactive:
-            raise ValueError("Revoke is only enabled under interactive mode.")
+            raise exceptions.InvalidValue(
+                "Revoke is only enabled under interactive mode."
+            )
         self._validate_running_mode()
 
         if not _helpers.is_python_3():
@@ -307,7 +310,8 @@ def from_info(cls, info, **kwargs):
                 credentials.
 
         Raises:
-            ValueError: For invalid parameters.
+            google.auth.exceptions.InvalidValue: For invalid parameters.
+            google.auth.exceptions.MalformedError: For invalid parameters.
         """
         return super(Credentials, cls).from_info(info, **kwargs)
 
@@ -344,7 +348,7 @@ def _parse_subject_token(self, response):
         self._validate_response_schema(response)
         if not response["success"]:
             if "code" not in response or "message" not in response:
-                raise ValueError(
+                raise exceptions.MalformedError(
                     "Error code and message fields are required in the response."
                 )
             raise exceptions.RefreshError(
@@ -357,7 +361,9 @@ def _parse_subject_token(self, response):
                 "The token returned by the executable is expired."
             )
         if "token_type" not in response:
-            raise ValueError("The executable response is missing the token_type field.")
+            raise exceptions.MalformedError(
+                "The executable response is missing the token_type field."
+            )
         if (
             response["token_type"] == "urn:ietf:params:oauth:token-type:jwt"
             or response["token_type"] == "urn:ietf:params:oauth:token-type:id_token"
@@ -375,7 +381,9 @@ def _validate_revoke_response(self, response):
 
     def _validate_response_schema(self, response):
         if "version" not in response:
-            raise ValueError("The executable response is missing the version field.")
+            raise exceptions.MalformedError(
+                "The executable response is missing the version field."
+            )
         if response["version"] > EXECUTABLE_SUPPORTED_MAX_VERSION:
             raise exceptions.RefreshError(
                 "Executable returned unsupported version {}.".format(
@@ -384,19 +392,21 @@ def _validate_response_schema(self, response):
             )
 
         if "success" not in response:
-            raise ValueError("The executable response is missing the success field.")
+            raise exceptions.MalformedError(
+                "The executable response is missing the success field."
+            )
 
     def _validate_running_mode(self):
         env_allow_executables = os.environ.get(
             "GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES"
         )
         if env_allow_executables != "1":
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "Executables need to be explicitly allowed (set GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES to '1') to run."
             )
 
         if self.interactive and not self._credential_source_executable_output_file:
-            raise ValueError(
+            raise exceptions.MalformedError(
                 "An output_file must be specified in the credential configuration for interactive mode."
             )
 
@@ -404,9 +414,11 @@ def _validate_running_mode(self):
             self.interactive
             and not self._credential_source_executable_interactive_timeout_millis
         ):
-            raise ValueError(
+            raise exceptions.InvalidOperation(
                 "Interactive mode cannot run without an interactive timeout."
             )
 
         if self.interactive and not self.is_workforce_pool:
-            raise ValueError("Interactive mode is only enabled for workforce pool.")
+            raise exceptions.InvalidValue(
+                "Interactive mode is only enabled for workforce pool."
+            )
diff --git a/google/auth/transport/_aiohttp_requests.py b/google/auth/transport/_aiohttp_requests.py
index 179cadbdf..364570e31 100644
--- a/google/auth/transport/_aiohttp_requests.py
+++ b/google/auth/transport/_aiohttp_requests.py
@@ -140,7 +140,7 @@ class Request(transport.Request):
     def __init__(self, session=None):
         # TODO: Use auto_decompress property for aiohttp 3.7+
         if session is not None and session._auto_decompress:
-            raise ValueError(
+            raise exceptions.InvalidOperation(
                 "Client sessions with auto_decompress=True are not supported."
             )
         self.session = session
diff --git a/google/auth/transport/grpc.py b/google/auth/transport/grpc.py
index 87fa5042f..55764b6f6 100644
--- a/google/auth/transport/grpc.py
+++ b/google/auth/transport/grpc.py
@@ -255,7 +255,7 @@ def my_client_cert_callback():
     google_auth_credentials = grpc.metadata_call_credentials(metadata_plugin)
 
     if ssl_credentials and client_cert_callback:
-        raise ValueError(
+        raise exceptions.MalformedError(
             "Received both ssl_credentials and client_cert_callback; "
             "these are mutually exclusive."
         )