Merge pull request #12446 from leela/login-attempt-tracker

refactor: Track consecutive login attempts

frappe/__init__.py

@@ -196,17 +196,20 @@ def init(site, sites_path=None, new_site=False):
 
 	local.initialised = True
 
-def connect(site=None, db_name=None):
+def connect(site=None, db_name=None, set_admin_as_user=True):
 	"""Connect to site database instance.
 
 	:param site: If site is given, calls `frappe.init`.
-	:param db_name: Optional. Will use from `site_config.json`."""
+	:param db_name: Optional. Will use from `site_config.json`.
+	:param set_admin_as_user: Set Administrator as current user.
+	"""
 	from frappe.database import get_db
 	if site:
 		init(site)
 
 	local.db = get_db(user=db_name or local.conf.db_name)
-	set_user("Administrator")
+	if set_admin_as_user:
+		set_user("Administrator")
 
 def connect_replica():
 	from frappe.database import get_db

frappe/app.py

@@ -128,6 +128,8 @@ def init_request(request):
 	if frappe.local.conf.get('maintenance_mode'):
 		frappe.connect()
 		raise frappe.SessionStopped('Session Stopped')
+	else:
+		frappe.connect(set_admin_as_user=False)
 
 	make_form_dict(request)
 

frappe/auth.py

@@ -207,23 +207,44 @@ class LoginManager:
 		if frappe.session.user != "Guest":
 			clear_sessions(frappe.session.user, keep_current=True)
 
-	def authenticate(self, user=None, pwd=None):
+	def authenticate(self, user: str = None, pwd: str = None):
+		from frappe.core.doctype.user.user import User
+
 		if not (user and pwd):
 			user, pwd = frappe.form_dict.get('usr'), frappe.form_dict.get('pwd')
 		if not (user and pwd):
 			self.fail(_('Incomplete login details'), user=user)
 
-		if cint(frappe.db.get_value("System Settings", "System Settings", "allow_login_using_mobile_number")):
-			user = frappe.db.get_value("User", filters={"mobile_no": user}, fieldname="name") or user
+		# Ignore password check if tmp_id is set, 2FA takes care of authentication.
+		validate_password = not bool(frappe.form_dict.get('tmp_id'))
+		user = User.find_by_credentials(user, pwd, validate_password=validate_password)
+
+		if not user:
+			self.fail('Invalid login credentials')
+
+		sys_settings = frappe.get_doc("System Settings")
+		track_login_attempts = (sys_settings.allow_consecutive_login_attempts >0)
+
+		tracker_kwargs = {}
+		if track_login_attempts:
+			tracker_kwargs['lock_interval'] = sys_settings.allow_login_after_fail
+			tracker_kwargs['max_consecutive_login_attempts'] = sys_settings.allow_consecutive_login_attempts
 
-		if cint(frappe.db.get_value("System Settings", "System Settings", "allow_login_using_user_name")):
-			user = frappe.db.get_value("User", filters={"username": user}, fieldname="name") or user
+		tracker = LoginAttemptTracker(user.name, **tracker_kwargs)
 
-		self.check_if_enabled(user)
-		if not frappe.form_dict.get('tmp_id'):
-			self.user = self.check_password(user, pwd)
+		if track_login_attempts and not tracker.is_user_allowed():
+			frappe.throw(_("Your account has been locked and will resume after {0} seconds")
+				.format(sys_settings.allow_login_after_fail), frappe.SecurityException)
+
+		if not user.is_authenticated:
+			tracker.add_failure_attempt()
+			self.fail('Invalid login credentials', user=user.name)
+		elif not (user.name == 'Administrator' or user.enabled):
+			tracker.add_failure_attempt()
+			self.fail('User disabled or missing', user=user.name)
 		else:
-			self.user = user
+			tracker.add_success_attempt()
+		self.user = user.name
 
 	def force_user_to_reset_password(self):
 		if not self.user:
@@ -245,23 +266,12 @@ class LoginManager:
 			if last_pwd_reset_days > reset_pwd_after_days:
 				return True
 
-	def check_if_enabled(self, user):
-		"""raise exception if user not enabled"""
-		doc = frappe.get_doc("System Settings")
-		if cint(doc.allow_consecutive_login_attempts) > 0:
-			check_consecutive_login_attempts(user, doc)
-
-		if user=='Administrator': return
-		if not cint(frappe.db.get_value('User', user, 'enabled')):
-			self.fail('User disabled or missing', user=user)
-
 	def check_password(self, user, pwd):
 		"""check password"""
 		try:
 			# returns user in correct case
 			return check_password(user, pwd)
 		except frappe.AuthenticationError:
-			self.update_invalid_login(user)
 			self.fail('Incorrect password', user=user)
 
 	def fail(self, message, user=None):
@@ -272,15 +282,6 @@ class LoginManager:
 		frappe.db.commit()
 		raise frappe.AuthenticationError
 
-	def update_invalid_login(self, user):
-		last_login_tried = get_last_tried_login_data(user)
-
-		failed_count = 0
-		if last_login_tried > get_datetime():
-			failed_count = get_login_failed_count(user)
-
-		frappe.cache().hset('login_failed_count', user, failed_count + 1)
-
 	def run_trigger(self, event='on_login'):
 		for method in frappe.get_hooks().get(event, []):
 			frappe.call(frappe.get_attr(method), login_manager=self)
@@ -383,38 +384,6 @@ def clear_cookies():
 		frappe.session.sid = ""
 	frappe.local.cookie_manager.delete_cookie(["full_name", "user_id", "sid", "user_image", "system_user"])
 
-def get_last_tried_login_data(user, get_last_login=False):
-	locked_account_time = frappe.cache().hget('locked_account_time', user)
-	if get_last_login and locked_account_time:
-		return locked_account_time
-
-	last_login_tried = frappe.cache().hget('last_login_tried', user)
-	if not last_login_tried or last_login_tried < get_datetime():
-		last_login_tried = get_datetime() + datetime.timedelta(seconds=60)
-
-	frappe.cache().hset('last_login_tried', user, last_login_tried)
-
-	return last_login_tried
-
-def get_login_failed_count(user):
-	return cint(frappe.cache().hget('login_failed_count', user)) or 0
-
-def check_consecutive_login_attempts(user, doc):
-	login_failed_count = get_login_failed_count(user)
-	last_login_tried = (get_last_tried_login_data(user, True)
-		+ datetime.timedelta(seconds=doc.allow_login_after_fail))
-
-	if login_failed_count >= cint(doc.allow_consecutive_login_attempts):
-		locked_account_time = frappe.cache().hget('locked_account_time', user)
-		if not locked_account_time:
-			frappe.cache().hset('locked_account_time', user, get_datetime())
-
-		if last_login_tried > get_datetime():
-			frappe.throw(_("Your account has been locked and will resume after {0} seconds")
-				.format(doc.allow_login_after_fail), frappe.SecurityException)
-		else:
-			delete_login_failed_cache(user)
-
 def validate_ip_address(user):
 	"""check if IP Address is valid"""
 	user = frappe.get_cached_doc("User", user) if not frappe.flags.in_test else frappe.get_doc("User", user)
@@ -436,3 +405,87 @@ def validate_ip_address(user):
 			return
 
 	frappe.throw(_("Access not allowed from this IP Address"), frappe.AuthenticationError)
+
+
+class LoginAttemptTracker(object):
+	"""Track login attemts of a user.
+
+	Lock the account for s number of seconds if there have been n consecutive unsuccessful attempts to log in.
+	"""
+	def __init__(self, user_name: str, max_consecutive_login_attempts: int=3, lock_interval:int = 5*60):
+		""" Initialize the tracker.
+
+		:param user_name: Name of the loggedin user
+		:param max_consecutive_login_attempts: Maximum allowed consecutive failed login attempts
+		:param lock_interval: Locking interval incase of maximum failed attempts
+		"""
+		self.user_name = user_name
+		self.lock_interval = datetime.timedelta(seconds=lock_interval)
+		self.max_failed_logins = max_consecutive_login_attempts
+
+	@property
+	def login_failed_count(self):
+		return frappe.cache().hget('login_failed_count', self.user_name)
+
+	@login_failed_count.setter
+	def login_failed_count(self, count):
+		frappe.cache().hset('login_failed_count', self.user_name, count)
+
+	@login_failed_count.deleter
+	def login_failed_count(self):
+		frappe.cache().hdel('login_failed_count', self.user_name)
+
+	@property
+	def login_failed_time(self):
+		"""First failed login attempt time within lock interval.
+
+		For every user we track only First failed login attempt time within lock interval of time.
+		"""
+		return frappe.cache().hget('login_failed_time', self.user_name)
+
+	@login_failed_time.setter
+	def login_failed_time(self, timestamp):
+		frappe.cache().hset('login_failed_time', self.user_name, timestamp)
+
+	@login_failed_time.deleter
+	def login_failed_time(self):
+		frappe.cache().hdel('login_failed_time', self.user_name)
+
+	def add_failure_attempt(self):
+		""" Log user failure attempts into the system.
+
+		Increase the failure count if new failure is with in current lock interval time period, if not reset the login failure count.
+		"""
+		login_failed_time = self.login_failed_time
+		login_failed_count = self.login_failed_count # Consecutive login failure count
+		current_time = get_datetime()
+
+		if not (login_failed_time and login_failed_count):
+			login_failed_time, login_failed_count = current_time, 0
+
+		if login_failed_time + self.lock_interval > current_time:
+			login_failed_count += 1
+		else:
+			login_failed_time, login_failed_count = current_time, 1
+
+		self.login_failed_time = login_failed_time
+		self.login_failed_count = login_failed_count
+
+	def add_success_attempt(self):
+		"""Reset login failures.
+		"""
+		del self.login_failed_count
+		del self.login_failed_time
+
+	def is_user_allowed(self) -> bool:
+		"""Is user allowed to login
+
+		User is not allowed to login if login failures are greater than threshold within in lock interval from first login failure.
+		"""
+		login_failed_time = self.login_failed_time
+		login_failed_count = self.login_failed_count or 0
+		current_time = get_datetime()
+
+		if login_failed_time and login_failed_time + self.lock_interval > current_time and login_failed_count > self.max_failed_logins:
+			return False
+		return True

frappe/core/doctype/activity_log/test_activity_log.py

@@ -77,6 +77,10 @@ class TestActivityLog(unittest.TestCase):
 		self.assertRaises(frappe.AuthenticationError, LoginManager)
 		self.assertRaises(frappe.AuthenticationError, LoginManager)
 		self.assertRaises(frappe.AuthenticationError, LoginManager)
+
+		# REMOVE ME: current logic allows allow_consecutive_login_attempts+1 attempts
+		# before raising security exception, remove below line when that is fixed.
+		self.assertRaises(frappe.AuthenticationError, LoginManager)
 		self.assertRaises(frappe.SecurityException, LoginManager)
 		time.sleep(5)
 		self.assertRaises(frappe.AuthenticationError, LoginManager)

frappe/core/doctype/user/user.py

@@ -6,7 +6,7 @@ import frappe
 from frappe.model.document import Document
 from frappe.utils import cint, flt, has_gravatar, escape_html, format_datetime, now_datetime, get_formatted_email, today
 from frappe import throw, msgprint, _
-from frappe.utils.password import update_password as _update_password
+from frappe.utils.password import update_password as _update_password, check_password
 from frappe.desk.notifications import clear_notifications
 from frappe.desk.doctype.notification_settings.notification_settings import create_notification_settings
 from frappe.utils.user import get_system_managers
@@ -527,6 +527,27 @@ class User(Document):
 
 		return [i.strip() for i in self.restrict_ip.split(",")]
 
+	@classmethod
+	def find_by_credentials(cls, user_name: str, password: str, validate_password: bool = True):
+		"""Find the user by credentials.
+		"""
+		login_with_mobile = cint(frappe.db.get_value("System Settings", "System Settings", "allow_login_using_mobile_number"))
+		filter = {"mobile_no": user_name} if login_with_mobile else {"name": user_name}
+
+		user = frappe.db.get_value("User", filters=filter, fieldname=['name', 'enabled'], as_dict=True) or {}
+		if not user:
+			return
+
+		user['is_authenticated'] = True
+		if validate_password:
+			try:
+				check_password(user_name, password)
+			except frappe.AuthenticationError:
+				user['is_authenticated'] = False
+
+		return user
+
+
 @frappe.whitelist()
 def get_timezones():
 	import pytz

frappe/integrations/doctype/connected_app/test_connected_app.py

@@ -108,13 +108,8 @@ class TestConnectedApp(unittest.TestCase):
 
 		session = requests.Session()
 
-		# first login of a new user on a new site fails with "401 UNAUTHORIZED"
-		# when anybody fixes that, the two lines below can be removed
 		first_login = login()
-		self.assertEqual(first_login.status_code, 401)
-
-		second_login = login()
-		self.assertEqual(second_login.status_code, 200)
+		self.assertEqual(first_login.status_code, 200)
 
 		authorization_url = self.connected_app.initiate_web_application_flow(user=self.user_name)
 

frappe/tests/test_auth.py

@@ -0,0 +1,47 @@
+# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and Contributors
+# MIT License. See license.txt
+from __future__ import unicode_literals
+
+import time
+import unittest
+from frappe.auth import LoginAttemptTracker
+
+class TestLoginAttemptTracker(unittest.TestCase):
+	def test_account_lock(self):
+		"""Make sure that account locks after `n consecutive failures
+		"""
+		tracker = LoginAttemptTracker(user_name='tester', max_consecutive_login_attempts=3, lock_interval=60)
+		# Clear the cache by setting attempt as success
+		tracker.add_success_attempt()
+
+		tracker.add_failure_attempt()
+		self.assertTrue(tracker.is_user_allowed())
+
+		tracker.add_failure_attempt()
+		self.assertTrue(tracker.is_user_allowed())
+
+		tracker.add_failure_attempt()
+		self.assertTrue(tracker.is_user_allowed())
+
+		tracker.add_failure_attempt()
+		self.assertFalse(tracker.is_user_allowed())
+
+	def test_account_unlock(self):
+		"""Make sure that locked account gets unlocked after lock_interval of time.
+		"""
+		lock_interval = 10 # In sec
+		tracker = LoginAttemptTracker(user_name='tester', max_consecutive_login_attempts=1, lock_interval=lock_interval)
+		# Clear the cache by setting attempt as success
+		tracker.add_success_attempt()
+
+		tracker.add_failure_attempt()
+		self.assertTrue(tracker.is_user_allowed())
+
+		tracker.add_failure_attempt()
+		self.assertFalse(tracker.is_user_allowed())
+
+		# Sleep for lock_interval of time, so that next request con unlock the user access.
+		time.sleep(lock_interval)
+
+		tracker.add_failure_attempt()
+		self.assertTrue(tracker.is_user_allowed())

frappe/utils/data.py

@@ -550,13 +550,13 @@ def flt(s, precision=None):
 
 	return num
 
-def cint(s):
+def cint(s, default=0):
 	"""Convert to integer
 
 		:param s: Number in string or other numeric format.
 		:returns: Converted number in python integer type.
 
-		Returns 0 if input can not be converted to integer.
+		Returns default if input can not be converted to integer.
 
 		Examples:
 		>>> cint("100")
@@ -565,9 +565,10 @@ def cint(s):
 		0
 
 	"""
-	try: num = int(float(s))
-	except: num = 0
-	return num
+	try:
+		return int(float(s))
+	except Exception:
+		return default
 
 def floor(s):
 	"""