{
    "file_diffs": [
        {
            "old_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Interface information about credentials\n\nProvides minimalistic interface to deal (query, request, store) with most common\ntypes of credentials.  To be used by Authenticators\n\"\"\"\n\n__dev_doc__ = \"\"\"\nPossibly useful in the future 3rd part developments\n\nhttps://github.com/omab/python-social-auth\n    social authentication/registration mechanism with support for several\n    frameworks and auth providers\n\"\"\"\n\nimport time\n\nfrom collections import OrderedDict\n\nfrom ..dochelpers import exc_str\nfrom ..support.exceptions import AccessDeniedError\nfrom ..support.keyring_ import keyring as keyring_\nfrom ..ui import ui\nfrom ..utils import auto_repr\nfrom ..support.network import iso8601_to_epoch\n\nfrom logging import getLogger\nlgr = getLogger('datalad.downloaders.credentials')\n\n\n@auto_repr\nclass Credential(object):\n    \"\"\"Base class for different types of credentials\n    \"\"\"\n\n    # Should be defined in a subclass as an OrderedDict of fields\n    # name -> dict(attributes)\n    _FIELDS = None\n    _KNOWN_ATTRS = {\n        'hidden',    # UI should not display the value\n        'repeat',    # UI should repeat entry or not. Set to False to override default logic\n        'optional',  # Not mandatory thus not requested if not set\n    }\n\n    def __init__(self, name, url=None, keyring=None):\n        \"\"\"\n        Parameters\n        ----------\n        name : str\n            Name of the credential, as it would be identified by in the centralized\n            storage of credentials\n        url : str, optional\n            URL string to point users to where to seek obtaining the credentials\n        keyring : a keyring\n            An object providing (g|s)et_password.  If None, keyring module is used\n            as is\n        \"\"\"\n        self.name = name\n        self.url = url\n        self._keyring = keyring or keyring_\n        self._prepare()\n\n    def _prepare(self):\n        \"\"\"Additional house-keeping possibly to be performed in subclasses\n\n        Created to avoid all the passing args/kwargs of __init__ all the time\n        \"\"\"\n        # Basic checks\n        for f, fattrs in self._FIELDS.items():\n            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)\n            if unknown_attrs:\n                raise ValueError(\"Unknown attributes %s. Known are: %s\"\n                                 % (unknown_attrs, self._KNOWN_ATTRS))\n\n    def _is_field_optional(self, f):\n        return self._FIELDS[f].get('optional', False)\n\n    @property\n    def is_known(self):\n        \"\"\"Return True if values for all fields of the credential are known\"\"\"\n        try:\n            return all(\n                self._is_field_optional(f) or self._get_field_value(f) is not None\n                for f in self._FIELDS)\n        except Exception as exc:\n            lgr.warning(\"Failed to query keyring: %s\" % exc_str(exc))\n            return False\n\n    def _get_field_value(self, field):\n        return self._keyring.get(self.name, field)\n\n    def _ask_field_value(self, f, instructions=None):\n        msg = instructions if instructions else \\\n            (\"You need to authenticate with %r credentials.\" % self.name +\n                  (\" %s provides information on how to gain access\"\n                   % self.url if self.url else ''))\n\n        # provide custom options only if set for the field\n        f_props = self._FIELDS[f]\n        kwargs = {}\n        for p in ('hidden', 'repeat'):\n            if p in f_props:\n                kwargs[p] = f_props[p]\n        return ui.question(\n            f,\n            title=msg,\n            **kwargs\n        )\n\n    def _ask_and_set(self, f, instructions=None):\n        v = self._ask_field_value(f, instructions=instructions)\n        self.set(**{f: v})\n        return v\n\n    def enter_new(self, instructions=None, **kwargs):\n        \"\"\"Enter new values for the credential fields\n\n        Parameters\n        ----------\n        instructions : str, optional\n          If given, the auto-generated instructions based on a login-URL are\n          replaced by the given string\n        **kwargs\n          Any given key value pairs with non-None values are used to set the\n          field `key` to the given value, without asking for user input\n        \"\"\"\n        if kwargs:\n            unknown_fields = set(kwargs).difference(self._FIELDS)\n            known_fields = set(self._FIELDS).difference(kwargs)\n            if unknown_fields:\n                raise ValueError(\n                    \"Unknown to %s field(s): %s.  Known but not specified: %s\"\n                    % (self,\n                       ', '.join(sorted(unknown_fields)),\n                       ', '.join(sorted(known_fields))\n                       ))\n        # Use ui., request credential fields corresponding to the type\n        for f in self._FIELDS:\n            if kwargs.get(f, None):\n                # use given value, don't ask\n                self.set(**{f: kwargs[f]})\n            elif not self._is_field_optional(f):\n                self._ask_and_set(f, instructions=instructions)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        fields = {}\n        for f in self._FIELDS:\n            v = self._get_field_value(f)\n            if not self._is_field_optional(f):\n                while v is None:  # was not known\n                    v = self._ask_and_set(f)\n                fields[f] = v\n            elif v is not None:\n                fields[f] = v\n        return fields\n\n    def set(self, **kwargs):\n        \"\"\"Set field(s) of the credential\"\"\"\n        for f, v in kwargs.items():\n            if f not in self._FIELDS:\n                raise ValueError(\"Unknown field %s. Known are: %s\"\n                                 % (f, self._FIELDS.keys()))\n            self._keyring.set(self.name, f, v)\n\n    def get(self, f, default=None):\n        \"\"\"Get a field of the credential\"\"\"\n        if f not in self._FIELDS:\n            raise ValueError(\"Unknown field %s. Known are: %s\"\n                             % (f, self._FIELDS.keys()))\n        try:\n            return self._get_field_value(f)\n        except:  # MIH: what could even happen? _keyring not a dict?\n            return default\n\n    def delete(self):\n        \"\"\"Deletes credential values from the keyring\"\"\"\n        for f in self._FIELDS:\n            self._keyring.delete(self.name, f)\n\n\nclass UserPassword(Credential):\n    \"\"\"Simple type of a credential which consists of user/password pair\"\"\"\n\n    _FIELDS = OrderedDict([('user', {}),\n                           ('password', {'hidden': True})])\n\n    is_expired = False  # no expiration provisioned\n\n\nclass Token(Credential):\n    \"\"\"Simple type of a credential which provides a single token\"\"\"\n\n    _FIELDS = OrderedDict([('token', {'hidden': True, 'repeat': False})])\n\n    is_expired = False  # no expiration provisioned\n\n\nclass AWS_S3(Credential):\n    \"\"\"Credential for AWS S3 service\"\"\"\n\n    _FIELDS = OrderedDict([('key_id', {'repeat': False}),\n                           ('secret_id', {'hidden': True, 'repeat': False}),\n                           ('session', {'optional': True}),\n                           ('expiration', {'optional': True}),\n                           ])\n\n    @property\n    def is_expired(self):\n        exp = self.get('expiration', None)\n        if not exp:\n            return False\n        exp_epoch = iso8601_to_epoch(exp)\n        # -2 to artificially shorten duration of the allotment to avoid\n        # possible race conditions between us checking either it has\n        # already expired before submitting a request.\n        expire_in = (exp_epoch - time.time() - 2) / 3600.\n\n        lgr.debug(\n            (\"Credential %s has expired %.2fh ago\"\n                if expire_in <= 0 else \"Credential %s will expire in %.2fh\")\n            % (self.name, expire_in))\n        return expire_in <= 0\n\n\n@auto_repr\nclass CompositeCredential(Credential):\n    \"\"\"Credential which represent a sequence of Credentials where front one is exposed to user\n    \"\"\"\n\n    # To be defined in sub-classes\n    _CREDENTIAL_CLASSES = None\n    _CREDENTIAL_ADAPTERS = None\n\n    def _prepare(self):\n        assert len(self._CREDENTIAL_CLASSES) > 1, \"makes sense only if there is > 1 credential\"\n        assert len(self._CREDENTIAL_CLASSES) == len(self._CREDENTIAL_ADAPTERS) + 1, \\\n            \"there should be 1 less of adapter than _CREDENTIAL_CLASSES\"\n\n        for C in self._CREDENTIAL_CLASSES:\n            assert issubclass(C, Credential), \"%s must be a subclass of Credential\" % C\n\n        # First credential should bear the name and url\n        credentials = [self._CREDENTIAL_CLASSES[0](self.name, url=self.url, keyring=self._keyring)]\n        # and we just reuse its _FIELDS for _ask_field_value etc\n        self._FIELDS = credentials[0]._FIELDS\n        # the rest with index suffix, but storing themselves in the same keyring\n        for iC, C in enumerate(self._CREDENTIAL_CLASSES[1:]):\n            credentials.append(\n                C(name=\"%s:%d\" % (self.name, iC + 1), url=None, keyring=self._keyring)\n            )\n        self._credentials = credentials\n\n        super(CompositeCredential, self)._prepare()\n\n    # Here it becomes tricky, since theoretically it is the \"tail\"\n    # ones which might expire etc, so we wouldn't exactly know what\n    # new credentials outside process wanted -- it would be silly to ask\n    # right away the \"entry\" credentials if it is just the expiration of the\n    # tail credentials\n    def enter_new(self):\n        # should invalidate/remove all tail credentials to avoid failing attempts to login\n        self._credentials[0].enter_new()\n        self.refresh()\n\n    def refresh(self):\n        \"\"\"Re-establish \"dependent\" credentials\n\n        E.g. if code outside was reported that it expired somehow before known expiration datetime\n        \"\"\"\n        for c in self._credentials[1:]:\n            c.delete()\n        # trigger re-establishing the chain\n        _ = self()\n        if self.is_expired:\n            raise RuntimeError(\"Credential %s expired right upon refresh: should have not happened\")\n\n    @property\n    def is_expired(self):\n        return any(c.is_expired for c in self._credentials)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        # Start from the tail until we have credentials set\n        idx = len(self._credentials) - 1\n        for c in self._credentials[::-1]:\n            if c.is_known and not c.is_expired:\n                break\n            idx -= 1\n\n        if idx < 0:\n            # none was known, all the same -- start with the first one\n            idx = 0\n\n        # TODO: consider moving logic of traversal into authenticator since it is\n        # the one spitting out authentication error etc\n        # Theoretically we could just reuse 'fields' from adapter in the next step\n        # but let's do full circle, so that if any \"normalization\" is done by\n        # Credentials we take that into account\n        for c, adapter, next_c in zip(\n                self._credentials[idx:],\n                self._CREDENTIAL_ADAPTERS[idx:],\n                self._credentials[idx + 1:]):\n            fields = c()\n            next_fields = adapter(self, **fields)\n            next_c.set(**next_fields)\n\n        return self._credentials[-1]()\n\n\ndef _nda_adapter(composite, user=None, password=None):\n    from datalad.support.third.nda_aws_token_generator import NDATokenGenerator\n    from .. import cfg\n    nda_auth_url = cfg.obtain('datalad.externals.nda.dbserver')\n    gen = NDATokenGenerator(nda_auth_url)\n    lgr.debug(\"Generating token for NDA user %s using %s talking to %s\",\n              user, gen, nda_auth_url)\n    try:\n        token = gen.generate_token(user, password)\n    except Exception as exc:  # it is really just an \"Exception\"\n        exc_str = str(exc).lower()\n        # ATM it is \"Invalid username and/or password\"\n        # but who knows what future would bring\n        if \"invalid\" in exc_str and (\"user\" in exc_str or \"password\" in exc_str):\n            raise AccessDeniedError(exc_str)\n        raise\n    # There are also session and expiration we ignore... TODO anything about it?!!!\n    # we could create a derived AWS_S3 which would also store session and expiration\n    # and then may be Composite could use those????\n    return dict(key_id=token.access_key, secret_id=token.secret_key,\n                session=token.session, expiration=token.expiration)\n\n\nclass NDA_S3(CompositeCredential):\n    \"\"\"Credential to access NDA AWS\n\n    So for NDA we need a credential which is a composite credential.\n    User provides UserPassword and then some adapter generates AWS_S3\n    out of it\n    \"\"\"\n    _CREDENTIAL_CLASSES = (UserPassword, AWS_S3)\n    _CREDENTIAL_ADAPTERS = (_nda_adapter,)\n\n\ndef _loris_adapter(composite, user=None, password=None, **kwargs):\n    from datalad.support.third.loris_token_generator import LORISTokenGenerator\n\n    gen = LORISTokenGenerator(url=composite.url)\n    token = gen.generate_token(user, password)\n\n    return dict(token=token)\n\n\nclass LORIS_Token(CompositeCredential):\n    _CREDENTIAL_CLASSES = (UserPassword, Token)\n    _CREDENTIAL_ADAPTERS = (_loris_adapter,)\n\n    def __init__(self, name, url=None, keyring=None):\n        super(CompositeCredential, self).__init__(name, url, keyring)\n\n\nCREDENTIAL_TYPES = {\n    'user_password': UserPassword,\n    'aws-s3': AWS_S3,\n    'nda-s3': NDA_S3,\n    'token': Token,\n    'loris-token': LORIS_Token\n}\n",
            "new_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Interface information about credentials\n\nProvides minimalistic interface to deal (query, request, store) with most common\ntypes of credentials.  To be used by Authenticators\n\"\"\"\n\n__dev_doc__ = \"\"\"\nPossibly useful in the future 3rd part developments\n\nhttps://github.com/omab/python-social-auth\n    social authentication/registration mechanism with support for several\n    frameworks and auth providers\n\"\"\"\n\nimport time\n\nfrom collections import OrderedDict\n\nfrom ..dochelpers import exc_str\nfrom ..support.exceptions import AccessDeniedError\nfrom ..support.keyring_ import keyring as keyring_\nfrom ..ui import ui\nfrom ..utils import auto_repr\nfrom ..support.network import iso8601_to_epoch\n\nfrom datalad import cfg as dlcfg\n\nfrom logging import getLogger\nlgr = getLogger('datalad.downloaders.credentials')\n\n\n@auto_repr\nclass Credential(object):\n    \"\"\"Base class for different types of credentials\n    \"\"\"\n\n    # Should be defined in a subclass as an OrderedDict of fields\n    # name -> dict(attributes)\n    _FIELDS = None\n    _KNOWN_ATTRS = {\n        'hidden',    # UI should not display the value\n        'repeat',    # UI should repeat entry or not. Set to False to override default logic\n        'optional',  # Not mandatory thus not requested if not set\n    }\n\n    def __init__(self, name, url=None, keyring=None):\n        \"\"\"\n        Parameters\n        ----------\n        name : str\n            Name of the credential, as it would be identified by in the centralized\n            storage of credentials\n        url : str, optional\n            URL string to point users to where to seek obtaining the credentials\n        keyring : a keyring\n            An object providing (g|s)et_password.  If None, keyring module is used\n            as is\n        \"\"\"\n        self.name = name\n        self.url = url\n        self._keyring = keyring or keyring_\n        self._prepare()\n\n    def _prepare(self):\n        \"\"\"Additional house-keeping possibly to be performed in subclasses\n\n        Created to avoid all the passing args/kwargs of __init__ all the time\n        \"\"\"\n        # Basic checks\n        for f, fattrs in self._FIELDS.items():\n            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)\n            if unknown_attrs:\n                raise ValueError(\"Unknown attributes %s. Known are: %s\"\n                                 % (unknown_attrs, self._KNOWN_ATTRS))\n\n    def _is_field_optional(self, f):\n        return self._FIELDS[f].get('optional', False)\n\n    @property\n    def is_known(self):\n        \"\"\"Return True if values for all fields of the credential are known\"\"\"\n        try:\n            return all(\n                self._is_field_optional(f) or self._get_field_value(f) is not None\n                for f in self._FIELDS)\n        except Exception as exc:\n            lgr.warning(\"Failed to query keyring: %s\" % exc_str(exc))\n            return False\n\n    def _get_field_value(self, field):\n        return dlcfg.get('datalad.credential.{name}.{field}'.format(\n            name=self.name,\n            field=field.replace('_', '-')\n        )) or self._keyring.get(self.name, field)\n\n    def _ask_field_value(self, f, instructions=None):\n        msg = instructions if instructions else \\\n            (\"You need to authenticate with %r credentials.\" % self.name +\n                  (\" %s provides information on how to gain access\"\n                   % self.url if self.url else ''))\n\n        # provide custom options only if set for the field\n        f_props = self._FIELDS[f]\n        kwargs = {}\n        for p in ('hidden', 'repeat'):\n            if p in f_props:\n                kwargs[p] = f_props[p]\n        return ui.question(\n            f,\n            title=msg,\n            **kwargs\n        )\n\n    def _ask_and_set(self, f, instructions=None):\n        v = self._ask_field_value(f, instructions=instructions)\n        self.set(**{f: v})\n        return v\n\n    def enter_new(self, instructions=None, **kwargs):\n        \"\"\"Enter new values for the credential fields\n\n        Parameters\n        ----------\n        instructions : str, optional\n          If given, the auto-generated instructions based on a login-URL are\n          replaced by the given string\n        **kwargs\n          Any given key value pairs with non-None values are used to set the\n          field `key` to the given value, without asking for user input\n        \"\"\"\n        if kwargs:\n            unknown_fields = set(kwargs).difference(self._FIELDS)\n            known_fields = set(self._FIELDS).difference(kwargs)\n            if unknown_fields:\n                raise ValueError(\n                    \"Unknown to %s field(s): %s.  Known but not specified: %s\"\n                    % (self,\n                       ', '.join(sorted(unknown_fields)),\n                       ', '.join(sorted(known_fields))\n                       ))\n        # Use ui., request credential fields corresponding to the type\n        for f in self._FIELDS:\n            if kwargs.get(f, None):\n                # use given value, don't ask\n                self.set(**{f: kwargs[f]})\n            elif not self._is_field_optional(f):\n                self._ask_and_set(f, instructions=instructions)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        fields = {}\n        for f in self._FIELDS:\n            v = self._get_field_value(f)\n            if not self._is_field_optional(f):\n                while v is None:  # was not known\n                    v = self._ask_and_set(f)\n                fields[f] = v\n            elif v is not None:\n                fields[f] = v\n        return fields\n\n    def set(self, **kwargs):\n        \"\"\"Set field(s) of the credential\"\"\"\n        for f, v in kwargs.items():\n            if f not in self._FIELDS:\n                raise ValueError(\"Unknown field %s. Known are: %s\"\n                                 % (f, self._FIELDS.keys()))\n            self._keyring.set(self.name, f, v)\n\n    def get(self, f, default=None):\n        \"\"\"Get a field of the credential\"\"\"\n        if f not in self._FIELDS:\n            raise ValueError(\"Unknown field %s. Known are: %s\"\n                             % (f, self._FIELDS.keys()))\n        try:\n            return self._get_field_value(f)\n        except:  # MIH: what could even happen? _keyring not a dict?\n            return default\n\n    def delete(self):\n        \"\"\"Deletes credential values from the keyring\"\"\"\n        for f in self._FIELDS:\n            self._keyring.delete(self.name, f)\n\n\nclass UserPassword(Credential):\n    \"\"\"Simple type of a credential which consists of user/password pair\"\"\"\n\n    _FIELDS = OrderedDict([('user', {}),\n                           ('password', {'hidden': True})])\n\n    is_expired = False  # no expiration provisioned\n\n\nclass Token(Credential):\n    \"\"\"Simple type of a credential which provides a single token\"\"\"\n\n    _FIELDS = OrderedDict([('token', {'hidden': True, 'repeat': False})])\n\n    is_expired = False  # no expiration provisioned\n\n\nclass AWS_S3(Credential):\n    \"\"\"Credential for AWS S3 service\"\"\"\n\n    _FIELDS = OrderedDict([('key_id', {'repeat': False}),\n                           ('secret_id', {'hidden': True, 'repeat': False}),\n                           ('session', {'optional': True}),\n                           ('expiration', {'optional': True}),\n                           ])\n\n    @property\n    def is_expired(self):\n        exp = self.get('expiration', None)\n        if not exp:\n            return False\n        exp_epoch = iso8601_to_epoch(exp)\n        # -2 to artificially shorten duration of the allotment to avoid\n        # possible race conditions between us checking either it has\n        # already expired before submitting a request.\n        expire_in = (exp_epoch - time.time() - 2) / 3600.\n\n        lgr.debug(\n            (\"Credential %s has expired %.2fh ago\"\n                if expire_in <= 0 else \"Credential %s will expire in %.2fh\")\n            % (self.name, expire_in))\n        return expire_in <= 0\n\n\n@auto_repr\nclass CompositeCredential(Credential):\n    \"\"\"Credential which represent a sequence of Credentials where front one is exposed to user\n    \"\"\"\n\n    # To be defined in sub-classes\n    _CREDENTIAL_CLASSES = None\n    _CREDENTIAL_ADAPTERS = None\n\n    def _prepare(self):\n        assert len(self._CREDENTIAL_CLASSES) > 1, \"makes sense only if there is > 1 credential\"\n        assert len(self._CREDENTIAL_CLASSES) == len(self._CREDENTIAL_ADAPTERS) + 1, \\\n            \"there should be 1 less of adapter than _CREDENTIAL_CLASSES\"\n\n        for C in self._CREDENTIAL_CLASSES:\n            assert issubclass(C, Credential), \"%s must be a subclass of Credential\" % C\n\n        # First credential should bear the name and url\n        credentials = [self._CREDENTIAL_CLASSES[0](self.name, url=self.url, keyring=self._keyring)]\n        # and we just reuse its _FIELDS for _ask_field_value etc\n        self._FIELDS = credentials[0]._FIELDS\n        # the rest with index suffix, but storing themselves in the same keyring\n        for iC, C in enumerate(self._CREDENTIAL_CLASSES[1:]):\n            credentials.append(\n                C(name=\"%s:%d\" % (self.name, iC + 1), url=None, keyring=self._keyring)\n            )\n        self._credentials = credentials\n\n        super(CompositeCredential, self)._prepare()\n\n    # Here it becomes tricky, since theoretically it is the \"tail\"\n    # ones which might expire etc, so we wouldn't exactly know what\n    # new credentials outside process wanted -- it would be silly to ask\n    # right away the \"entry\" credentials if it is just the expiration of the\n    # tail credentials\n    def enter_new(self):\n        # should invalidate/remove all tail credentials to avoid failing attempts to login\n        self._credentials[0].enter_new()\n        self.refresh()\n\n    def refresh(self):\n        \"\"\"Re-establish \"dependent\" credentials\n\n        E.g. if code outside was reported that it expired somehow before known expiration datetime\n        \"\"\"\n        for c in self._credentials[1:]:\n            c.delete()\n        # trigger re-establishing the chain\n        _ = self()\n        if self.is_expired:\n            raise RuntimeError(\"Credential %s expired right upon refresh: should have not happened\")\n\n    @property\n    def is_expired(self):\n        return any(c.is_expired for c in self._credentials)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        # Start from the tail until we have credentials set\n        idx = len(self._credentials) - 1\n        for c in self._credentials[::-1]:\n            if c.is_known and not c.is_expired:\n                break\n            idx -= 1\n\n        if idx < 0:\n            # none was known, all the same -- start with the first one\n            idx = 0\n\n        # TODO: consider moving logic of traversal into authenticator since it is\n        # the one spitting out authentication error etc\n        # Theoretically we could just reuse 'fields' from adapter in the next step\n        # but let's do full circle, so that if any \"normalization\" is done by\n        # Credentials we take that into account\n        for c, adapter, next_c in zip(\n                self._credentials[idx:],\n                self._CREDENTIAL_ADAPTERS[idx:],\n                self._credentials[idx + 1:]):\n            fields = c()\n            next_fields = adapter(self, **fields)\n            next_c.set(**next_fields)\n\n        return self._credentials[-1]()\n\n\ndef _nda_adapter(composite, user=None, password=None):\n    from datalad.support.third.nda_aws_token_generator import NDATokenGenerator\n    from .. import cfg\n    nda_auth_url = cfg.obtain('datalad.externals.nda.dbserver')\n    gen = NDATokenGenerator(nda_auth_url)\n    lgr.debug(\"Generating token for NDA user %s using %s talking to %s\",\n              user, gen, nda_auth_url)\n    try:\n        token = gen.generate_token(user, password)\n    except Exception as exc:  # it is really just an \"Exception\"\n        exc_str = str(exc).lower()\n        # ATM it is \"Invalid username and/or password\"\n        # but who knows what future would bring\n        if \"invalid\" in exc_str and (\"user\" in exc_str or \"password\" in exc_str):\n            raise AccessDeniedError(exc_str)\n        raise\n    # There are also session and expiration we ignore... TODO anything about it?!!!\n    # we could create a derived AWS_S3 which would also store session and expiration\n    # and then may be Composite could use those????\n    return dict(key_id=token.access_key, secret_id=token.secret_key,\n                session=token.session, expiration=token.expiration)\n\n\nclass NDA_S3(CompositeCredential):\n    \"\"\"Credential to access NDA AWS\n\n    So for NDA we need a credential which is a composite credential.\n    User provides UserPassword and then some adapter generates AWS_S3\n    out of it\n    \"\"\"\n    _CREDENTIAL_CLASSES = (UserPassword, AWS_S3)\n    _CREDENTIAL_ADAPTERS = (_nda_adapter,)\n\n\ndef _loris_adapter(composite, user=None, password=None, **kwargs):\n    from datalad.support.third.loris_token_generator import LORISTokenGenerator\n\n    gen = LORISTokenGenerator(url=composite.url)\n    token = gen.generate_token(user, password)\n\n    return dict(token=token)\n\n\nclass LORIS_Token(CompositeCredential):\n    _CREDENTIAL_CLASSES = (UserPassword, Token)\n    _CREDENTIAL_ADAPTERS = (_loris_adapter,)\n\n    def __init__(self, name, url=None, keyring=None):\n        super(CompositeCredential, self).__init__(name, url, keyring)\n\n\nCREDENTIAL_TYPES = {\n    'user_password': UserPassword,\n    'aws-s3': AWS_S3,\n    'nda-s3': NDA_S3,\n    'token': Token,\n    'loris-token': LORIS_Token\n}\n",
            "header": {
                "file": {
                    "path": "datalad/downloaders/credentials.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "73403dd56",
                "new_commit_hash": "063038936",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/datalad/downloaders/credentials.py"
            },
            "plus_file": {
                "path": "b/datalad/downloaders/credentials.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 31,
                            "length": 6
                        },
                        "new_range": {
                            "start": 31,
                            "length": 8
                        },
                        "section": "from ..ui import ui"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "from ..utils import auto_repr",
                                "type": "context"
                            },
                            {
                                "content": "from ..support.network import iso8601_to_epoch",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "from datalad import cfg as dlcfg",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "from logging import getLogger",
                                "type": "context"
                            },
                            {
                                "content": "lgr = getLogger('datalad.downloaders.credentials')",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "datalad/downloaders/credentials.py",
                            "type": "statement",
                            "name": "",
                            "content": "from datalad import cfg as dlcfg",
                            "ast_type_str": "ImportFrom",
                            "start_lineno": 34,
                            "end_lineno": 34,
                            "parent": null
                        }
                    ],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 94,
                            "length": 7
                        },
                        "new_range": {
                            "start": 96,
                            "length": 10
                        },
                        "section": "class Credential(object):"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "            return False",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def _get_field_value(self, field):",
                                "type": "context"
                            },
                            {
                                "content": "        return self._keyring.get(self.name, field)",
                                "type": "deleted"
                            },
                            {
                                "content": "        return dlcfg.get('datalad.credential.{name}.{field}'.format(",
                                "type": "added"
                            },
                            {
                                "content": "            name=self.name,",
                                "type": "added"
                            },
                            {
                                "content": "            field=field.replace('_', '-')",
                                "type": "added"
                            },
                            {
                                "content": "        )) or self._keyring.get(self.name, field)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    def _ask_field_value(self, f, instructions=None):",
                                "type": "context"
                            },
                            {
                                "content": "        msg = instructions if instructions else \\",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/downloaders/credentials.py",
                            "type": "class",
                            "name": "Credential",
                            "content": "class Credential(object):\n    \"\"\"Base class for different types of credentials\n    \"\"\"\n\n    # Should be defined in a subclass as an OrderedDict of fields\n    # name -> dict(attributes)\n    _FIELDS = None\n    _KNOWN_ATTRS = {\n        'hidden',    # UI should not display the value\n        'repeat',    # UI should repeat entry or not. Set to False to override default logic\n        'optional',  # Not mandatory thus not requested if not set\n    }\n\n    def __init__(self, name, url=None, keyring=None):\n        \"\"\"\n        Parameters\n        ----------\n        name : str\n            Name of the credential, as it would be identified by in the centralized\n            storage of credentials\n        url : str, optional\n            URL string to point users to where to seek obtaining the credentials\n        keyring : a keyring\n            An object providing (g|s)et_password.  If None, keyring module is used\n            as is\n        \"\"\"\n        self.name = name\n        self.url = url\n        self._keyring = keyring or keyring_\n        self._prepare()\n\n    def _prepare(self):\n        \"\"\"Additional house-keeping possibly to be performed in subclasses\n\n        Created to avoid all the passing args/kwargs of __init__ all the time\n        \"\"\"\n        # Basic checks\n        for f, fattrs in self._FIELDS.items():\n            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)\n            if unknown_attrs:\n                raise ValueError(\"Unknown attributes %s. Known are: %s\"\n                                 % (unknown_attrs, self._KNOWN_ATTRS))\n\n    def _is_field_optional(self, f):\n        return self._FIELDS[f].get('optional', False)\n\n    @property\n    def is_known(self):\n        \"\"\"Return True if values for all fields of the credential are known\"\"\"\n        try:\n            return all(\n                self._is_field_optional(f) or self._get_field_value(f) is not None\n                for f in self._FIELDS)\n        except Exception as exc:\n            lgr.warning(\"Failed to query keyring: %s\" % exc_str(exc))\n            return False\n\n    def _get_field_value(self, field):\n        return self._keyring.get(self.name, field)\n\n    def _ask_field_value(self, f, instructions=None):\n        msg = instructions if instructions else \\\n            (\"You need to authenticate with %r credentials.\" % self.name +\n                  (\" %s provides information on how to gain access\"\n                   % self.url if self.url else ''))\n\n        # provide custom options only if set for the field\n        f_props = self._FIELDS[f]\n        kwargs = {}\n        for p in ('hidden', 'repeat'):\n            if p in f_props:\n                kwargs[p] = f_props[p]\n        return ui.question(\n            f,\n            title=msg,\n            **kwargs\n        )\n\n    def _ask_and_set(self, f, instructions=None):\n        v = self._ask_field_value(f, instructions=instructions)\n        self.set(**{f: v})\n        return v\n\n    def enter_new(self, instructions=None, **kwargs):\n        \"\"\"Enter new values for the credential fields\n\n        Parameters\n        ----------\n        instructions : str, optional\n          If given, the auto-generated instructions based on a login-URL are\n          replaced by the given string\n        **kwargs\n          Any given key value pairs with non-None values are used to set the\n          field `key` to the given value, without asking for user input\n        \"\"\"\n        if kwargs:\n            unknown_fields = set(kwargs).difference(self._FIELDS)\n            known_fields = set(self._FIELDS).difference(kwargs)\n            if unknown_fields:\n                raise ValueError(\n                    \"Unknown to %s field(s): %s.  Known but not specified: %s\"\n                    % (self,\n                       ', '.join(sorted(unknown_fields)),\n                       ', '.join(sorted(known_fields))\n                       ))\n        # Use ui., request credential fields corresponding to the type\n        for f in self._FIELDS:\n            if kwargs.get(f, None):\n                # use given value, don't ask\n                self.set(**{f: kwargs[f]})\n            elif not self._is_field_optional(f):\n                self._ask_and_set(f, instructions=instructions)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        fields = {}\n        for f in self._FIELDS:\n            v = self._get_field_value(f)\n            if not self._is_field_optional(f):\n                while v is None:  # was not known\n                    v = self._ask_and_set(f)\n                fields[f] = v\n            elif v is not None:\n                fields[f] = v\n        return fields\n\n    def set(self, **kwargs):\n        \"\"\"Set field(s) of the credential\"\"\"\n        for f, v in kwargs.items():\n            if f not in self._FIELDS:\n                raise ValueError(\"Unknown field %s. Known are: %s\"\n                                 % (f, self._FIELDS.keys()))\n            self._keyring.set(self.name, f, v)\n\n    def get(self, f, default=None):\n        \"\"\"Get a field of the credential\"\"\"\n        if f not in self._FIELDS:\n            raise ValueError(\"Unknown field %s. Known are: %s\"\n                             % (f, self._FIELDS.keys()))\n        try:\n            return self._get_field_value(f)\n        except:  # MIH: what could even happen? _keyring not a dict?\n            return default\n\n    def delete(self):\n        \"\"\"Deletes credential values from the keyring\"\"\"\n        for f in self._FIELDS:\n            self._keyring.delete(self.name, f)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 39,
                            "end_lineno": 186,
                            "parent": null
                        },
                        {
                            "file_name": "datalad/downloaders/credentials.py",
                            "type": "method",
                            "name": "Credential._get_field_value",
                            "content": "class Credential(object):\n    \"\"\"Base class for different types of credentials\n    \"\"\"\n\n    # Should be defined in a subclass as an OrderedDict of fields\n    # name -> dict(attributes)\n    _FIELDS = None\n    _KNOWN_ATTRS = {\n        'hidden',    # UI should not display the value\n        'repeat',    # UI should repeat entry or not. Set to False to override default logic\n        'optional',  # Not mandatory thus not requested if not set\n    }\n\n    def __init__(self, name, url=None, keyring=None):\n        \"\"\"\n        Parameters\n        ----------\n        name : str\n            Name of the credential, as it would be identified by in the centralized\n            storage of credentials\n        url : str, optional\n            URL string to point users to where to seek obtaining the credentials\n        keyring : a keyring\n            An object providing (g|s)et_password.  If None, keyring module is used\n            as is\n        \"\"\"\n        self.name = name\n        self.url = url\n        self._keyring = keyring or keyring_\n        self._prepare()\n\n    def _prepare(self):\n        \"\"\"Additional house-keeping possibly to be performed in subclasses\n\n        Created to avoid all the passing args/kwargs of __init__ all the time\n        \"\"\"\n        # Basic checks\n        for f, fattrs in self._FIELDS.items():\n            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)\n            if unknown_attrs:\n                raise ValueError(\"Unknown attributes %s. Known are: %s\"\n                                 % (unknown_attrs, self._KNOWN_ATTRS))\n\n    def _is_field_optional(self, f):\n        return self._FIELDS[f].get('optional', False)\n\n    @property\n    def is_known(self):\n        \"\"\"Return True if values for all fields of the credential are known\"\"\"\n        try:\n            return all(\n                self._is_field_optional(f) or self._get_field_value(f) is not None\n                for f in self._FIELDS)\n        except Exception as exc:\n            lgr.warning(\"Failed to query keyring: %s\" % exc_str(exc))\n            return False\n\n    def _get_field_value(self, field):\n        return self._keyring.get(self.name, field)\n\n    def _ask_field_value(self, f, instructions=None):\n        msg = instructions if instructions else \\\n            (\"You need to authenticate with %r credentials.\" % self.name +\n                  (\" %s provides information on how to gain access\"\n                   % self.url if self.url else ''))\n\n        # provide custom options only if set for the field\n        f_props = self._FIELDS[f]\n        kwargs = {}\n        for p in ('hidden', 'repeat'):\n            if p in f_props:\n                kwargs[p] = f_props[p]\n        return ui.question(\n            f,\n            title=msg,\n            **kwargs\n        )\n\n    def _ask_and_set(self, f, instructions=None):\n        v = self._ask_field_value(f, instructions=instructions)\n        self.set(**{f: v})\n        return v\n\n    def enter_new(self, instructions=None, **kwargs):\n        \"\"\"Enter new values for the credential fields\n\n        Parameters\n        ----------\n        instructions : str, optional\n          If given, the auto-generated instructions based on a login-URL are\n          replaced by the given string\n        **kwargs\n          Any given key value pairs with non-None values are used to set the\n          field `key` to the given value, without asking for user input\n        \"\"\"\n        if kwargs:\n            unknown_fields = set(kwargs).difference(self._FIELDS)\n            known_fields = set(self._FIELDS).difference(kwargs)\n            if unknown_fields:\n                raise ValueError(\n                    \"Unknown to %s field(s): %s.  Known but not specified: %s\"\n                    % (self,\n                       ', '.join(sorted(unknown_fields)),\n                       ', '.join(sorted(known_fields))\n                       ))\n        # Use ui., request credential fields corresponding to the type\n        for f in self._FIELDS:\n            if kwargs.get(f, None):\n                # use given value, don't ask\n                self.set(**{f: kwargs[f]})\n            elif not self._is_field_optional(f):\n                self._ask_and_set(f, instructions=instructions)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        fields = {}\n        for f in self._FIELDS:\n            v = self._get_field_value(f)\n            if not self._is_field_optional(f):\n                while v is None:  # was not known\n                    v = self._ask_and_set(f)\n                fields[f] = v\n            elif v is not None:\n                fields[f] = v\n        return fields\n\n    def set(self, **kwargs):\n        \"\"\"Set field(s) of the credential\"\"\"\n        for f, v in kwargs.items():\n            if f not in self._FIELDS:\n                raise ValueError(\"Unknown field %s. Known are: %s\"\n                                 % (f, self._FIELDS.keys()))\n            self._keyring.set(self.name, f, v)\n\n    def get(self, f, default=None):\n        \"\"\"Get a field of the credential\"\"\"\n        if f not in self._FIELDS:\n            raise ValueError(\"Unknown field %s. Known are: %s\"\n                             % (f, self._FIELDS.keys()))\n        try:\n            return self._get_field_value(f)\n        except:  # MIH: what could even happen? _keyring not a dict?\n            return default\n\n    def delete(self):\n        \"\"\"Deletes credential values from the keyring\"\"\"\n        for f in self._FIELDS:\n            self._keyring.delete(self.name, f)",
                            "ast_type_str": "ClassDef",
                            "start_lineno": 96,
                            "end_lineno": 97,
                            "parent": {
                                "file_name": "datalad/downloaders/credentials.py",
                                "type": "class",
                                "name": "Credential",
                                "content": "class Credential(object):\n    \"\"\"Base class for different types of credentials\n    \"\"\"\n\n    # Should be defined in a subclass as an OrderedDict of fields\n    # name -> dict(attributes)\n    _FIELDS = None\n    _KNOWN_ATTRS = {\n        'hidden',    # UI should not display the value\n        'repeat',    # UI should repeat entry or not. Set to False to override default logic\n        'optional',  # Not mandatory thus not requested if not set\n    }\n\n    def __init__(self, name, url=None, keyring=None):\n        \"\"\"\n        Parameters\n        ----------\n        name : str\n            Name of the credential, as it would be identified by in the centralized\n            storage of credentials\n        url : str, optional\n            URL string to point users to where to seek obtaining the credentials\n        keyring : a keyring\n            An object providing (g|s)et_password.  If None, keyring module is used\n            as is\n        \"\"\"\n        self.name = name\n        self.url = url\n        self._keyring = keyring or keyring_\n        self._prepare()\n\n    def _prepare(self):\n        \"\"\"Additional house-keeping possibly to be performed in subclasses\n\n        Created to avoid all the passing args/kwargs of __init__ all the time\n        \"\"\"\n        # Basic checks\n        for f, fattrs in self._FIELDS.items():\n            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)\n            if unknown_attrs:\n                raise ValueError(\"Unknown attributes %s. Known are: %s\"\n                                 % (unknown_attrs, self._KNOWN_ATTRS))\n\n    def _is_field_optional(self, f):\n        return self._FIELDS[f].get('optional', False)\n\n    @property\n    def is_known(self):\n        \"\"\"Return True if values for all fields of the credential are known\"\"\"\n        try:\n            return all(\n                self._is_field_optional(f) or self._get_field_value(f) is not None\n                for f in self._FIELDS)\n        except Exception as exc:\n            lgr.warning(\"Failed to query keyring: %s\" % exc_str(exc))\n            return False\n\n    def _get_field_value(self, field):\n        return self._keyring.get(self.name, field)\n\n    def _ask_field_value(self, f, instructions=None):\n        msg = instructions if instructions else \\\n            (\"You need to authenticate with %r credentials.\" % self.name +\n                  (\" %s provides information on how to gain access\"\n                   % self.url if self.url else ''))\n\n        # provide custom options only if set for the field\n        f_props = self._FIELDS[f]\n        kwargs = {}\n        for p in ('hidden', 'repeat'):\n            if p in f_props:\n                kwargs[p] = f_props[p]\n        return ui.question(\n            f,\n            title=msg,\n            **kwargs\n        )\n\n    def _ask_and_set(self, f, instructions=None):\n        v = self._ask_field_value(f, instructions=instructions)\n        self.set(**{f: v})\n        return v\n\n    def enter_new(self, instructions=None, **kwargs):\n        \"\"\"Enter new values for the credential fields\n\n        Parameters\n        ----------\n        instructions : str, optional\n          If given, the auto-generated instructions based on a login-URL are\n          replaced by the given string\n        **kwargs\n          Any given key value pairs with non-None values are used to set the\n          field `key` to the given value, without asking for user input\n        \"\"\"\n        if kwargs:\n            unknown_fields = set(kwargs).difference(self._FIELDS)\n            known_fields = set(self._FIELDS).difference(kwargs)\n            if unknown_fields:\n                raise ValueError(\n                    \"Unknown to %s field(s): %s.  Known but not specified: %s\"\n                    % (self,\n                       ', '.join(sorted(unknown_fields)),\n                       ', '.join(sorted(known_fields))\n                       ))\n        # Use ui., request credential fields corresponding to the type\n        for f in self._FIELDS:\n            if kwargs.get(f, None):\n                # use given value, don't ask\n                self.set(**{f: kwargs[f]})\n            elif not self._is_field_optional(f):\n                self._ask_and_set(f, instructions=instructions)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        fields = {}\n        for f in self._FIELDS:\n            v = self._get_field_value(f)\n            if not self._is_field_optional(f):\n                while v is None:  # was not known\n                    v = self._ask_and_set(f)\n                fields[f] = v\n            elif v is not None:\n                fields[f] = v\n        return fields\n\n    def set(self, **kwargs):\n        \"\"\"Set field(s) of the credential\"\"\"\n        for f, v in kwargs.items():\n            if f not in self._FIELDS:\n                raise ValueError(\"Unknown field %s. Known are: %s\"\n                                 % (f, self._FIELDS.keys()))\n            self._keyring.set(self.name, f, v)\n\n    def get(self, f, default=None):\n        \"\"\"Get a field of the credential\"\"\"\n        if f not in self._FIELDS:\n            raise ValueError(\"Unknown field %s. Known are: %s\"\n                             % (f, self._FIELDS.keys()))\n        try:\n            return self._get_field_value(f)\n        except:  # MIH: what could even happen? _keyring not a dict?\n            return default\n\n    def delete(self):\n        \"\"\"Deletes credential values from the keyring\"\"\"\n        for f in self._FIELDS:\n            self._keyring.delete(self.name, f)",
                                "ast_type_str": "ClassDef",
                                "start_lineno": 39,
                                "end_lineno": 186,
                                "parent": null
                            }
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        },
        {
            "old_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Tests for credentials\"\"\"\n\nfrom unittest.mock import patch\nfrom datalad.tests.utils import (\n    assert_equal,\n    assert_false,\n    assert_in,\n    assert_raises,\n    assert_true,\n    SkipTest,\n    with_testsui,\n)\nfrom datalad.support.keyring_ import (\n    Keyring,\n    MemoryKeyring,\n)\nfrom ..credentials import (\n    AWS_S3,\n    CompositeCredential,\n    UserPassword,\n)\n\n\n@with_testsui(responses=[\n    'user1', 'password1',\n    # when we do provide user to enter_new\n    'newpassword',\n])\ndef test_cred1_enter_new():\n    keyring = MemoryKeyring()\n    cred = UserPassword(\"name\", keyring=keyring)\n    assert_false(cred.is_known)\n    assert_equal(cred.enter_new(), None)\n    assert_true(cred.is_known)\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_equal(keyring.get('name', 'password'), 'password1')\n    keyring.delete('name')\n    assert_raises(KeyError, keyring.delete, 'name', 'user')\n    assert_raises(KeyError, keyring.delete, 'name')\n    assert_equal(keyring.get('name', 'user'), None)\n\n    # Test it blowing up if we provide unknown field\n    with assert_raises(ValueError) as cme:\n        cred.enter_new(username='user')\n    assert_in('field(s): username.  Known but not specified: password, user',\n              str(cme.exception))\n\n    # Test that if user is provided, it is not asked\n    cred.enter_new(user='user2')\n    assert_equal(keyring.get('name', 'user'), 'user2')\n    assert_equal(keyring.get('name', 'password'), 'newpassword')\n\n@with_testsui(responses=['password1'])\ndef test_cred1_call():\n    keyring = MemoryKeyring()\n    cred = UserPassword(\"name\", keyring=keyring)\n    # we will set the name but not the password, expecting UI\n    # requesting it\n    assert_equal(keyring.set('name', 'user', 'user1'), None)\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_equal(cred(), {'user': 'user1', 'password': 'password1'})\n    assert_equal(keyring.get('name', 'password'), 'password1')\n\n\ndef test_keyring():\n    # mock out keyring methods and test that we are providing correct values\n    # with 'datalad-' prefix\n    raise SkipTest(\"provide tests for Keyring which interfaces keyring module\")\n\n\ndef _cred1_adapter(composite, user=None, password=None):\n    \"\"\"Just a sample adapter from one user/pw type to another\"\"\"\n    return dict(user=user + \"_1\", password=password + \"_2\")\n\n\nclass _CCred1(CompositeCredential):\n    \"\"\"A Simple composite credential which will do some entries transformation\n    \"\"\"\n    _CREDENTIAL_CLASSES = (UserPassword, UserPassword)\n    _CREDENTIAL_ADAPTERS = (_cred1_adapter,)\n\n\n@with_testsui(responses=['user1', 'password1',\n                         'user2', 'password2'])\ndef test_composite_credential1():\n    # basic test of composite credential\n    keyring = MemoryKeyring()\n    cred = _CCred1(\"name\", keyring=keyring)\n    # When queried, does the chain\n    assert_equal(cred(), {'user': 'user1_1', 'password': 'password1_2'})\n    # But the \"Front\" credential is exposed to the user\n    assert_equal(cred.get('user'), 'user1')\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_raises(ValueError, cred.get, 'unknown_field')\n    assert_equal(cred.get('password'), 'password1')\n    assert_equal(keyring.get('name', 'password'), 'password1')\n    # ATM composite credential stores \"derived\" ones unconditionally in the\n    # keyring as well\n    assert_equal(keyring.get('name:1', 'user'), 'user1_1')\n    assert_equal(keyring.get('name:1', 'password'), 'password1_2')\n\n    # and now enter new should remove \"derived\" entries\n    cred.enter_new()\n    assert_equal(keyring.get('name', 'user'), 'user2')\n    assert_equal(keyring.get('name', 'password'), 'password2')\n    # we immediately refresh all credentials in the chain\n    assert_equal(keyring.get('name:1', 'user'), 'user2_1')\n    assert_equal(keyring.get('name:1', 'password'), 'password2_2')\n    assert_equal(cred(), {'user': 'user2_1', 'password': 'password2_2'})\n\n\ndef test_credentials_from_env():\n    keyring = Keyring()\n    cred = AWS_S3(\"test-s3\", keyring=keyring)\n    assert_false(cred.is_known)\n    assert_equal(cred.get('key_id'), None)\n    assert_equal(cred.get('secret_id'), None)\n\n    with patch.dict('os.environ', {'DATALAD_test_s3_key_id': '1'}):\n        assert_equal(cred.get('key_id'), '1')\n        assert_false(cred.is_known)\n        with patch.dict('os.environ', {'DATALAD_test_s3_secret_id': '2'}):\n            assert_equal(cred.get('key_id'), '1')\n            assert_equal(cred.get('secret_id'), '2')\n            assert_true(cred.is_known)\n        assert_false(cred.is_known)  # no memory of the past\n",
            "new_file_content": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Tests for credentials\"\"\"\n\nfrom unittest.mock import patch\nfrom datalad.tests.utils import (\n    assert_equal,\n    assert_false,\n    assert_in,\n    assert_raises,\n    assert_true,\n    SkipTest,\n    with_testsui,\n)\nfrom datalad.support.keyring_ import (\n    Keyring,\n    MemoryKeyring,\n)\nfrom ..credentials import (\n    AWS_S3,\n    CompositeCredential,\n    UserPassword,\n)\nfrom datalad.config import ConfigManager\n\n\n@with_testsui(responses=[\n    'user1', 'password1',\n    # when we do provide user to enter_new\n    'newpassword',\n])\ndef test_cred1_enter_new():\n    keyring = MemoryKeyring()\n    cred = UserPassword(\"name\", keyring=keyring)\n    assert_false(cred.is_known)\n    assert_equal(cred.enter_new(), None)\n    assert_true(cred.is_known)\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_equal(keyring.get('name', 'password'), 'password1')\n    keyring.delete('name')\n    assert_raises(KeyError, keyring.delete, 'name', 'user')\n    assert_raises(KeyError, keyring.delete, 'name')\n    assert_equal(keyring.get('name', 'user'), None)\n\n    # Test it blowing up if we provide unknown field\n    with assert_raises(ValueError) as cme:\n        cred.enter_new(username='user')\n    assert_in('field(s): username.  Known but not specified: password, user',\n              str(cme.exception))\n\n    # Test that if user is provided, it is not asked\n    cred.enter_new(user='user2')\n    assert_equal(keyring.get('name', 'user'), 'user2')\n    assert_equal(keyring.get('name', 'password'), 'newpassword')\n\n@with_testsui(responses=['password1'])\ndef test_cred1_call():\n    keyring = MemoryKeyring()\n    cred = UserPassword(\"name\", keyring=keyring)\n    # we will set the name but not the password, expecting UI\n    # requesting it\n    assert_equal(keyring.set('name', 'user', 'user1'), None)\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_equal(cred(), {'user': 'user1', 'password': 'password1'})\n    assert_equal(keyring.get('name', 'password'), 'password1')\n\n\ndef test_keyring():\n    # mock out keyring methods and test that we are providing correct values\n    # with 'datalad-' prefix\n    raise SkipTest(\"provide tests for Keyring which interfaces keyring module\")\n\n\ndef _cred1_adapter(composite, user=None, password=None):\n    \"\"\"Just a sample adapter from one user/pw type to another\"\"\"\n    return dict(user=user + \"_1\", password=password + \"_2\")\n\n\nclass _CCred1(CompositeCredential):\n    \"\"\"A Simple composite credential which will do some entries transformation\n    \"\"\"\n    _CREDENTIAL_CLASSES = (UserPassword, UserPassword)\n    _CREDENTIAL_ADAPTERS = (_cred1_adapter,)\n\n\n@with_testsui(responses=['user1', 'password1',\n                         'user2', 'password2'])\ndef test_composite_credential1():\n    # basic test of composite credential\n    keyring = MemoryKeyring()\n    cred = _CCred1(\"name\", keyring=keyring)\n    # When queried, does the chain\n    assert_equal(cred(), {'user': 'user1_1', 'password': 'password1_2'})\n    # But the \"Front\" credential is exposed to the user\n    assert_equal(cred.get('user'), 'user1')\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_raises(ValueError, cred.get, 'unknown_field')\n    assert_equal(cred.get('password'), 'password1')\n    assert_equal(keyring.get('name', 'password'), 'password1')\n    # ATM composite credential stores \"derived\" ones unconditionally in the\n    # keyring as well\n    assert_equal(keyring.get('name:1', 'user'), 'user1_1')\n    assert_equal(keyring.get('name:1', 'password'), 'password1_2')\n\n    # and now enter new should remove \"derived\" entries\n    cred.enter_new()\n    assert_equal(keyring.get('name', 'user'), 'user2')\n    assert_equal(keyring.get('name', 'password'), 'password2')\n    # we immediately refresh all credentials in the chain\n    assert_equal(keyring.get('name:1', 'user'), 'user2_1')\n    assert_equal(keyring.get('name:1', 'password'), 'password2_2')\n    assert_equal(cred(), {'user': 'user2_1', 'password': 'password2_2'})\n\n\ndef test_credentials_from_env():\n    keyring = Keyring()\n    cred = AWS_S3(\"test-s3\", keyring=keyring)\n    assert_false(cred.is_known)\n    assert_equal(cred.get('key_id'), None)\n    assert_equal(cred.get('secret_id'), None)\n\n    def _check1():\n        assert_equal(cred.get('key_id'), '1')\n        assert_false(cred.is_known)\n\n    def _check2():\n        assert_equal(cred.get('key_id'), '1')\n        assert_equal(cred.get('secret_id'), '2')\n        assert_true(cred.is_known)\n\n    # this is the old way, should still work\n    with patch.dict('os.environ', {'DATALAD_test_s3_key_id': '1'}):\n        _check1()\n        with patch.dict('os.environ', {'DATALAD_test_s3_secret_id': '2'}):\n            _check2()\n        assert_false(cred.is_known)  # no memory of the past\n\n    # here is the new way\n    import datalad\n    try:\n        with patch.dict('os.environ', {'DATALAD_CREDENTIAL_test__s3_key__id': '1'}):\n            datalad.cfg.reload()\n            _check1()\n            with patch.dict('os.environ', {'DATALAD_CREDENTIAL_test__s3_secret__id': '2'}):\n                datalad.cfg.reload()\n                _check2()\n            datalad.cfg.reload()\n            assert_false(cred.is_known)  # no memory of the past\n    finally:\n        datalad.cfg.reload()\n",
            "header": {
                "file": {
                    "path": "datalad/downloaders/tests/test_credentials.py"
                },
                "misc_line": null
            },
            "index_line": {
                "old_commit_hash": "37bcc2dee",
                "new_commit_hash": "b5ccbbe7a",
                "mode": "100644"
            },
            "is_binary_file": false,
            "binary_line": null,
            "minus_file": {
                "path": "a/datalad/downloaders/tests/test_credentials.py"
            },
            "plus_file": {
                "path": "b/datalad/downloaders/tests/test_credentials.py"
            },
            "hunks": [
                {
                    "descriptor": {
                        "old_range": {
                            "start": 27,
                            "length": 6
                        },
                        "new_range": {
                            "start": 27,
                            "length": 7
                        },
                        "section": "from ..credentials import ("
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    CompositeCredential,",
                                "type": "context"
                            },
                            {
                                "content": "    UserPassword,",
                                "type": "context"
                            },
                            {
                                "content": ")",
                                "type": "context"
                            },
                            {
                                "content": "from datalad.config import ConfigManager",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "@with_testsui(responses=[",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [],
                    "added_entities": [
                        {
                            "file_name": "datalad/downloaders/tests/test_credentials.py",
                            "type": "statement",
                            "name": "",
                            "content": "from datalad.config import ConfigManager",
                            "ast_type_str": "ImportFrom",
                            "start_lineno": 30,
                            "end_lineno": 30,
                            "parent": null
                        }
                    ],
                    "deleted_entities": []
                },
                {
                    "descriptor": {
                        "old_range": {
                            "start": 124,
                            "length": 11
                        },
                        "new_range": {
                            "start": 125,
                            "length": 32
                        },
                        "section": "def test_credentials_from_env():"
                    },
                    "line_group": {
                        "all_lines": [
                            {
                                "content": "    assert_equal(cred.get('key_id'), None)",
                                "type": "context"
                            },
                            {
                                "content": "    assert_equal(cred.get('secret_id'), None)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "context"
                            },
                            {
                                "content": "    with patch.dict('os.environ', {'DATALAD_test_s3_key_id': '1'}):",
                                "type": "deleted"
                            },
                            {
                                "content": "    def _check1():",
                                "type": "added"
                            },
                            {
                                "content": "        assert_equal(cred.get('key_id'), '1')",
                                "type": "context"
                            },
                            {
                                "content": "        assert_false(cred.is_known)",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    def _check2():",
                                "type": "added"
                            },
                            {
                                "content": "        assert_equal(cred.get('key_id'), '1')",
                                "type": "added"
                            },
                            {
                                "content": "        assert_equal(cred.get('secret_id'), '2')",
                                "type": "added"
                            },
                            {
                                "content": "        assert_true(cred.is_known)",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    # this is the old way, should still work",
                                "type": "added"
                            },
                            {
                                "content": "    with patch.dict('os.environ', {'DATALAD_test_s3_key_id': '1'}):",
                                "type": "added"
                            },
                            {
                                "content": "        _check1()",
                                "type": "added"
                            },
                            {
                                "content": "        with patch.dict('os.environ', {'DATALAD_test_s3_secret_id': '2'}):",
                                "type": "context"
                            },
                            {
                                "content": "            assert_equal(cred.get('key_id'), '1')",
                                "type": "deleted"
                            },
                            {
                                "content": "            assert_equal(cred.get('secret_id'), '2')",
                                "type": "deleted"
                            },
                            {
                                "content": "            assert_true(cred.is_known)",
                                "type": "deleted"
                            },
                            {
                                "content": "            _check2()",
                                "type": "added"
                            },
                            {
                                "content": "        assert_false(cred.is_known)  # no memory of the past",
                                "type": "context"
                            },
                            {
                                "content": "",
                                "type": "added"
                            },
                            {
                                "content": "    # here is the new way",
                                "type": "added"
                            },
                            {
                                "content": "    import datalad",
                                "type": "added"
                            },
                            {
                                "content": "    try:",
                                "type": "added"
                            },
                            {
                                "content": "        with patch.dict('os.environ', {'DATALAD_CREDENTIAL_test__s3_key__id': '1'}):",
                                "type": "added"
                            },
                            {
                                "content": "            datalad.cfg.reload()",
                                "type": "added"
                            },
                            {
                                "content": "            _check1()",
                                "type": "added"
                            },
                            {
                                "content": "            with patch.dict('os.environ', {'DATALAD_CREDENTIAL_test__s3_secret__id': '2'}):",
                                "type": "added"
                            },
                            {
                                "content": "                datalad.cfg.reload()",
                                "type": "added"
                            },
                            {
                                "content": "                _check2()",
                                "type": "added"
                            },
                            {
                                "content": "            datalad.cfg.reload()",
                                "type": "added"
                            },
                            {
                                "content": "            assert_false(cred.is_known)  # no memory of the past",
                                "type": "added"
                            },
                            {
                                "content": "    finally:",
                                "type": "added"
                            },
                            {
                                "content": "        datalad.cfg.reload()",
                                "type": "added"
                            },
                            {
                                "content": "",
                                "type": "context"
                            }
                        ]
                    },
                    "modified_entities": [
                        {
                            "file_name": "datalad/downloaders/tests/test_credentials.py",
                            "type": "function",
                            "name": "test_credentials_from_env",
                            "content": "def test_credentials_from_env():\n    keyring = Keyring()\n    cred = AWS_S3(\"test-s3\", keyring=keyring)\n    assert_false(cred.is_known)\n    assert_equal(cred.get('key_id'), None)\n    assert_equal(cred.get('secret_id'), None)\n\n    with patch.dict('os.environ', {'DATALAD_test_s3_key_id': '1'}):\n        assert_equal(cred.get('key_id'), '1')\n        assert_false(cred.is_known)\n        with patch.dict('os.environ', {'DATALAD_test_s3_secret_id': '2'}):\n            assert_equal(cred.get('key_id'), '1')\n            assert_equal(cred.get('secret_id'), '2')\n            assert_true(cred.is_known)\n        assert_false(cred.is_known)",
                            "ast_type_str": "FunctionDef",
                            "start_lineno": 120,
                            "end_lineno": 134,
                            "parent": null
                        }
                    ],
                    "added_entities": [],
                    "deleted_entities": []
                }
            ]
        }
    ],
    "old_commit_hash": "f75c73119236c3ff8e99800d5cb9ca4559b3b7ee^",
    "new_commit_hash": "f75c73119236c3ff8e99800d5cb9ca4559b3b7ee",
    "commit_message": "NF: Support credential query via ConfigManager\n\nFixes gh-5677",
    "commit_date": "2021-05-20T09:59:26+02:00",
    "metadata": {}
}
{
    "repo_name": "datalad",
    "new_commit_hash": "f75c73119236c3ff8e99800d5cb9ca4559b3b7ee",
    "test_file_codes": [
        "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Tests for credentials\"\"\"\n\nfrom unittest.mock import patch\nfrom datalad.tests.utils import (\n    assert_equal,\n    assert_false,\n    assert_in,\n    assert_raises,\n    assert_true,\n    SkipTest,\n    with_testsui,\n)\nfrom datalad.support.keyring_ import (\n    Keyring,\n    MemoryKeyring,\n)\nfrom datalad.downloaders.credentials import (\n    AWS_S3,\n    CompositeCredential,\n    UserPassword,\n)\nfrom datalad.config import ConfigManager\n\n\n@with_testsui(responses=[\n    'user1', 'password1',\n    # when we do provide user to enter_new\n    'newpassword',\n])\ndef test_cred1_enter_new():\n    keyring = MemoryKeyring()\n    cred = UserPassword(\"name\", keyring=keyring)\n    assert_false(cred.is_known)\n    assert_equal(cred.enter_new(), None)\n    assert_true(cred.is_known)\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_equal(keyring.get('name', 'password'), 'password1')\n    keyring.delete('name')\n    assert_raises(KeyError, keyring.delete, 'name', 'user')\n    assert_raises(KeyError, keyring.delete, 'name')\n    assert_equal(keyring.get('name', 'user'), None)\n\n    # Test it blowing up if we provide unknown field\n    with assert_raises(ValueError) as cme:\n        cred.enter_new(username='user')\n    assert_in('field(s): username.  Known but not specified: password, user',\n              str(cme.exception))\n\n    # Test that if user is provided, it is not asked\n    cred.enter_new(user='user2')\n    assert_equal(keyring.get('name', 'user'), 'user2')\n    assert_equal(keyring.get('name', 'password'), 'newpassword')\n\n@with_testsui(responses=['password1'])\ndef test_cred1_call():\n    keyring = MemoryKeyring()\n    cred = UserPassword(\"name\", keyring=keyring)\n    # we will set the name but not the password, expecting UI\n    # requesting it\n    assert_equal(keyring.set('name', 'user', 'user1'), None)\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_equal(cred(), {'user': 'user1', 'password': 'password1'})\n    assert_equal(keyring.get('name', 'password'), 'password1')\n\n\ndef test_keyring():\n    # mock out keyring methods and test that we are providing correct values\n    # with 'datalad-' prefix\n    raise SkipTest(\"provide tests for Keyring which interfaces keyring module\")\n\n\ndef _cred1_adapter(composite, user=None, password=None):\n    \"\"\"Just a sample adapter from one user/pw type to another\"\"\"\n    return dict(user=user + \"_1\", password=password + \"_2\")\n\n\nclass _CCred1(CompositeCredential):\n    \"\"\"A Simple composite credential which will do some entries transformation\n    \"\"\"\n    _CREDENTIAL_CLASSES = (UserPassword, UserPassword)\n    _CREDENTIAL_ADAPTERS = (_cred1_adapter,)\n\n\n@with_testsui(responses=['user1', 'password1',\n                         'user2', 'password2'])\ndef test_composite_credential1():\n    # basic test of composite credential\n    keyring = MemoryKeyring()\n    cred = _CCred1(\"name\", keyring=keyring)\n    # When queried, does the chain\n    assert_equal(cred(), {'user': 'user1_1', 'password': 'password1_2'})\n    # But the \"Front\" credential is exposed to the user\n    assert_equal(cred.get('user'), 'user1')\n    assert_equal(keyring.get('name', 'user'), 'user1')\n    assert_raises(ValueError, cred.get, 'unknown_field')\n    assert_equal(cred.get('password'), 'password1')\n    assert_equal(keyring.get('name', 'password'), 'password1')\n    # ATM composite credential stores \"derived\" ones unconditionally in the\n    # keyring as well\n    assert_equal(keyring.get('name:1', 'user'), 'user1_1')\n    assert_equal(keyring.get('name:1', 'password'), 'password1_2')\n\n    # and now enter new should remove \"derived\" entries\n    cred.enter_new()\n    assert_equal(keyring.get('name', 'user'), 'user2')\n    assert_equal(keyring.get('name', 'password'), 'password2')\n    # we immediately refresh all credentials in the chain\n    assert_equal(keyring.get('name:1', 'user'), 'user2_1')\n    assert_equal(keyring.get('name:1', 'password'), 'password2_2')\n    assert_equal(cred(), {'user': 'user2_1', 'password': 'password2_2'})\n\n\ndef test_credentials_from_env():\n    keyring = Keyring()\n    cred = AWS_S3(\"test-s3\", keyring=keyring)\n    assert_false(cred.is_known)\n    assert_equal(cred.get('key_id'), None)\n    assert_equal(cred.get('secret_id'), None)\n\n    def _check1():\n        assert_equal(cred.get('key_id'), '1')\n        assert_false(cred.is_known)\n\n    def _check2():\n        assert_equal(cred.get('key_id'), '1')\n        assert_equal(cred.get('secret_id'), '2')\n        assert_true(cred.is_known)\n\n    # this is the old way, should still work\n    with patch.dict('os.environ', {'DATALAD_test_s3_key_id': '1'}):\n        _check1()\n        with patch.dict('os.environ', {'DATALAD_test_s3_secret_id': '2'}):\n            _check2()\n        assert_false(cred.is_known)  # no memory of the past\n\n    # here is the new way\n    import datalad\n    try:\n        with patch.dict('os.environ', {'DATALAD_CREDENTIAL_test__s3_key__id': '1'}):\n            datalad.cfg.reload()\n            _check1()\n            with patch.dict('os.environ', {'DATALAD_CREDENTIAL_test__s3_secret__id': '2'}):\n                datalad.cfg.reload()\n                _check2()\n            datalad.cfg.reload()\n            assert_false(cred.is_known)  # no memory of the past\n    finally:\n        datalad.cfg.reload()\n",
        "import pytest\nimport tempfile\nimport os\n\n\n@pytest.fixture\ndef path():\n    with tempfile.TemporaryDirectory() as tmpdir:\n        yield tmpdir\n\n\n@pytest.fixture(autouse=True)\ndef setup_git_config():\n    with tempfile.TemporaryDirectory() as home:\n        old_home = os.environ.get(\"HOME\")\n        os.environ[\"HOME\"] = home\n        os.system('git config --global user.name \"DataLad Tester\"')\n        os.system('git config --global user.email \"test@example.com\"')\n        yield\n        if old_home:\n            os.environ[\"HOME\"] = old_home\n"
    ],
    "test_file_names": [
        "test_1.py",
        "conftest.py"
    ],
    "setup_res_code": 0,
    "setup_res_stdout": "Starting Datalad installation attempts...\nVerifying installation...\nDatalad version: 0.14.4.dev311\n✅ Installation successful!\nSuccessfully installed Datalad using Python 3.9\n",
    "setup_res_stderr": "Using CPython 3.9.21\nCreating virtual environment at: .venv\nActivate with: source .venv/bin/activate\nResolved 25 packages in 148ms\nInstalled 25 packages in 2.64s\n + coverage==7.6.10\n + docopt==0.6.2\n + exceptiongroup==1.2.2\n + grabbit==0.2.6\n + importlib-resources==6.4.5\n + iniconfig==2.0.0\n + nibabel==5.3.2\n + num2words==0.5.14\n + numpy==2.0.2\n + packaging==24.2\n + pandas==2.2.3\n + patsy==1.0.1\n + pluggy==1.5.0\n + pybids==0.6.5\n + pytest==8.3.4\n + pytest-cov==6.0.0\n + python-dateutil==2.9.0.post0\n + pytz==2024.2\n + scipy==1.13.1\n + setuptools==75.6.0\n + six==1.17.0\n + tomli==2.2.1\n + typing-extensions==4.12.2\n + tzdata==2024.2\n + zipp==3.21.0\nResolved 62 packages in 3.36s\nPrepared 1 package in 3.68s\nInstalled 56 packages in 236ms\n + annexremote==1.6.6\n + appdirs==1.4.4\n + backports-tarfile==1.2.0\n + beautifulsoup4==4.12.3\n + boto==2.49.0\n + certifi==2024.12.14\n + cffi==1.17.1\n + chardet==5.2.0\n + charset-normalizer==3.4.1\n + citeproc-py==0.6.0\n + cryptography==44.0.0\n + datalad==0.14.4 (from file:///home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_f75c73119236c3ff8e99800d5cb9ca4559b3b7ee)\n + deprecated==1.2.15\n + distro==1.9.0\n + duecredit==0.10.2\n + exifread==3.0.0\n + fasteners==0.19\n + httpretty==1.1.4\n + humanize==4.11.0\n + idna==3.10\n + importlib-metadata==8.5.0\n + iso8601==2.1.0\n + jaraco-classes==3.4.0\n + jaraco-context==6.0.1\n + jaraco-functools==4.1.0\n + jeepney==0.8.0\n + keyring==25.6.0\n + keyrings-alt==5.0.2\n + looseversion==1.3.0\n + lxml==5.3.0\n + more-itertools==10.5.0\n + msgpack==1.1.0\n + multidict==6.1.0\n + mutagen==1.47.0\n + nose==1.3.7\n + patool==1.15.0\n + pillow==11.1.0\n + propcache==0.2.1\n + pycparser==2.22\n + pygithub==2.5.0\n + pyjwt==2.10.1\n + pynacl==1.5.0\n + pyperclip==1.9.0\n + python-xmp-toolkit==2.0.2\n + pyyaml==6.0.2\n + requests==2.32.3\n + requests-ftp==0.3.1\n + secretstorage==3.3.3\n + simplejson==3.19.3\n + soupsieve==2.6\n + tqdm==4.67.1\n + urllib3==1.26.20\n + vcrpy==7.0.0\n + whoosh==2.7.4\n + wrapt==1.17.0\n + yarl==1.18.3\nResolved 96 packages in 104ms\nInstalled 32 packages in 565ms\n + alabaster==0.7.16\n + asv==0.6.4\n + asv-runner==0.2.1\n + babel==2.16.0\n + build==1.2.2.post1\n + distlib==0.3.9\n + docutils==0.21.2\n + filelock==3.16.1\n + gprof2dot==2024.6.6\n + imagesize==1.4.1\n + jinja2==3.1.5\n + json5==0.10.0\n + markupsafe==3.0.2\n + nose-timer==1.0.1\n + platformdirs==4.3.6\n + psutil==6.1.1\n + pygments==2.18.0\n + pympler==1.1\n + pypandoc==1.14\n + pyproject-hooks==1.2.0\n + snowballstemmer==2.2.0\n + sphinx==7.4.7\n + sphinx-rtd-theme==3.0.2\n + sphinxcontrib-applehelp==2.0.0\n + sphinxcontrib-devhelp==2.0.0\n + sphinxcontrib-htmlhelp==2.1.0\n + sphinxcontrib-jquery==4.1\n + sphinxcontrib-jsmath==1.0.1\n + sphinxcontrib-qthelp==2.0.0\n + sphinxcontrib-serializinghtml==2.0.0\n + tabulate==0.9.0\n + virtualenv==20.28.0\n",
    "new_commit_res_code": 0,
    "new_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.9.21, pytest-8.3.4, pluggy-1.5.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_f75c73119236c3ff8e99800d5cb9ca4559b3b7ee\nconfigfile: pyproject.toml\nplugins: cov-6.0.0\ncollected 5 items\n\nr2e_tests/test_1.py ..s..                                                [100%]\n\n==================================== PASSES ====================================\n=========================== short test summary info ============================\nPASSED r2e_tests/test_1.py::test_cred1_enter_new\nPASSED r2e_tests/test_1.py::test_cred1_call\nPASSED r2e_tests/test_1.py::test_composite_credential1\nPASSED r2e_tests/test_1.py::test_credentials_from_env\nSKIPPED [1] .venv/lib/python3.9/site-packages/_pytest/unittest.py:385: provide tests for Keyring which interfaces keyring module\n========================= 4 passed, 1 skipped in 3.87s =========================\n",
    "new_commit_res_stderr": "",
    "old_commit_res_code": 1,
    "old_commit_res_stdout": "============================= test session starts ==============================\nplatform linux -- Python 3.9.21, pytest-8.3.4, pluggy-1.5.0\nrootdir: /home/gcpuser/buckets/local_repoeval_bucket/repos/datalad_f75c73119236c3ff8e99800d5cb9ca4559b3b7ee\nconfigfile: pyproject.toml\nplugins: cov-6.0.0\ncollected 5 items\n\nr2e_tests/test_1.py ..s.F                                                [100%]\n\n=================================== FAILURES ===================================\n__________________________ test_credentials_from_env ___________________________\n\n    def test_credentials_from_env():\n        keyring = Keyring()\n        cred = AWS_S3(\"test-s3\", keyring=keyring)\n        assert_false(cred.is_known)\n        assert_equal(cred.get('key_id'), None)\n        assert_equal(cred.get('secret_id'), None)\n    \n        def _check1():\n            assert_equal(cred.get('key_id'), '1')\n            assert_false(cred.is_known)\n    \n        def _check2():\n            assert_equal(cred.get('key_id'), '1')\n            assert_equal(cred.get('secret_id'), '2')\n            assert_true(cred.is_known)\n    \n        # this is the old way, should still work\n        with patch.dict('os.environ', {'DATALAD_test_s3_key_id': '1'}):\n            _check1()\n            with patch.dict('os.environ', {'DATALAD_test_s3_secret_id': '2'}):\n                _check2()\n            assert_false(cred.is_known)  # no memory of the past\n    \n        # here is the new way\n        import datalad\n        try:\n            with patch.dict('os.environ', {'DATALAD_CREDENTIAL_test__s3_key__id': '1'}):\n                datalad.cfg.reload()\n>               _check1()\n\nr2e_tests/test_1.py:149: \n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \nr2e_tests/test_1.py:129: in _check1\n    assert_equal(cred.get('key_id'), '1')\n../../../../.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py:837: in assertEqual\n    assertion_func(first, second, msg=msg)\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \n\nself = <nose.tools.trivial.Dummy testMethod=nop>, first = None, second = '1'\nmsg = \"None != '1'\"\n\n    def _baseAssertEqual(self, first, second, msg=None):\n        \"\"\"The default assertEqual implementation, not type specific.\"\"\"\n        if not first == second:\n            standardMsg = '%s != %s' % _common_shorten_repr(first, second)\n            msg = self._formatMessage(msg, standardMsg)\n>           raise self.failureException(msg)\nE           AssertionError: None != '1'\n\n../../../../.local/share/uv/python/cpython-3.9.21-linux-x86_64-gnu/lib/python3.9/unittest/case.py:830: AssertionError\n==================================== PASSES ====================================\n=========================== short test summary info ============================\nPASSED r2e_tests/test_1.py::test_cred1_enter_new\nPASSED r2e_tests/test_1.py::test_cred1_call\nPASSED r2e_tests/test_1.py::test_composite_credential1\nSKIPPED [1] .venv/lib/python3.9/site-packages/_pytest/unittest.py:385: provide tests for Keyring which interfaces keyring module\nFAILED r2e_tests/test_1.py::test_credentials_from_env - AssertionError: None ...\n==================== 1 failed, 3 passed, 1 skipped in 0.72s ====================\n",
    "old_commit_res_stderr": ""
}
datalad/downloaders/credentials.py datalad/downloaders/tests/test_credentials.py
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title**: Credential values from environment variables are not retrieved correctly

