From f5b66a21f5056658a04b68b0fb78e4365cf96e92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Janier=20Rodr=C3=ADguez?= Date: Thu, 20 Aug 2026 01:13:08 +0200 Subject: [PATCH 1/2] chore(mfa): trace TOTP validation to diagnose rejected codes A rejected TOTP code left no trace, so a clock drift between the client and the server could not be told apart from a wrong or a replayed code. Traces the result, the valid window and the server time; the code itself is never traced. --- server/src/uds/mfas/TOTP/mfa.py | 35 +++++++++- server/tests/mfas/__init__.py | 31 +++++++++ server/tests/mfas/test_totp.py | 116 ++++++++++++++++++++++++++++++++ 3 files changed, 179 insertions(+), 3 deletions(-) create mode 100644 server/tests/mfas/__init__.py create mode 100644 server/tests/mfas/test_totp.py diff --git a/server/src/uds/mfas/TOTP/mfa.py b/server/src/uds/mfas/TOTP/mfa.py index 4423fa465..0b02189c6 100644 --- a/server/src/uds/mfas/TOTP/mfa.py +++ b/server/src/uds/mfas/TOTP/mfa.py @@ -27,6 +27,7 @@ # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """ Author: Adolfo Gómez, dkmaster at dkmon dot com +Author: Janier Rodríguez, jrodriguez at virtualcable dot es """ import typing import logging @@ -126,6 +127,7 @@ def _user_data(self, userid: str) -> tuple[str, bool]: if data is None: data = (pyotp.random_base32(), False) self._save_user_data(userid, data) + logger.info("TOTP: New secret issued for user [%s]", userid) return data def _save_user_data(self, userid: str, data: tuple[str, bool]) -> None: @@ -176,6 +178,11 @@ def process( validity: typing.Optional[int] = None, ) -> 'mfas.MFA.RESULT': if self.ask_for_otp(request) is False: + logger.info( + "TOTP: Access allowed to user [%s] from IP [%s] without OTP, network is on the allowed list", + userid, + request.ip, + ) return mfas.MFA.RESULT.ALLOWED # The data is provided by an external source, so we need to process anything on the request @@ -191,18 +198,40 @@ def validate( validity: typing.Optional[int] = None, ) -> None: if self.ask_for_otp(request) is False: + logger.info( + "TOTP: Validation skipped for user [%s] from IP [%s], network is on the allowed list", + userid, + request.ip, + ) return if self.cache.get(userid + code) is not None: + logger.warning( + "TOTP: Code already used by user [%s] from IP [%s], length [%d]", userid, request.ip, len(code) + ) raise exceptions.auth.MFAError(gettext('Code is already used. Wait a minute and try again.')) # Get data from storage related to this user secret, qr_has_been_shown = self._user_data(userid) # Validate code - if not self.get_totp(userid, username).verify( - code, valid_window=self.valid_window.as_int(), for_time=sql_now() - ): + # The server time and the valid window are traced because a rejected code is, most of the time, + # a clock drift between the client and this server. The code itself is never traced, only its length. + now = sql_now() + is_valid = self.get_totp(userid, username).verify( + code, valid_window=self.valid_window.as_int(), for_time=now + ) + logger.info( + "TOTP: Validation of user [%s] from IP [%s] returned [%s]. Code length [%d], " + "valid window [%d], server time [%s]", + userid, + request.ip, + is_valid, + len(code), + self.valid_window.as_int(), + now.isoformat(), + ) + if not is_valid: raise exceptions.auth.MFAError(gettext('Invalid code')) self.cache.put(userid + code, True, self.valid_window.as_int() * (TOTP_INTERVAL + 1)) diff --git a/server/tests/mfas/__init__.py b/server/tests/mfas/__init__.py new file mode 100644 index 000000000..8506270e4 --- /dev/null +++ b/server/tests/mfas/__init__.py @@ -0,0 +1,31 @@ +# -*- coding: utf-8 -*- +# +# Copyright (c) 2026 Virtual Cable S.L. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without modification, +# are permitted provided that the following conditions are met: +# +# * Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# * Neither the name of Virtual Cable S.L. nor the names of its contributors +# may be used to endorse or promote products derived from this software +# without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +""" +Author: Janier Rodríguez, jrodriguez at virtualcable dot es +""" diff --git a/server/tests/mfas/test_totp.py b/server/tests/mfas/test_totp.py new file mode 100644 index 000000000..1d0b38f6b --- /dev/null +++ b/server/tests/mfas/test_totp.py @@ -0,0 +1,116 @@ +# -*- coding: utf-8 -*- +# +# Copyright (c) 2026 Virtual Cable S.L. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without modification, +# are permitted provided that the following conditions are met: +# +# * Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# * Neither the name of Virtual Cable S.L. nor the names of its contributors +# may be used to endorse or promote products derived from this software +# without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +""" +Author: Janier Rodríguez, jrodriguez at virtualcable dot es +""" +import typing +from unittest import mock + +import pyotp + +from uds.core import exceptions +from uds.mfas.TOTP.mfa import TOTP_MFA, TOTP_INTERVAL + +from ..utils.test import UDSTestCase + +LOGGER_NAME: typing.Final[str] = 'uds.mfas.TOTP.mfa' + + +class TOTPValidateTest(UDSTestCase): + """ + Validation of TOTP codes and of the traces left to diagnose a rejected one + """ + + secret: str + + def setUp(self) -> None: + self.secret = pyotp.random_base32() + + def _mfa(self) -> TOTP_MFA: + mfa = TOTP_MFA(self.create_environment(), None) + mfa.valid_window.value = 1 + return mfa + + def _request(self) -> typing.Any: + return mock.MagicMock(ip='127.0.0.1') + + def _current_code(self) -> str: + return pyotp.TOTP(self.secret, interval=TOTP_INTERVAL).now() + + def test_valid_code_passes_and_is_traced(self) -> None: + mfa = self._mfa() + code = self._current_code() + + with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=True), mock.patch.object( + TOTP_MFA, '_user_data', return_value=(self.secret, True) + ): + with self.assertLogs(LOGGER_NAME, level='INFO') as logs: + mfa.validate(self._request(), 'user1', 'user1', 'ident', code) + + self.assertTrue(any('returned [True]' in line for line in logs.output)) + self.assertFalse(any(code in line for line in logs.output), 'The code must never be traced') + + def test_invalid_code_raises_and_is_traced(self) -> None: + mfa = self._mfa() + + with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=True), mock.patch.object( + TOTP_MFA, '_user_data', return_value=(self.secret, True) + ): + with self.assertLogs(LOGGER_NAME, level='INFO') as logs: + with self.assertRaises(exceptions.auth.MFAError): + mfa.validate(self._request(), 'user1', 'user1', 'ident', '000000') + + self.assertTrue(any('returned [False]' in line for line in logs.output)) + self.assertFalse(any("'000000'" in line for line in logs.output), 'The code must never be traced') + + def test_replayed_code_is_rejected_and_traced(self) -> None: + mfa = self._mfa() + code = self._current_code() + + with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=True), mock.patch.object( + TOTP_MFA, '_user_data', return_value=(self.secret, True) + ): + mfa.validate(self._request(), 'user1', 'user1', 'ident', code) + + with self.assertLogs(LOGGER_NAME, level='WARNING') as logs: + with self.assertRaises(exceptions.auth.MFAError): + mfa.validate(self._request(), 'user1', 'user1', 'ident', code) + + self.assertTrue(any('already used' in line for line in logs.output)) + self.assertFalse(any(code in line for line in logs.output), 'The code must never be traced') + + def test_allowed_network_skips_validation_and_is_traced(self) -> None: + mfa = self._mfa() + + with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=False): + with self.assertLogs(LOGGER_NAME, level='INFO') as logs: + # An invalid code must be accepted: the network, not the code, is what allows the login + mfa.validate(self._request(), 'user1', 'user1', 'ident', 'not-a-code') + + self.assertTrue(any('Validation skipped' in line for line in logs.output)) From 9922bd270b80ef2e3572c90a4c8941c32e07f820 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Janier=20Rodr=C3=ADguez?= Date: Thu, 20 Aug 2026 17:52:39 +0200 Subject: [PATCH 2/2] chore(mfa): log only rejected TOTP codes Logging every validation fills the log with one line per successful login, which on a deployment with many users is noise. Only the failures stay: an invalid code and an already used one, both at warning level, with the code length, the valid window and the server time. A code that passes leaves no trace, and the code itself is never logged. --- server/src/uds/mfas/TOTP/mfa.py | 40 ++++++++++++--------------------- server/tests/mfas/test_totp.py | 19 +++++++--------- 2 files changed, 22 insertions(+), 37 deletions(-) diff --git a/server/src/uds/mfas/TOTP/mfa.py b/server/src/uds/mfas/TOTP/mfa.py index 0b02189c6..8e0aae882 100644 --- a/server/src/uds/mfas/TOTP/mfa.py +++ b/server/src/uds/mfas/TOTP/mfa.py @@ -127,7 +127,6 @@ def _user_data(self, userid: str) -> tuple[str, bool]: if data is None: data = (pyotp.random_base32(), False) self._save_user_data(userid, data) - logger.info("TOTP: New secret issued for user [%s]", userid) return data def _save_user_data(self, userid: str, data: tuple[str, bool]) -> None: @@ -178,11 +177,6 @@ def process( validity: typing.Optional[int] = None, ) -> 'mfas.MFA.RESULT': if self.ask_for_otp(request) is False: - logger.info( - "TOTP: Access allowed to user [%s] from IP [%s] without OTP, network is on the allowed list", - userid, - request.ip, - ) return mfas.MFA.RESULT.ALLOWED # The data is provided by an external source, so we need to process anything on the request @@ -198,11 +192,6 @@ def validate( validity: typing.Optional[int] = None, ) -> None: if self.ask_for_otp(request) is False: - logger.info( - "TOTP: Validation skipped for user [%s] from IP [%s], network is on the allowed list", - userid, - request.ip, - ) return if self.cache.get(userid + code) is not None: @@ -215,23 +204,22 @@ def validate( secret, qr_has_been_shown = self._user_data(userid) # Validate code - # The server time and the valid window are traced because a rejected code is, most of the time, - # a clock drift between the client and this server. The code itself is never traced, only its length. now = sql_now() - is_valid = self.get_totp(userid, username).verify( + if not self.get_totp(userid, username).verify( code, valid_window=self.valid_window.as_int(), for_time=now - ) - logger.info( - "TOTP: Validation of user [%s] from IP [%s] returned [%s]. Code length [%d], " - "valid window [%d], server time [%s]", - userid, - request.ip, - is_valid, - len(code), - self.valid_window.as_int(), - now.isoformat(), - ) - if not is_valid: + ): + # Only rejected codes are traced. The server time and the valid window go with them because a + # rejection is, most of the time, a clock drift between the client and this server. + # The code itself is never traced, only its length. + logger.warning( + "TOTP: Invalid code from user [%s] at IP [%s]. Code length [%d], " + "valid window [%d], server time [%s]", + userid, + request.ip, + len(code), + self.valid_window.as_int(), + now.isoformat(), + ) raise exceptions.auth.MFAError(gettext('Invalid code')) self.cache.put(userid + code, True, self.valid_window.as_int() * (TOTP_INTERVAL + 1)) diff --git a/server/tests/mfas/test_totp.py b/server/tests/mfas/test_totp.py index 1d0b38f6b..4a4121ff3 100644 --- a/server/tests/mfas/test_totp.py +++ b/server/tests/mfas/test_totp.py @@ -44,7 +44,8 @@ class TOTPValidateTest(UDSTestCase): """ - Validation of TOTP codes and of the traces left to diagnose a rejected one + Validation of TOTP codes and of the traces left by a rejected one. + A code that passes leaves no trace: only failures are logged. """ secret: str @@ -63,30 +64,27 @@ def _request(self) -> typing.Any: def _current_code(self) -> str: return pyotp.TOTP(self.secret, interval=TOTP_INTERVAL).now() - def test_valid_code_passes_and_is_traced(self) -> None: + def test_valid_code_passes_without_tracing(self) -> None: mfa = self._mfa() code = self._current_code() with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=True), mock.patch.object( TOTP_MFA, '_user_data', return_value=(self.secret, True) ): - with self.assertLogs(LOGGER_NAME, level='INFO') as logs: + with self.assertNoLogs(LOGGER_NAME, level='INFO'): mfa.validate(self._request(), 'user1', 'user1', 'ident', code) - self.assertTrue(any('returned [True]' in line for line in logs.output)) - self.assertFalse(any(code in line for line in logs.output), 'The code must never be traced') - def test_invalid_code_raises_and_is_traced(self) -> None: mfa = self._mfa() with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=True), mock.patch.object( TOTP_MFA, '_user_data', return_value=(self.secret, True) ): - with self.assertLogs(LOGGER_NAME, level='INFO') as logs: + with self.assertLogs(LOGGER_NAME, level='WARNING') as logs: with self.assertRaises(exceptions.auth.MFAError): mfa.validate(self._request(), 'user1', 'user1', 'ident', '000000') - self.assertTrue(any('returned [False]' in line for line in logs.output)) + self.assertTrue(any('Invalid code from user' in line for line in logs.output)) self.assertFalse(any("'000000'" in line for line in logs.output), 'The code must never be traced') def test_replayed_code_is_rejected_and_traced(self) -> None: @@ -105,12 +103,11 @@ def test_replayed_code_is_rejected_and_traced(self) -> None: self.assertTrue(any('already used' in line for line in logs.output)) self.assertFalse(any(code in line for line in logs.output), 'The code must never be traced') - def test_allowed_network_skips_validation_and_is_traced(self) -> None: + def test_allowed_network_skips_validation_without_tracing(self) -> None: mfa = self._mfa() with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=False): - with self.assertLogs(LOGGER_NAME, level='INFO') as logs: + with self.assertNoLogs(LOGGER_NAME, level='INFO'): # An invalid code must be accepted: the network, not the code, is what allows the login mfa.validate(self._request(), 'user1', 'user1', 'ident', 'not-a-code') - self.assertTrue(any('Validation skipped' in line for line in logs.output))