FM / botocore

/usr/lib/rads/venv/lib/python3.13/site-packages/botocore/

/usr/lib/rads/venv/lib/python3.13/site-packages/botocore

Name	Size	Mode	Actions
crt/	-	0755	rm
data/	-	0755	rm
docs/	-	0755	rm
retries/	-	0755	rm
vendored/	-	0755	rm
__pycache__/	-	0755	rm
args.py	38460	0644	edit dl rm
auth.py	46062	0644	edit dl rm
awsrequest.py	23178	0644	edit dl rm
cacert.pem	266617	0644	edit dl rm
client.py	54154	0644	edit dl rm
compat.py	11607	0644	edit dl rm
compress.py	4533	0644	edit dl rm
config.py	20092	0644	edit dl rm
configloader.py	10039	0644	edit dl rm
configprovider.py	38092	0644	edit dl rm
context.py	3782	0644	edit dl rm
credentials.py	88913	0644	edit dl rm
discovery.py	11069	0644	edit dl rm
endpoint.py	16739	0644	edit dl rm
endpoint_provider.py	23005	0644	edit dl rm
errorfactory.py	3722	0644	edit dl rm
eventstream.py	20279	0644	edit dl rm
exceptions.py	23333	0644	edit dl rm
handlers.py	66230	0644	edit dl rm
history.py	1744	0644	edit dl rm
hooks.py	25053	0644	edit dl rm
httpchecksum.py	19452	0644	edit dl rm
httpsession.py	18696	0644	edit dl rm
loaders.py	18834	0644	edit dl rm
model.py	32347	0644	edit dl rm
monitoring.py	20594	0644	edit dl rm
paginate.py	27909	0644	edit dl rm
parsers.py	60400	0644	edit dl rm
plugin.py	2534	0644	edit dl rm
regions.py	33932	0644	edit dl rm
response.py	7865	0644	edit dl rm
retryhandler.py	14702	0644	edit dl rm
serialize.py	49017	0644	edit dl rm
session.py	51661	0644	edit dl rm
signers.py	34519	0644	edit dl rm
stub.py	16193	0644	edit dl rm
tokens.py	11764	0644	edit dl rm
translate.py	3406	0644	edit dl rm
useragent.py	23171	0644	edit dl rm
utils.py	136764	0644	edit dl rm
validate.py	13759	0644	edit dl rm
waiter.py	14711	0644	edit dl rm
__init__.py	8019	0644	edit dl rm

Edit: /usr/lib/rads/venv/lib/python3.13/site-packages/botocore/tokens.py (11764B)

# Copyright 2022 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
# http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.
import json
import logging
import os
import threading
from datetime import datetime, timedelta
from typing import NamedTuple, Optional

import dateutil.parser
from dateutil.tz import tzutc

from botocore import UNSIGNED
from botocore.compat import total_seconds
from botocore.config import Config
from botocore.exceptions import (
    ClientError,
    InvalidConfigError,
    TokenRetrievalError,
)
from botocore.utils import (
    CachedProperty,
    JSONFileCache,
    SSOTokenLoader,
    create_nested_client,
    get_token_from_environment,
)

logger = logging.getLogger(__name__)

def _utc_now():
    return datetime.now(tzutc())

def create_token_resolver(session):
    providers = [
        ScopedEnvTokenProvider(session),
        SSOTokenProvider(session),
    ]
    return TokenProviderChain(providers=providers)

def _serialize_utc_timestamp(obj):
    if isinstance(obj, datetime):
        return obj.strftime("%Y-%m-%dT%H:%M:%SZ")
    return obj

def _sso_json_dumps(obj):
    return json.dumps(obj, default=_serialize_utc_timestamp)

class FrozenAuthToken(NamedTuple):
    token: str
    expiration: Optional[datetime] = None

class DeferredRefreshableToken:
    # The time at which we'll attempt to refresh, but not block if someone else
    # is refreshing.
    _advisory_refresh_timeout = 15 * 60
    # The time at which all threads will block waiting for a refreshed token
    _mandatory_refresh_timeout = 10 * 60
    # Refresh at most once every minute to avoid blocking every request
    _attempt_timeout = 60

