kallithea Changeset - c9d859a89a88

Changeset - c9d859a89a88

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

Mads Kiilerich - 7 years ago 2018-12-26 01:54:23
mads@kiilerich.com

auth: move 'active' handling out of the individual auth modules

The 'active' flag in the Kallithea user database is very fundamental and should
not be specific to auth modules. Modules should only care about whether the
user is active in the external authentication system.

user_activation_state is thus removed, and 'hg.extern_activate.auto' is now
consistently checked for all kinds of external authentication.

6 files changed with 17 insertions and 48 deletions:

kallithea/lib/auth_modules/__init__.py

kallithea/lib/auth_modules/auth_container.py

kallithea/lib/auth_modules/auth_crowd.py

kallithea/lib/auth_modules/auth_internal.py

kallithea/lib/auth_modules/auth_ldap.py

kallithea/lib/auth_modules/auth_pam.py

0 comments (0 inline, 0 general)

kallithea/lib/auth_modules/__init__.py

➞

Show inline comments

@@ @@ -51,13 +51,12 @@ class KallitheaAuthPluginBase(object): @@
         "firstname": "first name",
         "lastname": "last name",
         "email": "email address",
         "groups": '["list", "of", "groups"]',
         "extern_name": "name in external source of record",
         "admin": 'True|False defines if user should be Kallithea admin',
         "active": 'True|False defines active state of user in Kallithea',
+    }
     @property
     def validators(self):
         """
         Exposes Kallithea validators modules
@@ @@ -193,20 +192,12 @@ class KallitheaAuthPluginBase(object): @@
             "description": "Enable or Disable this Authentication Plugin",
             "formname": "Enabled"
+            }
+        )
         return rcsettings
     def user_activation_state(self):
         """
         Defines user activation state when creating new users
         :returns: boolean
         """
         raise NotImplementedError("Not implemented in base class")
     def auth(self, userobj, username, passwd, settings, **kwargs):
         """
         Given a user object (which may be None), username, a plaintext password,
         and a settings object (containing all the keys needed as listed in settings()),
         authenticate this user's login attempt.
@@ @@ -217,17 +208,12 @@ class KallitheaAuthPluginBase(object): @@
         """
         raise NotImplementedError("not implemented in base class")
     def _authenticate(self, userobj, username, passwd, settings, **kwargs):
         """
         Wrapper to call self.auth() that validates call on it
         :param userobj: userobj
         :param username: username
         :param passwd: plaintext password
         :param settings: plugin settings
         """
         user_data = self.auth(userobj, username, passwd, settings, **kwargs)
         if user_data is not None:
             return self._validate_auth_return(user_data)
         return None
@@ @@ -252,12 +238,18 @@ class KallitheaExternalAuthPlugin(Kallit @@
         raise NotImplementedError("Not implemented in base class")
     def _authenticate(self, userobj, username, passwd, settings, **kwargs):
         user_data = super(KallitheaExternalAuthPlugin, self)._authenticate(
             userobj, username, passwd, settings, **kwargs)
         if user_data is not None:
             if userobj is None: # external authentication of unknown user that will be created soon
                 def_user_perms = User.get_default_user().AuthUser.permissions['global']
                 active = 'hg.extern_activate.auto' in def_user_perms
             else:
                 active = userobj.active
             if self.use_fake_password():
                 # Randomize the PW because we don't need it, but don't want
                 # them blank either
                 passwd = PasswordGenerator().gen_password(length=8)
             log.debug('Updating or creating user info from %s plugin',
@@ @@ -265,16 +257,16 @@ class KallitheaExternalAuthPlugin(Kallit @@
             user = UserModel().create_or_update(
                 username=user_data['username'],
                 password=passwd,
                 email=user_data["email"],
                 firstname=user_data["firstname"],
                 lastname=user_data["lastname"],
-                active=user_data["active"],
                 active=active,
                 admin=user_data["admin"],
                 extern_name=user_data["extern_name"],
                 extern_type=self.name
+                extern_type=self.name,
+            )
             # enforce user is just in given groups, all of them has to be ones
             # created from plugins. We store this info in _group_data JSON field
             groups = user_data['groups'] or []
             UserGroupModel().enforce_groups(user, groups, self.name)
             Session().commit()
@@ @@ -366,12 +358,17 @@ def authenticate(username, password, env @@
             continue
         # use plugin's method of user extraction.
         user = plugin.get_user(username, environ=environ,
                                settings=plugin_settings)
         log.debug('Plugin %s extracted user `%s`', module, user)
         if user is not None and not user.active:
             log.error("Rejecting authentication of in-active user %s", user)
             continue
         if not plugin.accepts(user):
             log.debug('Plugin %s does not accept user `%s` for authentication',
                       module, user)
             continue
         else:
             log.debug('Plugin %s accepted user `%s` for authentication',
@@ @@ -405,13 +402,13 @@ def authenticate(username, password, env @@
                         username, module)
     return None
 def get_managed_fields(user):
     """return list of fields that are managed by the user's auth source, usually some of