**Description**:
When environment variables for credentials are set, the system fails to retrieve the expected values, resulting in credentials being marked as unknown.

**Bug Example**:
```python
import os
from datalad.downloaders.credentials import AWS_S3, Keyring

keyring = Keyring()
cred = AWS_S3('test-s3', keyring=keyring)

# Initially, credentials are unknown
print(cred.is_known)  # Expected: False

# Setting environment variable for key_id
os.environ['DATALAD_test_s3_key_id'] = '1'
print(cred.get('key_id'))  # Expected: '1'

# Setting environment variable for secret_id
os.environ['DATALAD_test_s3_secret_id'] = '2'
print(cred.get('secret_id'))  # Expected: '2'
print(cred.is_known)  # Expected: True
```

**Expected Behavior**:
When the environment variables `DATALAD_test_s3_key_id` and `DATALAD_test_s3_secret_id` are set to `'1'` and `'2'` respectively, the methods `cred.get('key_id')` and `cred.get('secret_id')` should return `'1'` and `'2'`. Additionally, `cred.is_known` should return `True`, indicating that the credentials have been successfully retrieved.

**Actual Behavior**:
After setting the environment variables, `cred.get('key_id')` returns `None` instead of `'1'`, and `cred.is_known` remains `False`. This means the credentials are not recognized even though the environment variables are correctly set.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### datalad/downloaders/credentials.py
# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-
# ex: set sts=4 ts=4 sw=4 noet:
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the datalad package for the
#   copyright and license terms.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
"""Interface information about credentials

Provides minimalistic interface to deal (query, request, store) with most common
types of credentials.  To be used by Authenticators
"""

