Log only rejected TOTP codes

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
@@ -194,15 +195,31 @@ def validate(
             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
+        now = sql_now()
         if not self.get_totp(userid, username).verify(
-            code, valid_window=self.valid_window.as_int(), for_time=sql_now()
+            code, valid_window=self.valid_window.as_int(), for_time=now
         ):
+            # 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))

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
+"""

server/tests/mfas/test_totp.py

@@ -0,0 +1,113 @@
+# -*- 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 by a rejected one.
+    A code that passes leaves no trace: only failures are logged.
+    """
+
+    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_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.assertNoLogs(LOGGER_NAME, level='INFO'):
+                mfa.validate(self._request(), 'user1', 'user1', 'ident', code)
+
+    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='WARNING') as logs:
+                with self.assertRaises(exceptions.auth.MFAError):
+                    mfa.validate(self._request(), 'user1', 'user1', 'ident', '000000')
+
+        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:
+        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_without_tracing(self) -> None:
+        mfa = self._mfa()
+
+        with mock.patch.object(TOTP_MFA, 'ask_for_otp', return_value=False):
+            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')
+