# © 2025 EarthDaily Analytics Corp.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License 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 base64
import datetime
import errno
import json
import os
import random
import stat
import tempfile
import threading
import warnings
from hashlib import sha1
from descarteslabs.exceptions import AuthError, OauthError
try:
# public client
from ..core.common.http import Retry, Session
except ImportError:
# inside monorepo
from ..common.http import Retry, Session
# This is only for the existing DL production tenant, and must remain in place
# until the tenant is completely replaced, if ever.
LEGACY_DELEGATION_CLIENT_IDS = ["ZOBAi4UROl5gKZIpxxlwOEfx8KpqXf2c"]
# copied from descarteslabs/common/threading/local.py, but we need
# it standalone here to avoid any dependencies on our own packages
# for client configuration purposes
class ThreadLocalWrapper(object):
"""
A wrapper around a thread-local object that gets created lazily in every
thread of every process via the given factory callable when it is
accessed. I.e., at most one instance per thread exists.
In contrast to standard thread-locals this is compatible with multiple
processes.
"""
def __init__(self, factory):
self._factory = factory
self._create_local(os.getpid())
def get(self):
self._init_local()
if not hasattr(self._local, "wrapped"):
self._local.wrapped = self._factory()
return self._local.wrapped
def _init_local(self):
local_pid = os.getpid()
previous_pid = getattr(self._local, "_pid", None)
if previous_pid is None:
self._local._pid = local_pid
elif local_pid != previous_pid:
self._create_local(local_pid)
def _create_local(self, pid):
self._local = threading.local()
self._local._pid = pid
DEFAULT_TOKEN_INFO_DIR = os.path.join(os.path.expanduser("~"), ".descarteslabs")
DEFAULT_TOKEN_INFO_PATH = os.path.join(DEFAULT_TOKEN_INFO_DIR, "token_info.json")
JWT_TOKEN_PREFIX = "jwt_token_"
DESCARTESLABS_CLIENT_ID = "DESCARTESLABS_CLIENT_ID"
DESCARTESLABS_CLIENT_SECRET = "DESCARTESLABS_CLIENT_SECRET"
DESCARTESLABS_REFRESH_TOKEN = "DESCARTESLABS_REFRESH_TOKEN"
DESCARTESLABS_TOKEN = "DESCARTESLABS_TOKEN"
DESCARTESLABS_TOKEN_INFO_PATH = "DESCARTESLABS_TOKEN_INFO_PATH"
DESCARTESLABS_CUSTOM_CLAIM_PREFIX = "earthdaily__dl__"
def base64url_decode(input):
"""Helper method to base64url_decode a string.
Parameter
---------
input : str
A base64url_encoded string to decode.
"""
rem = len(input) % 4
if rem > 0:
input += b"=" * (4 - rem)
return base64.urlsafe_b64decode(input)
def makedirs_if_not_exists(path):
if not os.path.exists(path):
try:
os.makedirs(path)
os.chmod(path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
except OSError as ex:
if ex.errno == errno.EEXIST:
pass
else:
raise
def get_default_domain():
# See if we know the environment we're in, and if so use the
# correct `iam_url`. Use a default if we don't know the environment
from descarteslabs.config import peek_settings
class DummyAuth:
payload = {}
return peek_settings().iam_url
def get_app_domain():
from descarteslabs.config import peek_settings
class DummyAuth:
payload = {}
return peek_settings().app_url
[docs]class Auth:
"""Client used to authenticate with all Descartes Labs service APIs."""
RETRY_CONFIG = Retry(
total=5,
backoff_factor=random.uniform(1, 10),
allowed_methods=frozenset(["GET", "POST"]),
status_forcelist=[429, 500, 502, 503, 504],
)
AUTHORIZATION_ERROR = (
"No valid authentication info found{}. "
"See https://docs.descarteslabs.com/authentication.html."
)
KEY_CLIENT_ID = "client_id"
KEY_CLIENT_SECRET = "client_secret"
KEY_REFRESH_TOKEN = "refresh_token"
KEY_SCOPE = "scope"
KEY_GRANT_TYPE = "grant_type"
KEY_TARGET = "target"
KEY_API_TYPE = "api_type"
KEY_JWT_TOKEN = "jwt_token"
KEY_ALT_JWT_TOKEN = "JWT_TOKEN"
# The various prefixes that can be used in Catalog ACLs.
ACL_PREFIX_USER = "user:" # Followed by the user's sha1 hash
ACL_PREFIX_EMAIL = "email:" # Followed by the user's email
ACL_PREFIX_GROUP = "group:" # Followed by a lowercase group
ACL_PREFIX_ORG = "org:" # Followed by a lowercase org name
ACL_PREFIX_ACCESS = "access-id:" # Followed by the purchase-specific access id
# Note that the access-id, including the prefix `access_id:`, is matched against
# a group with the same name. In other words `group:access-id:<access-id>` will
# match against `access-id:<access-id>` (assuming the `<access_id>` is identical).
# these match the values in descarteslabs/common/services/python_auth/groups.py
ORG_ADMIN_SUFFIX = ":org-admin"
RESOURCE_ADMIN_SUFFIX = ":resource-admin"
# These are cache keys for caching various data in the object's __dict__.
# These are scrubbed out with `_clear_cache()` when retrieving a new token.
KEY_PAYLOAD = "_payload"
KEY_ALL_ACL_SUBJECTS = "_aas"
KEY_ALL_ACL_SUBJECTS_AS_SET = "_aasas"
KEY_ALL_OWNER_ACL_SUBJECTS = "_aoas"
KEY_ALL_OWNER_ACL_SUBJECTS_AS_SET = "_aoasas"
__attrs__ = [
"domain",
"scope",
"leeway",
"token_info_path",
"client_id",
"client_secret",
"refresh_token",
"_token",
"_namespace",
"RETRY_CONFIG",
]
_default_token_info_path = object() # Just any unique object
_instance = None # the default Auth instance
def __init__(
self,
domain=None,
scope=None,
leeway=500,
token_info_path=_default_token_info_path,
client_id=None,
client_secret=None,
jwt_token=None,
refresh_token=None,
retries=None,
_suppress_warning=False,
):
"""Retrieves a JWT access token from a client id and refresh token for cli usage.
By default and without arguments the credentials are retrieved from a
config file named ``token_info.json``. This file can be created by running
``descarteslabs auth login`` from the command line.
You can change the default location by setting the environment variable
``DESCARTESLABS_TOKEN_INFO_PATH``. Make sure you do this **before** running
``descarteslabs auth login`` so the credentials will be saved to the file
specified in the environment variable, and when still set when instantiating
this class, the credentials will be read from that file.
To use a short-lived access token that will not be refreshed, either set the
environment variable ``DESCARTESLABS_TOKEN`` or use the ``jwt_token`` parameter.
To use a long-lived refresh token that will be refreshed, either set the
environment variables ``DESCARTESLABS_CLIENT_ID`` and
``DESCARTESLABS_CLIENT_SECRET`` or use the parameters ``client_id`` and
``client_secret``. This will retrieve an access token which will be cached
between instances for the same combination of client id and client secret.
If in addition to the client id and client secret you also specify a valid
short-lived access token, it will be used until it expires.
Note that the environment variable ``DESCARTESLABS_REFRESH_TOKEN`` is identical
to ``DESCARTESLABS_CLIENT_SECRET`` and the parameter ``refresh_token`` is
identical to ``client_secret``. Use one or the other but not both.
Although discouraged, it is possible to set one value as environment variable,
and pass the other value in as parameter. For example, one could set the
environment variable ``DESCARTESLABS_CLIENT_ID`` and only pass in the parameter
``client_secret``.
If you also specify a ``token_info_path`` that indicates which file to
read the credentials from. If used by itself, it works the same as
``DESCARTESLABS_TOKEN_INFO_PATH`` and assuming the file exists and contains
valid credentials, you could switch between accounts this way.
If you specify the ``token_info_path`` together with an additional
client id and client secret (whether retrieved through environment
variables or given using parameters), the given credentials will be
written to the given file. If this file already exists and contains
matching credentials, it will be used to retrieve the short-lived
access token and refreshes it when it expires. If the file already
exists and contains conflicting credentials, it will be overwritten
with the new credentials.
Parameters
----------
domain : str, default ``descarteslabs.config.get_settings().IAM_URL``
The domain used for the credentials. You should normally never
change this.
scope : list(str), optional
The JWT access token fields to be included. You should normally
never have to use this.
leeway : int, default 500
The leeway is given in seconds and is used as a safety cushion
for the expiration. If the expiration falls within the leeway,
the JWT access token will be renewed.
token_info_path : str, default ``~/.descarteslabs/token_info.json``
Path to a JSON file holding the credentials. If not set and
credentials are provided through environment variables or through
parameters, this parameter will **not** be used. However, if no
credentials are provided through environment variables or through
parameters, it will default to ``~/.descarteslabs/token_info.json``
and credentials will be retrieved from that file if present. If
explicitly set to ``None``, credentials will never be retrieved
from file and **must** be provided through environment variables
or parameters.
client_id : str, optional
The JWT client id. If provided it will take precedence over the
corresponding environment variable, or the credentials retrieved through
the file specified in ``token_info_path``. If this parameter is provided,
you **must** either provide a ``client_secret`` or ``refresh_token`` (but not
both). Access tokens retrieved this way will be cached without revealing
the client secret.
client_secret : str, optional
The refresh token used to retrieve short-lived access tokens. If provided
it will take precedence over the corresponding environment variable, or the
credentials retrieved through the file specified in ``token_info_path``. If
this parameter is provided, you **must** also provide a client id either as
a parameter or through an environment variable. Access tokens retrieved this
way will be cached without revealing the client secret.
jwt_token : str, optional
A short-lived JWT access token. If valid and used without other parameters,
it will be used for access. If used with a client id, the access token must
match or it will be discarded. If the access token is discarded either
because it expired or didn't match the given client id, and no client secret
has been given, no new access token can be retrieved and access will be
denied. If used with both client id and client secret, the token will be
cached and updated as needed without revealing the client secret.
refresh_token : str, optional
Identical to the ``client_secret``. You can only specify one or the other,
or if specified both, they must match. The refresh token takes precedence
over the client secret.
retries : Retry or int, optional
The number of retries and backoff policy;
by default 5 retries with a random backoff policy between 1 and 10 seconds.
Raises
------
UserWarning
In case the refresh token and client secret differ.
In case the defailt or given ``token_info_path`` cannot be found.
In case no credentials can be found.
Examples
--------
>>> import descarteslabs
>>> # Use default credentials obtained through 'descarteslabs auth login'
>>> auth = descarteslabs.auth.Auth()
>>> # Your Descartes Labs user id
>>> auth.namespace # doctest: +SKIP
'a54d88e06612d820bc3be72877c74f257b561b19'
>>> auth = descarteslabs.auth.Auth(
... client_id="some-client-id",
... client_secret="some-client-secret",
... )
>>> auth.namespace # doctest: +SKIP
'67f21eb1040f978fe1da32e5e33501d0f4a604ac'
>>>
"""
# The logic here is murky and changed over time. Initially, the logic would
# retrieve *any* of the information from *any* of the sources. This resulted in
# the `token_info.json` being overwritten when you would use a different refresh
# token set in the environment or passed in. This was changed to make a
# distinction between data that is provided through the environment or as
# arguments, versus the data that is retrieved from `token_info.json`. This still
# allows arbitrary combinations of data provided through the environment and
# passed in as arguments.
# In addition there are duplicate keys and arguments, which makes things even
# more unnecessarily complicated. For backward compatibility reasons we keep it
# as-is. Overall the core information consists of:
# client_id: The oauth application id.
# client_secret: Same as refresh_token.
# refresh_token: The oauth application refresh token. Refresh token has
# precedence over client_secret.
# _token: The short-lived jwt id token that can be generated from the
# refresh token if present.
self.token_info_path = token_info_path
if token_info_path is Auth._default_token_info_path:
token_info_path = None
self.token_info_path = os.environ.get(
DESCARTESLABS_TOKEN_INFO_PATH, DEFAULT_TOKEN_INFO_PATH
)
token_info = {}
# First determine if we are getting our info from the args or environment
self.client_id = next(
(
x
for x in (
client_id,
os.environ.get(DESCARTESLABS_CLIENT_ID),
os.environ.get("CLIENT_ID"),
)
if x is not None
),
None,
)
self.client_secret = next(
(
x
for x in (
client_secret,
os.environ.get(DESCARTESLABS_CLIENT_SECRET),
os.environ.get("CLIENT_SECRET"),
)
if x is not None
),
None,
)
self.refresh_token = next(
(
x
for x in (
refresh_token,
os.environ.get(DESCARTESLABS_REFRESH_TOKEN),
)
if x is not None
),
None,
)
self._token = next(
(
x
for x in (
jwt_token,
os.environ.get(DESCARTESLABS_TOKEN),
)
if x is not None
),
None,
)
# Make sure self.refresh_token is set
if not self.refresh_token:
self.refresh_token = self.client_secret
if self.client_id or self.refresh_token or self._token:
# Information is provided through the environment or as argument
if token_info_path:
# Explicit token_info.json file; see if we can use it...
if os.path.exists(self.token_info_path):
token_info = self._read_token_info(self.token_info_path)
if (
not self._token
and self.client_id == token_info.get(self.KEY_CLIENT_ID)
and self.refresh_token == token_info.get(self.KEY_REFRESH_TOKEN)
):
self._token = token_info.get(self.KEY_JWT_TOKEN)
elif self.refresh_token and self.token_info_path:
# Make the saved JWT token file unique to the refresh token
token = self.refresh_token
token_sha1 = sha1(token.encode("utf-8")).hexdigest()
self.token_info_path = os.path.join(
DEFAULT_TOKEN_INFO_DIR, f"{JWT_TOKEN_PREFIX}{token_sha1}.json"
)
if self._token:
self._write_token_info(
self.token_info_path, {self.KEY_JWT_TOKEN: self._token}
)
else:
self._token = self._read_token_info(
self.token_info_path, suppress_warning=True
).get(self.KEY_JWT_TOKEN)
elif self.token_info_path:
# All information comes from the cached token_info.json file
token_info = self._read_token_info(self.token_info_path, _suppress_warning)
self.client_id = token_info.get(self.KEY_CLIENT_ID)
self.client_secret = token_info.get(self.KEY_CLIENT_SECRET)
self.refresh_token = token_info.get(self.KEY_REFRESH_TOKEN)
self._token = next(
(
x
for x in (
token_info.get(self.KEY_ALT_JWT_TOKEN),
token_info.get(self.KEY_JWT_TOKEN),
)
if x is not None
),
None,
)
# The refresh token and client secret should be identical if both set
if (
self.client_secret
and self.refresh_token
and self.client_secret != self.refresh_token
):
warnings.warn(
"Authentication token mismatch: both the client secret and the "
"refresh token are provided but differ in value; "
"the refresh token will be used for authentication.",
stacklevel=2,
)
# Make sure they're identical. Refresh token has precedence.
if self.refresh_token:
self.client_secret = self.refresh_token
elif self.client_secret:
self.refresh_token = self.client_secret
self.scope = next(
(x for x in (scope, token_info.get(self.KEY_SCOPE)) if x is not None), None
)
# Verify that the token is valid; otherwise clear it
if self._token:
try:
payload = self._get_payload(self._token)
except AuthError:
self._token = None
else:
if self._token_expired(payload) or (
self.client_id and payload.get("aud") != self.client_id
):
self._token = None
if not _suppress_warning and not (
self._token or (self.client_id and self.refresh_token)
):
# Won't authn if we don't have a token or a client_id/refresh_token pair
warnings.warn(self.AUTHORIZATION_ERROR.format(""), stacklevel=2)
self._namespace = None
if retries is None:
retries = self.RETRY_CONFIG
self._retry_config = retries
self._init_session()
self.leeway = leeway
if domain is None:
domain = get_default_domain()
self.domain = domain
[docs] @classmethod
def from_environment_or_token_json(cls, **kwargs):
"""Creates an Auth object from the given arguments.
Creates an Auth object from the given arguments,
environment variables, or stored credentials.
See :py:class:`Auth` for details.
"""
return Auth(**kwargs)
def _init_session(self):
# Sessions can't be shared across threads or processes because the underlying
# SSL connection pool can't be shared. We create them thread-local to avoid
# intractable exceptions when users naively share clients e.g. when using
# multiprocessing.
self._session = ThreadLocalWrapper(self.build_session)
def _token_expired(self, payload, leeway=0):
exp = payload.get("exp")
if exp is not None:
now = (
datetime.datetime.now(datetime.timezone.utc)
- datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
).total_seconds()
return now + leeway > exp
return True # Must have exp
@property
def token(self):
"""Gets the short-lived JWT access token.
Returns
-------
str
The JWT token string.
Raises
------
AuthError
Raised when incomplete credentials were provided.
OauthError
Raised when a token cannot be obtained or refreshed.
"""
if self._token is None:
self._get_token()
else: # might have token but could be close to expiration
payload = self._get_payload(self._token)
if self._token_expired(payload, self.leeway):
try:
self._get_token()
except AuthError as e:
# Unable to refresh, raise if truly expired
if self._token_expired(payload):
raise e
return self._token
@property
def payload(self):
"""Gets the token payload.
Returns
-------
dict
Dictionary containing the fields specified by scope, which may include:
.. highlight:: none
::
name: The name of the user.
groups: Groups to which the user belongs.
org: The organization to which the user belongs.
email: The email address of the user.
email_verified: True if the user's email has been verified.
sub: The user identifier.
exp: The expiration time of the token, in seconds since
the start of the unix epoch.
Raises
------
AuthError
Raised when incomplete credentials were provided.
OauthError
Raised when a token cannot be obtained or refreshed.
"""
payload = self.__dict__.get(self.KEY_PAYLOAD)
if payload is None:
payload = self._get_payload(self.token)
# doctor custom claims
if DESCARTESLABS_CUSTOM_CLAIM_PREFIX:
for key in list(payload.keys()):
if key.startswith(DESCARTESLABS_CUSTOM_CLAIM_PREFIX):
payload[key[len(DESCARTESLABS_CUSTOM_CLAIM_PREFIX) :]] = (
payload.pop(key)
)
self.__dict__[self.KEY_PAYLOAD] = payload
return payload
@staticmethod
def _get_payload(token):
if isinstance(token, str):
token = token.encode("utf-8")
try:
# Anything that goes wrong here means it's a bad token
claims = token.split(b".")[1]
return json.loads(base64url_decode(claims).decode("utf-8"))
except Exception as e:
raise AuthError("Unable to read token {}: {}".format(token, e))
@property
def session(self):
return self._session.get()
def build_session(self):
session = Session(self.domain, retries=self._retry_config)
# local testing will not have necessary certs
if self.domain.startswith("https://dev.localhost"):
session.verify = False
return session
[docs] @staticmethod
def get_default_auth():
"""Retrieve the default Auth.
This Auth is used whenever you don't explicitly set the Auth
when creating clients, etc.
"""
if Auth._instance is None:
Auth._instance = Auth()
return Auth._instance
[docs] @staticmethod
def set_default_auth(auth):
"""Change the default Auth to the given Auth.
This is the Auth that will be used whenever you don't explicitly set the
Auth when creating clients, etc.
"""
Auth._instance = auth
@staticmethod
def _read_token_info(path, suppress_warning=False):
if os.environ.get("DESCARTESLABS_NO_JWT_CACHE", "").lower() == "true":
return {}
try:
with open(path) as fp:
return json.load(fp)
except Exception as e:
if not suppress_warning:
warnings.warn(
"Unable to read token_info from {} with error {}.".format(
path, str(e)
),
stacklevel=3,
)
return {}
@staticmethod
def _write_token_info(path, token_info):
token_info_directory = os.path.dirname(path)
temp_prefix = ".{}.".format(os.path.basename(path))
fd = None
temp_path = None
suppress_warning = False
try:
if Auth.KEY_JWT_TOKEN in token_info:
token = token_info[Auth.KEY_JWT_TOKEN]
if isinstance(token, bytes):
token_info[Auth.KEY_JWT_TOKEN] = token.decode("utf-8")
makedirs_if_not_exists(token_info_directory)
fd, temp_path = tempfile.mkstemp(
prefix=temp_prefix, dir=token_info_directory
)
if JWT_TOKEN_PREFIX in path:
token_info = {Auth.KEY_JWT_TOKEN: token_info[Auth.KEY_JWT_TOKEN]}
suppress_warning = True
try:
with os.fdopen(fd, "w+") as fp:
json.dump(token_info, fp)
finally:
fd = None # Closed now
os.chmod(temp_path, stat.S_IRUSR | stat.S_IWUSR)
try:
os.rename(temp_path, path)
except FileExistsError:
# On windows remove the file first
os.remove(path)
os.rename(temp_path, path)
except Exception as e:
if not suppress_warning:
warnings.warn(
"Failed to save token: {}".format(e),
stacklevel=3,
)
finally:
if fd is not None:
os.close(fd)
if temp_path is not None and os.path.exists(temp_path):
os.remove(temp_path)
def _get_token(self, timeout=100):
if self.client_id is None:
raise AuthError(self.AUTHORIZATION_ERROR.format(" (no client_id)"))
if self.client_secret is None and self.refresh_token is None:
raise AuthError(
self.AUTHORIZATION_ERROR.format(" (no client_secret or refresh_token)")
)
if self.client_id in LEGACY_DELEGATION_CLIENT_IDS:
if self.scope is None:
scope = ["openid", "name", "groups", "org", "email"]
else:
scope = self.scope
params = {
self.KEY_SCOPE: " ".join(scope),
self.KEY_CLIENT_ID: self.client_id,
self.KEY_GRANT_TYPE: "urn:ietf:params:oauth:grant-type:jwt-bearer",
self.KEY_TARGET: self.client_id,
self.KEY_API_TYPE: "app",
self.KEY_REFRESH_TOKEN: self.refresh_token,
}
else:
params = {
self.KEY_CLIENT_ID: self.client_id,
self.KEY_GRANT_TYPE: "refresh_token",
self.KEY_REFRESH_TOKEN: self.refresh_token,
}
if self.scope is not None:
params[self.KEY_SCOPE] = " ".join(self.scope)
r = self.session.post("/token", json=params, timeout=timeout)
if r.status_code != 200:
raise OauthError("Could not retrieve token: {}".format(r.text.strip()))
data = r.json()
access_token = data.get("access_token")
id_token = data.get("id_token") # TODO(justin) remove legacy id_token usage
if access_token is not None:
self._token = access_token
elif id_token is not None:
self._token = id_token
else:
raise OauthError("Could not retrieve token")
# clear out payload and subjects cache
self._clear_cache()
token_info = {}
# Read the token from the token_info_path, and save it again
if self.token_info_path:
token_info = self._read_token_info(
self.token_info_path, suppress_warning=True
)
if (
token_info.get(self.KEY_CLIENT_ID) != self.client_id
or token_info.get(self.KEY_CLIENT_SECRET) != self.client_secret
):
# Not matching; better rewrite!
token_info = {
self.KEY_CLIENT_ID: self.client_id,
self.KEY_CLIENT_SECRET: self.client_secret,
self.KEY_REFRESH_TOKEN: self.refresh_token,
}
token_info[self.KEY_JWT_TOKEN] = self._token
token_info.pop(self.KEY_ALT_JWT_TOKEN, None) # Remove alt key
self._write_token_info(self.token_info_path, token_info)
@property
def namespace(self):
"""Gets the user namespace (the Descartes Labs user id).
Returns
-------
str
The user namespace.
Raises
------
AuthError
Raised when incomplete credentials were provided.
OauthError
Raised when a token cannot be obtained or refreshed.
"""
namespace = self._namespace
if namespace is None:
namespace = self.payload.get("userid")
if not namespace:
# legacy, compute it on the fly
namespace = sha1(self.payload["sub"].encode("utf-8")).hexdigest()
self._namespace = namespace
return namespace
@property
def all_acl_subjects(self):
"""
A list of all ACL subjects identifying this user (the user itself, the org, the
groups) which can be used in ACL queries.
"""
subjects = self.__dict__.get(self.KEY_ALL_ACL_SUBJECTS)
if subjects is None:
subjects = [self.ACL_PREFIX_USER + self.namespace]
if email := self.payload.get("email"):
subjects.append(self.ACL_PREFIX_EMAIL + email.lower())
if org := self.payload.get("org"):
subjects.append(self.ACL_PREFIX_ORG + org)
subjects += [
self.ACL_PREFIX_GROUP + group for group in self._active_groups()
]
self.__dict__[self.KEY_ALL_ACL_SUBJECTS] = subjects
return subjects
@property
def all_acl_subjects_as_set(self):
subjects_as_set = self.__dict__.get(self.KEY_ALL_ACL_SUBJECTS_AS_SET)
if subjects_as_set is None:
subjects_as_set = set(self.all_acl_subjects)
self.__dict__[self.KEY_ALL_ACL_SUBJECTS_AS_SET] = subjects_as_set
return subjects_as_set
@property
def all_owner_acl_subjects(self):
"""
A list of ACL subjects identifying this user (the user itself, the org,
org admin and catalog admins) which can be used in owner ACL queries.
"""
subjects = self.__dict__.get(self.KEY_ALL_OWNER_ACL_SUBJECTS)
if subjects is None:
subjects = [self.ACL_PREFIX_USER + self.namespace]
subjects.extend(
[self.ACL_PREFIX_ORG + org for org in self.get_org_admins() if org]
)
subjects.extend(
[
self.ACL_PREFIX_ACCESS + access_id
for access_id in self.get_resource_admins()
if access_id
]
)
self.__dict__[self.KEY_ALL_OWNER_ACL_SUBJECTS] = subjects
return subjects
@property
def all_owner_acl_subjects_as_set(self):
subjects_as_set = self.__dict__.get(self.KEY_ALL_OWNER_ACL_SUBJECTS_AS_SET)
if subjects_as_set is None:
subjects_as_set = set(self.all_owner_acl_subjects)
self.__dict__[self.KEY_ALL_OWNER_ACL_SUBJECTS_AS_SET] = subjects_as_set
return subjects_as_set
def get_org_admins(self):
# This retrieves the value of the org to be added if the user has one or
# more org-admin groups, otherwise the empty list.
return [
group[: -len(self.ORG_ADMIN_SUFFIX)]
for group in self.payload.get("groups", [])
if group.endswith(self.ORG_ADMIN_SUFFIX)
]
def get_resource_admins(self):
# This retrieves the value of the access-id to be added if the user has one or
# more resource-admin groups, otherwise the empty list.
return [
group[: -len(self.RESOURCE_ADMIN_SUFFIX)]
for group in self.payload.get("groups", [])
if group.endswith(self.RESOURCE_ADMIN_SUFFIX)
]
def _active_groups(self):
"""
Attempts to filter groups to just the ones that are currently valid for this
user. If they have a colon, the prefix leading up to the colon must be the
user's current org, otherwise the user should not actually have rights with
this group.
"""
org = self.payload.get("org")
for group in self.payload.get("groups", []):
parts = group.split(":")
if len(parts) == 1:
yield group
elif org and parts[0] == org:
yield group
def _clear_cache(self):
for key in (
self.KEY_PAYLOAD,
self.KEY_ALL_ACL_SUBJECTS,
self.KEY_ALL_ACL_SUBJECTS_AS_SET,
self.KEY_ALL_OWNER_ACL_SUBJECTS,
self.KEY_ALL_OWNER_ACL_SUBJECTS_AS_SET,
):
if key in self.__dict__:
del self.__dict__[key]
self._namespace = None
def __getstate__(self):
return dict((attr, getattr(self, attr)) for attr in self.__attrs__)
def __setstate__(self, state):
for name, value in state.items():
setattr(self, name, value)
self._init_session()
if __name__ == "__main__":
auth = Auth.get_default_auth()
print(auth.token)