__dev_doc__ = """
Possibly useful in the future 3rd part developments

https://github.com/omab/python-social-auth
    social authentication/registration mechanism with support for several
    frameworks and auth providers
"""

import time

from collections import OrderedDict

from ..dochelpers import exc_str
from ..support.exceptions import AccessDeniedError
from ..support.keyring_ import keyring as keyring_
from ..ui import ui
from ..utils import auto_repr
from ..support.network import iso8601_to_epoch

from logging import getLogger
lgr = getLogger('datalad.downloaders.credentials')


@auto_repr
class Credential(object):
    """Base class for different types of credentials
    """

    # Should be defined in a subclass as an OrderedDict of fields
    # name -> dict(attributes)
    _FIELDS = None
    _KNOWN_ATTRS = {
        'hidden',    # UI should not display the value
        'repeat',    # UI should repeat entry or not. Set to False to override default logic
        'optional',  # Not mandatory thus not requested if not set
    }

    def __init__(self, name, url=None, keyring=None):
        """
        Parameters
        ----------
        name : str
            Name of the credential, as it would be identified by in the centralized
            storage of credentials
        url : str, optional
            URL string to point users to where to seek obtaining the credentials
        keyring : a keyring
            An object providing (g|s)et_password.  If None, keyring module is used
            as is
        """
        self.name = name
        self.url = url
        self._keyring = keyring or keyring_
        self._prepare()

    def _prepare(self):
        """Additional house-keeping possibly to be performed in subclasses

        Created to avoid all the passing args/kwargs of __init__ all the time
        """
        # Basic checks
        for f, fattrs in self._FIELDS.items():
            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)
            if unknown_attrs:
                raise ValueError("Unknown attributes %s. Known are: %s"
                                 % (unknown_attrs, self._KNOWN_ATTRS))

    def _is_field_optional(self, f):
        return self._FIELDS[f].get('optional', False)

    @property
    def is_known(self):
        """Return True if values for all fields of the credential are known"""
        try:
            return all(
                self._is_field_optional(f) or self._get_field_value(f) is not None
                for f in self._FIELDS)
        except Exception as exc:
            lgr.warning("Failed to query keyring: %s" % exc_str(exc))
            return False

    def _get_field_value(self, field):
        return self._keyring.get(self.name, field)

    def _ask_field_value(self, f, instructions=None):
        msg = instructions if instructions else \
            ("You need to authenticate with %r credentials." % self.name +
                  (" %s provides information on how to gain access"
                   % self.url if self.url else ''))

        # provide custom options only if set for the field
        f_props = self._FIELDS[f]
        kwargs = {}
        for p in ('hidden', 'repeat'):
            if p in f_props:
                kwargs[p] = f_props[p]
        return ui.question(
            f,
            title=msg,
            **kwargs
        )

    def _ask_and_set(self, f, instructions=None):
        v = self._ask_field_value(f, instructions=instructions)
        self.set(**{f: v})
        return v

    def enter_new(self, instructions=None, **kwargs):
        """Enter new values for the credential fields

        Parameters
        ----------
        instructions : str, optional
          If given, the auto-generated instructions based on a login-URL are
          replaced by the given string
        **kwargs
          Any given key value pairs with non-None values are used to set the
          field `key` to the given value, without asking for user input
        """
        if kwargs:
            unknown_fields = set(kwargs).difference(self._FIELDS)
            known_fields = set(self._FIELDS).difference(kwargs)
            if unknown_fields:
                raise ValueError(
                    "Unknown to %s field(s): %s.  Known but not specified: %s"
                    % (self,
                       ', '.join(sorted(unknown_fields)),
                       ', '.join(sorted(known_fields))
                       ))
        # Use ui., request credential fields corresponding to the type
        for f in self._FIELDS:
            if kwargs.get(f, None):
                # use given value, don't ask
                self.set(**{f: kwargs[f]})
            elif not self._is_field_optional(f):
                self._ask_and_set(f, instructions=instructions)

    def __call__(self):
        """Obtain credentials from a keyring and if any is not known -- ask"""
        fields = {}
        for f in self._FIELDS:
            v = self._get_field_value(f)
            if not self._is_field_optional(f):
                while v is None:  # was not known
                    v = self._ask_and_set(f)
                fields[f] = v
            elif v is not None:
                fields[f] = v
        return fields

    def set(self, **kwargs):
        """Set field(s) of the credential"""
        for f, v in kwargs.items():
            if f not in self._FIELDS:
                raise ValueError("Unknown field %s. Known are: %s"
                                 % (f, self._FIELDS.keys()))
            self._keyring.set(self.name, f, v)

    def get(self, f, default=None):
        """Get a field of the credential"""
        if f not in self._FIELDS:
            raise ValueError("Unknown field %s. Known are: %s"
                             % (f, self._FIELDS.keys()))
        try:
            return self._get_field_value(f)
        except:  # MIH: what could even happen? _keyring not a dict?
            return default

    def delete(self):
        """Deletes credential values from the keyring"""
        for f in self._FIELDS:
            self._keyring.delete(self.name, f)