-    'username', 'firstname', 'lastname', 'email', 'active', 'password'
     'username', 'firstname', 'lastname', 'email', 'password'
     """
     auth_plugins = get_auth_plugins()
     for plugin in auth_plugins:
         module = plugin.__class__.__module__
         log.debug('testing %s (%s) with auth plugin %s', user, user.extern_type, module)
         if plugin.name == user.extern_type:

kallithea/lib/auth_modules/auth_container.py

➞

Show inline comments

@@ @@ -102,16 +102,12 @@ class KallitheaAuthPlugin(auth_modules.K @@
+        ]
         return settings
     def use_fake_password(self):
         return True
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser.permissions['global']
         return 'hg.extern_activate.auto' in def_user_perms
     def _clean_username(self, username):
         # Removing realm and domain from username
         username = username.partition('@')[0]
         username = username.rpartition('\\')[2]
         return username
@@ @@ -192,25 +188,23 @@ class KallitheaAuthPlugin(auth_modules.K @@
         # if cannot fetch username, it's a no-go for this plugin to proceed
         if not username:
             return None
         # old attrs fetched from Kallithea database
         admin = getattr(userobj, 'admin', False)
         active = getattr(userobj, 'active', True)
         email = environ.get(settings.get('email_header'), getattr(userobj, 'email', ''))
         firstname = environ.get(settings.get('firstname_header'), getattr(userobj, 'firstname', ''))
         lastname = environ.get(settings.get('lastname_header'), getattr(userobj, 'lastname', ''))
         user_data = {
             'username': username,
             'firstname': safe_unicode(firstname or username),
             'lastname': safe_unicode(lastname or ''),
             'groups': [],
             'email': email or '',
             'admin': admin or False,
             'active': active,
             'extern_name': username,
+        }
         log.info('user `%s` authenticated correctly', user_data['username'])
         return user_data

kallithea/lib/auth_modules/auth_crowd.py

➞

Show inline comments

@@ @@ -190,16 +190,12 @@ class KallitheaAuthPlugin(auth_modules.K @@
+        ]
         return settings
     def use_fake_password(self):
         return True
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser.permissions['global']
         return 'hg.extern_activate.auto' in def_user_perms
     def auth(self, userobj, username, password, settings, **kwargs):
         """
         Given a user object (which may be null), username, a plaintext password,
         and a settings object (containing all the keys needed as listed in settings()),
         authenticate this user's login attempt.
