from keystoneauth1.exceptions.http import Conflict
from keystoneclient.v3 import client
from http import HTTPStatus
-from osm_nbi.validation import is_valid_uuid
+from osm_nbi.validation import is_valid_uuid, validate_input, http_schema
class AuthconnKeystone(Authconn):
self.domains_id2name = {}
self.domains_name2id = {}
- self.auth_url = "http://{0}:{1}/v3".format(config.get("auth_url", "keystone"), config.get("auth_port", "5000"))
+ self.auth_url = config.get("auth_url")
+ if config.get("auth_url"):
+ validate_input(self.auth_url, http_schema)
+ else:
+ self.auth_url = "http://{0}:{1}/v3".format(config.get("auth_host", "keystone"),
+ config.get("auth_port", "5000"))
self.user_domain_name_list = config.get("user_domain_name", "default")
self.user_domain_name_list = self.user_domain_name_list.split(",")
# read only domain list
project_name=self.admin_project,
auth_url=self.auth_url)
self.sess = session.Session(auth=self.auth)
- self.keystone = client.Client(session=self.sess)
+ self.keystone = client.Client(session=self.sess, endpoint_override=self.auth_url)
def authenticate(self, credentials, token_info=None):
"""
or user_info.get("add_project_role_mappings") or user_info.get("remove_project_role_mappings"):
# if user_index>0, it is an external domain, that should not be updated
ctime = user_obj._admin.get("created", 0) if hasattr(user_obj, "_admin") else 0
- self.keystone.users.update(user_id, password=user_info.get("password"), name=user_info.get("username"),
- _admin={"created": ctime, "modified": time.time()})
+ try:
+ self.keystone.users.update(user_id, password=user_info.get("password"),
+ name=user_info.get("username"),
+ _admin={"created": ctime, "modified": time.time()})
+ except Exception as e:
+ if user_info.get("username") or user_info.get("password"):
+ raise AuthconnOperationException("Error during username/password change: {}".format(str(e)))
+ self.logger.error("Error during updating user profile: {}".format(str(e)))
for mapping in user_info.get("remove_project_role_mappings", []):
self.remove_role_from_user(user_obj, mapping["project"], mapping["role"])