class UserPassword(Credential):
    """Simple type of a credential which consists of user/password pair"""

    _FIELDS = OrderedDict([('user', {}),
                           ('password', {'hidden': True})])

    is_expired = False  # no expiration provisioned


class Token(Credential):
    """Simple type of a credential which provides a single token"""

    _FIELDS = OrderedDict([('token', {'hidden': True, 'repeat': False})])

    is_expired = False  # no expiration provisioned


class AWS_S3(Credential):
    """Credential for AWS S3 service"""

    _FIELDS = OrderedDict([('key_id', {'repeat': False}),
                           ('secret_id', {'hidden': True, 'repeat': False}),
                           ('session', {'optional': True}),
                           ('expiration', {'optional': True}),
                           ])

    @property
    def is_expired(self):
        exp = self.get('expiration', None)
        if not exp:
            return False
        exp_epoch = iso8601_to_epoch(exp)
        # -2 to artificially shorten duration of the allotment to avoid
        # possible race conditions between us checking either it has
        # already expired before submitting a request.
        expire_in = (exp_epoch - time.time() - 2) / 3600.

        lgr.debug(
            ("Credential %s has expired %.2fh ago"
                if expire_in <= 0 else "Credential %s will expire in %.2fh")
            % (self.name, expire_in))
        return expire_in <= 0


