Add type stubs for google-auth-library-python

stubs/google-auth/METADATA.toml

@@ -0,0 +1,7 @@
+version = "2.48.*"
+upstream-repository = "https://github.com/googleapis/google-auth-library-python"
+dependencies = [
+    "types-requests",
+    "types-grpcio",
+    "cryptography",
+]

stubs/google-auth/google/auth/__init__.pyi

@@ -0,0 +1,9 @@
+from google.auth._default import (
+    default as default,
+    load_credentials_from_dict as load_credentials_from_dict,
+    load_credentials_from_file as load_credentials_from_file,
+)
+
+__all__ = ["default", "load_credentials_from_file", "load_credentials_from_dict"]
+
+class Python37DeprecationWarning(DeprecationWarning): ...

stubs/google-auth/google/auth/_agent_identity_utils.pyi

@@ -0,0 +1,10 @@
+from typing import Any
+
+CRYPTOGRAPHY_NOT_FOUND_ERROR: str
+
+def get_agent_identity_certificate_path() -> str | None: ...
+def get_and_parse_agent_identity_certificate() -> Any | None: ...
+def parse_certificate(cert_bytes: bytes) -> Any: ...
+def calculate_certificate_fingerprint(cert: Any) -> str: ...
+def should_request_bound_token(cert: Any) -> bool: ...
+def get_cached_cert_fingerprint(cached_cert: bytes | None) -> str: ...

stubs/google-auth/google/auth/_cache.pyi

@@ -0,0 +1,15 @@
+from typing import Hashable, TypeVar
+
+_Key = TypeVar("_Key", bound=Hashable)
+_Value = TypeVar("_Value")
+
+class LRUCache(dict[_Key, _Value]):
+    maxsize: int
+
+    def __init__(self, maxsize: int) -> None: ...
+    def clear(self) -> None: ...
+    def __getitem__(self, key: _Key) -> _Value: ...
+    def __setitem__(self, key: _Key, value: _Value) -> None: ...
+    def __delitem__(self, key: _Key) -> None: ...
+    def popitem(self) -> tuple[_Key, _Value]: ...
+    def _update(self, key: _Key) -> None: ...

stubs/google-auth/google/auth/_cloud_sdk.pyi

@@ -0,0 +1,6 @@
+CLOUD_SDK_CLIENT_ID: str
+
+def get_config_path() -> str: ...
+def get_application_default_credentials_path() -> str: ...
+def get_project_id() -> str | None: ...
+def get_auth_access_token(account: str | None = None) -> str: ...

stubs/google-auth/google/auth/_constants.pyi

@@ -0,0 +1,3 @@
+_SERVICE_ACCOUNT_TRUST_BOUNDARY_LOOKUP_ENDPOINT: str
+_WORKFORCE_POOL_TRUST_BOUNDARY_LOOKUP_ENDPOINT: str
+_WORKLOAD_IDENTITY_POOL_TRUST_BOUNDARY_LOOKUP_ENDPOINT: str

stubs/google-auth/google/auth/_credentials_async.pyi

@@ -0,0 +1,20 @@
+import abc
+from typing import Mapping, Sequence
+
+from google.auth import credentials
+from google.auth.transport import Request as _Request
+
+class Credentials(credentials.Credentials, metaclass=abc.ABCMeta):
+    async def before_request(self, request: _Request, method: str, url: str, headers: Mapping[str, str]) -> None: ...
+
+class CredentialsWithQuotaProject(credentials.CredentialsWithQuotaProject, metaclass=abc.ABCMeta): ...
+
+class AnonymousCredentials(credentials.AnonymousCredentials, Credentials):
+    async def before_request(self, request: _Request, method: str, url: str, headers: Mapping[str, str]) -> None: ...
+
+class ReadOnlyScoped(credentials.ReadOnlyScoped, metaclass=abc.ABCMeta): ...
+class Scoped(credentials.Scoped, metaclass=abc.ABCMeta): ...
+
+def with_scopes_if_required(credentials: Credentials, scopes: Sequence[str]) -> Credentials: ...
+
+class Signing(credentials.Signing, metaclass=abc.ABCMeta): ...

stubs/google-auth/google/auth/_credentials_base.pyi

@@ -0,0 +1,12 @@
+import abc
+from _typeshed import Incomplete
+from typing import Any, Coroutine
+
+from google.auth.transport import Request as _TransportRequest
+
+class _BaseCredentials(metaclass=abc.ABCMeta):
+    token: Incomplete
+
+    def __init__(self) -> None: ...
+    @abc.abstractmethod
+    def refresh(self, request: _TransportRequest) -> None | Coroutine[Any, Any, None]: ...

stubs/google-auth/google/auth/_default.pyi