def __init__(self, method, refresh_using, time_fetcher=_utc_now):
        self._time_fetcher = time_fetcher
        self._refresh_using = refresh_using
        self.method = method

# The frozen token is protected by this lock
        self._refresh_lock = threading.Lock()
        self._frozen_token = None
        self._next_refresh = None

def get_frozen_token(self):
        self._refresh()
        return self._frozen_token

def _refresh(self):
        # If we don't need to refresh just return
        refresh_type = self._should_refresh()
        if not refresh_type:
            return None

# Block for refresh if we're in the mandatory refresh window
        block_for_refresh = refresh_type == "mandatory"
        if self._refresh_lock.acquire(block_for_refresh):
            try:
                self._protected_refresh()
            finally:
                self._refresh_lock.release()

def _protected_refresh(self):
        # This should only be called after acquiring the refresh lock
        # Another thread may have already refreshed, double check refresh
        refresh_type = self._should_refresh()
        if not refresh_type:
            return None

try:
            now = self._time_fetcher()
            self._next_refresh = now + timedelta(seconds=self._attempt_timeout)
            self._frozen_token = self._refresh_using()
        except Exception:
            logger.warning(
                "Refreshing token failed during the %s refresh period.",
                refresh_type,
                exc_info=True,
            )
            if refresh_type == "mandatory":
                # This refresh was mandatory, error must be propagated back
                raise

if self._is_expired():
            # Fresh credentials should never be expired
            raise TokenRetrievalError(
                provider=self.method,
                error_msg="Token has expired and refresh failed",
            )

def _is_expired(self):
        if self._frozen_token is None:
            return False

expiration = self._frozen_token.expiration
        remaining = total_seconds(expiration - self._time_fetcher())
        return remaining <= 0

def _should_refresh(self):
        if self._frozen_token is None:
            # We don't have a token yet, mandatory refresh
            return "mandatory"

expiration = self._frozen_token.expiration
        if expiration is None:
            # No expiration, so assume we don't need to refresh.
            return None

now = self._time_fetcher()
        if now < self._next_refresh:
            return None

remaining = total_seconds(expiration - now)

if remaining < self._mandatory_refresh_timeout:
            return "mandatory"
        elif remaining < self._advisory_refresh_timeout:
            return "advisory"

return None

class TokenProviderChain:
    def __init__(self, providers=None):
        if providers is None:
            providers = []
        self._providers = providers

def load_token(self, **kwargs):
        for provider in self._providers:
            token = provider.load_token(**kwargs)
            if token is not None:
                return token
        return None

class SSOTokenProvider:
    METHOD = "sso"
    _REFRESH_WINDOW = 15 * 60
    _SSO_TOKEN_CACHE_DIR = os.path.expanduser(
        os.path.join("~", ".aws", "sso", "cache")
    )
    _SSO_CONFIG_VARS = [
        "sso_start_url",
        "sso_region",
    ]
    _GRANT_TYPE = "refresh_token"
    DEFAULT_CACHE_CLS = JSONFileCache

def __init__(
        self, session, cache=None, time_fetcher=_utc_now, profile_name=None
    ):
        self._session = session
        if cache is None:
            cache = self.DEFAULT_CACHE_CLS(
                self._SSO_TOKEN_CACHE_DIR,
                dumps_func=_sso_json_dumps,
            )
        self._now = time_fetcher
        self._cache = cache
        self._token_loader = SSOTokenLoader(cache=self._cache)
        self._profile_name = (
            profile_name
            or self._session.get_config_variable("profile")
            or 'default'
        )

def _load_sso_config(self):
        loaded_config = self._session.full_config
        profiles = loaded_config.get("profiles", {})
        sso_sessions = loaded_config.get("sso_sessions", {})
        profile_config = profiles.get(self._profile_name, {})

if "sso_session" not in profile_config:
            return

sso_session_name = profile_config["sso_session"]
        sso_config = sso_sessions.get(sso_session_name, None)