@auto_repr
class CompositeCredential(Credential):
    """Credential which represent a sequence of Credentials where front one is exposed to user
    """

    # To be defined in sub-classes
    _CREDENTIAL_CLASSES = None
    _CREDENTIAL_ADAPTERS = None

    def _prepare(self):
        assert len(self._CREDENTIAL_CLASSES) > 1, "makes sense only if there is > 1 credential"
        assert len(self._CREDENTIAL_CLASSES) == len(self._CREDENTIAL_ADAPTERS) + 1, \
            "there should be 1 less of adapter than _CREDENTIAL_CLASSES"

        for C in self._CREDENTIAL_CLASSES:
            assert issubclass(C, Credential), "%s must be a subclass of Credential" % C

        # First credential should bear the name and url
        credentials = [self._CREDENTIAL_CLASSES[0](self.name, url=self.url, keyring=self._keyring)]
        # and we just reuse its _FIELDS for _ask_field_value etc
        self._FIELDS = credentials[0]._FIELDS
        # the rest with index suffix, but storing themselves in the same keyring
        for iC, C in enumerate(self._CREDENTIAL_CLASSES[1:]):
            credentials.append(
                C(name="%s:%d" % (self.name, iC + 1), url=None, keyring=self._keyring)
            )
        self._credentials = credentials

        super(CompositeCredential, self)._prepare()

    # Here it becomes tricky, since theoretically it is the "tail"
    # ones which might expire etc, so we wouldn't exactly know what
    # new credentials outside process wanted -- it would be silly to ask
    # right away the "entry" credentials if it is just the expiration of the
    # tail credentials
    def enter_new(self):
        # should invalidate/remove all tail credentials to avoid failing attempts to login
        self._credentials[0].enter_new()
        self.refresh()

    def refresh(self):
        """Re-establish "dependent" credentials

        E.g. if code outside was reported that it expired somehow before known expiration datetime
        """
        for c in self._credentials[1:]:
            c.delete()
        # trigger re-establishing the chain
        _ = self()
        if self.is_expired:
            raise RuntimeError("Credential %s expired right upon refresh: should have not happened")

    @property
    def is_expired(self):
        return any(c.is_expired for c in self._credentials)

    def __call__(self):
        """Obtain credentials from a keyring and if any is not known -- ask"""
        # Start from the tail until we have credentials set
        idx = len(self._credentials) - 1
        for c in self._credentials[::-1]:
            if c.is_known and not c.is_expired:
                break
            idx -= 1

        if idx < 0:
            # none was known, all the same -- start with the first one
            idx = 0

        # TODO: consider moving logic of traversal into authenticator since it is
        # the one spitting out authentication error etc
        # Theoretically we could just reuse 'fields' from adapter in the next step
        # but let's do full circle, so that if any "normalization" is done by
        # Credentials we take that into account
        for c, adapter, next_c in zip(
                self._credentials[idx:],
                self._CREDENTIAL_ADAPTERS[idx:],
                self._credentials[idx + 1:]):
            fields = c()
            next_fields = adapter(self, **fields)
            next_c.set(**next_fields)

        return self._credentials[-1]()