@@ -0,0 +1,29 @@
+from typing import TYPE_CHECKING, Any, Mapping, Sequence
+
+from google.auth.credentials import Credentials as Credentials
+from google.auth.transport import Request as Request
+
+if TYPE_CHECKING:
+    from google.auth.api_key import Credentials as _ApiKeyCredentials
+
+def load_credentials_from_file(
+    filename: str,
+    scopes: Sequence[str] | None = None,
+    default_scopes: Sequence[str] | None = None,
+    quota_project_id: str | None = None,
+    request: Request | None = None,
+) -> tuple[Credentials, str | None]: ...
+def load_credentials_from_dict(
+    info: Mapping[str, Any],
+    scopes: Sequence[str] | None = None,
+    default_scopes: Sequence[str] | None = None,
+    quota_project_id: str | None = None,
+    request: Request | None = None,
+) -> tuple[Credentials, str | None]: ...
+def get_api_key_credentials(key: str) -> _ApiKeyCredentials: ...
+def default(
+    scopes: Sequence[str] | None = None,
+    request: Request | None = None,
+    quota_project_id: str | None = None,
+    default_scopes: Sequence[str] | None = None,
+) -> tuple[Credentials, str | None]: ...

stubs/google-auth/google/auth/_default_async.pyi

@@ -0,0 +1,11 @@
+from typing import Sequence
+
+from google.auth.credentials import Credentials as Credentials
+from google.auth.transport import Request as _Request
+
+def load_credentials_from_file(
+    filename: str, scopes: Sequence[str] | None = None, quota_project_id: str | None = None
+) -> tuple[Credentials, str | None]: ...
+def default_async(
+    scopes: Sequence[str] | None = None, request: _Request | None = None, quota_project_id: str | None = None
+) -> tuple[Credentials, str | None]: ...

stubs/google-auth/google/auth/_exponential_backoff.pyi

@@ -0,0 +1,22 @@
+class _BaseExponentialBackoff:
+    def __init__(
+        self,
+        total_attempts: int = 3,
+        initial_wait_seconds: float = 1.0,
+        randomization_factor: float = 0.1,
+        multiplier: float = 2.0,
+    ) -> None: ...
+    @property
+    def total_attempts(self) -> int: ...
+    @property
+    def backoff_count(self) -> int: ...
+
+class ExponentialBackoff(_BaseExponentialBackoff):
+    def __init__(self, *args: object, **kwargs: object) -> None: ...
+    def __iter__(self) -> ExponentialBackoff: ...
+    def __next__(self) -> int: ...
+
+class AsyncExponentialBackoff(_BaseExponentialBackoff):
+    def __init__(self, *args: object, **kwargs: object) -> None: ...
+    def __aiter__(self) -> AsyncExponentialBackoff: ...
+    async def __anext__(self) -> int: ...

stubs/google-auth/google/auth/_helpers.pyi

@@ -0,0 +1,23 @@
+import datetime
+import logging
+from typing import Any, Callable, Mapping, Sequence
+
+REFRESH_THRESHOLD: datetime.timedelta
+
+def copy_docstring(source_class: type) -> Callable[[Any], Any]: ...
+def parse_content_type(header_value: str) -> str: ...
+def utcnow() -> datetime.datetime: ...
+def utcfromtimestamp(timestamp: float) -> datetime.datetime: ...
+def datetime_to_secs(value: datetime.datetime) -> int: ...
+def to_bytes(value: str | bytes, encoding: str = "utf-8") -> bytes: ...
+def from_bytes(value: str | bytes) -> str: ...
+def update_query(url: str, params: Mapping[str, str], remove: Sequence[str] | None = None) -> str: ...
+def scopes_to_string(scopes: Sequence[str]) -> str: ...
+def string_to_scopes(scopes: Sequence[str] | str) -> list[str]: ...
+def padded_urlsafe_b64decode(value: str | bytes) -> bytes: ...
+def unpadded_urlsafe_b64encode(value: str | bytes) -> str | bytes: ...
+def get_bool_from_env(variable_name: str, default: bool = False) -> bool: ...
+def is_python_3() -> bool: ...
+def is_logging_enabled(logger: logging.Logger) -> bool: ...
+def request_log(logger: logging.Logger, method: str, url: str, body: bytes | None, headers: Mapping[str, str] | None) -> None: ...
+def response_log(logger: logging.Logger, response: Any) -> None: ...

stubs/google-auth/google/auth/_jwt_async.pyi