@@ @@ -228,25 +224,23 @@ class KallitheaAuthPlugin(auth_modules.K @@
         res = server.user_groups(crowd_user["name"])
         log.debug("Crowd groups: \n%s", formatted_json(res))
         crowd_user["groups"] = [x["name"] for x in res["groups"]]
         # old attrs fetched from Kallithea database
         admin = getattr(userobj, 'admin', False)
         active = getattr(userobj, 'active', True)
         email = getattr(userobj, 'email', '')
         firstname = getattr(userobj, 'firstname', '')
         lastname = getattr(userobj, 'lastname', '')
         user_data = {
             'username': crowd_user["name"] or username,
             'firstname': crowd_user["first-name"] or firstname,
             'lastname': crowd_user["last-name"] or lastname,
             'groups': crowd_user["groups"],
             'email': crowd_user["email"] or email,
             'admin': admin,
             'active': active,
             'extern_name': crowd_user["name"],
+        }
         # set an admin if we're in admin_groups of crowd
         for group in settings["admin_groups"].split(","):
             if group in user_data["groups"]:

kallithea/lib/auth_modules/auth_internal.py

➞

Show inline comments

@@ @@ -44,16 +44,12 @@ class KallitheaAuthPlugin(auth_modules.K @@
         # Also found as kallithea.lib.model.db.User.DEFAULT_AUTH_TYPE
         return 'internal'
     def settings(self):
         return []
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser.permissions['global']
         return 'hg.register.auto_activate' in def_user_perms
     def accepts(self, user, accepts_empty=True):
         """
         Custom accepts for this auth that doesn't accept empty users. We
         know that user exists in database.
         """
         return super(KallitheaAuthPlugin, self).accepts(user,
@@ @@ -75,31 +71,27 @@ class KallitheaAuthPlugin(auth_modules.K @@
             "username": userobj.username,
             "firstname": userobj.firstname,
             "lastname": userobj.lastname,
             "groups": [],
             "email": userobj.email,
             "admin": userobj.admin,
             "active": userobj.active,
             "extern_name": userobj.user_id,
+        }
         log.debug(formatted_json(user_data))
         log.debug(formatted_json(user_data))
         if userobj.active:
             from kallithea.lib import auth
             password_match = auth.check_password(password, userobj.password)
-            if userobj.is_default_user and userobj.active:
         if userobj.is_default_user:
                 log.info('user %s authenticated correctly as anonymous user',
                          username)
                 return user_data
             elif userobj.username == username and password_match:
                 log.info('user %s authenticated correctly', user_data['username'])
                 return user_data
             log.error("user %s had a bad password", username)
             return None
         else:
             log.warning('user %s tried auth but is disabled', username)
             return None
     def get_managed_fields(self):
         # Note: 'username' should only be editable (at least for user) if self registration is enabled
         return []

kallithea/lib/auth_modules/auth_ldap.py

➞

Show inline comments

@@ @@ -286,16 +286,12 @@ class KallitheaAuthPlugin(auth_modules.K @@
+        ]
         return settings
     def use_fake_password(self):
         return True
     def user_activation_state(self):
         def_user_perms = User.get_default_user().AuthUser.permissions['global']
         return 'hg.extern_activate.auto' in def_user_perms
     def auth(self, userobj, username, password, settings, **kwargs):
         """
         Given a user object (which may be null), username, a plaintext password,
         and a settings object (containing all the keys needed as listed in settings()),
         authenticate this user's login attempt.
@@ @@ -336,25 +332,23 @@ class KallitheaAuthPlugin(auth_modules.K @@
             log.debug('Got ldap DN response %s', user_dn)
             get_ldap_attr = lambda k: ldap_attrs.get(settings.get(k), [''])[0]
             # old attrs fetched from Kallithea database
             admin = getattr(userobj, 'admin', False)
             active = getattr(userobj, 'active', self.user_activation_state())
             email = getattr(userobj, 'email', '')
             firstname = getattr(userobj, 'firstname', '')
             lastname = getattr(userobj, 'lastname', '')
             user_data = {
                 'username': username,
                 'firstname': safe_unicode(get_ldap_attr('attr_firstname') or firstname),
                 'lastname': safe_unicode(get_ldap_attr('attr_lastname') or lastname),
                 'groups': [],
                 'email': get_ldap_attr('attr_email') or email,
                 'admin': admin,
                 'active': active,
                 'extern_name': user_dn,
+            }
             log.info('user %s authenticated correctly', user_data['username'])
             return user_data
         except LdapUsernameError:

kallithea/lib/auth_modules/auth_pam.py

➞

Show inline comments

@@ @@ -112,25 +112,23 @@ class KallitheaAuthPlugin(auth_modules.K @@
                 return None
         else:
             log.debug("Using cached auth for user: %s", username)
         # old attrs fetched from Kallithea database
         admin = getattr(userobj, 'admin', False)
         active = getattr(userobj, 'active', True)
         email = getattr(userobj, 'email', '') or "%s@%s" % (username, socket.gethostname())
         firstname = getattr(userobj, 'firstname', '')
         lastname = getattr(userobj, 'lastname', '')
         user_data = {
             'username': username,
             'firstname': firstname,
             'lastname': lastname,
             'groups': [g.gr_name for g in grp.getgrall() if username in g.gr_mem],
             'email': email,
             'admin': admin,
             'active': active,
             'extern_name': username,
+        }
         try:
             user_pw_data = pwd.getpwnam(username)
             regex = settings["gecos"]

0 comments (0 inline, 0 general)