def _nda_adapter(composite, user=None, password=None):
    from datalad.support.third.nda_aws_token_generator import NDATokenGenerator
    from .. import cfg
    nda_auth_url = cfg.obtain('datalad.externals.nda.dbserver')
    gen = NDATokenGenerator(nda_auth_url)
    lgr.debug("Generating token for NDA user %s using %s talking to %s",
              user, gen, nda_auth_url)
    try:
        token = gen.generate_token(user, password)
    except Exception as exc:  # it is really just an "Exception"
        exc_str = str(exc).lower()
        # ATM it is "Invalid username and/or password"
        # but who knows what future would bring
        if "invalid" in exc_str and ("user" in exc_str or "password" in exc_str):
            raise AccessDeniedError(exc_str)
        raise
    # There are also session and expiration we ignore... TODO anything about it?!!!
    # we could create a derived AWS_S3 which would also store session and expiration
    # and then may be Composite could use those????
    return dict(key_id=token.access_key, secret_id=token.secret_key,
                session=token.session, expiration=token.expiration)


class NDA_S3(CompositeCredential):
    """Credential to access NDA AWS

    So for NDA we need a credential which is a composite credential.
    User provides UserPassword and then some adapter generates AWS_S3
    out of it
    """
    _CREDENTIAL_CLASSES = (UserPassword, AWS_S3)
    _CREDENTIAL_ADAPTERS = (_nda_adapter,)


def _loris_adapter(composite, user=None, password=None, **kwargs):
    from datalad.support.third.loris_token_generator import LORISTokenGenerator

    gen = LORISTokenGenerator(url=composite.url)
    token = gen.generate_token(user, password)

    return dict(token=token)


class LORIS_Token(CompositeCredential):
    _CREDENTIAL_CLASSES = (UserPassword, Token)
    _CREDENTIAL_ADAPTERS = (_loris_adapter,)

    def __init__(self, name, url=None, keyring=None):
        super(CompositeCredential, self).__init__(name, url, keyring)


CREDENTIAL_TYPES = {
    'user_password': UserPassword,
    'aws-s3': AWS_S3,
    'nda-s3': NDA_S3,
    'token': Token,
    'loris-token': LORIS_Token
}

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
[ISSUE]
**Title**: Credential values from environment variables are not retrieved correctly

**Description**:
When environment variables for credentials are set, the system fails to retrieve the expected values, resulting in credentials being marked as unknown.

**Bug Example**:
```python
import os
from datalad.downloaders.credentials import AWS_S3, Keyring

keyring = Keyring()
cred = AWS_S3('test-s3', keyring=keyring)

# Initially, credentials are unknown
print(cred.is_known)  # Expected: False

# Setting environment variable for key_id
os.environ['DATALAD_test_s3_key_id'] = '1'
print(cred.get('key_id'))  # Expected: '1'

# Setting environment variable for secret_id
os.environ['DATALAD_test_s3_secret_id'] = '2'
print(cred.get('secret_id'))  # Expected: '2'
print(cred.is_known)  # Expected: True
```