@@ -0,0 +1,84 @@
+from typing import Any, Mapping
+
+from google.auth import _credentials_async, jwt
+from google.auth.credentials import Signing
+from google.auth.crypt import Signer as _Signer
+from google.auth.transport import Request as _Request
+
+_DEFAULT_TOKEN_LIFETIME_SECS: int
+_DEFAULT_MAX_CACHE_SIZE: int
+_ALGORITHM_TO_VERIFIER_CLASS: dict[str, type]
+_CRYPTOGRAPHY_BASED_ALGORITHMS: frozenset[str]
+
+def encode(
+    signer: _Signer, payload: Mapping[str, str], header: Mapping[str, str] | None = None, key_id: str | None = None
+) -> bytes: ...
+def _decode_jwt_segment(encoded_section: bytes) -> Mapping[str, Any]: ...
+def _unverified_decode(token: str | bytes) -> tuple[Mapping[str, Any], Mapping[str, Any], bytes, bytes]: ...
+def decode_header(token: str | bytes) -> Mapping[str, Any]: ...
+def _verify_iat_and_exp(payload: Mapping[str, str], clock_skew_in_seconds: int = 0) -> None: ...
+def decode(
+    token: str,
+    certs: str | bytes | Mapping[str, str | bytes] | None = None,
+    verify: bool = True,
+    audience: str | list[str] | None = None,
+    clock_skew_in_seconds: int = 0,
+) -> Mapping[str, Any]: ...
+
+class Credentials(jwt.Credentials, _credentials_async.Credentials):
+    expiry: Any
+
+    def __init__(
+        self,
+        signer: _Signer,
+        issuer: str,
+        subject: str,
+        audience: str,
+        additional_claims: Mapping[str, str] | None = None,
+        token_lifetime: int = ...,
+        quota_project_id: str | None = None,
+    ) -> None: ...
+    @classmethod
+    def _from_signer_and_info(cls, signer: _Signer, info: Mapping[str, str], **kwargs: Any) -> Credentials: ...
+    @classmethod
+    def from_service_account_info(cls, info: Mapping[str, str], **kwargs: Any) -> Credentials: ...
+    @classmethod
+    def from_service_account_file(cls, filename: str, **kwargs: Any) -> Credentials: ...
+    @classmethod
+    def from_signing_credentials(cls, credentials: Signing, audience: str, **kwargs: Any) -> Credentials: ...
+    def with_claims(
+        self,
+        issuer: str | None = None,
+        subject: str | None = None,
+        audience: str | None = None,
+        additional_claims: Mapping[str, str] | None = None,
+    ) -> Credentials: ...
+    def with_quota_project(self, quota_project_id: str | None) -> Credentials: ...
+    def refresh(self, request: Any) -> None: ...
+    def sign_bytes(self, message: bytes) -> bytes: ...
+    @property
+    def signer_email(self) -> str: ...
+    @property
+    def signer(self) -> _Signer: ...
+    @property
+    def additional_claims(self) -> Mapping[str, str]: ...
+
+class OnDemandCredentials(jwt.OnDemandCredentials, _credentials_async.Credentials):
+    """On-demand JWT credentials.
+
+    Like :class:`Credentials`, this class uses a JWT as the bearer token for
+    authentication. However, this class does not require the audience at
+    construction time. Instead, it will generate a new token on-demand for
+    each request using the request URI as the audience. It caches tokens
+    so that multiple requests to the same URI do not incur the overhead
+    of generating a new token every time.
+
+    This behavior is especially useful for `gRPC`_ clients. A gRPC service may
+    have multiple audience and gRPC clients may not know all of the audiences
+    required for accessing a particular service. With these credentials,
+    no knowledge of the audiences is required ahead of time.
+
+    .. _grpc: http://www.grpc.io/
+    """
+
+    async def before_request(self, request: _Request, method: str, url: str, headers: Mapping[str, str]) -> None: ...

stubs/google-auth/google/auth/_oauth2client.pyi

@@ -0,0 +1,3 @@
+from typing import Any
+
+def convert(credentials: Any) -> Any: ...

stubs/google-auth/google/auth/_refresh_worker.pyi

@@ -0,0 +1,11 @@
+import threading
+from typing import Any
+
+class RefreshThreadManager:
+    def __init__(self) -> None: ...
+    def start_refresh(self, cred: Any, request: Any) -> bool: ...
+    def clear_error(self) -> None: ...
+
+class RefreshThread(threading.Thread):
+    def __init__(self, cred: Any, request: Any, **kwargs: Any) -> None: ...
+    def run(self) -> None: ...

stubs/google-auth/google/auth/_service_account_info.pyi

@@ -0,0 +1,8 @@
+from typing import Mapping, Sequence
+
+from google.auth import crypt as _crypt
+
+def from_dict(data: Mapping[str, str], require: Sequence[str] | None = None, use_rsa_signer: bool = True) -> _crypt.Signer: ...
+def from_filename(
+    filename: str, require: Sequence[str] | None = None, use_rsa_signer: bool = True
+) -> tuple[Mapping[str, str], _crypt.Signer]: ...

