From aeb19a36dd91eed933d2e3e0851febaf0e102578 Mon Sep 17 00:00:00 2001
From: Timm Fitschen <t.fitschen@indiscale.com>
Date: Thu, 7 May 2020 15:09:26 +0200
Subject: [PATCH] WIP: auth_token password method
---
src/caosdb/common/models.py | 12 +-
.../connection/authentication/auth_token.py | 93 +++++++++
.../connection/authentication/interface.py | 10 +-
src/caosdb/connection/connection.py | 6 +-
unittests/test_add_property.py | 185 ++++++++++--------
unittests/test_connection.py | 21 +-
6 files changed, 229 insertions(+), 98 deletions(-)
create mode 100644 src/caosdb/connection/authentication/auth_token.py
diff --git a/src/caosdb/common/models.py b/src/caosdb/common/models.py
index 6c4ed8b4..293806a3 100644
--- a/src/caosdb/common/models.py
+++ b/src/caosdb/common/models.py
@@ -3481,6 +3481,11 @@ class DropOffBox(list):
return self
+class UserInfo():
+
+ def __init__(self, xml):
+ self.roles = [ role.text for role in xml.findall("Roles/Role")]
+
class Info():
def __init__(self):
@@ -3499,7 +3504,10 @@ class Info():
for e in xml:
m = _parse_single_xml_element(e)
- self.messages.append(m)
+ if isinstance(m, UserInfo):
+ self.user_info = m
+ else:
+ self.messages.append(m)
def __str__(self):
if "counts" not in self.messages:
@@ -3622,6 +3630,8 @@ def _parse_single_xml_element(elem):
return ACL(xml=elem)
elif elem.tag == "Permissions":
return Permissions(xml=elem)
+ elif elem.tag == "UserInfo":
+ return UserInfo(xml=elem)
else:
return Message(type=elem.tag, code=elem.get(
"code"), description=elem.get("description"), body=elem.text)
diff --git a/src/caosdb/connection/authentication/auth_token.py b/src/caosdb/connection/authentication/auth_token.py
new file mode 100644
index 00000000..4cff568d
--- /dev/null
+++ b/src/caosdb/connection/authentication/auth_token.py
@@ -0,0 +1,93 @@
+# -*- coding: utf-8 -*-
+#
+# ** header v3.0
+# This file is a part of the CaosDB Project.
+#
+# Copyright (C) 2018 Research Group Biomedical Physics,
+# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
+# Copyright (C) 2020 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2020 Timm Fitschen <f.fitschen@indiscale.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <https://www.gnu.org/licenses/>.
+#
+# ** end header
+#
+"""auth_token.
+
+A Authentictor which only uses only a pre-supplied authentication token.
+"""
+from __future__ import absolute_import, unicode_literals, print_function
+from .interface import AbstractAuthenticator, CaosDBServerConnection
+from caosdb.exceptions import LoginFailedException
+
+
+def get_authentication_provider():
+ """get_authentication_provider.
+
+ Return an authenticator which only uses a pre-supplied authentication
+ token.
+
+ Returns
+ -------
+ AuthTokenAuthenticator
+ """
+ return AuthTokenAuthenticator()
+
+
+class AuthTokenAuthenticator(AbstractAuthenticator):
+ """AuthTokenAuthenticator.
+
+ Subclass of AbstractAuthenticator which provides authentication only via
+ a given authentication token.
+
+ Methods
+ -------
+ login
+ logout
+ configure
+ """
+
+ def __init__(self):
+ super(AuthTokenAuthenticator, self).__init__()
+ self.auth_token = None
+ self._connection = None
+
+ def login(self):
+ self._login()
+
+ def _login(self):
+ raise LoginFailedException("The authentication token is expired or you "
+ "have been logged out otherwise. The "
+ "auth_token authenticator cannot log in "
+ "again. You must provide a new "
+ "authentication token.")
+
+ def logout(self):
+ self._logout()
+
+ def _logout(self):
+ self.logger.debug("[LOGOUT]")
+ if self.auth_token is not None:
+ self._connection.request(method="DELETE", path="logout")
+ self.auth_token = None
+
+ def configure(self, **config):
+ if "auth_token" in config:
+ self.auth_token = config["auth_token"]
+ if "connection" in config:
+ self._connection = config["connection"]
+ if not isinstance(self._connection, CaosDBServerConnection):
+ raise Exception("""Bad configuration of the caosdb connection.
+ The `connection` must be an instance of
+ `CaosDBConnection`.""")
diff --git a/src/caosdb/connection/authentication/interface.py b/src/caosdb/connection/authentication/interface.py
index f156345a..d9a9b430 100644
--- a/src/caosdb/connection/authentication/interface.py
+++ b/src/caosdb/connection/authentication/interface.py
@@ -50,6 +50,8 @@ class AbstractAuthenticator(ABC):
logger : Logger
A logger which should be used for all logging which has to do with
authentication.
+ auth_token : str
+ A string representation of a CaosDB Auth Token.
Methods
-------
@@ -59,10 +61,6 @@ class AbstractAuthenticator(ABC):
on_request
on_response
- Attributes
- ----------
- auth_token : str
- A string representation of a CaosDB Auth Token.
"""
def __init__(self):
@@ -187,7 +185,7 @@ class CredentialsAuthenticator(AbstractAuthenticator):
self._logout()
def _logout(self):
- _LOGGER.debug("[LOGOUT]")
+ self.logger.debug("[LOGOUT]")
if self.auth_token is not None:
self._connection.request(method="DELETE", path="logout")
self.auth_token = None
@@ -195,7 +193,7 @@ class CredentialsAuthenticator(AbstractAuthenticator):
def _login(self):
username = self._credentials_provider.username
password = self._credentials_provider.password
- _LOGGER.debug("[LOGIN] %s", username)
+ self.logger.debug("[LOGIN] %s", username)
# we need a username for this:
if username is None:
diff --git a/src/caosdb/connection/connection.py b/src/caosdb/connection/connection.py
index 3553c86c..5c6d3005 100644
--- a/src/caosdb/connection/connection.py
+++ b/src/caosdb/connection/connection.py
@@ -333,6 +333,9 @@ def configure_connection(**kwargs):
Whether SSL certificate warnings should be ignored. Only use this for
development purposes! (Default: False)
+ auth_token : str
+ An authentication token which has been issued by the CaosDB Server.
+
implementation : CaosDBServerConnection
The class which implements the connection. (Default:
_DefaultCaosDBServerConnection)
@@ -473,9 +476,6 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
self._authenticator = _get_authenticator(
connection=self._delegate_connection, **config)
- if "auth_token" in config:
- self._authenticator.auth_token = config["auth_token"]
-
return self
def retrieve(self, entity_uri_segments=None, query_dict=None, **kwargs):
diff --git a/unittests/test_add_property.py b/unittests/test_add_property.py
index bd68f31b..917ee68b 100644
--- a/unittests/test_add_property.py
+++ b/unittests/test_add_property.py
@@ -21,68 +21,82 @@
#
# ** end header
#
+from pytest import raises
import caosdb as db
-from nose.tools import assert_is, assert_is_none, assert_equals, assert_is_not_none, assert_raises
def test_no_parameter():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
- with assert_raises(UserWarning) as cm:
+ with raises(UserWarning) as cm:
rec.add_property()
- assert_equals(
- cm.exception.args[0],
- "This method expects you to pass at least an entity, a name or an id.")
- assert_equals(0, len(rec.get_properties()))
+ assert cm.value.args[0] == ("This method expects you to pass at "
+ "least an entity, a name or an id.")
+ assert 0 == len(rec.get_properties())
def test_only_value_parameter():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
- with assert_raises(UserWarning) as cm:
+ with raises(UserWarning) as cm:
rec.add_property(value="bla")
- assert_equals(
- cm.exception.args[0],
- "This method expects you to pass at least an entity, a name or an id.")
- assert_equals(0, len(rec.get_properties()))
+ assert cm.value.args[0] == ("This method expects you to pass at "
+ "least an entity, a name or an id.")
+ assert 0 == len(rec.get_properties())
def test_property_name_ambiguity_1():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
- with assert_raises(UserWarning) as cm:
+ with raises(UserWarning) as cm:
rec.add_property("one_name", name="another_name")
- assert_equals(
- cm.exception.args[0],
- "The first parameter was neither an instance of Entity nor an integer. Therefore the string representation of your first parameter would normally be interpreted name of the property which is to be added. But you have also specified a parameter 'name' in the method call. This is ambiguous and cannot be processed.")
- assert_equals(0, len(rec.get_properties()))
+ assert cm.value.args[0] == ("The first parameter was neither an "
+ "instance of Entity nor an integer. "
+ "Therefore the string representation of "
+ "your first parameter would normally be "
+ "interpreted name of the property which "
+ "is to be added. But you have also "
+ "specified a parameter 'name' in the "
+ "method call. This is ambiguous and "
+ "cannot be processed.")
+ assert 0 == len(rec.get_properties())
def test_property_name_ambiguity_2():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
- with assert_raises(UserWarning) as cm:
+ with raises(UserWarning) as cm:
rec.add_property({}, name="another_name")
- assert_equals(
- cm.exception.args[0],
- "The first parameter was neither an instance of Entity nor an integer. Therefore the string representation of your first parameter would normally be interpreted name of the property which is to be added. But you have also specified a parameter 'name' in the method call. This is ambiguous and cannot be processed.")
- assert_equals(0, len(rec.get_properties()))
+ assert cm.value.args[0] == ("The first parameter was neither an "
+ "instance of Entity nor an integer. "
+ "Therefore the string representation of "
+ "your first parameter would normally be "
+ "interpreted name of the property which "
+ "is to be added. But you have also "
+ "specified a parameter 'name' in the "
+ "method call. This is ambiguous and "
+ "cannot be processed.")
+ assert 0 == len(rec.get_properties())
def test_property_id_ambiguity():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
- with assert_raises(UserWarning) as cm:
+ with raises(UserWarning) as cm:
rec.add_property(25, id=26)
- assert_equals(
- cm.exception.args[0],
- "The first parameter was an integer which would normally be interpreted as the id of the property which is to be added. But you have also specified a parameter 'id' in the method call. This is ambiguous and cannot be processed.")
- assert_equals(0, len(rec.get_properties()))
+
+ assert cm.value.args[0] == ("The first parameter was an integer which "
+ "would normally be interpreted as the id of "
+ "the property which is to be added. But you "
+ "have also specified a parameter 'id' in the "
+ "method call. This is ambiguous and cannot be "
+ "processed.")
+ assert 0 == len(rec.get_properties())
def test_property_parameter_with_entity():
@@ -94,18 +108,17 @@ def test_property_parameter_with_entity():
unit="m",
description="This is the length of something.")
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(abstract_property)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property("length")
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.name, "length")
- assert_equals(concrete_property.id, 512)
- assert_equals(concrete_property.description,
- "This is the length of something.")
- assert_equals(concrete_property.unit, "m")
- assert_equals(concrete_property.datatype, db.DOUBLE)
- assert_is(concrete_property._wrapped_entity, abstract_property)
+ assert concrete_property is not None
+ assert concrete_property.name == "length"
+ assert concrete_property.id == 512
+ assert concrete_property.description == "This is the length of something."
+ assert concrete_property.unit == "m"
+ assert concrete_property.datatype == db.DOUBLE
+ assert concrete_property._wrapped_entity == abstract_property
def test_property_parameter_with_entity_and_value():
@@ -117,54 +130,53 @@ def test_property_parameter_with_entity_and_value():
unit="m",
description="This is the length of something.")
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(abstract_property, 3.14)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property("length")
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.name, "length")
- assert_equals(concrete_property.id, 512)
- assert_equals(concrete_property.description,
- "This is the length of something.")
- assert_equals(concrete_property.unit, "m")
- assert_equals(concrete_property.value, 3.14)
- assert_equals(concrete_property.datatype, db.DOUBLE)
- assert_is(concrete_property._wrapped_entity, abstract_property)
+ assert concrete_property is not None
+ assert concrete_property.name == "length"
+ assert concrete_property.id == 512
+ assert concrete_property.description == "This is the length of something."
+ assert concrete_property.unit == "m"
+ assert concrete_property.value == 3.14
+ assert concrete_property.datatype == db.DOUBLE
+ assert concrete_property._wrapped_entity == abstract_property
def test_property_parameter_with_id():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(512)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property(512)
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.id, 512)
+ assert concrete_property is not None
+ assert concrete_property.id == 512
def test_property_parameter_with_id_and_value():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(512, 3.14)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property(512)
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.id, 512)
- assert_equals(concrete_property.value, 3.14)
+ assert concrete_property is not None
+ assert concrete_property.id == 512
+ assert concrete_property.value == 3.14
def test_datatype():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(512, 3.14)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property(512)
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.id, 512)
- assert_equals(concrete_property.value, 3.14)
+ assert concrete_property is not None
+ assert concrete_property.id == 512
+ assert concrete_property.value == 3.14
def test_property_parameter_with_entity_and_datatype():
@@ -176,41 +188,40 @@ def test_property_parameter_with_entity_and_datatype():
unit="m",
description="This is the length of something.")
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(abstract_property, 3.14, datatype=db.INTEGER)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property("length")
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.name, "length")
- assert_equals(concrete_property.id, 512)
- assert_equals(concrete_property.description,
- "This is the length of something.")
- assert_equals(concrete_property.unit, "m")
- assert_equals(concrete_property.value, 3.14)
- assert_equals(concrete_property.datatype, db.INTEGER)
- assert_is(concrete_property._wrapped_entity, abstract_property)
+ assert concrete_property is not None
+ assert concrete_property.name == "length"
+ assert concrete_property.id == 512
+ assert concrete_property.description == "This is the length of something."
+ assert concrete_property.unit == "m"
+ assert concrete_property.value == 3.14
+ assert concrete_property.datatype == db.INTEGER
+ assert concrete_property._wrapped_entity == abstract_property
def test_kw_name_and_value():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(name="length", value=3.14)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property("length")
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.value, 3.14)
+ assert concrete_property is not None
+ assert concrete_property.value == 3.14
def test_kw_id_and_value():
rec = db.Record()
- assert_equals(0, len(rec.get_properties()))
+ assert 0 == len(rec.get_properties())
rec.add_property(id=512, value=3.14)
- assert_equals(1, len(rec.get_properties()))
+ assert 1 == len(rec.get_properties())
concrete_property = rec.get_property(512)
- assert_is_not_none(concrete_property)
- assert_equals(concrete_property.value, 3.14)
+ assert concrete_property is not None
+ assert concrete_property.value == 3.14
def test_add_list_of_entitities():
@@ -220,7 +231,7 @@ def test_add_list_of_entitities():
values.append(db.Record(name=str(i)))
rec.add_property("listOfEntities", values)
for e in rec.get_property("listOfEntities").value:
- assert_is_none(e.id)
+ assert e.id is None
i = 0
for val in values:
@@ -229,5 +240,5 @@ def test_add_list_of_entitities():
i = 0
for e in rec.get_property("listOfEntities").value:
- assert_equals(i, e.id)
+ assert i == e.id
i += 1
diff --git a/unittests/test_connection.py b/unittests/test_connection.py
index c1f62088..2323ff97 100644
--- a/unittests/test_connection.py
+++ b/unittests/test_connection.py
@@ -261,9 +261,28 @@ def test_missing_auth_method():
def test_missing_password():
connection = configure_connection()
connection.configure(implementation=setup_two_resources,
- password_method="plain", auth_token="[test-auth-token]")
+ password_method="plain")
+ connection._authenticator.auth_token="[test-auth-token]"
assert connection.retrieve(["some"]).headers["Cookie"] == "SessionToken=%5Btest-auth-token%5D;"
+
connection.configure(implementation=setup_two_resources,
password_method="plain")
with raises(LoginFailedException):
connection.delete(["401"])
+
+
+def test_auth_token_connection():
+ connection = configure_connection(auth_token="blablabla",
+ password_method="auth_token",
+ implementation=setup_two_resources)
+ connection.retrieve(["some"]).headers["Cookie"] == "SessionToken=blablabla;"
+
+ connection._logout()
+ with raises(LoginFailedException) as cm:
+ connection.retrieve(["some"]).headers["Cookie"] == "SessionToken=blablabla;"
+ assert cm.value.args[0] == ("The authentication token is expired or you "
+ "have been logged out otherwise. The "
+ "auth_token authenticator cannot log in "
+ "again. You must provide a new authentication "
+ "token.")
+
--
GitLab