**Expected Behavior**:
When the environment variables `DATALAD_test_s3_key_id` and `DATALAD_test_s3_secret_id` are set to `'1'` and `'2'` respectively, the methods `cred.get('key_id')` and `cred.get('secret_id')` should return `'1'` and `'2'`. Additionally, `cred.is_known` should return `True`, indicating that the credentials have been successfully retrieved.

**Actual Behavior**:
After setting the environment variables, `cred.get('key_id')` returns `None` instead of `'1'`, and `cred.is_known` remains `False`. This means the credentials are not recognized even though the environment variables are correctly set.

[/ISSUE]
{
    "test_cred1_enter_new": "PASSED",
    "test_cred1_call": "PASSED",
    "test_composite_credential1": "PASSED",
    "test_credentials_from_env": "PASSED"
}
diff --git a/datalad/downloaders/credentials.py b/datalad/downloaders/credentials.py
index 73403dd56..063038936 100644
--- a/datalad/downloaders/credentials.py
+++ b/datalad/downloaders/credentials.py
@@ -31,6 +31,8 @@ from ..ui import ui
 from ..utils import auto_repr
 from ..support.network import iso8601_to_epoch
 
+from datalad import cfg as dlcfg
+
 from logging import getLogger
 lgr = getLogger('datalad.downloaders.credentials')
 
@@ -94,7 +96,10 @@ class Credential(object):
             return False
 
     def _get_field_value(self, field):
-        return self._keyring.get(self.name, field)
+        return dlcfg.get('datalad.credential.{name}.{field}'.format(
+            name=self.name,
+            field=field.replace('_', '-')
+        )) or self._keyring.get(self.name, field)
 
     def _ask_field_value(self, f, instructions=None):
         msg = instructions if instructions else \

{"datalad/downloaders/credentials.py": "# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-\n# ex: set sts=4 ts=4 sw=4 noet:\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n#\n#   See COPYING file distributed along with the datalad package for the\n#   copyright and license terms.\n#\n# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##\n\"\"\"Interface information about credentials\n\nProvides minimalistic interface to deal (query, request, store) with most common\ntypes of credentials.  To be used by Authenticators\n\"\"\"\n\n__dev_doc__ = \"\"\"\nPossibly useful in the future 3rd part developments\n\nhttps://github.com/omab/python-social-auth\n    social authentication/registration mechanism with support for several\n    frameworks and auth providers\n\"\"\"\n\nimport time\n\nfrom collections import OrderedDict\n\nfrom ..dochelpers import exc_str\nfrom ..support.exceptions import AccessDeniedError\nfrom ..support.keyring_ import keyring as keyring_\nfrom ..ui import ui\nfrom ..utils import auto_repr\nfrom ..support.network import iso8601_to_epoch\n\nfrom logging import getLogger\nlgr = getLogger('datalad.downloaders.credentials')\n\n\n@auto_repr\nclass Credential(object):\n    \"\"\"Base class for different types of credentials\n    \"\"\"\n\n    # Should be defined in a subclass as an OrderedDict of fields\n    # name -> dict(attributes)\n    _FIELDS = None\n    _KNOWN_ATTRS = {\n        'hidden',    # UI should not display the value\n        'repeat',    # UI should repeat entry or not. Set to False to override default logic\n        'optional',  # Not mandatory thus not requested if not set\n    }\n\n    def __init__(self, name, url=None, keyring=None):\n        \"\"\"\n        Parameters\n        ----------\n        name : str\n            Name of the credential, as it would be identified by in the centralized\n            storage of credentials\n        url : str, optional\n            URL string to point users to where to seek obtaining the credentials\n        keyring : a keyring\n            An object providing (g|s)et_password.  If None, keyring module is used\n            as is\n        \"\"\"\n        self.name = name\n        self.url = url\n        self._keyring = keyring or keyring_\n        self._prepare()\n\n    def _prepare(self):\n        \"\"\"Additional house-keeping possibly to be performed in subclasses\n\n        Created to avoid all the passing args/kwargs of __init__ all the time\n        \"\"\"\n        # Basic checks\n        for f, fattrs in self._FIELDS.items():\n            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)\n            if unknown_attrs:\n                raise ValueError(\"Unknown attributes %s. Known are: %s\"\n                                 % (unknown_attrs, self._KNOWN_ATTRS))\n\n    def _is_field_optional(self, f):\n        return self._FIELDS[f].get('optional', False)\n\n    @property\n    def is_known(self):\n        \"\"\"Return True if values for all fields of the credential are known\"\"\"\n        try:\n            return all(\n                self._is_field_optional(f) or self._get_field_value(f) is not None\n                for f in self._FIELDS)\n        except Exception as exc:\n            lgr.warning(\"Failed to query keyring: %s\" % exc_str(exc))\n            return False\n\n    def _get_field_value(self, field):\n        return self._keyring.get(self.name, field)\n\n    def _ask_field_value(self, f, instructions=None):\n        msg = instructions if instructions else \\\n            (\"You need to authenticate with %r credentials.\" % self.name +\n                  (\" %s provides information on how to gain access\"\n                   % self.url if self.url else ''))\n\n        # provide custom options only if set for the field\n        f_props = self._FIELDS[f]\n        kwargs = {}\n        for p in ('hidden', 'repeat'):\n            if p in f_props:\n                kwargs[p] = f_props[p]\n        return ui.question(\n            f,\n            title=msg,\n            **kwargs\n        )\n\n    def _ask_and_set(self, f, instructions=None):\n        v = self._ask_field_value(f, instructions=instructions)\n        self.set(**{f: v})\n        return v\n\n    def enter_new(self, instructions=None, **kwargs):\n        \"\"\"Enter new values for the credential fields\n\n        Parameters\n        ----------\n        instructions : str, optional\n          If given, the auto-generated instructions based on a login-URL are\n          replaced by the given string\n        **kwargs\n          Any given key value pairs with non-None values are used to set the\n          field `key` to the given value, without asking for user input\n        \"\"\"\n        if kwargs:\n            unknown_fields = set(kwargs).difference(self._FIELDS)\n            known_fields = set(self._FIELDS).difference(kwargs)\n            if unknown_fields:\n                raise ValueError(\n                    \"Unknown to %s field(s): %s.  Known but not specified: %s\"\n                    % (self,\n                       ', '.join(sorted(unknown_fields)),\n                       ', '.join(sorted(known_fields))\n                       ))\n        # Use ui., request credential fields corresponding to the type\n        for f in self._FIELDS:\n            if kwargs.get(f, None):\n                # use given value, don't ask\n                self.set(**{f: kwargs[f]})\n            elif not self._is_field_optional(f):\n                self._ask_and_set(f, instructions=instructions)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        fields = {}\n        for f in self._FIELDS:\n            v = self._get_field_value(f)\n            if not self._is_field_optional(f):\n                while v is None:  # was not known\n                    v = self._ask_and_set(f)\n                fields[f] = v\n            elif v is not None:\n                fields[f] = v\n        return fields\n\n    def set(self, **kwargs):\n        \"\"\"Set field(s) of the credential\"\"\"\n        for f, v in kwargs.items():\n            if f not in self._FIELDS:\n                raise ValueError(\"Unknown field %s. Known are: %s\"\n                                 % (f, self._FIELDS.keys()))\n            self._keyring.set(self.name, f, v)\n\n    def get(self, f, default=None):\n        \"\"\"Get a field of the credential\"\"\"\n        if f not in self._FIELDS:\n            raise ValueError(\"Unknown field %s. Known are: %s\"\n                             % (f, self._FIELDS.keys()))\n        try:\n            return self._get_field_value(f)\n        except:  # MIH: what could even happen? _keyring not a dict?\n            return default\n\n    def delete(self):\n        \"\"\"Deletes credential values from the keyring\"\"\"\n        for f in self._FIELDS:\n            self._keyring.delete(self.name, f)\n\n\nclass UserPassword(Credential):\n    \"\"\"Simple type of a credential which consists of user/password pair\"\"\"\n\n    _FIELDS = OrderedDict([('user', {}),\n                           ('password', {'hidden': True})])\n\n    is_expired = False  # no expiration provisioned\n\n\nclass Token(Credential):\n    \"\"\"Simple type of a credential which provides a single token\"\"\"\n\n    _FIELDS = OrderedDict([('token', {'hidden': True, 'repeat': False})])\n\n    is_expired = False  # no expiration provisioned\n\n\nclass AWS_S3(Credential):\n    \"\"\"Credential for AWS S3 service\"\"\"\n\n    _FIELDS = OrderedDict([('key_id', {'repeat': False}),\n                           ('secret_id', {'hidden': True, 'repeat': False}),\n                           ('session', {'optional': True}),\n                           ('expiration', {'optional': True}),\n                           ])\n\n    @property\n    def is_expired(self):\n        exp = self.get('expiration', None)\n        if not exp:\n            return False\n        exp_epoch = iso8601_to_epoch(exp)\n        # -2 to artificially shorten duration of the allotment to avoid\n        # possible race conditions between us checking either it has\n        # already expired before submitting a request.\n        expire_in = (exp_epoch - time.time() - 2) / 3600.\n\n        lgr.debug(\n            (\"Credential %s has expired %.2fh ago\"\n                if expire_in <= 0 else \"Credential %s will expire in %.2fh\")\n            % (self.name, expire_in))\n        return expire_in <= 0\n\n\n@auto_repr\nclass CompositeCredential(Credential):\n    \"\"\"Credential which represent a sequence of Credentials where front one is exposed to user\n    \"\"\"\n\n    # To be defined in sub-classes\n    _CREDENTIAL_CLASSES = None\n    _CREDENTIAL_ADAPTERS = None\n\n    def _prepare(self):\n        assert len(self._CREDENTIAL_CLASSES) > 1, \"makes sense only if there is > 1 credential\"\n        assert len(self._CREDENTIAL_CLASSES) == len(self._CREDENTIAL_ADAPTERS) + 1, \\\n            \"there should be 1 less of adapter than _CREDENTIAL_CLASSES\"\n\n        for C in self._CREDENTIAL_CLASSES:\n            assert issubclass(C, Credential), \"%s must be a subclass of Credential\" % C\n\n        # First credential should bear the name and url\n        credentials = [self._CREDENTIAL_CLASSES[0](self.name, url=self.url, keyring=self._keyring)]\n        # and we just reuse its _FIELDS for _ask_field_value etc\n        self._FIELDS = credentials[0]._FIELDS\n        # the rest with index suffix, but storing themselves in the same keyring\n        for iC, C in enumerate(self._CREDENTIAL_CLASSES[1:]):\n            credentials.append(\n                C(name=\"%s:%d\" % (self.name, iC + 1), url=None, keyring=self._keyring)\n            )\n        self._credentials = credentials\n\n        super(CompositeCredential, self)._prepare()\n\n    # Here it becomes tricky, since theoretically it is the \"tail\"\n    # ones which might expire etc, so we wouldn't exactly know what\n    # new credentials outside process wanted -- it would be silly to ask\n    # right away the \"entry\" credentials if it is just the expiration of the\n    # tail credentials\n    def enter_new(self):\n        # should invalidate/remove all tail credentials to avoid failing attempts to login\n        self._credentials[0].enter_new()\n        self.refresh()\n\n    def refresh(self):\n        \"\"\"Re-establish \"dependent\" credentials\n\n        E.g. if code outside was reported that it expired somehow before known expiration datetime\n        \"\"\"\n        for c in self._credentials[1:]:\n            c.delete()\n        # trigger re-establishing the chain\n        _ = self()\n        if self.is_expired:\n            raise RuntimeError(\"Credential %s expired right upon refresh: should have not happened\")\n\n    @property\n    def is_expired(self):\n        return any(c.is_expired for c in self._credentials)\n\n    def __call__(self):\n        \"\"\"Obtain credentials from a keyring and if any is not known -- ask\"\"\"\n        # Start from the tail until we have credentials set\n        idx = len(self._credentials) - 1\n        for c in self._credentials[::-1]:\n            if c.is_known and not c.is_expired:\n                break\n            idx -= 1\n\n        if idx < 0:\n            # none was known, all the same -- start with the first one\n            idx = 0\n\n        # TODO: consider moving logic of traversal into authenticator since it is\n        # the one spitting out authentication error etc\n        # Theoretically we could just reuse 'fields' from adapter in the next step\n        # but let's do full circle, so that if any \"normalization\" is done by\n        # Credentials we take that into account\n        for c, adapter, next_c in zip(\n                self._credentials[idx:],\n                self._CREDENTIAL_ADAPTERS[idx:],\n                self._credentials[idx + 1:]):\n            fields = c()\n            next_fields = adapter(self, **fields)\n            next_c.set(**next_fields)\n\n        return self._credentials[-1]()\n\n\ndef _nda_adapter(composite, user=None, password=None):\n    from datalad.support.third.nda_aws_token_generator import NDATokenGenerator\n    from .. import cfg\n    nda_auth_url = cfg.obtain('datalad.externals.nda.dbserver')\n    gen = NDATokenGenerator(nda_auth_url)\n    lgr.debug(\"Generating token for NDA user %s using %s talking to %s\",\n              user, gen, nda_auth_url)\n    try:\n        token = gen.generate_token(user, password)\n    except Exception as exc:  # it is really just an \"Exception\"\n        exc_str = str(exc).lower()\n        # ATM it is \"Invalid username and/or password\"\n        # but who knows what future would bring\n        if \"invalid\" in exc_str and (\"user\" in exc_str or \"password\" in exc_str):\n            raise AccessDeniedError(exc_str)\n        raise\n    # There are also session and expiration we ignore... TODO anything about it?!!!\n    # we could create a derived AWS_S3 which would also store session and expiration\n    # and then may be Composite could use those????\n    return dict(key_id=token.access_key, secret_id=token.secret_key,\n                session=token.session, expiration=token.expiration)\n\n\nclass NDA_S3(CompositeCredential):\n    \"\"\"Credential to access NDA AWS\n\n    So for NDA we need a credential which is a composite credential.\n    User provides UserPassword and then some adapter generates AWS_S3\n    out of it\n    \"\"\"\n    _CREDENTIAL_CLASSES = (UserPassword, AWS_S3)\n    _CREDENTIAL_ADAPTERS = (_nda_adapter,)\n\n\ndef _loris_adapter(composite, user=None, password=None, **kwargs):\n    from datalad.support.third.loris_token_generator import LORISTokenGenerator\n\n    gen = LORISTokenGenerator(url=composite.url)\n    token = gen.generate_token(user, password)\n\n    return dict(token=token)\n\n\nclass LORIS_Token(CompositeCredential):\n    _CREDENTIAL_CLASSES = (UserPassword, Token)\n    _CREDENTIAL_ADAPTERS = (_loris_adapter,)\n\n    def __init__(self, name, url=None, keyring=None):\n        super(CompositeCredential, self).__init__(name, url, keyring)\n\n\nCREDENTIAL_TYPES = {\n    'user_password': UserPassword,\n    'aws-s3': AWS_S3,\n    'nda-s3': NDA_S3,\n    'token': Token,\n    'loris-token': LORIS_Token\n}\n"}
A user will ask you to solve a task. You should first draft your thinking process (inner monologue). Then, generate the solution.