stubs/google-auth/google/auth/aio/__init__.pyi

@@ -0,0 +1 @@
+__version__: str

stubs/google-auth/google/auth/aio/_helpers.pyi

@@ -0,0 +1,4 @@
+import logging
+from typing import Any
+
+async def response_log_async(logger: logging.Logger, response: Any) -> None: ...

stubs/google-auth/google/auth/aio/credentials.pyi

@@ -0,0 +1,23 @@
+from _typeshed import Incomplete
+from typing import Mapping
+
+from google.auth._credentials_base import _BaseCredentials
+from google.auth.transport import Request as _TransportRequest
+
+class Credentials(_BaseCredentials):
+    def __init__(self) -> None: ...
+    async def apply(self, headers: Mapping[str, str], token: str | None = None) -> None: ...
+    async def refresh(self, request: _TransportRequest) -> None: ...
+    async def before_request(self, request: _TransportRequest, method: str, url: str, headers: Mapping[str, str]) -> None: ...
+
+class StaticCredentials(Credentials):
+    token: Incomplete
+
+    def __init__(self, token: str) -> None: ...
+    async def refresh(self, request: _TransportRequest) -> None: ...
+    async def before_request(self, request: _TransportRequest, method: str, url: str, headers: Mapping[str, str]) -> None: ...
+
+class AnonymousCredentials(Credentials):
+    async def refresh(self, request: _TransportRequest) -> None: ...
+    async def apply(self, headers: Mapping[str, str], token: str | None = None) -> None: ...
+    async def before_request(self, request: _TransportRequest, method: str, url: str, headers: Mapping[str, str]) -> None: ...

stubs/google-auth/google/auth/aio/transport/__init__.pyi

@@ -0,0 +1,26 @@
+import abc
+from typing import Any, AsyncGenerator, Mapping, Sequence
+
+DEFAULT_RETRYABLE_STATUS_CODES: Sequence[int]
+DEFAULT_MAX_RETRY_ATTEMPTS: int
+
+class Response(metaclass=abc.ABCMeta):
+    @property
+    @abc.abstractmethod
+    def status_code(self) -> int: ...
+    @property
+    @abc.abstractmethod
+    def headers(self) -> Mapping[str, str]: ...
+    @abc.abstractmethod
+    async def content(self, chunk_size: int) -> AsyncGenerator[bytes]: ...
+    @abc.abstractmethod
+    async def read(self) -> bytes: ...
+    @abc.abstractmethod
+    async def close(self) -> None: ...
+
+class Request(metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    async def __call__(
+        self, url: str, method: str, body: bytes | None, headers: Mapping[str, str] | None, timeout: float, **kwargs: Any
+    ) -> Response: ...
+    async def close(self) -> None: ...

stubs/google-auth/google/auth/aio/transport/aiohttp.pyi

@@ -0,0 +1,29 @@
+from typing import Any, AsyncGenerator, Mapping
+
+from google.auth.aio import transport
+
+ClientTimeout = Any
+
+class Response(transport.Response):
+    def __init__(self, response: Any) -> None: ...
+    @property
+    def status_code(self) -> int: ...
+    @property
+    def headers(self) -> Mapping[str, str]: ...
+    async def content(self, chunk_size: int = 1024) -> AsyncGenerator[bytes]: ...
+    async def read(self) -> bytes: ...
+    async def close(self) -> None: ...
+
+class Request(transport.Request):
+    def __init__(self, session: Any | None = None) -> None: ...
+    async def __call__(
+        self,
+        url: str,
+        method: str = "GET",
+        body: bytes | None = None,
+        headers: Mapping[str, str] | None = None,
+        timeout: float | ClientTimeout = ...,
+        *args: Any,
+        **kwargs: Any,
+    ) -> transport.Response: ...
+    async def close(self) -> None: ...

stubs/google-auth/google/auth/aio/transport/mtls.pyi

@@ -0,0 +1,11 @@
+import ssl
+from typing import Callable
+
+def make_client_cert_ssl_context(cert_bytes: bytes, key_bytes: bytes, passphrase: bytes | None = None) -> ssl.SSLContext: ...
+def default_client_cert_source() -> Callable[[], tuple[bytes, bytes]]: ...
+async def get_client_ssl_credentials(
+    certificate_config_path: str | None = None,
+) -> tuple[bool, bytes | None, bytes | None, bytes | None]: ...
+async def get_client_cert_and_key(
+    client_cert_callback: Callable[[], tuple[bytes, bytes]] | None = None,
+) -> tuple[bool, bytes | None, bytes | None]: ...