if not sso_config:
            error_msg = (
                f'The profile "{self._profile_name}" is configured to use the SSO '
                f'token provider but the "{sso_session_name}" sso_session '
                f"configuration does not exist."
            )
            raise InvalidConfigError(error_msg=error_msg)

missing_configs = []
        for var in self._SSO_CONFIG_VARS:
            if var not in sso_config:
                missing_configs.append(var)

if missing_configs:
            error_msg = (
                f'The profile "{self._profile_name}" is configured to use the SSO '
                f"token provider but is missing the following configuration: "
                f"{missing_configs}."
            )
            raise InvalidConfigError(error_msg=error_msg)

return {
            "session_name": sso_session_name,
            "sso_region": sso_config["sso_region"],
            "sso_start_url": sso_config["sso_start_url"],
        }

@CachedProperty
    def _sso_config(self):
        return self._load_sso_config()

@CachedProperty
    def _client(self):
        config = Config(
            region_name=self._sso_config["sso_region"],
            signature_version=UNSIGNED,
        )
        return create_nested_client(self._session, "sso-oidc", config=config)

def _attempt_create_token(self, token):
        response = self._client.create_token(
            grantType=self._GRANT_TYPE,
            clientId=token["clientId"],
            clientSecret=token["clientSecret"],
            refreshToken=token["refreshToken"],
        )
        expires_in = timedelta(seconds=response["expiresIn"])
        new_token = {
            "startUrl": self._sso_config["sso_start_url"],
            "region": self._sso_config["sso_region"],
            "accessToken": response["accessToken"],
            "expiresAt": self._now() + expires_in,
            # Cache the registration alongside the token
            "clientId": token["clientId"],
            "clientSecret": token["clientSecret"],
            "registrationExpiresAt": token["registrationExpiresAt"],
        }
        if "refreshToken" in response:
            new_token["refreshToken"] = response["refreshToken"]
        logger.info("SSO Token refresh succeeded")
        return new_token

def _refresh_access_token(self, token):
        keys = (
            "refreshToken",
            "clientId",
            "clientSecret",
            "registrationExpiresAt",
        )
        missing_keys = [k for k in keys if k not in token]
        if missing_keys:
            msg = f"Unable to refresh SSO token: missing keys: {missing_keys}"
            logger.info(msg)
            return None

expiry = dateutil.parser.parse(token["registrationExpiresAt"])
        if total_seconds(expiry - self._now()) <= 0:
            logger.info("SSO token registration expired at %s", expiry)
            return None

try:
            return self._attempt_create_token(token)
        except ClientError:
            logger.warning("SSO token refresh attempt failed", exc_info=True)
            return None

def _refresher(self):
        start_url = self._sso_config["sso_start_url"]
        session_name = self._sso_config["session_name"]
        logger.info("Loading cached SSO token for %s", session_name)
        token_dict = self._token_loader(start_url, session_name=session_name)
        expiration = dateutil.parser.parse(token_dict["expiresAt"])
        logger.debug("Cached SSO token expires at %s", expiration)

remaining = total_seconds(expiration - self._now())
        if remaining < self._REFRESH_WINDOW:
            new_token_dict = self._refresh_access_token(token_dict)
            if new_token_dict is not None:
                token_dict = new_token_dict
                expiration = token_dict["expiresAt"]
                self._token_loader.save_token(
                    start_url, token_dict, session_name=session_name
                )

return FrozenAuthToken(
            token_dict["accessToken"], expiration=expiration
        )

def load_token(self, **kwargs):
        if self._sso_config is None:
            return None

return DeferredRefreshableToken(
            self.METHOD, self._refresher, time_fetcher=self._now
        )

class ScopedEnvTokenProvider:
    """
    Token provider that loads tokens from environment variables scoped to
    a specific `signing_name`.
    """

METHOD = 'env'

def __init__(self, session, environ=None):
        self._session = session
        if environ is None:
            environ = os.environ
        self.environ = environ

def load_token(self, **kwargs):
        signing_name = kwargs.get("signing_name")
        if signing_name is None:
            return None

token = get_token_from_environment(signing_name, self.environ)

if token is not None:
            logger.info("Found token in environment variables.")
            return FrozenAuthToken(token)

cmd: