Merge pull request #9587 from nuwang/set_password_if_null

[20.05] Assign random password in user manager if none specified
This commit is contained in:
Marius van den Beek
2020-06-02 12:04:50 +02:00
committed by GitHub
11 changed files with 85 additions and 71 deletions
-1
View File
@@ -125,7 +125,6 @@ class CustosAuthnz(IdentityProvider):
raise exceptions.AuthenticationFailed(message)
else:
user = trans.app.user_manager.create(email=email, username=username)
user.set_random_password()
trans.sa_session.add(user)
trans.sa_session.flush()
custos_authnz_token = CustosAuthnzToken(user=user,
+4 -1
View File
@@ -93,7 +93,10 @@ class UserManager(base.ModelManager, deletable.PurgableManagerMixin):
"""
self._error_on_duplicate_email(email)
user = self.model_class(email=email)
user.set_password_cleartext(password)
if password:
user.set_password_cleartext(password)
else:
user.set_random_password()
user.username = username
if self.app.config.user_activation_on:
user.active = False
+4
View File
@@ -51,6 +51,7 @@ import galaxy.security.passwords
import galaxy.util
from galaxy.model.item_attrs import get_item_annotation_str, UsesAnnotations
from galaxy.security import get_permitted_actions
from galaxy.security.validate_user_input import validate_password_str
from galaxy.util import (directory_hash_id, ready_name_for_url,
unicodify, unique_id)
from galaxy.util.bunch import Bunch
@@ -387,6 +388,9 @@ class User(Dictifiable, RepresentById):
"""
Set user password to the digest of `cleartext`.
"""
message = validate_password_str(cleartext, cleartext)
if message:
raise Exception("Invalid password: %s" % message)
if User.use_pbkdf2:
self.password = galaxy.security.passwords.hash_password(cleartext)
else:
+2
View File
@@ -18,6 +18,8 @@ def hash_password(password):
"""
Hash a password, currently will use the PBKDF2 scheme.
"""
if password is None:
raise Exception("password cannot be None")
return hash_password_PBKDF2(password)
+37 -17
View File
@@ -26,17 +26,44 @@ FILL_CHAR = '-'
PASSWORD_MIN_LEN = 6
def validate_email(trans, email, user=None, check_dup=True, allow_empty=False):
"""
Validates the email format, also checks whether the domain is blacklisted in the disposable domains configuration.
"""
def validate_email_str(email):
"""Validates a string containing an email address."""
message = ''
if (user and user.email == email) or (email == "" and allow_empty):
return message
if not(VALID_EMAIL_RE.match(email)):
message = "The format of the email address is not correct."
elif len(email) > EMAIL_MAX_LEN:
message = "Email address cannot be more than %d characters in length." % EMAIL_MAX_LEN
return message
def validate_password_str(password, confirm):
if password != confirm:
return "Passwords do not match."
if not password or len(password) < PASSWORD_MIN_LEN:
return "Use a password of at least %d characters." % PASSWORD_MIN_LEN
return ""
def validate_publicname_str(publicname):
"""Validates a string containing a public username."""
if len(publicname) < PUBLICNAME_MIN_LEN:
return "Public name must be at least %d characters in length." % (PUBLICNAME_MIN_LEN)
if len(publicname) > PUBLICNAME_MAX_LEN:
return "Public name cannot be more than %d characters in length." % (PUBLICNAME_MAX_LEN)
if not(VALID_PUBLICNAME_RE.match(publicname)):
return "Public name must contain only lower-case letters, numbers, '.', '_' and '-'."
return ''
def validate_email(trans, email, user=None, check_dup=True, allow_empty=False):
"""
Validates the email format, also checks whether the domain is blacklisted in the disposable domains configuration.
"""
if (user and user.email == email) or (email == "" and allow_empty):
return ''
message = validate_email_str(email)
if message:
pass
elif check_dup and trans.sa_session.query(trans.app.model.User).filter(func.lower(trans.app.model.User.table.c.email) == email.lower()).first():
message = "User with email '%s' already exists." % email
# If the whitelist is not empty filter out any domain not in the list and ignore blacklist.
@@ -63,12 +90,9 @@ def validate_publicname(trans, publicname, user=None):
"""
if user and user.username == publicname:
return ''
if len(publicname) < PUBLICNAME_MIN_LEN:
return "Public name must be at least %d characters in length." % (PUBLICNAME_MIN_LEN)
if len(publicname) > PUBLICNAME_MAX_LEN:
return "Public name cannot be more than %d characters in length." % (PUBLICNAME_MAX_LEN)
if not(VALID_PUBLICNAME_RE.match(publicname)):
return "Public name must contain only lower-case letters, numbers, '.', '_' and '-'."
message = validate_publicname_str(publicname)
if message:
return message
if trans.sa_session.query(trans.app.model.User).filter_by(username=publicname).first():
return "Public name is taken; please choose another."
return ''
@@ -89,8 +113,4 @@ def transform_publicname(publicname):
def validate_password(trans, password, confirm):
if len(password) < PASSWORD_MIN_LEN:
return "Use a password of at least %d characters." % PASSWORD_MIN_LEN
if password != confirm:
return "Passwords do not match."
return ""
return validate_password_str(password, confirm)
@@ -15,6 +15,7 @@ from galaxy import (
from galaxy.actions.admin import AdminActions
from galaxy.exceptions import ActionInputError, MessageException
from galaxy.model import tool_shed_install as install_model
from galaxy.security.validate_user_input import validate_password
from galaxy.tool_shed.util.repository_util import get_ids_of_tool_shed_repositories_being_installed
from galaxy.util import (
nice_size,
@@ -1387,10 +1388,9 @@ class AdminGalaxy(controller.JSAppLauncher, AdminActions, UsesQuotaMixin, QuotaP
else:
password = payload.get('password')
confirm = payload.get('confirm')
if len(password) < 6:
return self.message_exception(trans, 'Use a password of at least 6 characters.')
elif password != confirm:
return self.message_exception(trans, 'Passwords do not match.')
message = validate_password(trans, password, confirm)
if message:
return self.message_exception(trans, message)
for user in users.values():
user.set_password_cleartext(password)
trans.sa_session.add(user)
+3 -8
View File
@@ -4,12 +4,12 @@ import time
from sqlalchemy import false, func
from galaxy import util, web
from galaxy.security.validate_user_input import validate_password
from galaxy.util import inflector
from galaxy.util.hash_util import new_secure_hash
from galaxy.web.form_builder import CheckboxField
from tool_shed.util.web_util import escape
log = logging.getLogger(__name__)
compliance_log = logging.getLogger('COMPLIANCE')
@@ -685,13 +685,8 @@ class Admin(object):
user = get_user(trans, user_id)
password = kwd.get('password', None)
confirm = kwd.get('confirm', None)
if len(password) < 6:
message = "Use a password of at least 6 characters."
status = 'error'
break
elif password != confirm:
message = "Passwords do not match."
status = 'error'
message = validate_password(trans, password, confirm)
if message:
break
else:
user.set_password_cleartext(password)
+4
View File
@@ -9,6 +9,7 @@ from datetime import (
import tool_shed.repository_types.util as rt_util
from galaxy import util
from galaxy.model.orm.now import now
from galaxy.security.validate_user_input import validate_password_str
from galaxy.util import unique_id
from galaxy.util.bunch import Bunch
from galaxy.util.dictifiable import Dictifiable
@@ -67,6 +68,9 @@ class User(Dictifiable):
total_disk_usage = property(get_disk_usage, set_disk_usage)
def set_password_cleartext(self, cleartext):
message = validate_password_str(cleartext, cleartext)
if message:
raise Exception("Invalid password: %s" % message)
"""Set 'self.password' to the digest of 'cleartext'."""
self.password = new_secure_hash(text_type=cleartext)
@@ -4,7 +4,6 @@ from __future__ import print_function
import logging
import optparse
import os
import re
import sys
from six.moves.configparser import ConfigParser
@@ -14,16 +13,17 @@ sys.path.insert(1, os.path.join(os.path.dirname(__file__)))
import tool_shed.webapp.config as tool_shed_config
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.security.validate_user_input import (
validate_email_str,
validate_password_str,
validate_publicname_str
)
from tool_shed.webapp.model import mapping
from bootstrap_util import admin_user_info # noqa: I100,I201
log = logging.getLogger(__name__)
VALID_PUBLICNAME_RE = re.compile(r"^[a-z0-9\-]+$")
VALID_EMAIL_RE = re.compile(r"[^@]+@[^@]+\.[^@]+")
class BootstrapApplication(object):
"""
Creates a basic Tool Shed application in order to discover the database connection and use SQL
@@ -84,44 +84,15 @@ def create_user(app):
def validate(email, password, username):
message = validate_email(email)
if not message:
message = validate_password(password)
if not message:
message = validate_publicname(username)
message = "\n".join([validate_email_str(email),
validate_password_str(password, password),
validate_publicname_str(username)]).rstrip()
return message
def validate_email(email):
"""Validates the email format."""
message = ''
if not(VALID_EMAIL_RE.match(email)):
message = "Please enter a real email address."
elif len(email) > 255:
message = "Email address exceeds maximum allowable length."
return message
def validate_password(password):
if len(password) < 6:
return "Use a password of at least 6 characters"
return ''
def validate_publicname(username):
"""Validates the public username."""
if len(username) < 3:
return "Public name must be at least 3 characters in length"
if len(username) > 255:
return "Public name cannot be more than 255 characters in length"
if not(VALID_PUBLICNAME_RE.match(username)):
return "Public name must contain only lower-case letters, numbers and '-'"
return ''
if __name__ == "__main__":
parser = optparse.OptionParser(description='Create a user with API key.')
parser.add_option('-c', dest='config', action='store', help='.ini file to retried toolshed configuration from')
parser.add_option('-c', dest='config', action='store', help='.ini file to retrieve toolshed configuration from')
(args, options) = parser.parse_args()
ini_file = args.config
config_parser = ConfigParser({'here': os.getcwd()})
+9
View File
@@ -185,6 +185,15 @@ class UserManagerTestCase(BaseTestCase):
response = json.loads(controller.login(self.trans, payload={"login": user2.username, "password": default_password}))
self.assertEqual(response["message"], "Success.")
def test_empty_password(self):
self.log("should be able to create a user with no password")
user = self.user_manager.create(email='user@nopassword.com', username='nopassword')
self.assertIsNotNone(user.id)
self.assertIsNotNone(user.password)
# should not be able to login with a null or empty password
self.assertFalse(check_password("", user.password))
self.assertFalse(check_password(None, user.password))
# =============================================================================
class UserSerializerTestCase(BaseTestCase):
+9 -2
View File
@@ -10,9 +10,16 @@ def test_alwaysreject():
def test_localdb():
user = User(email='testmail', username='tester')
user.set_password_cleartext('test')
user.set_password_cleartext('test123')
t = LocalDB()
reject = t.authenticate_user(user, 'wrong', {'redact_username_in_logs': False})
accept = t.authenticate_user(user, 'test', {'redact_username_in_logs': False})
accept = t.authenticate_user(user, 'test123', {'redact_username_in_logs': False})
assert reject is False
assert accept is True
# Password must conform to policy (length etc)
try:
user.set_password_cleartext('test')
except Exception:
pass
else:
raise Exception("Password policy validation failed")