Your response format must follow the template below:
<think>
Your thoughts or/and draft, like working through an exercise on scratch paper. Be as casual and as long as you want until you are confident to generate a correct solution.
</think>
<solution>
Final solution presented to the user.
</solution>
system
We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[ISSUE]
**Title**: Credential values from environment variables are not retrieved correctly

**Description**:
When environment variables for credentials are set, the system fails to retrieve the expected values, resulting in credentials being marked as unknown.

**Bug Example**:
```python
import os
from datalad.downloaders.credentials import AWS_S3, Keyring

keyring = Keyring()
cred = AWS_S3('test-s3', keyring=keyring)

# Initially, credentials are unknown
print(cred.is_known)  # Expected: False

# Setting environment variable for key_id
os.environ['DATALAD_test_s3_key_id'] = '1'
print(cred.get('key_id'))  # Expected: '1'

# Setting environment variable for secret_id
os.environ['DATALAD_test_s3_secret_id'] = '2'
print(cred.get('secret_id'))  # Expected: '2'
print(cred.is_known)  # Expected: True
```

**Expected Behavior**:
When the environment variables `DATALAD_test_s3_key_id` and `DATALAD_test_s3_secret_id` are set to `'1'` and `'2'` respectively, the methods `cred.get('key_id')` and `cred.get('secret_id')` should return `'1'` and `'2'`. Additionally, `cred.is_known` should return `True`, indicating that the credentials have been successfully retrieved.

**Actual Behavior**:
After setting the environment variables, `cred.get('key_id')` returns `None` instead of `'1'`, and `cred.is_known` remains `False`. This means the credentials are not recognized even though the environment variables are correctly set.

[/ISSUE]
--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.

--- BEGIN FILE ---
```
### datalad/downloaders/credentials.py
# emacs: -*- mode: python; py-indent-offset: 4; tab-width: 4; indent-tabs-mode: nil -*-
# ex: set sts=4 ts=4 sw=4 noet:
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
#
#   See COPYING file distributed along with the datalad package for the
#   copyright and license terms.
#
# ## ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ### ##
"""Interface information about credentials

Provides minimalistic interface to deal (query, request, store) with most common
types of credentials.  To be used by Authenticators
"""

__dev_doc__ = """
Possibly useful in the future 3rd part developments

https://github.com/omab/python-social-auth
    social authentication/registration mechanism with support for several
    frameworks and auth providers
"""

import time

from collections import OrderedDict

from ..dochelpers import exc_str
from ..support.exceptions import AccessDeniedError
from ..support.keyring_ import keyring as keyring_
from ..ui import ui
from ..utils import auto_repr
from ..support.network import iso8601_to_epoch

from logging import getLogger
lgr = getLogger('datalad.downloaders.credentials')


@auto_repr
class Credential(object):
    """Base class for different types of credentials
    """

    # Should be defined in a subclass as an OrderedDict of fields
    # name -> dict(attributes)
    _FIELDS = None
    _KNOWN_ATTRS = {
        'hidden',    # UI should not display the value
        'repeat',    # UI should repeat entry or not. Set to False to override default logic
        'optional',  # Not mandatory thus not requested if not set
    }

    def __init__(self, name, url=None, keyring=None):
        """
        Parameters
        ----------
        name : str
            Name of the credential, as it would be identified by in the centralized
            storage of credentials
        url : str, optional
            URL string to point users to where to seek obtaining the credentials
        keyring : a keyring
            An object providing (g|s)et_password.  If None, keyring module is used
            as is
        """
        self.name = name
        self.url = url
        self._keyring = keyring or keyring_
        self._prepare()

    def _prepare(self):
        """Additional house-keeping possibly to be performed in subclasses

        Created to avoid all the passing args/kwargs of __init__ all the time
        """
        # Basic checks
        for f, fattrs in self._FIELDS.items():
            unknown_attrs = set(fattrs).difference(self._KNOWN_ATTRS)
            if unknown_attrs:
                raise ValueError("Unknown attributes %s. Known are: %s"
                                 % (unknown_attrs, self._KNOWN_ATTRS))

    def _is_field_optional(self, f):
        return self._FIELDS[f].get('optional', False)

    @property
    def is_known(self):
        """Return True if values for all fields of the credential are known"""
        try:
            return all(
                self._is_field_optional(f) or self._get_field_value(f) is not None
                for f in self._FIELDS)
        except Exception as exc:
            lgr.warning("Failed to query keyring: %s" % exc_str(exc))
            return False

    def _get_field_value(self, field):
        return self._keyring.get(self.name, field)

    def _ask_field_value(self, f, instructions=None):
        msg = instructions if instructions else \
            ("You need to authenticate with %r credentials." % self.name +
                  (" %s provides information on how to gain access"
                   % self.url if self.url else ''))

        # provide custom options only if set for the field
        f_props = self._FIELDS[f]
        kwargs = {}
        for p in ('hidden', 'repeat'):
            if p in f_props:
                kwargs[p] = f_props[p]
        return ui.question(
            f,
            title=msg,
            **kwargs
        )

    def _ask_and_set(self, f, instructions=None):
        v = self._ask_field_value(f, instructions=instructions)
        self.set(**{f: v})
        return v

    def enter_new(self, instructions=None, **kwargs):
        """Enter new values for the credential fields

        Parameters
        ----------
        instructions : str, optional
          If given, the auto-generated instructions based on a login-URL are
          replaced by the given string
        **kwargs
          Any given key value pairs with non-None values are used to set the
          field `key` to the given value, without asking for user input
        """
        if kwargs:
            unknown_fields = set(kwargs).difference(self._FIELDS)
            known_fields = set(self._FIELDS).difference(kwargs)
            if unknown_fields:
                raise ValueError(
                    "Unknown to %s field(s): %s.  Known but not specified: %s"
                    % (self,
                       ', '.join(sorted(unknown_fields)),
                       ', '.join(sorted(known_fields))
                       ))
        # Use ui., request credential fields corresponding to the type
        for f in self._FIELDS:
            if kwargs.get(f, None):
                # use given value, don't ask
                self.set(**{f: kwargs[f]})
            elif not self._is_field_optional(f):
                self._ask_and_set(f, instructions=instructions)

    def __call__(self):
        """Obtain credentials from a keyring and if any is not known -- ask"""
        fields = {}
        for f in self._FIELDS:
            v = self._get_field_value(f)
            if not self._is_field_optional(f):
                while v is None:  # was not known
                    v = self._ask_and_set(f)
                fields[f] = v
            elif v is not None:
                fields[f] = v
        return fields

    def set(self, **kwargs):
        """Set field(s) of the credential"""
        for f, v in kwargs.items():
            if f not in self._FIELDS:
                raise ValueError("Unknown field %s. Known are: %s"
                                 % (f, self._FIELDS.keys()))
            self._keyring.set(self.name, f, v)

    def get(self, f, default=None):
        """Get a field of the credential"""
        if f not in self._FIELDS:
            raise ValueError("Unknown field %s. Known are: %s"
                             % (f, self._FIELDS.keys()))
        try:
            return self._get_field_value(f)
        except:  # MIH: what could even happen? _keyring not a dict?
            return default

    def delete(self):
        """Deletes credential values from the keyring"""
        for f in self._FIELDS:
            self._keyring.delete(self.name, f)


class UserPassword(Credential):
    """Simple type of a credential which consists of user/password pair"""

    _FIELDS = OrderedDict([('user', {}),
                           ('password', {'hidden': True})])

    is_expired = False  # no expiration provisioned


class Token(Credential):
    """Simple type of a credential which provides a single token"""

    _FIELDS = OrderedDict([('token', {'hidden': True, 'repeat': False})])

    is_expired = False  # no expiration provisioned


class AWS_S3(Credential):
    """Credential for AWS S3 service"""

    _FIELDS = OrderedDict([('key_id', {'repeat': False}),
                           ('secret_id', {'hidden': True, 'repeat': False}),
                           ('session', {'optional': True}),
                           ('expiration', {'optional': True}),
                           ])

    @property
    def is_expired(self):
        exp = self.get('expiration', None)
        if not exp:
            return False
        exp_epoch = iso8601_to_epoch(exp)
        # -2 to artificially shorten duration of the allotment to avoid
        # possible race conditions between us checking either it has
        # already expired before submitting a request.
        expire_in = (exp_epoch - time.time() - 2) / 3600.

        lgr.debug(
            ("Credential %s has expired %.2fh ago"
                if expire_in <= 0 else "Credential %s will expire in %.2fh")
            % (self.name, expire_in))
        return expire_in <= 0


@auto_repr
class CompositeCredential(Credential):
    """Credential which represent a sequence of Credentials where front one is exposed to user
    """

    # To be defined in sub-classes
    _CREDENTIAL_CLASSES = None
    _CREDENTIAL_ADAPTERS = None

    def _prepare(self):
        assert len(self._CREDENTIAL_CLASSES) > 1, "makes sense only if there is > 1 credential"
        assert len(self._CREDENTIAL_CLASSES) == len(self._CREDENTIAL_ADAPTERS) + 1, \
            "there should be 1 less of adapter than _CREDENTIAL_CLASSES"

        for C in self._CREDENTIAL_CLASSES:
            assert issubclass(C, Credential), "%s must be a subclass of Credential" % C

        # First credential should bear the name and url
        credentials = [self._CREDENTIAL_CLASSES[0](self.name, url=self.url, keyring=self._keyring)]
        # and we just reuse its _FIELDS for _ask_field_value etc
        self._FIELDS = credentials[0]._FIELDS
        # the rest with index suffix, but storing themselves in the same keyring
        for iC, C in enumerate(self._CREDENTIAL_CLASSES[1:]):
            credentials.append(
                C(name="%s:%d" % (self.name, iC + 1), url=None, keyring=self._keyring)
            )
        self._credentials = credentials

        super(CompositeCredential, self)._prepare()

    # Here it becomes tricky, since theoretically it is the "tail"
    # ones which might expire etc, so we wouldn't exactly know what
    # new credentials outside process wanted -- it would be silly to ask
    # right away the "entry" credentials if it is just the expiration of the
    # tail credentials
    def enter_new(self):
        # should invalidate/remove all tail credentials to avoid failing attempts to login
        self._credentials[0].enter_new()
        self.refresh()

    def refresh(self):
        """Re-establish "dependent" credentials

        E.g. if code outside was reported that it expired somehow before known expiration datetime
        """
        for c in self._credentials[1:]:
            c.delete()
        # trigger re-establishing the chain
        _ = self()
        if self.is_expired:
            raise RuntimeError("Credential %s expired right upon refresh: should have not happened")

    @property
    def is_expired(self):
        return any(c.is_expired for c in self._credentials)

    def __call__(self):
        """Obtain credentials from a keyring and if any is not known -- ask"""
        # Start from the tail until we have credentials set
        idx = len(self._credentials) - 1
        for c in self._credentials[::-1]:
            if c.is_known and not c.is_expired:
                break
            idx -= 1

        if idx < 0:
            # none was known, all the same -- start with the first one
            idx = 0

        # TODO: consider moving logic of traversal into authenticator since it is
        # the one spitting out authentication error etc
        # Theoretically we could just reuse 'fields' from adapter in the next step
        # but let's do full circle, so that if any "normalization" is done by
        # Credentials we take that into account
        for c, adapter, next_c in zip(
                self._credentials[idx:],
                self._CREDENTIAL_ADAPTERS[idx:],
                self._credentials[idx + 1:]):
            fields = c()
            next_fields = adapter(self, **fields)
            next_c.set(**next_fields)

        return self._credentials[-1]()


def _nda_adapter(composite, user=None, password=None):
    from datalad.support.third.nda_aws_token_generator import NDATokenGenerator
    from .. import cfg
    nda_auth_url = cfg.obtain('datalad.externals.nda.dbserver')
    gen = NDATokenGenerator(nda_auth_url)
    lgr.debug("Generating token for NDA user %s using %s talking to %s",
              user, gen, nda_auth_url)
    try:
        token = gen.generate_token(user, password)
    except Exception as exc:  # it is really just an "Exception"
        exc_str = str(exc).lower()
        # ATM it is "Invalid username and/or password"
        # but who knows what future would bring
        if "invalid" in exc_str and ("user" in exc_str or "password" in exc_str):
            raise AccessDeniedError(exc_str)
        raise
    # There are also session and expiration we ignore... TODO anything about it?!!!
    # we could create a derived AWS_S3 which would also store session and expiration
    # and then may be Composite could use those????
    return dict(key_id=token.access_key, secret_id=token.secret_key,
                session=token.session, expiration=token.expiration)


class NDA_S3(CompositeCredential):
    """Credential to access NDA AWS

    So for NDA we need a credential which is a composite credential.
    User provides UserPassword and then some adapter generates AWS_S3
    out of it
    """
    _CREDENTIAL_CLASSES = (UserPassword, AWS_S3)
    _CREDENTIAL_ADAPTERS = (_nda_adapter,)


def _loris_adapter(composite, user=None, password=None, **kwargs):
    from datalad.support.third.loris_token_generator import LORISTokenGenerator

    gen = LORISTokenGenerator(url=composite.url)
    token = gen.generate_token(user, password)

    return dict(token=token)


class LORIS_Token(CompositeCredential):
    _CREDENTIAL_CLASSES = (UserPassword, Token)
    _CREDENTIAL_ADAPTERS = (_loris_adapter,)

    def __init__(self, name, url=None, keyring=None):
        super(CompositeCredential, self).__init__(name, url, keyring)


CREDENTIAL_TYPES = {
    'user_password': UserPassword,
    'aws-s3': AWS_S3,
    'nda-s3': NDA_S3,
    'token': Token,
    'loris-token': LORIS_Token
}

```
--- END FILE ---

Please first localize the bug based on the issue statement, and then generate *SEARCH/REPLACE* edits to fix the issue.

Every *SEARCH/REPLACE* edit must use this format:
1. The file path
2. The start of search block: <<<<<<< SEARCH
3. A contiguous chunk of lines to search for in the existing source code
4. The dividing line: =======
5. The lines to replace into the source code
6. The end of the replace block: >>>>>>> REPLACE

Here is an example:

```python
### mathweb/flask/app.py
<<<<<<< SEARCH
from flask import Flask
=======
import math
from flask import Flask
>>>>>>> REPLACE
```
</solution>

Please note that the *SEARCH/REPLACE* edit REQUIRES PROPER INDENTATION. If you would like to add the line '        print(x)', you must fully write that out, with all those spaces before the code!
Wrap each *SEARCH/REPLACE* edit in a code block as shown in the example above. If you have multiple *SEARCH/REPLACE* edits, use a separate code block for each one.
user