@@ -78,64 +78,64 @@ class KallitheaAuthPlugin(auth_modules.K
"default": "(?P<last_name>.+),\s*(?P<first_name>\w+)",
"formname": "Gecos Regex"
}
]
return settings
def use_fake_password(self):
return True
def auth(self, userobj, username, password, settings, **kwargs):
if username not in _auth_cache:
# Need lock here, as PAM authentication is not thread safe
_pam_lock.acquire()
try:
auth_result = pam.authenticate(username, password,
settings["service"])
# cache result only if we properly authenticated
if auth_result:
_auth_cache[username] = time.time()
finally:
_pam_lock.release()
if not auth_result:
log.error("PAM was unable to authenticate user: %s", username)
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,
"active_from_extern": None,
'extern_name': username,
user_data = pwd.getpwnam(username)
user_pw_data = pwd.getpwnam(username)
regex = settings["gecos"]
match = re.search(regex, user_data.pw_gecos)
match = re.search(regex, user_pw_data.pw_gecos)
if match:
user_data["firstname"] = match.group('first_name')
user_data["lastname"] = match.group('last_name')
except Exception:
log.warning("Cannot extract additional info for PAM user %s", username)
pass
log.debug("pamuser: \n%s", formatted_json(user_data))
log.info('user %s authenticated correctly', user_data['username'])
return user_data
def get_managed_fields(self):
return ['username', 'password']
Status change: