Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion server/mergin/auth/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
#
# SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial

from decouple import config
from decouple import config, Csv


class Configuration(object):
Expand All @@ -19,3 +19,9 @@ class Configuration(object):
LOCKOUT_POLICY = config("LOCKOUT_POLICY", default="5:300,10:3600")
# trailing window in seconds over which failed login attempts are counted
LOCKOUT_WINDOW = config("LOCKOUT_WINDOW", default=3600, cast=int)
# comma-separated substrings (case-insensitive) of user agents excluded from login history logging
LOGIN_HISTORY_EXCLUDED_USER_AGENTS = config(
"LOGIN_HISTORY_EXCLUDED_USER_AGENTS",
default="media-sync,work-packages,DB-sync",
cast=Csv(),
)
14 changes: 14 additions & 0 deletions server/mergin/auth/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,9 @@ def record_failed_login(self) -> Optional[int]:
counting only failed attempts within the trailing LOCKOUT_WINDOW and
since the last successful login (whichever bound is more recent).

Note: failed attempts from a user agent excluded from login_history logging
are not recorded at all, so they do not count toward lockout either.

Returns the lockout duration in seconds if a new lock was just applied, else None.
"""
LoginHistory.add_record(self.id, request, successful=False)
Expand Down Expand Up @@ -369,9 +372,20 @@ def __init__(
self.successful = successful
self.timestamp = datetime.datetime.now(tz=datetime.timezone.utc)

@staticmethod
def is_excluded_user_agent(ua: Optional[str]) -> bool:
"""Return True if the user agent matches one of the configured exclusions
and should not be logged in the login history."""
if not ua:
return False
excluded = current_app.config.get("LOGIN_HISTORY_EXCLUDED_USER_AGENTS", [])
return any(pattern.lower() in ua.lower() for pattern in excluded)

@staticmethod
def add_record(user_id: int, req: request, successful: bool = True) -> None:
ua = get_user_agent(req)
if LoginHistory.is_excluded_user_agent(ua):
return
ip = get_ip(req)
device_id = get_device_id(req)
lh = LoginHistory(user_id, ua, ip, device_id, successful=successful)
Expand Down
26 changes: 24 additions & 2 deletions server/mergin/tests/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -835,11 +835,33 @@ def test_api_login(client, data, headers, expected):
assert user.last_signed_in == login_history.timestamp


@pytest.mark.parametrize(
"ua", ["DB-sync/0.1", "media-sync/1.0", "work-packages-agent/2.0", "db-sync/0.1"]
)
def test_api_login_excluded_user_agent(client, ua):
"""Logins from user agents on the LOGIN_HISTORY_EXCLUDED_USER_AGENTS list (matched
case-insensitively as a substring) are not recorded in LoginHistory"""
with patch("mergin.auth.models.get_user_agent") as mock:
mock.return_value = ua
user_before = User.query.filter_by(username=DEFAULT_USER[0]).first()
last_signed_in_before = user_before.last_signed_in
resp = client.post(
"/v1/auth/login",
data=json.dumps({"login": "mergin", "password": "ilovemergin"}),
headers=json_headers,
)
assert resp.status_code == 200
user = User.query.filter_by(username=DEFAULT_USER[0]).first()
login_history = LoginHistory.query.filter_by(user_id=user.id).first()
assert login_history is None
assert user.last_signed_in == last_signed_in_before


def test_api_login_from_urllib(client):
"""DB-sync logins are recorded in LoginHistory just like any other client,
"""Non-excluded clients are recorded in LoginHistory just like any other client,
to keep a full picture of login activity (including for lockout purposes)."""
with patch("mergin.auth.models.get_user_agent") as mock:
mock.return_value = "DB-sync/0.1"
mock.return_value = "python-urllib/3.9"
resp = client.post(
"/v1/auth/login",
data=json.dumps({"login": "mergin", "password": "ilovemergin"}),
Expand Down
Loading